from uuid import UUID import pytest from app.services.runtime_injection import RuntimeInjectionError, RuntimeInjectionService @pytest.mark.asyncio async def test_resolve_configs_empty(db_session): result = await RuntimeInjectionService.resolve_configs( db_session, project_id=UUID(int=1), user_id=UUID(int=2), ) assert result == {} @pytest.mark.asyncio async def test_resolve_configs_global_only(db_session, sample_config): result = await RuntimeInjectionService.resolve_configs( db_session, project_id=UUID(int=1), user_id=UUID(int=2), ) assert result == {"test_key": "test_value"} @pytest.mark.asyncio async def test_resolve_configs_scope_override(db_session): from app.models.config import Config global_config = Config( scope_type="global", scope_id=UUID(int=0), key="shared_key", value="global_value", ) project_config = Config( scope_type="project", scope_id=UUID(int=1), key="shared_key", value="project_value", ) db_session.add_all([global_config, project_config]) await db_session.commit() result = await RuntimeInjectionService.resolve_configs( db_session, project_id=UUID(int=1), user_id=UUID(int=2), ) assert result["shared_key"] == "project_value" @pytest.mark.asyncio async def test_resolve_secrets_empty(db_session): result = await RuntimeInjectionService.resolve_secrets( db_session, project_id=UUID(int=1), user_id=UUID(int=2), ) assert result == {} @pytest.mark.asyncio async def test_resolve_secrets_decrypts(db_session, sample_secret): result = await RuntimeInjectionService.resolve_secrets( db_session, project_id=UUID(int=1), user_id=UUID(int=2), ) assert result == {"secret_key": "secret_value"} @pytest.mark.asyncio async def test_validate_secrets_exist_missing(db_session): with pytest.raises(RuntimeInjectionError, match="Missing required secrets"): await RuntimeInjectionService.validate_secrets_exist( db_session, required_secret_keys=["missing_secret"], project_id=UUID(int=1), user_id=UUID(int=2), ) @pytest.mark.asyncio async def test_validate_secrets_exist_found(db_session, sample_secret): await RuntimeInjectionService.validate_secrets_exist( db_session, required_secret_keys=["secret_key"], project_id=UUID(int=1), user_id=UUID(int=2), ) def test_generate_config_files(tmp_path): configs = {"app": {"port": 8080}, "debug": True} mounts = RuntimeInjectionService.generate_config_files(configs, tmp_path) assert len(mounts) == 2 assert (tmp_path / "app.json").exists() assert (tmp_path / "debug.json").exists() assert (tmp_path / "app.json").stat().st_mode & 0o777 == 0o400 def test_generate_secret_env_vars(): secrets = {"api_key": "abc123", "db_pass": "secret"} env_vars = RuntimeInjectionService.generate_secret_env_vars(secrets) assert env_vars == {"API_KEY": "abc123", "DB_PASS": "secret"}