Alex Blank
89ca9f10c7
- Change git mount schema from repo_id to remote_url - Remove database lookups for git mount resolution - Clone directly from URL at instance startup - Simplify frontend UI to text input for Git URL - Fix route ordering in git_repositories.py to prevent 422 errors - Update all tests to use remote_url field Breaking change: Git mounts now use remote_url instead of repo_id
488 lines
16 KiB
Python
488 lines
16 KiB
Python
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,
|
|
_merge_git_mounts,
|
|
)
|
|
|
|
|
|
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"}
|
|
|
|
def test_merge_git_mounts_basic(self) -> None:
|
|
"""Test basic git mount merging."""
|
|
result = _merge_git_mounts(
|
|
[],
|
|
[{"remote_url": "https://github.com/user/repo1.git", "source_path": ".", "target_path": "/app"}],
|
|
"source",
|
|
)
|
|
assert len(result) == 1
|
|
assert result[0]["remote_url"] == "https://github.com/user/repo1.git"
|
|
assert result[0]["target_path"] == "/app"
|
|
|
|
def test_merge_git_mounts_override_same_repo_target(self) -> None:
|
|
"""Test that git mounts with same repo+target override."""
|
|
result = _merge_git_mounts(
|
|
[{"remote_url": "https://github.com/user/repo1.git", "source_path": ".", "target_path": "/app", "branch": "main"}],
|
|
[{"remote_url": "https://github.com/user/repo1.git", "source_path": "src", "target_path": "/app", "branch": "dev"}],
|
|
"source",
|
|
)
|
|
assert len(result) == 1
|
|
assert result[0]["source_path"] == "src"
|
|
assert result[0]["branch"] == "dev"
|
|
|
|
def test_merge_git_mounts_different_targets(self) -> None:
|
|
"""Test that git mounts with different targets are preserved."""
|
|
result = _merge_git_mounts(
|
|
[{"remote_url": "https://github.com/user/repo1.git", "source_path": ".", "target_path": "/app"}],
|
|
[{"remote_url": "https://github.com/user/repo2.git", "source_path": ".", "target_path": "/config"}],
|
|
"source",
|
|
)
|
|
assert len(result) == 2
|
|
targets = {m["target_path"] for m in result}
|
|
assert targets == {"/app", "/config"}
|
|
|
|
|
|
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_with_git_mounts(self, db_session: AsyncSession) -> None:
|
|
"""Test resolving a profile with git mounts."""
|
|
user_id = uuid.uuid4()
|
|
|
|
profile = ConfigProfile(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
name="with-git-mounts",
|
|
env_vars={},
|
|
files={},
|
|
git_mounts=[
|
|
{"remote_url": "https://github.com/user/repo1.git", "source_path": ".", "target_path": "/app"},
|
|
],
|
|
)
|
|
db_session.add(profile)
|
|
await db_session.commit()
|
|
|
|
result = await resolve_profile(db_session, profile.id)
|
|
assert len(result.git_mounts) == 1
|
|
assert result.git_mounts[0]["remote_url"] == "https://github.com/user/repo1.git"
|
|
assert result.git_mounts[0]["target_path"] == "/app"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_profile_with_git_mount_includes(self, db_session: AsyncSession) -> None:
|
|
"""Test resolving a profile that includes another with git mounts."""
|
|
user_id = uuid.uuid4()
|
|
|
|
# Create base profile with git mount
|
|
base = ConfigProfile(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
name="base",
|
|
env_vars={},
|
|
files={},
|
|
git_mounts=[
|
|
{"remote_url": "https://github.com/user/repo1.git", "source_path": ".", "target_path": "/app"},
|
|
],
|
|
)
|
|
db_session.add(base)
|
|
|
|
# Create child profile with its own git mount
|
|
child = ConfigProfile(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
name="child",
|
|
env_vars={},
|
|
files={},
|
|
git_mounts=[
|
|
{"remote_url": "https://github.com/user/repo2.git", "source_path": "config", "target_path": "/config"},
|
|
],
|
|
)
|
|
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 len(result.git_mounts) == 2
|
|
targets = {m["target_path"] for m in result.git_mounts}
|
|
assert targets == {"/app", "/config"}
|
|
|
|
@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
|