alex
5331a0f110
Update files within profile directory mounts in place so editor saves reach running containers.
829 lines
27 KiB
Python
829 lines
27 KiB
Python
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.models.config.config_profile import ConfigProfile, ConfigProfileInclude
|
|
from src.services.config.config_profile_resolver import (
|
|
ConfigProfileCycleError,
|
|
ConfigProfileNotFoundError,
|
|
ResolvedMount,
|
|
ResolvedProfile,
|
|
apply_resolved_profile,
|
|
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: dict[str, str] = {}
|
|
_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.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.config_profile_resolver import ResolvedMount
|
|
|
|
overrides: dict[str, str] = {}
|
|
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 TestApplyResolvedProfile:
|
|
"""Unit tests for apply_resolved_profile directory-level mount behavior."""
|
|
|
|
def test_mounts_directory_not_individual_files(self, tmp_path) -> None:
|
|
"""Each ResolvedMount should produce one directory-level bind mount."""
|
|
resolved = ResolvedProfile(
|
|
profile_id=uuid.uuid4(),
|
|
profile_name="test",
|
|
mounts={
|
|
"/app": ResolvedMount(
|
|
target="/app",
|
|
mode="rw",
|
|
files={
|
|
"config.json": '{"key": "value"}',
|
|
"nested/file.txt": "hello",
|
|
},
|
|
)
|
|
},
|
|
)
|
|
env, files, volumes, hints = apply_resolved_profile(str(tmp_path), resolved)
|
|
|
|
assert len(volumes) == 1
|
|
assert volumes[0]["target"] == "/app"
|
|
assert Path(volumes[0]["source"]).name == "app"
|
|
assert Path(volumes[0]["source"]).is_dir()
|
|
assert (Path(volumes[0]["source"]) / "config.json").exists()
|
|
assert (Path(volumes[0]["source"]) / "nested" / "file.txt").exists()
|
|
|
|
def test_directory_mount_target(self, tmp_path) -> None:
|
|
"""A directory-level mount targets the configured directory path."""
|
|
resolved = ResolvedProfile(
|
|
profile_id=uuid.uuid4(),
|
|
profile_name="test",
|
|
mounts={
|
|
"/workspace/x/y": ResolvedMount(
|
|
target="/workspace/x/y",
|
|
mode="rw",
|
|
files={"z.json": "override"},
|
|
)
|
|
},
|
|
)
|
|
env, files, volumes, hints = apply_resolved_profile(str(tmp_path), resolved)
|
|
|
|
assert len(volumes) == 1
|
|
assert volumes[0]["target"] == "/workspace/x/y"
|
|
assert Path(volumes[0]["source"]).name == "workspace_x_y"
|
|
assert (Path(volumes[0]["source"]) / "z.json").exists()
|
|
|
|
def test_top_level_files_use_profile_scoped_direct_bind_mounts(self, tmp_path) -> None:
|
|
"""Top-level files are shared safely without mounting over a workspace."""
|
|
profile_id = uuid.uuid4()
|
|
resolved = ResolvedProfile(
|
|
profile_id=profile_id,
|
|
profile_name="test",
|
|
files={".tool/config.toml": "setting = true"},
|
|
)
|
|
|
|
instance_root = tmp_path / "instances"
|
|
_, files, volumes, _ = apply_resolved_profile(
|
|
str(instance_root / "instance-a"),
|
|
resolved,
|
|
working_dir="/workspace/project",
|
|
)
|
|
|
|
canonical_file = (
|
|
instance_root / "config-profiles" / str(profile_id) / "files" / ".tool" / "config.toml"
|
|
)
|
|
assert files == {}
|
|
assert volumes == [
|
|
{
|
|
"source": str(canonical_file),
|
|
"target": "/workspace/project/.tool/config.toml",
|
|
"type": "bind",
|
|
"readonly": False,
|
|
}
|
|
]
|
|
assert canonical_file.read_text() == "setting = true"
|
|
|
|
def test_top_level_file_update_preserves_bind_mount_inode(self, tmp_path) -> None:
|
|
"""An individually bind-mounted file must update in place."""
|
|
profile_id = uuid.uuid4()
|
|
instance_root = tmp_path / "instances"
|
|
resolved = ResolvedProfile(
|
|
profile_id=profile_id,
|
|
profile_name="test",
|
|
files={"settings.toml": "value = 1"},
|
|
)
|
|
|
|
apply_resolved_profile(str(instance_root / "instance-a"), resolved)
|
|
canonical_file = (
|
|
instance_root / "config-profiles" / str(profile_id) / "files" / "settings.toml"
|
|
)
|
|
original_inode = canonical_file.stat().st_ino
|
|
|
|
resolved.files["settings.toml"] = "value = 2"
|
|
apply_resolved_profile(str(instance_root / "instance-a"), resolved)
|
|
|
|
assert canonical_file.stat().st_ino == original_inode
|
|
assert canonical_file.read_text() == "value = 2"
|
|
|
|
def test_mounted_file_update_preserves_bind_mount_inode(self, tmp_path) -> None:
|
|
"""A file inside a profile directory mount must update in place."""
|
|
profile_id = uuid.uuid4()
|
|
instance_root = tmp_path / "instances"
|
|
resolved = ResolvedProfile(
|
|
profile_id=profile_id,
|
|
profile_name="test",
|
|
mounts={
|
|
"/etc/tool": ResolvedMount(
|
|
target="/etc/tool", mode="rw", files={"settings.toml": "value = 1"}
|
|
)
|
|
},
|
|
)
|
|
|
|
apply_resolved_profile(str(instance_root / "instance-a"), resolved)
|
|
canonical_file = (
|
|
instance_root
|
|
/ "config-profiles"
|
|
/ str(profile_id)
|
|
/ "mounts"
|
|
/ "etc_tool"
|
|
/ "settings.toml"
|
|
)
|
|
original_inode = canonical_file.stat().st_ino
|
|
|
|
resolved.mounts["/etc/tool"].files["settings.toml"] = "value = 2"
|
|
apply_resolved_profile(str(instance_root / "instance-a"), resolved)
|
|
|
|
assert canonical_file.stat().st_ino == original_inode
|
|
assert canonical_file.read_text() == "value = 2"
|
|
|
|
def test_instances_share_profile_scoped_mount_sources(self, tmp_path) -> None:
|
|
"""Different instance paths resolve a profile to one canonical source."""
|
|
profile_id = uuid.uuid4()
|
|
resolved = ResolvedProfile(
|
|
profile_id=profile_id,
|
|
profile_name="test",
|
|
mounts={"/app": ResolvedMount(target="/app", mode="rw", files={"config.ini": "x"})},
|
|
)
|
|
|
|
instance_root = tmp_path / "instances"
|
|
_, _, first_volumes, _ = apply_resolved_profile(str(instance_root / "instance-a"), resolved)
|
|
_, _, second_volumes, _ = apply_resolved_profile(str(instance_root / "instance-b"), resolved)
|
|
|
|
assert first_volumes[0]["source"] == second_volumes[0]["source"]
|
|
assert first_volumes[0]["source"] == str(
|
|
instance_root / "config-profiles" / str(profile_id) / "mounts" / "app"
|
|
)
|
|
|
|
def test_empty_mount_produces_no_volumes(self, tmp_path) -> None:
|
|
"""A mount with no files should not produce any volume entries."""
|
|
resolved = ResolvedProfile(
|
|
profile_id=uuid.uuid4(),
|
|
profile_name="test",
|
|
mounts={"/app": ResolvedMount(target="/app", mode="rw", files={})},
|
|
)
|
|
env, files, volumes, hints = apply_resolved_profile(str(tmp_path), resolved)
|
|
assert volumes == []
|
|
|
|
def test_home_expansion_in_directory_mount_target(self, tmp_path) -> None:
|
|
"""~ in mount target should be expanded to home_dir for directory mounts."""
|
|
resolved = ResolvedProfile(
|
|
profile_id=uuid.uuid4(),
|
|
profile_name="test",
|
|
mounts={
|
|
"~/.config": ResolvedMount(
|
|
target="~/.config",
|
|
mode="rw",
|
|
files={"app.toml": "setting = 1"},
|
|
)
|
|
},
|
|
)
|
|
env, files, volumes, hints = apply_resolved_profile(
|
|
str(tmp_path), resolved, home_dir="/home/user"
|
|
)
|
|
assert len(volumes) == 1
|
|
assert volumes[0]["target"] == "/home/user/.config"
|
|
assert (Path(volumes[0]["source"]) / "app.toml").exists()
|
|
|
|
def test_readonly_mount_sets_readonly_flag(self, tmp_path) -> None:
|
|
"""A mount with mode 'ro' should set readonly on the volume entry."""
|
|
resolved = ResolvedProfile(
|
|
profile_id=uuid.uuid4(),
|
|
profile_name="test",
|
|
mounts={
|
|
"/etc/app": ResolvedMount(
|
|
target="/etc/app",
|
|
mode="ro",
|
|
files={"config.cfg": "value"},
|
|
)
|
|
},
|
|
)
|
|
env, files, volumes, hints = apply_resolved_profile(str(tmp_path), resolved)
|
|
|
|
assert len(volumes) == 1
|
|
assert volumes[0]["target"] == "/etc/app"
|
|
assert volumes[0].get("readonly")
|
|
|
|
def test_writable_mount_does_not_set_readonly_flag(self, tmp_path) -> None:
|
|
"""A mount with mode 'rw' should not set readonly on the volume entry."""
|
|
resolved = ResolvedProfile(
|
|
profile_id=uuid.uuid4(),
|
|
profile_name="test",
|
|
mounts={
|
|
"/app": ResolvedMount(
|
|
target="/app",
|
|
mode="rw",
|
|
files={"config.json": "{}"},
|
|
)
|
|
},
|
|
)
|
|
env, files, volumes, hints = apply_resolved_profile(str(tmp_path), resolved)
|
|
|
|
assert len(volumes) == 1
|
|
assert volumes[0]["target"] == "/app"
|
|
assert not volumes[0].get("readonly")
|
|
|
|
|
|
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
|