michael/compose-farm
1
0
Fork 0
mirror of https://github.com/basnijholt/compose-farm.git synced 2026-02-03 14:13:26 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Rename services to stacks terminology (#79)

Browse Source
This commit is contained in:
Bas Nijholt
2025-12-20 16:00:41 -08:00
committed by GitHub
parent bb019bcae6
commit 350947ad12
71 changed files with 1752 additions and 1685 deletions
Download Patch File Download Diff File Expand all files Collapse all files

186
tests/test_cli_lifecycle.py
View File

@@ -11,18 +11,18 @@ from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
def _make_config(tmp_path: Path, services: dict[str, str] | None = None) -> Config:
def _make_config(tmp_path: Path, stacks: dict[str, str] | None = None) -> Config:
"""Create a minimal config for testing."""
compose_dir = tmp_path / "compose"
compose_dir.mkdir()
svc_dict: dict[str, str | list[str]] = (
dict(services) if services else {"svc1": "host1", "svc2": "host2"}
stack_dict: dict[str, str | list[str]] = (
dict(stacks) if stacks else {"svc1": "host1", "svc2": "host2"}
)
for svc in svc_dict:
svc_dir = compose_dir / svc
svc_dir.mkdir()
(svc_dir / "docker-compose.yml").write_text("services: {}\n")
for stack in stack_dict:
stack_dir = compose_dir / stack
stack_dir.mkdir()
(stack_dir / "docker-compose.yml").write_text("services: {}\n")
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
@@ -30,15 +30,15 @@ def _make_config(tmp_path: Path, services: dict[str, str] | None = None) -> Conf
return Config(
compose_dir=compose_dir,
hosts={"host1": Host(address="localhost"), "host2": Host(address="localhost")},
services=svc_dict,
stacks=stack_dict,
config_path=config_path,
)
def _make_result(service: str, success: bool = True) -> CommandResult:
def _make_result(stack: str, success: bool = True) -> CommandResult:
"""Create a command result."""
return CommandResult(
service=service,
stack=stack,
exit_code=0 if success else 1,
success=success,
stdout="",
@@ -50,14 +50,14 @@ class TestApplyCommand:
"""Tests for the apply command."""
def test_apply_nothing_to_do(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
"""When no migrations, orphans, or missing services, prints success message."""
"""When no migrations, orphans, or missing stacks, prints success message."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
):
apply(dry_run=False, no_orphans=False, full=False, config=None)
@@ -73,24 +73,24 @@ class TestApplyCommand:
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
"compose_farm.cli.lifecycle.get_orphaned_stacks",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host1"),
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stack_host", return_value="host1"),
patch("compose_farm.cli.lifecycle.stop_orphaned_stacks") as mock_stop,
patch("compose_farm.cli.lifecycle.up_stacks") as mock_up,
):
apply(dry_run=True, no_orphans=False, full=False, config=None)
captured = capsys.readouterr()
assert "Services to migrate" in captured.out
assert "Stacks to migrate" in captured.out
assert "svc1" in captured.out
assert "Orphaned services to stop" in captured.out
assert "Orphaned stacks to stop" in captured.out
assert "old-svc" in captured.out
assert "dry-run" in captured.out
@@ -99,24 +99,24 @@ class TestApplyCommand:
mock_up.assert_not_called()
def test_apply_executes_migrations(self, tmp_path: Path) -> None:
"""Apply runs migrations when services need migration."""
"""Apply runs migrations when stacks need migration."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host1"),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stack_host", return_value="host1"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.up_stacks") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
@@ -124,26 +124,26 @@ class TestApplyCommand:
mock_up.assert_called_once()
call_args = mock_up.call_args
assert call_args[0][1] == ["svc1"] # services list
assert call_args[0][1] == ["svc1"] # stacks list
def test_apply_executes_orphan_cleanup(self, tmp_path: Path) -> None:
"""Apply stops orphaned services."""
"""Apply stops orphaned stacks."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("old-svc@host1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
"compose_farm.cli.lifecycle.get_orphaned_stacks",
return_value={"old-svc": "host1"},
),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.stop_orphaned_stacks") as mock_stop,
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=False, config=None)
@@ -160,21 +160,21 @@ class TestApplyCommand:
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
"compose_farm.cli.lifecycle.get_orphaned_stacks",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host1"),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stack_host", return_value="host1"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.up_stacks") as mock_up,
patch("compose_farm.cli.lifecycle.stop_orphaned_stacks") as mock_stop,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
@@ -197,35 +197,35 @@ class TestApplyCommand:
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
"compose_farm.cli.lifecycle.get_orphaned_stacks",
return_value={"old-svc": "host1"},
),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
):
apply(dry_run=False, no_orphans=True, full=False, config=None)
captured = capsys.readouterr()
assert "Nothing to apply" in captured.out
def test_apply_starts_missing_services(self, tmp_path: Path) -> None:
"""Apply starts services that are in config but not in state."""
def test_apply_starts_missing_stacks(self, tmp_path: Path) -> None:
"""Apply starts stacks that are in config but not in state."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch(
"compose_farm.cli.lifecycle.get_services_not_in_state",
"compose_farm.cli.lifecycle.get_stacks_not_in_state",
return_value=["svc1"],
),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.up_stacks") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
@@ -235,43 +235,43 @@ class TestApplyCommand:
call_args = mock_up.call_args
assert call_args[0][1] == ["svc1"]
def test_apply_dry_run_shows_missing_services(
def test_apply_dry_run_shows_missing_stacks(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Dry run shows services that would be started."""
"""Dry run shows stacks that would be started."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch(
"compose_farm.cli.lifecycle.get_services_not_in_state",
"compose_farm.cli.lifecycle.get_stacks_not_in_state",
return_value=["svc1"],
),
):
apply(dry_run=True, no_orphans=False, full=False, config=None)
captured = capsys.readouterr()
assert "Services to start" in captured.out
assert "Stacks to start" in captured.out
assert "svc1" in captured.out
assert "dry-run" in captured.out
def test_apply_full_refreshes_all_services(self, tmp_path: Path) -> None:
"""--full runs up on all services to pick up config changes."""
def test_apply_full_refreshes_all_stacks(self, tmp_path: Path) -> None:
"""--full runs up on all stacks to pick up config changes."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1"), _make_result("svc2")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.up_stacks") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
@@ -279,57 +279,57 @@ class TestApplyCommand:
mock_up.assert_called_once()
call_args = mock_up.call_args
# Should refresh all services in config
# Should refresh all stacks in config
assert set(call_args[0][1]) == {"svc1", "svc2"}
def test_apply_full_dry_run_shows_refresh(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""--full --dry-run shows services that would be refreshed."""
"""--full --dry-run shows stacks that would be refreshed."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch("compose_farm.cli.lifecycle.get_stacks_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_stacks_not_in_state", return_value=[]),
):
apply(dry_run=True, no_orphans=False, full=True, config=None)
captured = capsys.readouterr()
assert "Services to refresh" in captured.out
assert "Stacks to refresh" in captured.out
assert "svc1" in captured.out
assert "svc2" in captured.out
assert "dry-run" in captured.out
def test_apply_full_excludes_already_handled_services(self, tmp_path: Path) -> None:
"""--full doesn't double-process services that are migrating or starting."""
def test_apply_full_excludes_already_handled_stacks(self, tmp_path: Path) -> None:
"""--full doesn't double-process stacks that are migrating or starting."""
cfg = _make_config(tmp_path, {"svc1": "host1", "svc2": "host2", "svc3": "host1"})
mock_results = [_make_result("svc1"), _make_result("svc3")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
patch(
"compose_farm.cli.lifecycle.get_services_not_in_state",
"compose_farm.cli.lifecycle.get_stacks_not_in_state",
return_value=["svc2"],
),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host2"),
patch("compose_farm.cli.lifecycle.get_stack_host", return_value="host2"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.up_stacks") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=True, config=None)
# up_services should be called 3 times: migrate, start, refresh
# up_stacks should be called 3 times: migrate, start, refresh
assert mock_up.call_count == 3
# Get the third call (refresh) and check it only has svc3
refresh_call = mock_up.call_args_list[2]
@@ -347,40 +347,40 @@ class TestDownOrphaned:
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_orphaned_stacks", return_value={}),
):
down(
services=None,
all_services=False,
stacks=None,
all_stacks=False,
orphaned=True,
host=None,
config=None,
)
captured = capsys.readouterr()
assert "No orphaned services to stop" in captured.out
assert "No orphaned stacks to stop" in captured.out
def test_down_orphaned_stops_services(self, tmp_path: Path) -> None:
"""--orphaned stops orphaned services."""
def test_down_orphaned_stops_stacks(self, tmp_path: Path) -> None:
"""--orphaned stops orphaned stacks."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("old-svc@host1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
"compose_farm.cli.lifecycle.get_orphaned_stacks",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.stop_orphaned_stacks") as mock_stop,
patch("compose_farm.cli.lifecycle.report_results"),
):
down(
services=None,
all_services=False,
stacks=None,
all_stacks=False,
orphaned=True,
host=None,
config=None,
@@ -388,12 +388,12 @@ class TestDownOrphaned:
mock_stop.assert_called_once_with(cfg)
def test_down_orphaned_with_services_errors(self) -> None:
"""--orphaned cannot be combined with service arguments."""
def test_down_orphaned_with_stacks_errors(self) -> None:
"""--orphaned cannot be combined with stack arguments."""
with pytest.raises(typer.Exit) as exc_info:
down(
services=["svc1"],
all_services=False,
stacks=["svc1"],
all_stacks=False,
orphaned=True,
host=None,
config=None,
@@ -405,8 +405,8 @@ class TestDownOrphaned:
"""--orphaned cannot be combined with --all."""
with pytest.raises(typer.Exit) as exc_info:
down(
services=None,
all_services=True,
stacks=None,
all_stacks=True,
orphaned=True,
host=None,
config=None,
@@ -418,8 +418,8 @@ class TestDownOrphaned:
"""--orphaned cannot be combined with --host."""
with pytest.raises(typer.Exit) as exc_info:
down(
services=None,
all_services=False,
stacks=None,
all_stacks=False,
orphaned=True,
host="host1",
config=None,

62
tests/test_cli_logs.py
View File

@@ -25,20 +25,20 @@ def _make_config(tmp_path: Path) -> Config:
return Config(
compose_dir=compose_dir,
hosts={"local": Host(address="localhost"), "remote": Host(address="192.168.1.10")},
services={"svc1": "local", "svc2": "local", "svc3": "remote"},
stacks={"svc1": "local", "svc2": "local", "svc3": "remote"},
)
def _make_result(service: str) -> CommandResult:
def _make_result(stack: str) -> CommandResult:
"""Create a successful command result."""
return CommandResult(service=service, exit_code=0, success=True, stdout="", stderr="")
return CommandResult(stack=stack, exit_code=0, success=True, stdout="", stderr="")
def _mock_run_async_factory(
services: list[str],
stacks: list[str],
) -> tuple[Any, list[CommandResult]]:
"""Create a mock run_async that returns results for given services."""
results = [_make_result(s) for s in services]
"""Create a mock run_async that returns results for given stacks."""
results = [_make_result(s) for s in stacks]
def mock_run_async(_coro: Coroutine[Any, Any, Any]) -> list[CommandResult]:
return results
@@ -49,7 +49,7 @@ def _mock_run_async_factory(
class TestLogsContextualDefault:
"""Tests for logs --tail contextual default behavior."""
def test_logs_all_services_defaults_to_20(self, tmp_path: Path) -> None:
def test_logs_all_stacks_defaults_to_20(self, tmp_path: Path) -> None:
"""When --all is specified, default tail should be 20."""
cfg = _make_config(tmp_path)
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2", "svc3"])
@@ -58,18 +58,18 @@ class TestLogsContextualDefault:
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.run_on_stacks") as mock_run,
):
mock_run.return_value = None
logs(services=None, all_services=True, host=None, follow=False, tail=None, config=None)
logs(stacks=None, all_stacks=True, host=None, follow=False, tail=None, config=None)
mock_run.assert_called_once()
call_args = mock_run.call_args
assert call_args[0][2] == "logs --tail 20"
def test_logs_single_service_defaults_to_100(self, tmp_path: Path) -> None:
"""When specific services are specified, default tail should be 100."""
def test_logs_single_stack_defaults_to_100(self, tmp_path: Path) -> None:
"""When specific stacks are specified, default tail should be 100."""
cfg = _make_config(tmp_path)
mock_run_async, _ = _mock_run_async_factory(["svc1"])
@@ -77,11 +77,11 @@ class TestLogsContextualDefault:
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.run_on_stacks") as mock_run,
):
logs(
services=["svc1"],
all_services=False,
stacks=["svc1"],
all_stacks=False,
host=None,
follow=False,
tail=None,
@@ -101,11 +101,11 @@ class TestLogsContextualDefault:
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.run_on_stacks") as mock_run,
):
logs(
services=None,
all_services=True,
stacks=None,
all_stacks=True,
host=None,
follow=False,
tail=50,
@@ -125,11 +125,11 @@ class TestLogsContextualDefault:
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.run_on_stacks") as mock_run,
):
logs(
services=["svc1"],
all_services=False,
stacks=["svc1"],
all_stacks=False,
host=None,
follow=True,
tail=None,
@@ -144,19 +144,19 @@ class TestLogsContextualDefault:
class TestLogsHostFilter:
"""Tests for logs --host filter behavior."""
def test_logs_host_filter_selects_services_on_host(self, tmp_path: Path) -> None:
"""When --host is specified, only services on that host are included."""
def test_logs_host_filter_selects_stacks_on_host(self, tmp_path: Path) -> None:
"""When --host is specified, only stacks on that host are included."""
cfg = _make_config(tmp_path)
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2"])
with (
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.run_on_stacks") as mock_run,
):
logs(
services=None,
all_services=False,
stacks=None,
all_stacks=False,
host="local",
follow=False,
tail=None,
@@ -169,18 +169,18 @@ class TestLogsHostFilter:
assert set(call_args[0][1]) == {"svc1", "svc2"}
def test_logs_host_filter_defaults_to_20_lines(self, tmp_path: Path) -> None:
"""When --host is specified, default tail should be 20 (multiple services)."""
"""When --host is specified, default tail should be 20 (multiple stacks)."""
cfg = _make_config(tmp_path)
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2"])
with (
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.run_on_stacks") as mock_run,
):
logs(
services=None,
all_services=False,
stacks=None,
all_stacks=False,
host="local",
follow=False,
tail=None,
@@ -196,8 +196,8 @@ class TestLogsHostFilter:
# No config mock needed - error is raised before config is loaded
with pytest.raises(typer.Exit) as exc_info:
logs(
services=None,
all_services=True,
stacks=None,
all_stacks=True,
host="local",
follow=False,
tail=None,

6
tests/test_cli_ssh.py
View File

@@ -53,7 +53,7 @@ class TestSshStatus:
hosts:
local:
address: localhost
services:
stacks:
test: local
""")
@@ -69,7 +69,7 @@ services:
hosts:
local:
address: localhost
services:
stacks:
test: local
""")
@@ -92,7 +92,7 @@ class TestSshSetup:
hosts:
local:
address: localhost
services:
stacks:
test: local
""")

