alex
51d93d9dc6
- Add RuntimeInjectionService for scope-based config/secret resolution - Mount configs as JSON files at /app/config/ with 0400 permissions - Inject secrets as environment variables with uppercase keys - Implement scope hierarchy: instance > project > user > global - Create ConfigListPage and SecretListPage frontend components - Mask secret values in API responses (never expose decrypted) - Validate secrets exist before spawning containers - Add comprehensive tests for runtime injection service - Update documentation with config/secrets workflow
111 lines
3.1 KiB
Python
111 lines
3.1 KiB
Python
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"}
|