464 lines
15 KiB
Python
464 lines
15 KiB
Python
"""Unit tests for the profile resolver service."""
|
|
|
|
import uuid
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from src.services.profile_resolver import (
|
|
ProfileCycleError,
|
|
ResolvedProfileOutput,
|
|
resolve_profile,
|
|
)
|
|
|
|
|
|
def _make_profile(
|
|
name: str,
|
|
env_vars: dict[str, str] | None = None,
|
|
start_command: str | None = None,
|
|
working_directory: str | None = None,
|
|
port: int | None = None,
|
|
mounts: list[MagicMock] | None = None,
|
|
includes: list[MagicMock] | None = None,
|
|
) -> MagicMock:
|
|
"""Create a mock ConfigProfile for testing."""
|
|
profile = MagicMock()
|
|
profile.id = uuid.uuid4()
|
|
profile.name = name
|
|
profile.environment_variables = env_vars or {}
|
|
profile.start_command = start_command
|
|
profile.working_directory = working_directory
|
|
profile.port = port
|
|
profile.mounts = mounts or []
|
|
profile.includes = includes or []
|
|
return profile
|
|
|
|
|
|
def _make_include(included_profile: MagicMock, order_index: int = 0) -> MagicMock:
|
|
"""Create a mock ConfigInclude for testing."""
|
|
include = MagicMock()
|
|
include.included_profile = included_profile
|
|
include.order_index = order_index
|
|
return include
|
|
|
|
|
|
def _make_mount(
|
|
target_path: str,
|
|
mode: str = "rw",
|
|
files: dict[str, str] | None = None,
|
|
order_index: int = 0,
|
|
) -> MagicMock:
|
|
"""Create a mock ConfigMount for testing."""
|
|
mount = MagicMock()
|
|
mount.target_path = target_path
|
|
mount.mode = mode
|
|
mount.files = files or {}
|
|
mount.order_index = order_index
|
|
return mount
|
|
|
|
|
|
class TestResolveProfileBasic:
|
|
"""Tests for basic profile resolution without includes."""
|
|
|
|
def test_empty_profile(self) -> None:
|
|
"""Resolving an empty profile returns empty output."""
|
|
profile = _make_profile("empty")
|
|
result = resolve_profile(profile)
|
|
|
|
assert isinstance(result, ResolvedProfileOutput)
|
|
assert result.profile_name == "empty"
|
|
assert result.environment_variables == {}
|
|
assert result.runtime_hints.start_command is None
|
|
assert result.runtime_hints.working_directory is None
|
|
assert result.runtime_hints.port is None
|
|
assert result.mounts == {}
|
|
assert result.resolution_order == ["empty"]
|
|
|
|
def test_env_vars_only(self) -> None:
|
|
"""Profile with env vars resolves correctly."""
|
|
profile = _make_profile(
|
|
"env-only",
|
|
env_vars={"FOO": "bar", "BAZ": "qux"},
|
|
)
|
|
result = resolve_profile(profile)
|
|
|
|
assert result.environment_variables == {"FOO": "bar", "BAZ": "qux"}
|
|
assert result.env_var_sources == {
|
|
"FOO": ["env-only"],
|
|
"BAZ": ["env-only"],
|
|
}
|
|
|
|
def test_runtime_hints_only(self) -> None:
|
|
"""Profile with runtime hints resolves correctly."""
|
|
profile = _make_profile(
|
|
"hints-only",
|
|
start_command="python app.py",
|
|
working_directory="/app",
|
|
port=8080,
|
|
)
|
|
result = resolve_profile(profile)
|
|
|
|
assert result.runtime_hints.start_command == "python app.py"
|
|
assert result.runtime_hints.working_directory == "/app"
|
|
assert result.runtime_hints.port == 8080
|
|
assert result.runtime_hints.overridden_hints == {
|
|
"start_command": "hints-only",
|
|
"working_directory": "hints-only",
|
|
"port": "hints-only",
|
|
}
|
|
|
|
def test_mounts_only(self) -> None:
|
|
"""Profile with mounts resolves correctly."""
|
|
profile = _make_profile(
|
|
"mounts-only",
|
|
mounts=[
|
|
_make_mount(
|
|
"/config",
|
|
mode="ro",
|
|
files={"settings.json": '{"key": "value"}'},
|
|
),
|
|
],
|
|
)
|
|
result = resolve_profile(profile)
|
|
|
|
assert "/config" in result.mounts
|
|
mount = result.mounts["/config"]
|
|
assert mount.target_path == "/config"
|
|
assert mount.mode == "ro"
|
|
assert mount.files == {"settings.json": '{"key": "value"}'}
|
|
|
|
|
|
class TestResolveProfileIncludes:
|
|
"""Tests for profile resolution with includes."""
|
|
|
|
def test_single_include(self) -> None:
|
|
"""Profile with one include resolves in correct order."""
|
|
base = _make_profile("base", env_vars={"FOO": "base"})
|
|
derived = _make_profile(
|
|
"derived",
|
|
env_vars={"BAR": "derived"},
|
|
includes=[_make_include(base, order_index=0)],
|
|
)
|
|
result = resolve_profile(derived)
|
|
|
|
assert result.resolution_order == ["derived", "base"]
|
|
assert result.environment_variables == {
|
|
"FOO": "base",
|
|
"BAR": "derived",
|
|
}
|
|
|
|
def test_multiple_includes_ordered(self) -> None:
|
|
"""Multiple includes are resolved in order_index order."""
|
|
first = _make_profile("first", env_vars={"KEY": "first"})
|
|
second = _make_profile("second", env_vars={"KEY": "second"})
|
|
main = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(first, order_index=0),
|
|
_make_include(second, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
assert result.resolution_order == ["main", "first", "second"]
|
|
# second overrides first
|
|
assert result.environment_variables == {"KEY": "second"}
|
|
assert result.env_var_sources["KEY"] == ["first", "second"]
|
|
|
|
def test_include_order_matters(self) -> None:
|
|
"""Changing include order changes resolution."""
|
|
a = _make_profile("a", env_vars={"KEY": "a"})
|
|
b = _make_profile("b", env_vars={"KEY": "b"})
|
|
main1 = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(a, order_index=0),
|
|
_make_include(b, order_index=1),
|
|
],
|
|
)
|
|
main2 = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(b, order_index=0),
|
|
_make_include(a, order_index=1),
|
|
],
|
|
)
|
|
|
|
result1 = resolve_profile(main1)
|
|
result2 = resolve_profile(main2)
|
|
|
|
assert result1.environment_variables["KEY"] == "b"
|
|
assert result2.environment_variables["KEY"] == "a"
|
|
|
|
def test_nested_includes(self) -> None:
|
|
"""Deeply nested includes resolve recursively."""
|
|
deep = _make_profile("deep", env_vars={"DEEP": "value"})
|
|
mid = _make_profile(
|
|
"mid",
|
|
env_vars={"MID": "value"},
|
|
includes=[_make_include(deep, order_index=0)],
|
|
)
|
|
top = _make_profile(
|
|
"top",
|
|
env_vars={"TOP": "value"},
|
|
includes=[_make_include(mid, order_index=0)],
|
|
)
|
|
result = resolve_profile(top)
|
|
|
|
assert result.resolution_order == ["top", "mid", "deep"]
|
|
assert result.environment_variables == {
|
|
"TOP": "value",
|
|
"MID": "value",
|
|
"DEEP": "value",
|
|
}
|
|
|
|
|
|
class TestResolveProfileOverrides:
|
|
"""Tests for deterministic override rules."""
|
|
|
|
def test_env_var_override(self) -> None:
|
|
"""Later layers override earlier env vars."""
|
|
base = _make_profile("base", env_vars={"KEY": "base"})
|
|
override = _make_profile("override", env_vars={"KEY": "override"})
|
|
main = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(base, order_index=0),
|
|
_make_include(override, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
assert result.environment_variables["KEY"] == "override"
|
|
assert result.env_var_sources["KEY"] == ["base", "override"]
|
|
|
|
def test_main_profile_wins_over_includes(self) -> None:
|
|
"""The main profile itself wins over all includes."""
|
|
base = _make_profile("base", env_vars={"KEY": "base"})
|
|
main = _make_profile(
|
|
"main",
|
|
env_vars={"KEY": "main"},
|
|
includes=[_make_include(base, order_index=0)],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
assert result.environment_variables["KEY"] == "main"
|
|
assert result.env_var_sources["KEY"] == ["base", "main"]
|
|
|
|
def test_runtime_hint_override(self) -> None:
|
|
"""Later layers override earlier runtime hints."""
|
|
base = _make_profile("base", start_command="python old.py")
|
|
override = _make_profile("override", start_command="python new.py")
|
|
main = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(base, order_index=0),
|
|
_make_include(override, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
assert result.runtime_hints.start_command == "python new.py"
|
|
assert result.runtime_hints.overridden_hints["start_command"] == "override"
|
|
|
|
def test_mount_file_override(self) -> None:
|
|
"""Later layers override earlier files in the same mount."""
|
|
base = _make_profile(
|
|
"base",
|
|
mounts=[
|
|
_make_mount(
|
|
"/config",
|
|
files={"app.json": '{"v": 1}'},
|
|
),
|
|
],
|
|
)
|
|
override = _make_profile(
|
|
"override",
|
|
mounts=[
|
|
_make_mount(
|
|
"/config",
|
|
files={"app.json": '{"v": 2}'},
|
|
),
|
|
],
|
|
)
|
|
main = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(base, order_index=0),
|
|
_make_include(override, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
mount = result.mounts["/config"]
|
|
assert mount.files["app.json"] == '{"v": 2}'
|
|
assert mount.overridden_files["app.json"] == ["override"]
|
|
|
|
def test_mount_mode_override(self) -> None:
|
|
"""Later layers override mount mode."""
|
|
base = _make_profile(
|
|
"base",
|
|
mounts=[_make_mount("/data", mode="ro")],
|
|
)
|
|
override = _make_profile(
|
|
"override",
|
|
mounts=[_make_mount("/data", mode="rw")],
|
|
)
|
|
main = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(base, order_index=0),
|
|
_make_include(override, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
assert result.mounts["/data"].mode == "rw"
|
|
assert result.mounts["/data"].mode_overridden_by == "override"
|
|
|
|
def test_mount_file_merge(self) -> None:
|
|
"""Different files in the same mount are merged."""
|
|
base = _make_profile(
|
|
"base",
|
|
mounts=[
|
|
_make_mount(
|
|
"/config",
|
|
files={"a.json": "1"},
|
|
),
|
|
],
|
|
)
|
|
override = _make_profile(
|
|
"override",
|
|
mounts=[
|
|
_make_mount(
|
|
"/config",
|
|
files={"b.json": "2"},
|
|
),
|
|
],
|
|
)
|
|
main = _make_profile(
|
|
"main",
|
|
includes=[
|
|
_make_include(base, order_index=0),
|
|
_make_include(override, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(main)
|
|
|
|
mount = result.mounts["/config"]
|
|
assert mount.files == {"a.json": "1", "b.json": "2"}
|
|
|
|
|
|
class TestResolveProfileCycles:
|
|
"""Tests for cycle detection during resolution."""
|
|
|
|
def test_direct_cycle(self) -> None:
|
|
"""A -> B -> A is detected."""
|
|
a = _make_profile("a")
|
|
b = _make_profile("b", includes=[_make_include(a, order_index=0)])
|
|
a.includes = [_make_include(b, order_index=0)]
|
|
|
|
with pytest.raises(ProfileCycleError) as exc_info:
|
|
resolve_profile(a)
|
|
|
|
assert "a" in exc_info.value.cycle_path
|
|
assert "b" in exc_info.value.cycle_path
|
|
|
|
def test_indirect_cycle(self) -> None:
|
|
"""A -> B -> C -> A is detected."""
|
|
a = _make_profile("a")
|
|
c = _make_profile("c")
|
|
b = _make_profile("b", includes=[_make_include(c, order_index=0)])
|
|
a.includes = [_make_include(b, order_index=0)]
|
|
c.includes = [_make_include(a, order_index=0)]
|
|
|
|
with pytest.raises(ProfileCycleError) as exc_info:
|
|
resolve_profile(a)
|
|
|
|
assert "a" in exc_info.value.cycle_path
|
|
assert "b" in exc_info.value.cycle_path
|
|
assert "c" in exc_info.value.cycle_path
|
|
|
|
def test_self_cycle(self) -> None:
|
|
"""A -> A is detected."""
|
|
a = _make_profile("a")
|
|
a.includes = [_make_include(a, order_index=0)]
|
|
|
|
with pytest.raises(ProfileCycleError) as exc_info:
|
|
resolve_profile(a)
|
|
|
|
assert exc_info.value.cycle_path == ["a", "a"]
|
|
|
|
def test_cycle_does_not_partially_resolve(self) -> None:
|
|
"""Cycle detection prevents any partial resolution."""
|
|
a = _make_profile("a", env_vars={"A": "a"})
|
|
b = _make_profile("b", env_vars={"B": "b"})
|
|
a.includes = [_make_include(b, order_index=0)]
|
|
b.includes = [_make_include(a, order_index=0)]
|
|
|
|
with pytest.raises(ProfileCycleError):
|
|
resolve_profile(a)
|
|
|
|
|
|
class TestResolveProfileDiamond:
|
|
"""Tests for diamond-shaped include graphs."""
|
|
|
|
def test_diamond_resolution(self) -> None:
|
|
"""Diamond graph resolves correctly without duplication issues."""
|
|
base = _make_profile("base", env_vars={"BASE": "base"})
|
|
left = _make_profile(
|
|
"left",
|
|
env_vars={"LEFT": "left"},
|
|
includes=[_make_include(base, order_index=0)],
|
|
)
|
|
right = _make_profile(
|
|
"right",
|
|
env_vars={"RIGHT": "right"},
|
|
includes=[_make_include(base, order_index=0)],
|
|
)
|
|
top = _make_profile(
|
|
"top",
|
|
env_vars={"TOP": "top"},
|
|
includes=[
|
|
_make_include(left, order_index=0),
|
|
_make_include(right, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(top)
|
|
|
|
# base should appear once (via left, then right skips because visited)
|
|
assert result.resolution_order == ["top", "left", "base", "right"]
|
|
assert result.environment_variables == {
|
|
"TOP": "top",
|
|
"LEFT": "left",
|
|
"RIGHT": "right",
|
|
"BASE": "base",
|
|
}
|
|
|
|
def test_diamond_override(self) -> None:
|
|
"""Diamond graph with conflicting overrides resolves correctly."""
|
|
base = _make_profile("base", env_vars={"KEY": "base"})
|
|
left = _make_profile(
|
|
"left",
|
|
env_vars={"KEY": "left"},
|
|
includes=[_make_include(base, order_index=0)],
|
|
)
|
|
right = _make_profile(
|
|
"right",
|
|
env_vars={"KEY": "right"},
|
|
includes=[_make_include(base, order_index=0)],
|
|
)
|
|
top = _make_profile(
|
|
"top",
|
|
includes=[
|
|
_make_include(left, order_index=0),
|
|
_make_include(right, order_index=1),
|
|
],
|
|
)
|
|
result = resolve_profile(top)
|
|
|
|
# right wins because it's later
|
|
assert result.environment_variables["KEY"] == "right"
|
|
assert result.env_var_sources["KEY"] == ["base", "left", "right"]
|
|
# Note: base appears once because visited set skips duplicate resolution in diamond graphs
|