26
tests/test_config.py
View File

@@ -36,43 +36,43 @@ class TestConfig:
config = Config(
compose_dir=Path("/opt/compose"),
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01"},
stacks={"plex": "nas01"},
)
assert config.compose_dir == Path("/opt/compose")
assert "nas01" in config.hosts
assert config.services["plex"] == "nas01"
assert config.stacks["plex"] == "nas01"
def test_config_invalid_service_host(self) -> None:
def test_config_invalid_stack_host(self) -> None:
with pytest.raises(ValueError, match="unknown host"):
Config(
compose_dir=Path("/opt/compose"),
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nonexistent"},
stacks={"plex": "nonexistent"},
)
def test_get_host(self) -> None:
config = Config(
compose_dir=Path("/opt/compose"),
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01"},
stacks={"plex": "nas01"},
)
host = config.get_host("plex")
assert host.address == "192.168.1.10"
def test_get_host_unknown_service(self) -> None:
def test_get_host_unknown_stack(self) -> None:
config = Config(
compose_dir=Path("/opt/compose"),
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01"},
stacks={"plex": "nas01"},
)
with pytest.raises(ValueError, match="Unknown service"):
with pytest.raises(ValueError, match="Unknown stack"):
config.get_host("unknown")
def test_get_compose_path(self) -> None:
config = Config(
compose_dir=Path("/opt/compose"),
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01"},
stacks={"plex": "nas01"},
)
path = config.get_compose_path("plex")
# Defaults to compose.yaml when no file exists
@@ -88,7 +88,7 @@ class TestLoadConfig:
"hosts": {
"nas01": {"address": "192.168.1.10", "user": "docker", "port": 2222},
},
"services": {"plex": "nas01"},
"stacks": {"plex": "nas01"},
}
config_file = tmp_path / "sdc.yaml"
config_file.write_text(yaml.dump(config_data))
@@ -102,7 +102,7 @@ class TestLoadConfig:
config_data = {
"compose_dir": "/opt/compose",
"hosts": {"nas01": "192.168.1.10"},
"services": {"plex": "nas01"},
"stacks": {"plex": "nas01"},
}
config_file = tmp_path / "sdc.yaml"
config_file.write_text(yaml.dump(config_data))
@@ -117,7 +117,7 @@ class TestLoadConfig:
"nas01": {"address": "192.168.1.10", "user": "docker"},
"nas02": "192.168.1.11",
},
"services": {"plex": "nas01", "jellyfin": "nas02"},
"stacks": {"plex": "nas01", "jellyfin": "nas02"},
}
config_file = tmp_path / "sdc.yaml"
config_file.write_text(yaml.dump(config_data))
@@ -137,7 +137,7 @@ class TestLoadConfig:
config_data = {
"compose_dir": "/opt/compose",
"hosts": {"local": "localhost"},
"services": {"test": "local"},
"stacks": {"test": "local"},
}
config_file = tmp_path / "sdc.yaml"
config_file.write_text(yaml.dump(config_data))

8
tests/test_config_cmd.py
View File

@@ -25,7 +25,7 @@ def valid_config_data() -> dict[str, Any]:
return {
"compose_dir": "/opt/compose",
"hosts": {"server1": "192.168.1.10"},
"services": {"nginx": "server1"},
"stacks": {"nginx": "server1"},
}
@@ -85,13 +85,13 @@ class TestGenerateTemplate:
data = yaml.safe_load(template)
assert "compose_dir" in data
assert "hosts" in data
assert "services" in data
assert "stacks" in data
def test_has_documentation_comments(self) -> None:
template = _generate_template()
assert "# Compose Farm configuration" in template
assert "hosts:" in template
assert "services:" in template
assert "stacks:" in template
class TestConfigInit:
@@ -210,7 +210,7 @@ class TestConfigValidate:
assert result.exit_code == 0
assert "Valid config" in result.stdout
assert "Hosts: 1" in result.stdout
assert "Services: 1" in result.stdout
assert "Stacks: 1" in result.stdout
def test_validate_invalid_config(self, runner: CliRunner, tmp_path: Path) -> None:
config_file = tmp_path / "invalid.yaml"

46
tests/test_executor.py
View File

