Alex Blank
29a12bb102
- Add expand_container_path() helper that resolves ~/ and $HOME/ prefixes
- Add get_manifest_home_dir() to compute /home/{user.name} or /root from manifest
- Set ENV HOME=... and ENV USER=... in generated Dockerfile for runtime compatibility
- Pass home_dir through instance creation and startup pipeline
- Expand mount targets in apply_resolved_profile() for regular profile mounts
- Expand mapping targets in _resolve_git_mount_mappings() for git mounts
- Expand working_directory and volume targets in _modify_compose_file()
- Update _prepare_manifest_instance to return home_dir alongside image tag
- Fetch tool_type early in start_instance to determine home_dir before profile application
Quality gates: pytest 188 passed, frontend typecheck clean
Addresses: home-path-expansion
606 lines
18 KiB
Python
606 lines
18 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,
|
|
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."""
|
|
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 normalizes to mappings format."""
|
|
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 "mappings" in result[0]
|
|
assert result[0]["mappings"] == [{"source_path": ".", "target_path": "/app"}]
|
|
|
|
def test_merge_git_mounts_concatenate_same_repo_branch(self) -> None:
|
|
"""Test that git mounts with same repo+branch concatenate mappings."""
|
|
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": "/src",
|
|
"branch": "main",
|
|
}
|
|
],
|
|
"source",
|
|
)
|
|
assert len(result) == 1
|
|
assert result[0]["branch"] == "main"
|
|
mappings: list[dict[str, str]] = result[0]["mappings"]
|
|
assert len(mappings) == 2
|
|
assert {"source_path": ".", "target_path": "/app"} in mappings
|
|
assert {"source_path": "src", "target_path": "/src"} in mappings
|
|
|
|
def test_merge_git_mounts_dedup_same_mapping(self) -> None:
|
|
"""Test that duplicate mappings are deduplicated."""
|
|
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": ".",
|
|
"target_path": "/app",
|
|
"branch": "main",
|
|
}
|
|
],
|
|
"source",
|
|
)
|
|
assert len(result) == 1
|
|
assert len(result[0]["mappings"]) == 1
|
|
|
|
def test_merge_git_mounts_different_repos(self) -> None:
|
|
"""Test that git mounts with different repos 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
|
|
urls = {m["remote_url"] for m in result}
|
|
assert urls == {
|
|
"https://github.com/user/repo1.git",
|
|
"https://github.com/user/repo2.git",
|
|
}
|
|
|
|
def test_merge_git_mounts_different_branches(self) -> None:
|
|
"""Test that same repo with different branches are kept separate."""
|
|
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": ".",
|
|
"target_path": "/app",
|
|
"branch": "dev",
|
|
}
|
|
],
|
|
"source",
|
|
)
|
|
assert len(result) == 2
|
|
branches = {m.get("branch") for m in result}
|
|
assert branches == {"main", "dev"}
|
|
|
|
|
|
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 normalizes to mappings."""
|
|
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 "mappings" in result.git_mounts[0]
|
|
assert result.git_mounts[0]["mappings"] == [
|
|
{"source_path": ".", "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
|
|
urls = {m["remote_url"] for m in result.git_mounts}
|
|
assert urls == {
|
|
"https://github.com/user/repo1.git",
|
|
"https://github.com/user/repo2.git",
|
|
}
|
|
for m in result.git_mounts:
|
|
assert "mappings" in m
|
|
|
|
@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
|