feat: add config profiles
- Add ConfigProfile and ConfigProfileInclude data models with migrations - Implement profile resolver service with ordered includes and merge rules - Add profile CRUD API with validation, compatibility, and cycle detection - Add instance API plumbing for profile selection on create/start/restart - Add resolved profile preview and default resolution APIs - Add frontend config profile API client and management UI - Add launch/restart profile selection UI - Add backend integration and unit tests (31 passing) OpenSpec: add-config-profiles Quality gates: ruff, TypeScript compile, 31 tests passing
This commit is contained in:
@@ -47,10 +47,8 @@ def test_client() -> Generator[TestClient, None, None]:
|
||||
app.dependency_overrides[get_db_session] = override_get_db_session
|
||||
|
||||
# Patch startup events to prevent PostgreSQL connection attempts
|
||||
with patch("src.main.init_database") as mock_init, \
|
||||
patch("src.main.seed_builtin_tool_types") as mock_seed:
|
||||
with patch("src.main.init_database") as mock_init:
|
||||
mock_init.return_value = True
|
||||
mock_seed.return_value = None
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
@@ -61,6 +59,31 @@ def test_client() -> Generator[TestClient, None, None]:
|
||||
asyncio.run(engine.dispose())
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session(test_client) -> AsyncGenerator[AsyncSession, None]:
|
||||
"""Provide an async database session for unit tests."""
|
||||
# Get the override function from the test_client fixture
|
||||
override_fn = app.dependency_overrides.get(get_db_session)
|
||||
if override_fn:
|
||||
gen = override_fn()
|
||||
session = await gen.asend(None)
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
await gen.aclose()
|
||||
else:
|
||||
# Fallback: create a new engine and session
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
async with async_sessionmaker(engine, expire_on_commit=False)() as session:
|
||||
yield session
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def authenticated_client(test_client) -> Generator[TestClient, None, None]:
|
||||
"""Provide an authenticated test client with a test user."""
|
||||
|
||||
@@ -0,0 +1,322 @@
|
||||
import uuid
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestConfigProfilesAPI:
|
||||
"""Integration tests for config profiles API."""
|
||||
|
||||
def test_list_config_profiles_requires_authentication(self, test_client: TestClient) -> None:
|
||||
"""Test that listing config profiles requires authentication."""
|
||||
response = test_client.get("/config-profiles")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_list_config_profiles_returns_user_profiles(self, authenticated_client: TestClient) -> None:
|
||||
"""Test that authenticated users can list their profiles."""
|
||||
response = authenticated_client.get("/config-profiles")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_create_config_profile_successfully(self, authenticated_client: TestClient) -> None:
|
||||
"""Test creating a config profile."""
|
||||
response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "test-profile",
|
||||
"description": "Test profile",
|
||||
"env_vars": {"VAR": "value"},
|
||||
"runtime_hints": {"start_command": "npm start"},
|
||||
"mounts": [{"target": "/app", "mode": "rw", "files": {}}],
|
||||
"files": {"test.txt": "hello"},
|
||||
},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["name"] == "test-profile"
|
||||
assert data["env_vars"] == {"VAR": "value"}
|
||||
assert data["files"] == {"test.txt": "hello"}
|
||||
assert data["mounts"][0]["target"] == "/app"
|
||||
|
||||
def test_create_config_profile_duplicate_name(self, authenticated_client: TestClient) -> None:
|
||||
"""Test that duplicate profile names are rejected."""
|
||||
# Create first profile
|
||||
response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "duplicate-profile",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
|
||||
# Try to create second with same name
|
||||
response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "duplicate-profile",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
assert response.status_code == 409
|
||||
|
||||
def test_create_config_profile_exceeds_size_limit(self, authenticated_client: TestClient) -> None:
|
||||
"""Test that profiles exceeding 10MB are rejected."""
|
||||
large_content = "x" * (11 * 1024 * 1024) # 11MB
|
||||
response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "large-profile",
|
||||
"env_vars": {},
|
||||
"files": {"large.txt": large_content},
|
||||
},
|
||||
)
|
||||
assert response.status_code == 413
|
||||
|
||||
def test_create_config_profile_invalid_file_path(self, authenticated_client: TestClient) -> None:
|
||||
"""Test that invalid file paths are rejected."""
|
||||
response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "bad-profile",
|
||||
"env_vars": {},
|
||||
"files": {"../../../etc/passwd": "malicious"},
|
||||
},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_create_config_profile_invalid_mount_target(self, authenticated_client: TestClient) -> None:
|
||||
"""Test that invalid mount targets are rejected."""
|
||||
response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "bad-mount-profile",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
"mounts": [{"target": "relative/path", "mode": "rw", "files": {}}],
|
||||
},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_get_config_profile_by_id(self, authenticated_client: TestClient) -> None:
|
||||
"""Test getting a config profile by ID."""
|
||||
# Create profile first
|
||||
create_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "get-test",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
profile_id = create_response.json()["id"]
|
||||
|
||||
# Get it back
|
||||
response = authenticated_client.get(f"/config-profiles/{profile_id}")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "get-test"
|
||||
|
||||
def test_get_config_profile_not_found(self, authenticated_client: TestClient) -> None:
|
||||
"""Test getting a non-existent profile."""
|
||||
response = authenticated_client.get(f"/config-profiles/{uuid.uuid4()}")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_update_config_profile_successfully(self, authenticated_client: TestClient) -> None:
|
||||
"""Test updating a config profile."""
|
||||
# Create profile first
|
||||
create_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "update-test",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
profile_id = create_response.json()["id"]
|
||||
|
||||
# Update it
|
||||
response = authenticated_client.put(
|
||||
f"/config-profiles/{profile_id}",
|
||||
json={
|
||||
"name": "updated-name",
|
||||
"env_vars": {"NEW_VAR": "new_value"},
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "updated-name"
|
||||
assert data["env_vars"] == {"NEW_VAR": "new_value"}
|
||||
|
||||
def test_delete_config_profile_successfully(self, authenticated_client: TestClient) -> None:
|
||||
"""Test deleting a config profile."""
|
||||
# Create profile first
|
||||
create_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "delete-test",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
profile_id = create_response.json()["id"]
|
||||
|
||||
# Delete it
|
||||
response = authenticated_client.delete(f"/config-profiles/{profile_id}")
|
||||
assert response.status_code == 204
|
||||
|
||||
# Verify it's gone
|
||||
get_response = authenticated_client.get(f"/config-profiles/{profile_id}")
|
||||
assert get_response.status_code == 404
|
||||
|
||||
def test_update_profile_includes_successfully(self, authenticated_client: TestClient) -> None:
|
||||
"""Test updating profile includes."""
|
||||
# Create base profile
|
||||
base_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "base-profile",
|
||||
"env_vars": {"BASE_VAR": "base_value"},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
base_id = base_response.json()["id"]
|
||||
|
||||
# Create child profile
|
||||
child_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "child-profile",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
child_id = child_response.json()["id"]
|
||||
|
||||
# Update includes
|
||||
response = authenticated_client.put(
|
||||
f"/config-profiles/{child_id}/includes",
|
||||
json={"includes": [base_id]},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
print(f"Response data: {data}")
|
||||
print(f"Includes: {data.get('includes', 'NO INCLUDES KEY')}")
|
||||
assert len(data["includes"]) == 1, f"Expected 1 include, got {len(data.get('includes', []))}: {data.get('includes', [])}"
|
||||
assert data["includes"][0]["included_profile_id"] == base_id
|
||||
|
||||
def test_update_profile_includes_cycle_detection(self, authenticated_client: TestClient) -> None:
|
||||
"""Test that include cycles are detected."""
|
||||
# Create profile A
|
||||
a_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "profile-a",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
a_id = a_response.json()["id"]
|
||||
|
||||
# Create profile B
|
||||
b_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "profile-b",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
b_id = b_response.json()["id"]
|
||||
|
||||
# Make B include A
|
||||
authenticated_client.put(
|
||||
f"/config-profiles/{b_id}/includes",
|
||||
json={"includes": [a_id]},
|
||||
)
|
||||
|
||||
# Try to make A include B (would create cycle)
|
||||
response = authenticated_client.put(
|
||||
f"/config-profiles/{a_id}/includes",
|
||||
json={"includes": [b_id]},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_preview_config_profile_successfully(self, authenticated_client: TestClient) -> None:
|
||||
"""Test previewing a resolved config profile."""
|
||||
# Create base profile
|
||||
base_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "preview-base",
|
||||
"env_vars": {"BASE_VAR": "base"},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
base_id = base_response.json()["id"]
|
||||
|
||||
# Create child profile
|
||||
child_response = authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "preview-child",
|
||||
"env_vars": {"CHILD_VAR": "child"},
|
||||
"files": {},
|
||||
},
|
||||
)
|
||||
child_id = child_response.json()["id"]
|
||||
|
||||
# Make child include base
|
||||
authenticated_client.put(
|
||||
f"/config-profiles/{child_id}/includes",
|
||||
json={"includes": [base_id]},
|
||||
)
|
||||
|
||||
# Preview child
|
||||
response = authenticated_client.get(f"/config-profiles/{child_id}/preview")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["profile_name"] == "preview-child"
|
||||
assert data["env_vars"]["BASE_VAR"] == "base"
|
||||
assert data["env_vars"]["CHILD_VAR"] == "child"
|
||||
assert len(data["included_profiles"]) == 1
|
||||
|
||||
def test_resolve_default_profile(self, authenticated_client: TestClient) -> None:
|
||||
"""Test resolving default profile for project/tool."""
|
||||
# Create a global default profile (no project/tool scoping)
|
||||
authenticated_client.post(
|
||||
"/config-profiles",
|
||||
json={
|
||||
"name": "default-profile",
|
||||
"env_vars": {},
|
||||
"files": {},
|
||||
"is_default": True,
|
||||
},
|
||||
)
|
||||
|
||||
# Resolve default with random project/tool (should fall back to global)
|
||||
project_id = str(uuid.uuid4())
|
||||
tool_type_id = str(uuid.uuid4())
|
||||
response = authenticated_client.get(
|
||||
"/config-profiles/defaults/resolve",
|
||||
params={"project_id": project_id, "tool_type_id": tool_type_id},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["profile_name"] == "default-profile"
|
||||
|
||||
def test_resolve_default_profile_no_match(self, authenticated_client: TestClient) -> None:
|
||||
"""Test resolving default profile when no profiles exist."""
|
||||
project_id = str(uuid.uuid4())
|
||||
tool_type_id = str(uuid.uuid4())
|
||||
|
||||
response = authenticated_client.get(
|
||||
"/config-profiles/defaults/resolve",
|
||||
params={"project_id": project_id, "tool_type_id": tool_type_id},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["profile_id"] is None
|
||||
@@ -0,0 +1,383 @@
|
||||
import uuid
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models.config_profile import ConfigProfile, ConfigProfileInclude
|
||||
from src.services.config_profile_resolver import (
|
||||
ConfigProfileCycleError,
|
||||
ConfigProfileNotFoundError,
|
||||
ResolvedProfile,
|
||||
check_include_cycle,
|
||||
resolve_profile,
|
||||
_merge_env_vars,
|
||||
_merge_files,
|
||||
_merge_mounts,
|
||||
_merge_runtime_hints,
|
||||
)
|
||||
|
||||
|
||||
class TestMergeFunctions:
|
||||
"""Unit tests for merge helper functions."""
|
||||
|
||||
def test_merge_env_vars_basic(self) -> None:
|
||||
"""Test basic env var merging."""
|
||||
result = _merge_env_vars(
|
||||
{"A": "1", "B": "2"},
|
||||
{"B": "3", "C": "4"},
|
||||
{},
|
||||
"source",
|
||||
)
|
||||
assert result == {"A": "1", "B": "3", "C": "4"}
|
||||
|
||||
def test_merge_env_vars_tracks_overrides(self) -> None:
|
||||
"""Test that env var overrides are tracked."""
|
||||
overrides = {}
|
||||
_merge_env_vars(
|
||||
{"A": "1"},
|
||||
{"A": "2"},
|
||||
overrides,
|
||||
"source",
|
||||
)
|
||||
assert overrides == {"A": "source"}
|
||||
|
||||
def test_merge_runtime_hints_basic(self) -> None:
|
||||
"""Test basic runtime hint merging."""
|
||||
result = _merge_runtime_hints(
|
||||
{"command": "old"},
|
||||
{"command": "new", "port": 8080},
|
||||
{},
|
||||
"source",
|
||||
)
|
||||
assert result == {"command": "new", "port": 8080}
|
||||
|
||||
def test_merge_files_basic(self) -> None:
|
||||
"""Test basic file merging."""
|
||||
result = _merge_files(
|
||||
{"a.txt": "old"},
|
||||
{"a.txt": "new", "b.txt": "content"},
|
||||
{},
|
||||
"source",
|
||||
)
|
||||
assert result == {"a.txt": "new", "b.txt": "content"}
|
||||
|
||||
def test_merge_mounts_basic(self) -> None:
|
||||
"""Test basic mount merging."""
|
||||
from src.services.config_profile_resolver import ResolvedMount
|
||||
result = _merge_mounts(
|
||||
{},
|
||||
[{"target": "/app", "mode": "rw", "files": {"a.txt": "content"}}],
|
||||
{},
|
||||
"source",
|
||||
)
|
||||
assert "/app" in result
|
||||
assert result["/app"].mode == "rw"
|
||||
assert result["/app"].files == {"a.txt": "content"}
|
||||
|
||||
def test_merge_mounts_file_override(self) -> None:
|
||||
"""Test mount file map merging with overrides."""
|
||||
from src.services.config_profile_resolver import ResolvedMount
|
||||
result = _merge_mounts(
|
||||
{"/app": ResolvedMount(target="/app", mode="rw", files={"a.txt": "old"})},
|
||||
[{"target": "/app", "mode": "rw", "files": {"a.txt": "new"}}],
|
||||
{},
|
||||
"source",
|
||||
)
|
||||
assert result["/app"].files == {"a.txt": "new"}
|
||||
|
||||
def test_merge_mounts_mode_conflict(self) -> None:
|
||||
"""Test that mount mode conflicts are resolved (later wins)."""
|
||||
from src.services.config_profile_resolver import ResolvedMount
|
||||
overrides = {}
|
||||
result = _merge_mounts(
|
||||
{"/app": ResolvedMount(target="/app", mode="rw", files={})},
|
||||
[{"target": "/app", "mode": "ro", "files": {}}],
|
||||
overrides,
|
||||
"source",
|
||||
)
|
||||
assert result["/app"].mode == "ro"
|
||||
assert overrides == {"/app": "source"}
|
||||
|
||||
|
||||
class TestResolveProfile:
|
||||
"""Unit tests for profile resolution."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_simple_profile(self, db_session: AsyncSession) -> None:
|
||||
"""Test resolving a profile with no includes."""
|
||||
user_id = uuid.uuid4()
|
||||
profile = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="simple",
|
||||
env_vars={"VAR": "value"},
|
||||
runtime_hints={"command": "run"},
|
||||
files={"test.txt": "content"},
|
||||
mounts=[{"target": "/app", "mode": "rw", "files": {}}],
|
||||
)
|
||||
db_session.add(profile)
|
||||
await db_session.commit()
|
||||
|
||||
result = await resolve_profile(db_session, profile.id)
|
||||
assert result.profile_name == "simple"
|
||||
assert result.env_vars == {"VAR": "value"}
|
||||
assert result.runtime_hints == {"command": "run"}
|
||||
assert result.files == {"test.txt": "content"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_profile_with_includes(self, db_session: AsyncSession) -> None:
|
||||
"""Test resolving a profile that includes another."""
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
# Create base profile
|
||||
base = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="base",
|
||||
env_vars={"BASE_VAR": "base_value"},
|
||||
files={},
|
||||
)
|
||||
db_session.add(base)
|
||||
|
||||
# Create child profile
|
||||
child = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="child",
|
||||
env_vars={"CHILD_VAR": "child_value"},
|
||||
files={},
|
||||
)
|
||||
db_session.add(child)
|
||||
await db_session.commit()
|
||||
|
||||
# Create include relationship
|
||||
include = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=child.id,
|
||||
included_profile_id=base.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include)
|
||||
await db_session.commit()
|
||||
|
||||
result = await resolve_profile(db_session, child.id)
|
||||
assert result.env_vars == {
|
||||
"BASE_VAR": "base_value",
|
||||
"CHILD_VAR": "child_value",
|
||||
}
|
||||
assert len(result.included_profiles) == 1
|
||||
assert result.included_profiles[0]["name"] == "base"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_profile_child_overrides_parent(self, db_session: AsyncSession) -> None:
|
||||
"""Test that child profile values override parent values."""
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
base = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="base",
|
||||
env_vars={"VAR": "base"},
|
||||
files={},
|
||||
)
|
||||
db_session.add(base)
|
||||
|
||||
child = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="child",
|
||||
env_vars={"VAR": "child"},
|
||||
files={},
|
||||
)
|
||||
db_session.add(child)
|
||||
await db_session.commit()
|
||||
|
||||
include = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=child.id,
|
||||
included_profile_id=base.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include)
|
||||
await db_session.commit()
|
||||
|
||||
result = await resolve_profile(db_session, child.id)
|
||||
assert result.env_vars == {"VAR": "child"}
|
||||
assert result.env_overrides == {"VAR": "child"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_profile_cycle_detection(self, db_session: AsyncSession) -> None:
|
||||
"""Test that cycles are detected during resolution."""
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
profile_a = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="a",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_a)
|
||||
|
||||
profile_b = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="b",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_b)
|
||||
await db_session.commit()
|
||||
|
||||
# A includes B
|
||||
include_ab = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=profile_a.id,
|
||||
included_profile_id=profile_b.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include_ab)
|
||||
|
||||
# B includes A (creates cycle)
|
||||
include_ba = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=profile_b.id,
|
||||
included_profile_id=profile_a.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include_ba)
|
||||
await db_session.commit()
|
||||
|
||||
with pytest.raises(ConfigProfileCycleError):
|
||||
await resolve_profile(db_session, profile_a.id)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_profile_not_found(self, db_session: AsyncSession) -> None:
|
||||
"""Test resolving a non-existent profile."""
|
||||
with pytest.raises(ConfigProfileNotFoundError):
|
||||
await resolve_profile(db_session, uuid.uuid4())
|
||||
|
||||
|
||||
class TestCheckIncludeCycle:
|
||||
"""Unit tests for include cycle checking."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_no_cycle(self, db_session: AsyncSession) -> None:
|
||||
"""Test checking when no cycle exists."""
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
profile_a = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="a",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_a)
|
||||
|
||||
profile_b = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="b",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_b)
|
||||
await db_session.commit()
|
||||
|
||||
# A includes B
|
||||
include = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=profile_a.id,
|
||||
included_profile_id=profile_b.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include)
|
||||
await db_session.commit()
|
||||
|
||||
result = await check_include_cycle(db_session, profile_a.id)
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_detects_cycle(self, db_session: AsyncSession) -> None:
|
||||
"""Test detecting an existing cycle."""
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
profile_a = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="a",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_a)
|
||||
|
||||
profile_b = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="b",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_b)
|
||||
await db_session.commit()
|
||||
|
||||
# A includes B
|
||||
include_ab = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=profile_a.id,
|
||||
included_profile_id=profile_b.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include_ab)
|
||||
|
||||
# B includes A
|
||||
include_ba = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=profile_b.id,
|
||||
included_profile_id=profile_a.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include_ba)
|
||||
await db_session.commit()
|
||||
|
||||
result = await check_include_cycle(db_session, profile_a.id)
|
||||
assert result is not None
|
||||
assert len(result) > 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_would_create_cycle(self, db_session: AsyncSession) -> None:
|
||||
"""Test detecting a cycle that would be created."""
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
profile_a = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="a",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_a)
|
||||
|
||||
profile_b = ConfigProfile(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user_id,
|
||||
name="b",
|
||||
env_vars={},
|
||||
files={},
|
||||
)
|
||||
db_session.add(profile_b)
|
||||
await db_session.commit()
|
||||
|
||||
# A includes B
|
||||
include = ConfigProfileInclude(
|
||||
id=uuid.uuid4(),
|
||||
profile_id=profile_a.id,
|
||||
included_profile_id=profile_b.id,
|
||||
order_index=0,
|
||||
)
|
||||
db_session.add(include)
|
||||
await db_session.commit()
|
||||
|
||||
# Check if adding B includes A would create cycle
|
||||
result = await check_include_cycle(db_session, profile_b.id, profile_a.id)
|
||||
assert result is not None
|
||||
Reference in New Issue
Block a user