@@ -14,7 +14,7 @@ from compose_farm.executor import (
is_local,
run_command,
run_compose,
run_on_services,
run_on_stacks,
)
# These tests run actual shell commands that only work on Linux
@@ -48,7 +48,7 @@ class TestRunLocalCommand:
result = await _run_local_command("echo hello", "test-service")
assert result.success is True
assert result.exit_code == 0
assert result.service == "test-service"
assert result.stack == "test-service"
async def test_run_local_command_failure(self) -> None:
result = await _run_local_command("exit 1", "test-service")
@@ -77,7 +77,7 @@ class TestRunCommand:
host = Host(address="local")
result = await run_command(host, "true", "my-service")
assert isinstance(result, CommandResult)
assert result.service == "my-service"
assert result.stack == "my-service"
assert result.exit_code == 0
assert result.success is True
@@ -88,40 +88,40 @@ class TestRunCompose:
async def test_run_compose_builds_correct_command(self, tmp_path: Path) -> None:
# Create a minimal compose file
compose_dir = tmp_path / "compose"
service_dir = compose_dir / "test-service"
service_dir.mkdir(parents=True)
compose_file = service_dir / "docker-compose.yml"
stack_dir = compose_dir / "test-service"
stack_dir.mkdir(parents=True)
compose_file = stack_dir / "docker-compose.yml"
compose_file.write_text("services: {}")
config = Config(
compose_dir=compose_dir,
hosts={"local": Host(address="localhost")},
services={"test-service": "local"},
stacks={"test-service": "local"},
)
# This will fail because docker compose isn't running,
# but we can verify the command structure works
result = await run_compose(config, "test-service", "config", stream=False)
# Command may fail due to no docker, but structure is correct
assert result.service == "test-service"
assert result.stack == "test-service"
class TestRunOnServices:
"""Tests for parallel service execution."""
class TestRunOnStacks:
"""Tests for parallel stack execution."""
async def test_run_on_services_parallel(self) -> None:
async def test_run_on_stacks_parallel(self) -> None:
config = Config(
compose_dir=Path("/tmp"),
hosts={"local": Host(address="localhost")},
services={"svc1": "local", "svc2": "local"},
stacks={"svc1": "local", "svc2": "local"},
)
# Use a simple command that will work without docker
# We'll test the parallelism structure
results = await run_on_services(config, ["svc1", "svc2"], "version", stream=False)
results = await run_on_stacks(config, ["svc1", "svc2"], "version", stream=False)
assert len(results) == 2
assert results[0].service == "svc1"
assert results[1].service == "svc2"
assert results[0].stack == "svc1"
assert results[1].stack == "svc2"
@linux_only
@@ -133,7 +133,7 @@ class TestCheckPathsExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
# Create test paths
(tmp_path / "dir1").mkdir()
@@ -151,7 +151,7 @@ class TestCheckPathsExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
result = await check_paths_exist(
@@ -166,7 +166,7 @@ class TestCheckPathsExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
(tmp_path / "exists").mkdir()
@@ -182,7 +182,7 @@ class TestCheckPathsExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
result = await check_paths_exist(config, "local", [])
@@ -198,7 +198,7 @@ class TestCheckNetworksExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
result = await check_networks_exist(config, "local", ["bridge"])
@@ -209,7 +209,7 @@ class TestCheckNetworksExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
result = await check_networks_exist(config, "local", ["nonexistent_network_xyz_123"])
@@ -220,7 +220,7 @@ class TestCheckNetworksExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
result = await check_networks_exist(
@@ -234,7 +234,7 @@ class TestCheckNetworksExist:
config = Config(
compose_dir=tmp_path,
hosts={"local": Host(address="localhost")},
services={},
stacks={},
)
result = await check_networks_exist(config, "local", [])

18
tests/test_logs.py
View File

@@ -11,7 +11,7 @@ from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
from compose_farm.logs import (
_parse_images_output,
collect_service_entries,
collect_stack_entries,
isoformat,
load_existing_entries,
merge_entries,
@@ -35,25 +35,25 @@ def test_parse_images_output_handles_list_and_lines() -> None:
async def test_snapshot_preserves_first_seen(tmp_path: Path) -> None:
compose_dir = tmp_path / "compose"
compose_dir.mkdir()
service_dir = compose_dir / "svc"
service_dir.mkdir()
(service_dir / "docker-compose.yml").write_text("services: {}\n")
stack_dir = compose_dir / "svc"
stack_dir.mkdir()
(stack_dir / "docker-compose.yml").write_text("services: {}\n")
config = Config(
compose_dir=compose_dir,
hosts={"local": Host(address="localhost")},
services={"svc": "local"},
stacks={"svc": "local"},
)
sample_output = json.dumps([{"Service": "svc", "Image": "redis", "Digest": "sha256:abc"}])
async def fake_run_compose(
_cfg: Config, service: str, compose_cmd: str, *, stream: bool = True
_cfg: Config, stack: str, compose_cmd: str, *, stream: bool = True
) -> CommandResult:
assert compose_cmd == "images --format json"
assert stream is False or stream is True
return CommandResult(
service=service,
stack=stack,
exit_code=0,
success=True,
stdout=sample_output,
@@ -64,7 +64,7 @@ async def test_snapshot_preserves_first_seen(tmp_path: Path) -> None:
# First snapshot
first_time = datetime(2025, 1, 1, tzinfo=UTC)
first_entries = await collect_service_entries(
first_entries = await collect_stack_entries(
config, "svc", now=first_time, run_compose_fn=fake_run_compose
)
first_iso = isoformat(first_time)
@@ -77,7 +77,7 @@ async def test_snapshot_preserves_first_seen(tmp_path: Path) -> None:
# Second snapshot
second_time = datetime(2025, 2, 1, tzinfo=UTC)
second_entries = await collect_service_entries(
second_entries = await collect_stack_entries(
config, "svc", now=second_time, run_compose_fn=fake_run_compose
)
second_iso = isoformat(second_time)

24
tests/test_operations.py
View File

@@ -11,23 +11,23 @@ import pytest
from compose_farm.cli import lifecycle
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
from compose_farm.operations import _migrate_service
from compose_farm.operations import _migrate_stack
@pytest.fixture
def basic_config(tmp_path: Path) -> Config:
"""Create a basic test config."""
compose_dir = tmp_path / "compose"
service_dir = compose_dir / "test-service"
service_dir.mkdir(parents=True)
(service_dir / "docker-compose.yml").write_text("services: {}")
stack_dir = compose_dir / "test-service"
stack_dir.mkdir(parents=True)
(stack_dir / "docker-compose.yml").write_text("services: {}")
return Config(
compose_dir=compose_dir,
hosts={
"host1": Host(address="localhost"),
"host2": Host(address="localhost"),
},
services={"test-service": "host2"},
stacks={"test-service": "host2"},
)
@@ -38,16 +38,16 @@ class TestMigrationCommands:
def config(self, tmp_path: Path) -> Config:
"""Create a test config."""
compose_dir = tmp_path / "compose"
service_dir = compose_dir / "test-service"
service_dir.mkdir(parents=True)
(service_dir / "docker-compose.yml").write_text("services: {}")
stack_dir = compose_dir / "test-service"
stack_dir.mkdir(parents=True)
(stack_dir / "docker-compose.yml").write_text("services: {}")
return Config(
compose_dir=compose_dir,
hosts={
"host1": Host(address="localhost"),
"host2": Host(address="localhost"),
},
services={"test-service": "host2"},
stacks={"test-service": "host2"},
)
async def test_migration_uses_pull_ignore_buildable(self, config: Config) -> None:
@@ -56,7 +56,7 @@ class TestMigrationCommands:
async def mock_run_compose_step(
cfg: Config,
service: str,
stack: str,
command: str,
*,
raw: bool,
@@ -64,7 +64,7 @@ class TestMigrationCommands:
) -> CommandResult:
commands_called.append(command)
return CommandResult(
service=service,
stack=stack,
exit_code=0,
success=True,
)
@@ -73,7 +73,7 @@ class TestMigrationCommands:
"compose_farm.operations._run_compose_step",
side_effect=mock_run_compose_step,
):
await _migrate_service(
await _migrate_stack(
config,
"test-service",
current_host="host1",

138
tests/test_refresh.py
View File

@@ -9,7 +9,7 @@ from compose_farm import executor as executor_module
from compose_farm import state as state_module
from compose_farm.cli import management as cli_management_module
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult, check_service_running
from compose_farm.executor import CommandResult, check_stack_running
@pytest.fixture
@@ -18,11 +18,11 @@ def mock_config(tmp_path: Path) -> Config:
compose_dir = tmp_path / "stacks"
compose_dir.mkdir()
# Create service directories with compose files
for service in ["plex", "jellyfin", "grafana"]:
svc_dir = compose_dir / service
svc_dir.mkdir()
(svc_dir / "compose.yaml").write_text(f"# {service} compose file\n")
# Create stack directories with compose files
for stack in ["plex", "jellyfin", "grafana"]:
stack_dir = compose_dir / stack
stack_dir.mkdir()
(stack_dir / "compose.yaml").write_text(f"# {stack} compose file\n")
return Config(
compose_dir=compose_dir,
@@ -30,7 +30,7 @@ def mock_config(tmp_path: Path) -> Config:
"nas01": Host(address="192.168.1.10", user="admin", port=22),
"nas02": Host(address="192.168.1.11", user="admin", port=22),
},
services={
stacks={
"plex": "nas01",
"jellyfin": "nas01",
"grafana": "nas02",
@@ -51,33 +51,33 @@ def state_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
return state_path
class TestCheckServiceRunning:
"""Tests for check_service_running function."""
class TestCheckStackRunning:
"""Tests for check_stack_running function."""
@pytest.mark.asyncio
async def test_service_running(self, mock_config: Config) -> None:
"""Returns True when service has running containers."""
async def test_stack_running(self, mock_config: Config) -> None:
"""Returns True when stack has running containers."""
with patch.object(executor_module, "run_command", new_callable=AsyncMock) as mock_run:
mock_run.return_value = CommandResult(
service="plex",
stack="plex",
exit_code=0,
success=True,
stdout="abc123\ndef456\n",
)
result = await check_service_running(mock_config, "plex", "nas01")
result = await check_stack_running(mock_config, "plex", "nas01")
assert result is True
@pytest.mark.asyncio
async def test_service_not_running(self, mock_config: Config) -> None:
"""Returns False when service has no running containers."""
async def test_stack_not_running(self, mock_config: Config) -> None:
"""Returns False when stack has no running containers."""
with patch.object(executor_module, "run_command", new_callable=AsyncMock) as mock_run:
mock_run.return_value = CommandResult(
service="plex",
stack="plex",
exit_code=0,
success=True,
stdout="",
)
result = await check_service_running(mock_config, "plex", "nas01")
result = await check_stack_running(mock_config, "plex", "nas01")
assert result is False
@pytest.mark.asyncio
@@ -85,19 +85,19 @@ class TestCheckServiceRunning:
"""Returns False when command fails."""
with patch.object(executor_module, "run_command", new_callable=AsyncMock) as mock_run:
mock_run.return_value = CommandResult(
service="plex",
stack="plex",
exit_code=1,
success=False,
)
result = await check_service_running(mock_config, "plex", "nas01")
result = await check_stack_running(mock_config, "plex", "nas01")
assert result is False
class TestMergeState:
"""Tests for _merge_state helper function."""
def test_merge_adds_new_services(self) -> None:
"""Merging adds newly discovered services to existing state."""
def test_merge_adds_new_stacks(self) -> None:
"""Merging adds newly discovered stacks to existing state."""
current: dict[str, str | list[str]] = {"plex": "nas01"}
discovered: dict[str, str | list[str]] = {"jellyfin": "nas02"}
removed: list[str] = []
@@ -106,8 +106,8 @@ class TestMergeState:
assert result == {"plex": "nas01", "jellyfin": "nas02"}
def test_merge_updates_existing_services(self) -> None:
"""Merging updates services that changed hosts."""
def test_merge_updates_existing_stacks(self) -> None:
"""Merging updates stacks that changed hosts."""
current: dict[str, str | list[str]] = {"plex": "nas01", "jellyfin": "nas01"}
discovered: dict[str, str | list[str]] = {"plex": "nas02"} # plex moved to nas02
removed: list[str] = []
@@ -116,8 +116,8 @@ class TestMergeState:
assert result == {"plex": "nas02", "jellyfin": "nas01"}
def test_merge_removes_stopped_services(self) -> None:
"""Merging removes services that were checked but not found."""
def test_merge_removes_stopped_stacks(self) -> None:
"""Merging removes stacks that were checked but not found."""
current: dict[str, str | list[str]] = {
"plex": "nas01",
"jellyfin": "nas01",
@@ -131,8 +131,8 @@ class TestMergeState:
# jellyfin removed, grafana untouched (wasn't in the refresh scope)
assert result == {"plex": "nas01", "grafana": "nas02"}
def test_merge_preserves_unrelated_services(self) -> None:
"""Merging preserves services that weren't part of the refresh."""
def test_merge_preserves_unrelated_stacks(self) -> None:
"""Merging preserves stacks that weren't part of the refresh."""
current: dict[str, str | list[str]] = {
"plex": "nas01",
"jellyfin": "nas01",
@@ -151,7 +151,7 @@ class TestReportSyncChanges:
"""Tests for _report_sync_changes function."""
def test_reports_added(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Reports newly discovered services."""
"""Reports newly discovered stacks."""
cli_management_module._report_sync_changes(
added=["plex", "jellyfin"],
removed=[],
@@ -160,12 +160,12 @@ class TestReportSyncChanges:
current_state={},
)
captured = capsys.readouterr()
assert "New services found (2)" in captured.out
assert "New stacks found (2)" in captured.out
assert "+ plex on nas01" in captured.out
assert "+ jellyfin on nas02" in captured.out
def test_reports_removed(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Reports services that are no longer running."""
"""Reports stacks that are no longer running."""
cli_management_module._report_sync_changes(
added=[],
removed=["grafana"],
@@ -174,11 +174,11 @@ class TestReportSyncChanges:
current_state={"grafana": "nas01"},
)
captured = capsys.readouterr()
assert "Services no longer running (1)" in captured.out
assert "Stacks no longer running (1)" in captured.out
assert "- grafana (was on nas01)" in captured.out
def test_reports_changed(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Reports services that moved to a different host."""
"""Reports stacks that moved to a different host."""
cli_management_module._report_sync_changes(
added=[],
removed=[],
@@ -187,23 +187,23 @@ class TestReportSyncChanges:
current_state={"plex": "nas01"},
)
captured = capsys.readouterr()
assert "Services on different hosts (1)" in captured.out
assert "Stacks on different hosts (1)" in captured.out
assert "~ plex: nas01 → nas02" in captured.out
class TestRefreshCommand:
"""Tests for the refresh command with service arguments."""
"""Tests for the refresh command with stack arguments."""
def test_refresh_specific_service_partial_merge(
def test_refresh_specific_stack_partial_merge(
self, mock_config: Config, capsys: pytest.CaptureFixture[str]
) -> None:
"""Refreshing specific services merges with existing state."""
"""Refreshing specific stacks merges with existing state."""
# Mock existing state
existing_state = {"plex": "nas01", "jellyfin": "nas01", "grafana": "nas02"}
with (
patch(
"compose_farm.cli.management.get_services",
"compose_farm.cli.management.get_stacks",
return_value=(["plex"], mock_config),
),
patch(
@@ -211,16 +211,16 @@ class TestRefreshCommand:
return_value=existing_state,
),
patch(
"compose_farm.cli.management._discover_services",
"compose_farm.cli.management._discover_stacks",
return_value={"plex": "nas02"}, # plex moved to nas02
),
patch("compose_farm.cli.management._snapshot_services"),
patch("compose_farm.cli.management._snapshot_stacks"),
patch("compose_farm.cli.management.save_state") as mock_save,
):
# services=["plex"], all_services=False -> partial refresh
# stacks=["plex"], all_stacks=False -> partial refresh
cli_management_module.refresh(
services=["plex"],
all_services=False,
stacks=["plex"],
all_stacks=False,
config=None,
log_path=None,
dry_run=False,
@@ -234,12 +234,12 @@ class TestRefreshCommand:
def test_refresh_all_replaces_state(
self, mock_config: Config, capsys: pytest.CaptureFixture[str]
) -> None:
"""Refreshing all services replaces the entire state."""
"""Refreshing all stacks replaces the entire state."""
existing_state = {"plex": "nas01", "jellyfin": "nas01", "old-service": "nas02"}
with (
patch(
"compose_farm.cli.management.get_services",
"compose_farm.cli.management.get_stacks",
return_value=(["plex", "jellyfin", "grafana"], mock_config),
),
patch(
@@ -247,33 +247,33 @@ class TestRefreshCommand:
return_value=existing_state,
),
patch(
"compose_farm.cli.management._discover_services",
"compose_farm.cli.management._discover_stacks",
return_value={"plex": "nas01", "grafana": "nas02"}, # jellyfin not running
),
patch("compose_farm.cli.management._snapshot_services"),
patch("compose_farm.cli.management._snapshot_stacks"),
patch("compose_farm.cli.management.save_state") as mock_save,
):
# services=None, all_services=False -> defaults to all (full refresh)
# stacks=None, all_stacks=False -> defaults to all (full refresh)
cli_management_module.refresh(
services=None,
all_services=False,
stacks=None,
all_stacks=False,
config=None,
log_path=None,
dry_run=False,
)
# Should have replaced: only discovered services remain
# Should have replaced: only discovered stacks remain
mock_save.assert_called_once()
saved_state = mock_save.call_args[0][1]
assert saved_state == {"plex": "nas01", "grafana": "nas02"}
def test_refresh_with_all_flag_full_refresh(self, mock_config: Config) -> None:
"""Using --all flag forces full refresh even with service names."""
"""Using --all flag forces full refresh even with stack names."""
existing_state = {"plex": "nas01", "jellyfin": "nas01"}
with (
patch(
"compose_farm.cli.management.get_services",
"compose_farm.cli.management.get_stacks",
return_value=(["plex", "jellyfin", "grafana"], mock_config),
),
patch(
@@ -281,16 +281,16 @@ class TestRefreshCommand:
return_value=existing_state,
),
patch(
"compose_farm.cli.management._discover_services",
"compose_farm.cli.management._discover_stacks",
return_value={"plex": "nas01"}, # only plex running
),
patch("compose_farm.cli.management._snapshot_services"),
patch("compose_farm.cli.management._snapshot_stacks"),
patch("compose_farm.cli.management.save_state") as mock_save,
):
# all_services=True -> full refresh (replaces state)
# all_stacks=True -> full refresh (replaces state)
cli_management_module.refresh(
services=["plex"], # ignored when --all is set
all_services=True,
stacks=["plex"], # ignored when --all is set
all_stacks=True,
config=None,
log_path=None,
dry_run=False,
@@ -298,16 +298,16 @@ class TestRefreshCommand:
mock_save.assert_called_once()
saved_state = mock_save.call_args[0][1]
# Full refresh: only discovered services
# Full refresh: only discovered stacks
assert saved_state == {"plex": "nas01"}
def test_refresh_partial_removes_stopped_service(self, mock_config: Config) -> None:
"""Partial refresh removes a service if it was checked but not found."""
def test_refresh_partial_removes_stopped_stack(self, mock_config: Config) -> None:
"""Partial refresh removes a stack if it was checked but not found."""
existing_state = {"plex": "nas01", "jellyfin": "nas01", "grafana": "nas02"}
with (
patch(
"compose_farm.cli.management.get_services",
"compose_farm.cli.management.get_stacks",
return_value=(["plex", "jellyfin"], mock_config),
),
patch(
@@ -315,15 +315,15 @@ class TestRefreshCommand:
return_value=existing_state,
),
patch(
"compose_farm.cli.management._discover_services",
"compose_farm.cli.management._discover_stacks",
return_value={"plex": "nas01"}, # jellyfin not running
),
patch("compose_farm.cli.management._snapshot_services"),
patch("compose_farm.cli.management._snapshot_stacks"),
patch("compose_farm.cli.management.save_state") as mock_save,
):
cli_management_module.refresh(
services=["plex", "jellyfin"],
all_services=False,
stacks=["plex", "jellyfin"],
all_stacks=False,
config=None,
log_path=None,
dry_run=False,
@@ -342,7 +342,7 @@ class TestRefreshCommand:
with (
patch(
"compose_farm.cli.management.get_services",
"compose_farm.cli.management.get_stacks",
return_value=(["plex"], mock_config),
),
patch(
@@ -350,14 +350,14 @@ class TestRefreshCommand:
return_value=existing_state,
),
patch(
"compose_farm.cli.management._discover_services",
"compose_farm.cli.management._discover_stacks",
return_value={"plex": "nas02"}, # would change
),
patch("compose_farm.cli.management.save_state") as mock_save,
):
cli_management_module.refresh(
services=["plex"],
all_services=False,
stacks=["plex"],
all_stacks=False,
config=None,
log_path=None,
dry_run=True,

114
tests/test_state.py
View File

@@ -6,13 +6,13 @@ import pytest
from compose_farm.config import Config, Host
from compose_farm.state import (
get_orphaned_services,
get_service_host,
get_services_not_in_state,
get_orphaned_stacks,
get_stack_host,
get_stacks_not_in_state,
load_state,
remove_service,
remove_stack,
save_state,
set_service_host,
set_stack_host,
)
@@ -24,7 +24,7 @@ def config(tmp_path: Path) -> Config:
return Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01"},
stacks={"plex": "nas01"},
config_path=config_path,
)
@@ -68,174 +68,174 @@ class TestSaveState:
assert "jellyfin: nas02" in content
class TestGetServiceHost:
"""Tests for get_service_host function."""
class TestGetStackHost:
"""Tests for get_stack_host function."""
def test_get_existing_service(self, config: Config) -> None:
"""Returns host for existing service."""
def test_get_existing_stack(self, config: Config) -> None:
"""Returns host for existing stack."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
host = get_service_host(config, "plex")
host = get_stack_host(config, "plex")
assert host == "nas01"
def test_get_nonexistent_service(self, config: Config) -> None:
"""Returns None for service not in state."""
def test_get_nonexistent_stack(self, config: Config) -> None:
"""Returns None for stack not in state."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
host = get_service_host(config, "unknown")
host = get_stack_host(config, "unknown")
assert host is None
class TestSetServiceHost:
"""Tests for set_service_host function."""
class TestSetStackHost:
"""Tests for set_stack_host function."""
def test_set_new_service(self, config: Config) -> None:
"""Adds new service to state."""
set_service_host(config, "plex", "nas01")
def test_set_new_stack(self, config: Config) -> None:
"""Adds new stack to state."""
set_stack_host(config, "plex", "nas01")
result = load_state(config)
assert result["plex"] == "nas01"
def test_update_existing_service(self, config: Config) -> None:
"""Updates host for existing service."""
def test_update_existing_stack(self, config: Config) -> None:
"""Updates host for existing stack."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
set_service_host(config, "plex", "nas02")
set_stack_host(config, "plex", "nas02")
result = load_state(config)
assert result["plex"] == "nas02"
class TestRemoveService:
"""Tests for remove_service function."""
class TestRemoveStack:
"""Tests for remove_stack function."""
def test_remove_existing_service(self, config: Config) -> None:
"""Removes service from state."""
def test_remove_existing_stack(self, config: Config) -> None:
"""Removes stack from state."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n jellyfin: nas02\n")
remove_service(config, "plex")
remove_stack(config, "plex")
result = load_state(config)
assert "plex" not in result
assert result["jellyfin"] == "nas02"
def test_remove_nonexistent_service(self, config: Config) -> None:
"""Removing nonexistent service doesn't error."""
def test_remove_nonexistent_stack(self, config: Config) -> None:
"""Removing nonexistent stack doesn't error."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
remove_service(config, "unknown") # Should not raise
remove_stack(config, "unknown") # Should not raise
result = load_state(config)
assert result["plex"] == "nas01"
class TestGetOrphanedServices:
"""Tests for get_orphaned_services function."""
class TestGetOrphanedStacks:
"""Tests for get_orphaned_stacks function."""
def test_no_orphans(self, config: Config) -> None:
"""Returns empty dict when all services in state are in config."""
"""Returns empty dict when all stacks in state are in config."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
result = get_orphaned_services(config)
result = get_orphaned_stacks(config)
assert result == {}
def test_finds_orphaned_service(self, config: Config) -> None:
"""Returns services in state but not in config."""
def test_finds_orphaned_stack(self, config: Config) -> None:
"""Returns stacks in state but not in config."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n jellyfin: nas02\n")
result = get_orphaned_services(config)
result = get_orphaned_stacks(config)
# plex is in config, jellyfin is not
assert result == {"jellyfin": "nas02"}
def test_finds_orphaned_multi_host_service(self, config: Config) -> None:
"""Returns multi-host orphaned services with host list."""
def test_finds_orphaned_multi_host_stack(self, config: Config) -> None:
"""Returns multi-host orphaned stacks with host list."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n dozzle:\n - nas01\n - nas02\n")
result = get_orphaned_services(config)
result = get_orphaned_stacks(config)
assert result == {"dozzle": ["nas01", "nas02"]}
def test_empty_state(self, config: Config) -> None:
"""Returns empty dict when state is empty."""
result = get_orphaned_services(config)
result = get_orphaned_stacks(config)
assert result == {}
def test_all_orphaned(self, tmp_path: Path) -> None:
"""Returns all services when none are in config."""
"""Returns all stacks when none are in config."""
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
cfg = Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={}, # No services in config
stacks={}, # No stacks in config
config_path=config_path,
)
state_file = cfg.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n jellyfin: nas02\n")
result = get_orphaned_services(cfg)
result = get_orphaned_stacks(cfg)
assert result == {"plex": "nas01", "jellyfin": "nas02"}
class TestGetServicesNotInState:
"""Tests for get_services_not_in_state function."""
class TestGetStacksNotInState:
"""Tests for get_stacks_not_in_state function."""
def test_all_in_state(self, config: Config) -> None:
"""Returns empty list when all services are in state."""
"""Returns empty list when all stacks are in state."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
result = get_services_not_in_state(config)
result = get_stacks_not_in_state(config)
assert result == []
def test_finds_missing_service(self, tmp_path: Path) -> None:
"""Returns services in config but not in state."""
"""Returns stacks in config but not in state."""
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
cfg = Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01", "jellyfin": "nas01"},
stacks={"plex": "nas01", "jellyfin": "nas01"},
config_path=config_path,
)
state_file = cfg.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
result = get_services_not_in_state(cfg)
result = get_stacks_not_in_state(cfg)
assert result == ["jellyfin"]
def test_empty_state(self, tmp_path: Path) -> None:
"""Returns all services when state is empty."""
"""Returns all stacks when state is empty."""
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
cfg = Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01", "jellyfin": "nas01"},
stacks={"plex": "nas01", "jellyfin": "nas01"},
config_path=config_path,
)
result = get_services_not_in_state(cfg)
result = get_stacks_not_in_state(cfg)
assert set(result) == {"plex", "jellyfin"}
def test_empty_config(self, config: Config) -> None:
"""Returns empty list when config has no services."""
"""Returns empty list when config has no stacks."""
# config fixture has plex: nas01, but we need empty config
config_path = config.config_path
config_path.write_text("")
cfg = Config(
compose_dir=config.compose_dir,
hosts={"nas01": Host(address="192.168.1.10")},
services={},
stacks={},
config_path=config_path,
)
result = get_services_not_in_state(cfg)
result = get_stacks_not_in_state(cfg)
assert result == []

22
tests/test_traefik.py
View File

@@ -18,7 +18,7 @@ def test_generate_traefik_config_with_published_port(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01"},
stacks={"plex": "nas01"},
)
compose_path = tmp_path / "plex" / "docker-compose.yml"
_write_compose(
@@ -56,7 +56,7 @@ def test_generate_traefik_config_without_published_port_warns(tmp_path: Path) ->
cfg = Config(
compose_dir=tmp_path,
hosts={"nas01": Host(address="192.168.1.10")},
services={"app": "nas01"},
stacks={"app": "nas01"},
)
compose_path = tmp_path / "app" / "docker-compose.yml"
_write_compose(
@@ -84,7 +84,7 @@ def test_generate_interpolates_env_and_infers_router_service(tmp_path: Path) ->
cfg = Config(
compose_dir=tmp_path,
hosts={"nas01": Host(address="192.168.1.10")},
services={"wakapi": "nas01"},
stacks={"wakapi": "nas01"},
)
compose_dir = tmp_path / "wakapi"
compose_dir.mkdir(parents=True, exist_ok=True)
@@ -126,7 +126,7 @@ def test_generate_interpolates_label_keys_and_ports(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"nas01": Host(address="192.168.1.10")},
services={"supabase": "nas01"},
stacks={"supabase": "nas01"},
)
compose_dir = tmp_path / "supabase"
compose_dir.mkdir(parents=True, exist_ok=True)
@@ -171,7 +171,7 @@ def test_generate_skips_services_with_enable_false(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"nas01": Host(address="192.168.1.10")},
services={"stack": "nas01"},
stacks={"stack": "nas01"},
)
compose_path = tmp_path / "stack" / "docker-compose.yml"
_write_compose(
@@ -201,7 +201,7 @@ def test_generate_follows_network_mode_service_for_ports(tmp_path: Path) -> None
cfg = Config(
compose_dir=tmp_path,
hosts={"nas01": Host(address="192.168.1.10")},
services={"vpn-stack": "nas01"},
stacks={"vpn-stack": "nas01"},
)
compose_path = tmp_path / "vpn-stack" / "docker-compose.yml"
_write_compose(
@@ -249,7 +249,7 @@ def test_parse_external_networks_single(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"host1": Host(address="192.168.1.10")},
services={"app": "host1"},
stacks={"app": "host1"},
)
compose_path = tmp_path / "app" / "compose.yaml"
_write_compose(
@@ -269,7 +269,7 @@ def test_parse_external_networks_multiple(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"host1": Host(address="192.168.1.10")},
services={"app": "host1"},
stacks={"app": "host1"},
)
compose_path = tmp_path / "app" / "compose.yaml"
_write_compose(
@@ -293,7 +293,7 @@ def test_parse_external_networks_none(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"host1": Host(address="192.168.1.10")},
services={"app": "host1"},
stacks={"app": "host1"},
)
compose_path = tmp_path / "app" / "compose.yaml"
_write_compose(
@@ -313,7 +313,7 @@ def test_parse_external_networks_no_networks_section(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"host1": Host(address="192.168.1.10")},
services={"app": "host1"},
stacks={"app": "host1"},
)
compose_path = tmp_path / "app" / "compose.yaml"
_write_compose(
@@ -330,7 +330,7 @@ def test_parse_external_networks_missing_compose(tmp_path: Path) -> None:
cfg = Config(
compose_dir=tmp_path,
hosts={"host1": Host(address="192.168.1.10")},
services={"app": "host1"},
stacks={"app": "host1"},
)
# Don't create compose file

8
tests/web/conftest.py
View File

@@ -13,11 +13,11 @@ if TYPE_CHECKING:
@pytest.fixture
def compose_dir(tmp_path: Path) -> Path:
"""Create a temporary compose directory with sample services."""
"""Create a temporary compose directory with sample stacks."""
compose_path = tmp_path / "compose"
compose_path.mkdir()
# Create a sample service
# Create a sample stack
plex_dir = compose_path / "plex"
plex_dir.mkdir()
(plex_dir / "compose.yaml").write_text("""
@@ -30,7 +30,7 @@ services:
""")
(plex_dir / ".env").write_text("PLEX_CLAIM=claim-xxx\n")
# Create another service
# Create another stack
sonarr_dir = compose_path / "sonarr"
sonarr_dir.mkdir()
(sonarr_dir / "compose.yaml").write_text("""
@@ -56,7 +56,7 @@ hosts:
server-2:
address: 192.168.1.11
services:
stacks:
plex: server-1
sonarr: server-2
""")

16
tests/web/test_helpers.py
View File

@@ -33,22 +33,22 @@ class TestValidateYaml:
assert "Invalid YAML" in exc_info.value.detail
class TestGetServiceComposePath:
"""Tests for _get_service_compose_path helper."""
class TestGetStackComposePath:
"""Tests for _get_stack_compose_path helper."""
def test_service_found(self, mock_config: Config) -> None:
from compose_farm.web.routes.api import _get_service_compose_path
def test_stack_found(self, mock_config: Config) -> None:
from compose_farm.web.routes.api import _get_stack_compose_path
path = _get_service_compose_path("plex")
path = _get_stack_compose_path("plex")
assert isinstance(path, Path)
assert path.name == "compose.yaml"
assert path.parent.name == "plex"
def test_service_not_found(self, mock_config: Config) -> None:
from compose_farm.web.routes.api import _get_service_compose_path
def test_stack_not_found(self, mock_config: Config) -> None:
from compose_farm.web.routes.api import _get_stack_compose_path
with pytest.raises(HTTPException) as exc_info:
_get_service_compose_path("nonexistent")
_get_stack_compose_path("nonexistent")
assert exc_info.value.status_code == 404
assert "not found" in exc_info.value.detail

364
tests/web/test_htmx_browser.py
View File

@@ -109,13 +109,13 @@ def browser_type_launch_args() -> dict[str, str]:
def test_config(tmp_path_factory: pytest.TempPathFactory) -> Path:
"""Create test config and compose files.
Creates a multi-host, multi-service config for comprehensive testing:
Creates a multi-host, multi-stack config for comprehensive testing:
- server-1: plex (running), sonarr (not started)
- server-2: radarr (running), jellyfin (not started)
"""
tmp: Path = tmp_path_factory.mktemp("data")
# Create compose dir with services
# Create compose dir with stacks
compose_dir = tmp / "compose"
compose_dir.mkdir()
for name in ["plex", "sonarr", "radarr", "jellyfin"]:
@@ -134,7 +134,7 @@ hosts:
server-2:
address: 192.168.1.20
user: docker
services:
stacks:
plex: server-1
sonarr: server-1
radarr: server-2
@@ -224,19 +224,19 @@ class TestHTMXSidebarLoading:
nav = page.locator("nav")
assert "Loading" in nav.inner_text() or nav.locator(".loading").count() > 0
def test_sidebar_loads_services_via_htmx(self, page: Page, server_url: str) -> None:
"""Sidebar fetches and displays services via hx-get on load."""
def test_sidebar_loads_stacks_via_htmx(self, page: Page, server_url: str) -> None:
"""Sidebar fetches and displays stacks via hx-get on load."""
page.goto(server_url)
# Wait for HTMX to load sidebar content
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Verify actual services from test config appear
services = page.locator("#sidebar-services li")
assert services.count() == 4 # plex, sonarr, radarr, jellyfin
# Verify actual stacks from test config appear
stacks = page.locator("#sidebar-stacks li")
assert stacks.count() == 4 # plex, sonarr, radarr, jellyfin
# Check specific services are present
content = page.locator("#sidebar-services").inner_text()
# Check specific stacks are present
content = page.locator("#sidebar-stacks").inner_text()
assert "plex" in content
assert "sonarr" in content
assert "radarr" in content
@@ -257,27 +257,27 @@ class TestHTMXSidebarLoading:
assert stats.is_visible()
# Wait for sidebar to fully load via HTMX
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Dashboard content must STILL be visible after sidebar loads
assert stats.is_visible(), "Dashboard disappeared after sidebar loaded"
assert page.locator("#stats-cards .card").count() >= 4
def test_sidebar_shows_running_status(self, page: Page, server_url: str) -> None:
"""Sidebar shows running/stopped status indicators for services."""
"""Sidebar shows running/stopped status indicators for stacks."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# plex and radarr are in state (running) - should have success status
plex_item = page.locator("#sidebar-services li", has_text="plex")
plex_item = page.locator("#sidebar-stacks li", has_text="plex")
assert plex_item.locator(".status-success").count() == 1
radarr_item = page.locator("#sidebar-services li", has_text="radarr")
radarr_item = page.locator("#sidebar-stacks li", has_text="radarr")
assert radarr_item.locator(".status-success").count() == 1
# sonarr and jellyfin are NOT in state (not started) - should have neutral status
sonarr_item = page.locator("#sidebar-services li", has_text="sonarr")
sonarr_item = page.locator("#sidebar-stacks li", has_text="sonarr")
assert sonarr_item.locator(".status-neutral").count() == 1
jellyfin_item = page.locator("#sidebar-services li", has_text="jellyfin")
jellyfin_item = page.locator("#sidebar-stacks li", has_text="jellyfin")
assert jellyfin_item.locator(".status-neutral").count() == 1
@@ -287,19 +287,19 @@ class TestHTMXBoostNavigation:
def test_navigation_updates_url_without_full_reload(self, page: Page, server_url: str) -> None:
"""Clicking boosted link updates URL without full page reload."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Add a marker to detect full page reload
page.evaluate("window.__htmxTestMarker = 'still-here'")
# Click a service link (boosted via hx-boost on parent)
page.locator("#sidebar-services a", has_text="plex").click()
# Click a stack link (boosted via hx-boost on parent)
page.locator("#sidebar-stacks a", has_text="plex").click()
# Wait for navigation
page.wait_for_url("**/service/plex", timeout=5000)
page.wait_for_url("**/stack/plex", timeout=5000)
# Verify URL changed
assert "/service/plex" in page.url
assert "/stack/plex" in page.url
# Verify NO full page reload (marker should still exist)
marker = page.evaluate("window.__htmxTestMarker")
@@ -308,17 +308,17 @@ class TestHTMXBoostNavigation:
def test_main_content_replaced_on_navigation(self, page: Page, server_url: str) -> None:
"""Navigation replaces #main-content via hx-target/hx-select."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Get initial main content
initial_content = page.locator("#main-content").inner_text()
assert "Compose Farm" in initial_content # Dashboard title
# Navigate to service page
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to stack page
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Main content should now show service page
# Main content should now show stack page
new_content = page.locator("#main-content").inner_text()
assert "plex" in new_content.lower()
assert "Compose Farm" not in new_content # Dashboard title should be gone
@@ -328,17 +328,17 @@ class TestDashboardContent:
"""Test dashboard displays correct data."""
def test_stats_show_correct_counts(self, page: Page, server_url: str) -> None:
"""Stats cards show accurate host/service counts from config."""
"""Stats cards show accurate host/stack counts from config."""
page.goto(server_url)
page.wait_for_selector("#stats-cards", timeout=5000)
stats = page.locator("#stats-cards").inner_text()
# From test config: 2 hosts, 4 services, 2 running (plex, radarr)
# From test config: 2 hosts, 4 stacks, 2 running (plex, radarr)
assert "2" in stats # hosts count
assert "4" in stats # services count
assert "4" in stats # stacks count
def test_pending_shows_not_started_services(self, page: Page, server_url: str) -> None:
def test_pending_shows_not_started_stacks(self, page: Page, server_url: str) -> None:
"""Pending operations shows sonarr and jellyfin as not started."""
page.goto(server_url)
page.wait_for_selector("#pending-operations", timeout=5000)
@@ -353,7 +353,7 @@ class TestDashboardContent:
def test_dashboard_monaco_loads(self, page: Page, server_url: str) -> None:
"""Dashboard page loads Monaco editor for config editing."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Wait for Monaco to load
page.wait_for_function("typeof monaco !== 'undefined'", timeout=5000)
@@ -391,19 +391,19 @@ class TestSaveConfigButton:
assert "Saved" in save_btn.inner_text()
class TestServiceDetailPage:
"""Test service detail page via HTMX navigation."""
class TestStackDetailPage:
"""Test stack detail page via HTMX navigation."""
def test_service_page_shows_service_info(self, page: Page, server_url: str) -> None:
"""Service page displays service information."""
def test_stack_page_shows_stack_info(self, page: Page, server_url: str) -> None:
"""Stack page displays stack information."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Should show service name and host info
# Should show stack name and host info
content = page.locator("#main-content").inner_text()
assert "plex" in content.lower()
assert "server-1" in content # assigned host from config
@@ -413,11 +413,11 @@ class TestServiceDetailPage:
def test_back_navigation_works(self, page: Page, server_url: str) -> None:
"""Browser back button works after HTMX navigation."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Go back
page.go_back()
@@ -426,14 +426,14 @@ class TestServiceDetailPage:
# Should be back on dashboard
assert page.url.rstrip("/") == server_url.rstrip("/")
def test_service_page_monaco_loads(self, page: Page, server_url: str) -> None:
"""Service page loads Monaco editor for compose/env editing."""
def test_stack_page_monaco_loads(self, page: Page, server_url: str) -> None:
"""Stack page loads Monaco editor for compose/env editing."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Wait for Monaco to load
page.wait_for_function("typeof monaco !== 'undefined'", timeout=5000)
@@ -459,33 +459,33 @@ class TestSidebarFilter:
filter_input.fill(text)
filter_input.dispatch_event("keyup")
def test_text_filter_hides_non_matching_services(self, page: Page, server_url: str) -> None:
"""Typing in filter input hides services that don't match."""
def test_text_filter_hides_non_matching_stacks(self, page: Page, server_url: str) -> None:
"""Typing in filter input hides stacks that don't match."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Initially all 4 services visible
visible_items = page.locator("#sidebar-services li:not([hidden])")
# Initially all 4 stacks visible
visible_items = page.locator("#sidebar-stacks li:not([hidden])")
assert visible_items.count() == 4
# Type in filter to match only "plex"
self._filter_sidebar(page, "plex")
# Only plex should be visible now
visible_after = page.locator("#sidebar-services li:not([hidden])")
visible_after = page.locator("#sidebar-stacks li:not([hidden])")
assert visible_after.count() == 1
assert "plex" in visible_after.first.inner_text()
def test_text_filter_updates_count_badge(self, page: Page, server_url: str) -> None:
"""Filter updates the service count badge."""
"""Filter updates the stack count badge."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Initial count should be (4)
count_badge = page.locator("#sidebar-count")
assert "(4)" in count_badge.inner_text()
# Filter to show only services containing "arr" (sonarr, radarr)
# Filter to show only stacks containing "arr" (sonarr, radarr)
self._filter_sidebar(page, "arr")
# Count should update to (2)
@@ -494,26 +494,26 @@ class TestSidebarFilter:
def test_text_filter_is_case_insensitive(self, page: Page, server_url: str) -> None:
"""Filter matching is case-insensitive."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Type uppercase
self._filter_sidebar(page, "PLEX")
# Should still match plex
visible = page.locator("#sidebar-services li:not([hidden])")
visible = page.locator("#sidebar-stacks li:not([hidden])")
assert visible.count() == 1
assert "plex" in visible.first.inner_text().lower()
def test_host_dropdown_filters_by_host(self, page: Page, server_url: str) -> None:
"""Host dropdown filters services by their assigned host."""
"""Host dropdown filters stacks by their assigned host."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Select server-1 from dropdown
page.locator("#sidebar-host-select").select_option("server-1")
# Only plex and sonarr (server-1 services) should be visible
visible = page.locator("#sidebar-services li:not([hidden])")
# Only plex and sonarr (server-1 stacks) should be visible
visible = page.locator("#sidebar-stacks li:not([hidden])")
assert visible.count() == 2
content = visible.all_inner_texts()
@@ -525,7 +525,7 @@ class TestSidebarFilter:
def test_combined_text_and_host_filter(self, page: Page, server_url: str) -> None:
"""Text filter and host filter work together."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Filter by server-2 host
page.locator("#sidebar-host-select").select_option("server-2")
@@ -533,24 +533,24 @@ class TestSidebarFilter:
# Then filter by text "arr" (should match only radarr on server-2)
self._filter_sidebar(page, "arr")
visible = page.locator("#sidebar-services li:not([hidden])")
visible = page.locator("#sidebar-stacks li:not([hidden])")
assert visible.count() == 1
assert "radarr" in visible.first.inner_text()
def test_clearing_filter_shows_all_services(self, page: Page, server_url: str) -> None:
"""Clearing filter restores all services."""
def test_clearing_filter_shows_all_stacks(self, page: Page, server_url: str) -> None:
"""Clearing filter restores all stacks."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Apply filter
self._filter_sidebar(page, "plex")
assert page.locator("#sidebar-services li:not([hidden])").count() == 1
assert page.locator("#sidebar-stacks li:not([hidden])").count() == 1
# Clear filter
self._filter_sidebar(page, "")
# All services visible again
assert page.locator("#sidebar-services li:not([hidden])").count() == 4
# All stacks visible again
assert page.locator("#sidebar-stacks li:not([hidden])").count() == 4
class TestCommandPalette:
@@ -559,7 +559,7 @@ class TestCommandPalette:
def test_cmd_k_opens_palette(self, page: Page, server_url: str) -> None:
"""Cmd+K keyboard shortcut opens the command palette."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Palette should be closed initially
assert not page.locator("#cmd-palette").is_visible()
@@ -574,7 +574,7 @@ class TestCommandPalette:
def test_palette_input_is_focused_on_open(self, page: Page, server_url: str) -> None:
"""Input field is focused when palette opens."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
@@ -586,7 +586,7 @@ class TestCommandPalette:
def test_palette_shows_navigation_commands(self, page: Page, server_url: str) -> None:
"""Palette shows Dashboard and Console navigation commands."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
@@ -595,23 +595,23 @@ class TestCommandPalette:
assert "Dashboard" in cmd_list
assert "Console" in cmd_list
def test_palette_shows_service_navigation(self, page: Page, server_url: str) -> None:
"""Palette includes service names for navigation."""
def test_palette_shows_stack_navigation(self, page: Page, server_url: str) -> None:
"""Palette includes stack names for navigation."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
cmd_list = page.locator("#cmd-list").inner_text()
# Services should appear as navigation options
# Stacks should appear as navigation options
assert "plex" in cmd_list
assert "radarr" in cmd_list
def test_palette_filters_on_input(self, page: Page, server_url: str) -> None:
"""Typing in palette filters the command list."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
@@ -627,7 +627,7 @@ class TestCommandPalette:
def test_arrow_down_moves_selection(self, page: Page, server_url: str) -> None:
"""Arrow down key moves selection to next item."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
@@ -648,25 +648,25 @@ class TestCommandPalette:
def test_enter_executes_and_closes_palette(self, page: Page, server_url: str) -> None:
"""Enter key executes selected command and closes palette."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
# Filter to plex service
# Filter to plex stack
page.locator("#cmd-input").fill("plex")
page.keyboard.press("Enter")
# Palette should close
page.wait_for_selector("#cmd-palette:not([open])", timeout=2000)
# Should navigate to plex service page
page.wait_for_url("**/service/plex", timeout=5000)
# Should navigate to plex stack page
page.wait_for_url("**/stack/plex", timeout=5000)
def test_click_executes_command(self, page: Page, server_url: str) -> None:
"""Clicking a command executes it."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
@@ -680,7 +680,7 @@ class TestCommandPalette:
def test_escape_closes_palette(self, page: Page, server_url: str) -> None:
"""Escape key closes the palette without executing."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
@@ -694,7 +694,7 @@ class TestCommandPalette:
def test_fab_button_opens_palette(self, page: Page, server_url: str) -> None:
"""Floating action button opens the command palette."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Click the FAB
page.locator("#cmd-fab").click()
@@ -709,7 +709,7 @@ class TestActionButtons:
def test_apply_button_makes_post_request(self, page: Page, server_url: str) -> None:
"""Apply button triggers POST to /api/apply."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Intercept the API call
api_calls: list[str] = []
@@ -738,7 +738,7 @@ class TestActionButtons:
def test_refresh_button_makes_post_request(self, page: Page, server_url: str) -> None:
"""Refresh button triggers POST to /api/refresh."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
api_calls: list[str] = []
@@ -761,7 +761,7 @@ class TestActionButtons:
def test_action_response_expands_terminal(self, page: Page, server_url: str) -> None:
"""Action button response with task_id expands terminal section."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Terminal should be collapsed initially
terminal_toggle = page.locator("#terminal-toggle")
@@ -786,16 +786,16 @@ class TestActionButtons:
timeout=3000,
)
def test_service_page_action_buttons(self, page: Page, server_url: str) -> None:
def test_stack_page_action_buttons(self, page: Page, server_url: str) -> None:
"""Service page has working action buttons."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Intercept service-specific API calls
# Intercept stack-specific API calls
api_calls: list[str] = []
def handle_route(route: Route) -> None:
@@ -806,14 +806,14 @@ class TestActionButtons:
body='{"task_id": "test-up-123"}',
)
page.route("**/api/service/plex/up", handle_route)
page.route("**/api/stack/plex/up", handle_route)
# Click Up button (use get_by_role for exact match, avoiding "Update")
page.get_by_role("button", name="Up", exact=True).click()
page.wait_for_timeout(500)
assert len(api_calls) == 1
assert "/api/service/plex/up" in api_calls[0]
assert "/api/stack/plex/up" in api_calls[0]
class TestKeyboardShortcuts:
@@ -855,50 +855,50 @@ class TestContentStability:
page.goto(server_url)
# Wait for all HTMX requests to complete
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
page.wait_for_load_state("networkidle")
# All major dashboard sections must be visible
assert page.locator("#stats-cards").is_visible(), "Stats cards missing"
assert page.locator("#stats-cards .card").count() >= 4, "Stats incomplete"
assert page.locator("#pending-operations").is_visible(), "Pending ops missing"
assert page.locator("#services-by-host").is_visible(), "Services by host missing"
assert page.locator("#sidebar-services").is_visible(), "Sidebar missing"
assert page.locator("#stacks-by-host").is_visible(), "Stacks by host missing"
assert page.locator("#sidebar-stacks").is_visible(), "Sidebar missing"
def test_sidebar_persists_after_navigation_and_back(self, page: Page, server_url: str) -> None:
"""Sidebar content persists through navigation cycle."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Remember sidebar state
initial_count = page.locator("#sidebar-services li").count()
initial_count = page.locator("#sidebar-stacks li").count()
assert initial_count == 4
# Navigate away
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Sidebar should still be there with same content
assert page.locator("#sidebar-services").is_visible()
assert page.locator("#sidebar-services li").count() == initial_count
assert page.locator("#sidebar-stacks").is_visible()
assert page.locator("#sidebar-stacks li").count() == initial_count
# Navigate back
page.go_back()
page.wait_for_url(server_url, timeout=5000)
# Sidebar still intact
assert page.locator("#sidebar-services").is_visible()
assert page.locator("#sidebar-services li").count() == initial_count
assert page.locator("#sidebar-stacks").is_visible()
assert page.locator("#sidebar-stacks li").count() == initial_count
def test_dashboard_sections_persist_after_save(self, page: Page, server_url: str) -> None:
"""Dashboard sections remain after save triggers cf:refresh event."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Capture initial state - all must be visible
assert page.locator("#stats-cards").is_visible()
assert page.locator("#pending-operations").is_visible()
assert page.locator("#services-by-host").is_visible()
assert page.locator("#stacks-by-host").is_visible()
# Trigger save (which dispatches cf:refresh)
page.locator("#save-config-btn").click()
@@ -913,15 +913,15 @@ class TestContentStability:
# All sections must still be visible
assert page.locator("#stats-cards").is_visible(), "Stats disappeared after save"
assert page.locator("#pending-operations").is_visible(), "Pending disappeared"
assert page.locator("#services-by-host").is_visible(), "Services disappeared"
assert page.locator("#sidebar-services").is_visible(), "Sidebar disappeared"
assert page.locator("#stacks-by-host").is_visible(), "Stacks disappeared"
assert page.locator("#sidebar-stacks").is_visible(), "Sidebar disappeared"
def test_filter_state_not_affected_by_other_htmx_requests(
self, page: Page, server_url: str
) -> None:
"""Sidebar filter state persists during other HTMX activity."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Apply a filter
filter_input = page.locator("#sidebar-filter")
@@ -929,7 +929,7 @@ class TestContentStability:
filter_input.dispatch_event("keyup")
# Verify filter is applied
assert page.locator("#sidebar-services li:not([hidden])").count() == 1
assert page.locator("#sidebar-stacks li:not([hidden])").count() == 1
# Trigger a save (causes cf:refresh on multiple elements)
page.locator("#save-config-btn").click()
@@ -937,15 +937,15 @@ class TestContentStability:
# Filter input should still have our text
# (Note: sidebar reloads so filter clears - this tests the sidebar reload works)
page.wait_for_selector("#sidebar-services", timeout=5000)
assert page.locator("#sidebar-services").is_visible()
page.wait_for_selector("#sidebar-stacks", timeout=5000)
assert page.locator("#sidebar-stacks").is_visible()
def test_main_content_not_affected_by_sidebar_refresh(
self, page: Page, server_url: str
) -> None:
"""Main content area stays intact when sidebar refreshes."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Get main content text
main_content = page.locator("#main-content")
@@ -965,11 +965,11 @@ class TestContentStability:
) -> None:
"""Multiple refresh cycles don't create duplicate elements."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Count initial elements
initial_stat_count = page.locator("#stats-cards .card").count()
initial_service_count = page.locator("#sidebar-services li").count()
initial_stack_count = page.locator("#sidebar-stacks li").count()
# Trigger multiple refreshes
for _ in range(3):
@@ -980,7 +980,7 @@ class TestContentStability:
# Counts should be same (no duplicates created)
assert page.locator("#stats-cards .card").count() == initial_stat_count
assert page.locator("#sidebar-services li").count() == initial_service_count
assert page.locator("#sidebar-stacks li").count() == initial_stack_count
class TestConsolePage:
@@ -1310,7 +1310,7 @@ class TestTerminalStreaming:
def test_terminal_stores_task_in_localstorage(self, page: Page, server_url: str) -> None:
"""Action response stores task ID in localStorage for reconnection."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Mock Apply API to return a task ID
page.route(
@@ -1318,7 +1318,7 @@ class TestTerminalStreaming:
lambda route: route.fulfill(
status=200,
content_type="application/json",
body='{"task_id": "test-task-123", "service": null, "command": "apply"}',
body='{"task_id": "test-task-123", "stack": null, "command": "apply"}',
),
)
@@ -1343,7 +1343,7 @@ class TestTerminalStreaming:
"""
# First, set up a task in localStorage before navigating
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Store a task ID in localStorage
page.evaluate("localStorage.setItem('cf_task:/', 'reconnect-test-123')")
@@ -1354,7 +1354,7 @@ class TestTerminalStreaming:
# Navigate back to dashboard
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Wait for xterm to load (reconnect uses whenXtermReady)
page.wait_for_function("typeof Terminal !== 'undefined'", timeout=5000)
@@ -1370,7 +1370,7 @@ class TestTerminalStreaming:
) -> None:
"""Action response with task_id triggers WebSocket connection to correct path."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Track WebSocket connections
ws_urls: list[str] = []
@@ -1386,7 +1386,7 @@ class TestTerminalStreaming:
lambda route: route.fulfill(
status=200,
content_type="application/json",
body='{"task_id": "ws-test-456", "service": null, "command": "apply"}',
body='{"task_id": "ws-test-456", "stack": null, "command": "apply"}',
),
)
@@ -1406,7 +1406,7 @@ class TestTerminalStreaming:
def test_terminal_displays_connected_message(self, page: Page, server_url: str) -> None:
"""Terminal shows [Connected] message after WebSocket opens."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Mock Apply API to return a task ID
page.route(
@@ -1414,7 +1414,7 @@ class TestTerminalStreaming:
lambda route: route.fulfill(
status=200,
content_type="application/json",
body='{"task_id": "connected-test", "service": null, "command": "apply"}',
body='{"task_id": "connected-test", "stack": null, "command": "apply"}',
),
)
@@ -1441,14 +1441,14 @@ class TestTerminalStreaming:
class TestExecTerminal:
"""Test exec terminal functionality for container shells."""
def test_service_page_has_exec_terminal_container(self, page: Page, server_url: str) -> None:
def test_stack_page_has_exec_terminal_container(self, page: Page, server_url: str) -> None:
"""Service page has exec terminal container (initially hidden)."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Exec terminal container should exist but be hidden
exec_container = page.locator("#exec-terminal-container")
@@ -1462,15 +1462,15 @@ class TestExecTerminal:
def test_exec_terminal_connects_websocket(self, page: Page, server_url: str) -> None:
"""Clicking Shell button triggers WebSocket to exec endpoint."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Mock containers API to return a container
page.route(
"**/api/service/plex/containers*",
"**/api/stack/plex/containers*",
lambda route: route.fulfill(
status=200,
content_type="text/html",
@@ -1489,7 +1489,7 @@ class TestExecTerminal:
# Reload to get mocked containers
page.reload()
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Track WebSocket connections
ws_urls: list[str] = []
@@ -1518,22 +1518,22 @@ class TestExecTerminal:
class TestServicePagePalette:
"""Test command palette behavior on service pages."""
"""Test command palette behavior on stack pages."""
def test_service_page_palette_has_action_commands(self, page: Page, server_url: str) -> None:
"""Command palette on service page shows service-specific actions."""
def test_stack_page_palette_has_action_commands(self, page: Page, server_url: str) -> None:
"""Command palette on stack page shows stack-specific actions."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Open command palette
page.keyboard.press("Control+k")
page.wait_for_selector("#cmd-palette[open]", timeout=2000)
# Verify service-specific action commands are visible
# Verify stack-specific action commands are visible
cmd_list = page.locator("#cmd-list").inner_text()
assert "Up" in cmd_list
assert "Down" in cmd_list
@@ -1542,14 +1542,14 @@ class TestServicePagePalette:
assert "Update" in cmd_list
assert "Logs" in cmd_list
def test_palette_action_triggers_service_api(self, page: Page, server_url: str) -> None:
"""Selecting action from palette triggers correct service API."""
def test_palette_action_triggers_stack_api(self, page: Page, server_url: str) -> None:
"""Selecting action from palette triggers correct stack API."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Track API calls
api_calls: list[str] = []
@@ -1559,10 +1559,10 @@ class TestServicePagePalette:
route.fulfill(
status=200,
content_type="application/json",
body='{"task_id": "palette-test", "service": "plex", "command": "up"}',
body='{"task_id": "palette-test", "stack": "plex", "command": "up"}',
)
page.route("**/api/service/plex/up", handle_route)
page.route("**/api/stack/plex/up", handle_route)
# Open command palette
page.keyboard.press("Control+k")
@@ -1577,16 +1577,16 @@ class TestServicePagePalette:
# Verify correct API was called
assert len(api_calls) >= 1
assert "/api/service/plex/up" in api_calls[0]
assert "/api/stack/plex/up" in api_calls[0]
def test_palette_apply_from_service_page(self, page: Page, server_url: str) -> None:
"""Selecting Apply from service page palette navigates to dashboard and triggers API."""
def test_palette_apply_from_stack_page(self, page: Page, server_url: str) -> None:
"""Selecting Apply from stack page palette navigates to dashboard and triggers API."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services a", timeout=5000)
page.wait_for_selector("#sidebar-stacks a", timeout=5000)
# Navigate to plex service
page.locator("#sidebar-services a", has_text="plex").click()
page.wait_for_url("**/service/plex", timeout=5000)
# Navigate to plex stack
page.locator("#sidebar-stacks a", has_text="plex").click()
page.wait_for_url("**/stack/plex", timeout=5000)
# Track API calls
api_calls: list[str] = []
@@ -1596,7 +1596,7 @@ class TestServicePagePalette:
route.fulfill(
status=200,
content_type="application/json",
body='{"task_id": "apply-test", "service": null, "command": "apply"}',
body='{"task_id": "apply-test", "stack": null, "command": "apply"}',
)
page.route("**/api/apply", handle_route)
@@ -1637,7 +1637,7 @@ class TestThemeSwitcher:
def test_theme_button_exists(self, page: Page, server_url: str) -> None:
"""Theme button exists in sidebar header."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Theme button should exist
assert page.locator("#theme-btn").count() == 1
@@ -1645,7 +1645,7 @@ class TestThemeSwitcher:
def test_theme_button_opens_palette_with_filter(self, page: Page, server_url: str) -> None:
"""Clicking theme button opens command palette pre-filtered to themes."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
self._open_theme_palette(page)
@@ -1660,7 +1660,7 @@ class TestThemeSwitcher:
def test_clicking_theme_changes_html_data_theme(self, page: Page, server_url: str) -> None:
"""Selecting a theme changes the data-theme attribute on <html>."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Get initial theme
initial_theme = page.locator("html").get_attribute("data-theme")
@@ -1677,7 +1677,7 @@ class TestThemeSwitcher:
def test_theme_persists_in_localstorage(self, page: Page, server_url: str) -> None:
"""Selected theme is saved to localStorage."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Select synthwave theme
self._open_theme_palette(page)
@@ -1690,7 +1690,7 @@ class TestThemeSwitcher:
def test_theme_restored_on_page_load(self, page: Page, server_url: str) -> None:
"""Theme is restored from localStorage on page load."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Set theme
self._open_theme_palette(page)
@@ -1698,7 +1698,7 @@ class TestThemeSwitcher:
# Reload page
page.reload()
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Theme should be restored
theme = page.locator("html").get_attribute("data-theme")
@@ -1707,7 +1707,7 @@ class TestThemeSwitcher:
def test_theme_can_be_changed_multiple_times(self, page: Page, server_url: str) -> None:
"""Theme can be changed multiple times in a session."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
themes_to_test = ["light", "dark", "nord", "sunset"]
@@ -1720,7 +1720,7 @@ class TestThemeSwitcher:
def test_themes_available_in_regular_palette(self, page: Page, server_url: str) -> None:
"""Themes are also available when opening regular command palette."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Open with Cmd+K
page.keyboard.press("Control+k")
@@ -1737,7 +1737,7 @@ class TestThemeSwitcher:
def test_theme_command_opens_theme_picker(self, page: Page, server_url: str) -> None:
"""Selecting 'Theme' command reopens palette with theme filter."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Open palette and select Theme command
page.keyboard.press("Control+k")
@@ -1758,7 +1758,7 @@ class TestThemeSwitcher:
def test_current_theme_is_preselected(self, page: Page, server_url: str) -> None:
"""Opening theme picker pre-selects the current theme."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Set a specific theme first
self._open_theme_palette(page)
@@ -1775,7 +1775,7 @@ class TestThemeSwitcher:
def test_theme_shows_color_swatches(self, page: Page, server_url: str) -> None:
"""Theme commands show color preview swatches."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
self._open_theme_palette(page)
@@ -1791,7 +1791,7 @@ class TestThemeSwitcher:
def test_theme_preview_on_arrow_navigation(self, page: Page, server_url: str) -> None:
"""Arrow key navigation previews themes without persisting."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Set initial theme
self._open_theme_palette(page)
@@ -1822,7 +1822,7 @@ class TestThemeSwitcher:
def test_theme_preview_restored_on_escape(self, page: Page, server_url: str) -> None:
"""Pressing Escape restores original theme."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Set initial theme
self._open_theme_palette(page)
@@ -1850,7 +1850,7 @@ class TestThemeSwitcher:
def test_theme_restored_after_non_theme_command(self, page: Page, server_url: str) -> None:
"""Theme restores when executing non-theme command after preview."""
page.goto(server_url)
page.wait_for_selector("#sidebar-services", timeout=5000)
page.wait_for_selector("#sidebar-stacks", timeout=5000)
# Set initial theme to dark
self._open_theme_palette(page)

22
tests/web/test_template_context.py
View File

@@ -21,10 +21,10 @@ def mock_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Config:
compose_dir = tmp_path / "compose"
compose_dir.mkdir()
# Create minimal service directory
svc_dir = compose_dir / "test-service"
svc_dir.mkdir()
(svc_dir / "compose.yaml").write_text("services:\n app:\n image: nginx\n")
# Create minimal stack directory
stack_dir = compose_dir / "test-service"
stack_dir.mkdir()
(stack_dir / "compose.yaml").write_text("services:\n app:\n image: nginx\n")
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text(f"""
@@ -32,7 +32,7 @@ compose_dir: {compose_dir}
hosts:
local-host:
address: localhost
services:
stacks:
test-service: local-host
""")
@@ -78,9 +78,9 @@ class TestPageTemplatesRender:
assert "Console" in response.text
assert "Terminal" in response.text
def test_service_detail_renders(self, client: TestClient) -> None:
"""Test service detail page renders without errors."""
response = client.get("/service/test-service")
def test_stack_detail_renders(self, client: TestClient) -> None:
"""Test stack detail page renders without errors."""
response = client.get("/stack/test-service")
assert response.status_code == 200
assert "test-service" in response.text
@@ -105,7 +105,7 @@ class TestPartialTemplatesRender:
response = client.get("/partials/pending")
assert response.status_code == 200
def test_services_by_host_renders(self, client: TestClient) -> None:
"""Test services_by_host partial renders without errors."""
response = client.get("/partials/services-by-host")
def test_stacks_by_host_renders(self, client: TestClient) -> None:
"""Test stacks_by_host partial renders without errors."""
response = client.get("/partials/stacks-by-host")
assert response.status_code == 200