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
Files
6b802106a9a5c65a75048e6f5fc289051858ee89
compose-farm/tests/test_cli_lifecycle.py
Bas Nijholt 6b802106a9 fix: Don't remove multi-host stacks from state when host-filtered 
When using `-H host` with a multi-host stack, we only stop one instance.
The stack is still running on other hosts, so we shouldn't remove it
from state entirely.

This prevents issues where:
- `cf apply` would try to re-start the stack
- `cf ps` would show incorrect running status
- Orphan detection would be confused

Added tests to verify state is preserved for host-filtered multi-host
operations and removed for full stack operations.
2026-02-01 12:37:47 -08:00

569 lines
23 KiB
Python
Raw Blame History

"""Tests for CLI lifecycle commands (apply, down --orphaned)."""
from pathlib import Path
from unittest.mock import patch
import pytest
import typer
from compose_farm.cli.lifecycle import apply, down
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
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()
stack_dict: dict[str, str | list[str]] = (
dict(stacks) if stacks else {"svc1": "host1", "svc2": "host2"}
)
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("")
return Config(
compose_dir=compose_dir,
hosts={"host1": Host(address="localhost"), "host2": Host(address="localhost")},
stacks=stack_dict,
config_path=config_path,
)
def _make_result(stack: str, success: bool = True) -> CommandResult:
"""Create a command result."""
return CommandResult(
stack=stack,
exit_code=0 if success else 1,
success=success,
stdout="",
stderr="",
)
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 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_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._discover_strays", return_value={}),
):
apply(dry_run=False, no_orphans=False, no_strays=False, full=False, config=None)
captured = capsys.readouterr()
assert "Nothing to apply" in captured.out
def test_apply_dry_run_shows_preview(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Dry run shows what would be done without executing."""
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_stacks",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
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._discover_strays", return_value={}),
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, no_strays=False, full=False, config=None)
captured = capsys.readouterr()
assert "Stacks to migrate" in captured.out
assert "svc1" in captured.out
assert "Orphaned stacks to stop" in captured.out
assert "old-svc" in captured.out
assert "dry-run" in captured.out
# Should not have called the actual operations
mock_stop.assert_not_called()
mock_up.assert_not_called()
def test_apply_executes_migrations(self, tmp_path: Path) -> None:
"""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_stacks", return_value={}),
patch(
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
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._discover_strays", return_value={}),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
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, no_strays=False, full=False, config=None)
mock_up.assert_called_once()
call_args = mock_up.call_args
assert call_args[0][1] == ["svc1"] # stacks list
def test_apply_executes_orphan_cleanup(self, tmp_path: Path) -> None:
"""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_stacks",
return_value={"old-svc": "host1"},
),
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._discover_strays", return_value={}),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
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, no_strays=False, full=False, config=None)
mock_stop.assert_called_once_with(cfg)
def test_apply_no_orphans_skips_orphan_cleanup(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""--no-orphans flag skips orphan cleanup."""
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_stacks",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
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._discover_strays", return_value={}),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
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"),
):
apply(dry_run=False, no_orphans=True, no_strays=False, full=False, config=None)
# Should run migrations but not orphan cleanup
mock_up.assert_called_once()
mock_stop.assert_not_called()
# Orphans should not appear in output
captured = capsys.readouterr()
assert "old-svc" not in captured.out
def test_apply_no_orphans_nothing_to_do(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""--no-orphans with only orphans means nothing to do."""
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_stacks",
return_value={"old-svc": "host1"},
),
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._discover_strays", return_value={}),
):
apply(dry_run=False, no_orphans=True, no_strays=False, full=False, config=None)
captured = capsys.readouterr()
assert "Nothing to apply" in captured.out
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_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=["svc1"],
),
patch("compose_farm.cli.lifecycle._discover_strays", return_value={}),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
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, no_strays=False, full=False, config=None)
mock_up.assert_called_once()
call_args = mock_up.call_args
assert call_args[0][1] == ["svc1"]
def test_apply_dry_run_shows_missing_stacks(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""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_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=["svc1"],
),
patch("compose_farm.cli.lifecycle._discover_strays", return_value={}),
):
apply(dry_run=True, no_orphans=False, no_strays=False, full=False, config=None)
captured = capsys.readouterr()
assert "Stacks to start" in captured.out
assert "svc1" in captured.out
assert "dry-run" in captured.out
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_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._discover_strays", return_value={}),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
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, no_strays=False, full=True, config=None)
mock_up.assert_called_once()
call_args = mock_up.call_args
# 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 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_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._discover_strays", return_value={}),
):
apply(dry_run=True, no_orphans=False, no_strays=False, full=True, config=None)
captured = capsys.readouterr()
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_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_stacks", return_value={}),
patch(
"compose_farm.cli.lifecycle.get_stacks_needing_migration",
return_value=["svc1"],
),
patch(
"compose_farm.cli.lifecycle.get_stacks_not_in_state",
return_value=["svc2"],
),
patch("compose_farm.cli.lifecycle.get_stack_host", return_value="host2"),
patch("compose_farm.cli.lifecycle._discover_strays", return_value={}),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
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, no_strays=False, full=True, config=None)
# 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]
assert refresh_call[0][1] == ["svc3"]
class TestDownOrphaned:
"""Tests for down --orphaned flag."""
def test_down_orphaned_no_orphans(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""When no orphans exist, 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_stacks", return_value={}),
):
down(
stacks=None,
all_stacks=False,
orphaned=True,
host=None,
config=None,
)
captured = capsys.readouterr()
assert "No orphaned stacks to stop" in captured.out
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_stacks",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.stop_orphaned_stacks") as mock_stop,
patch("compose_farm.cli.lifecycle.report_results"),
):
down(
stacks=None,
all_stacks=False,
orphaned=True,
host=None,
config=None,
)
mock_stop.assert_called_once_with(cfg)
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(
stacks=["svc1"],
all_stacks=False,
orphaned=True,
host=None,
config=None,
)
assert exc_info.value.exit_code == 1
def test_down_orphaned_with_all_errors(self) -> None:
"""--orphaned cannot be combined with --all."""
with pytest.raises(typer.Exit) as exc_info:
down(
stacks=None,
all_stacks=True,
orphaned=True,
host=None,
config=None,
)
assert exc_info.value.exit_code == 1
def test_down_orphaned_with_host_errors(self) -> None:
"""--orphaned cannot be combined with --host."""
with pytest.raises(typer.Exit) as exc_info:
down(
stacks=None,
all_stacks=False,
orphaned=True,
host="host1",
config=None,
)
assert exc_info.value.exit_code == 1
class TestHostFilterMultiHost:
"""Tests for --host filter with multi-host stacks."""
def _make_multi_host_config(self, tmp_path: Path) -> Config:
"""Create a config with a multi-host stack."""
compose_dir = tmp_path / "compose"
compose_dir.mkdir()
# Create stack directories
for stack in ["single-host", "multi-host"]:
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("")
return Config(
compose_dir=compose_dir,
hosts={
"host1": Host(address="192.168.1.1"),
"host2": Host(address="192.168.1.2"),
"host3": Host(address="192.168.1.3"),
},
stacks={
"single-host": "host1",
"multi-host": ["host1", "host2", "host3"], # runs on all 3 hosts
},
config_path=config_path,
)
def test_down_host_filter_limits_multi_host_stack(self, tmp_path: Path) -> None:
"""--host filter should only run down on that host for multi-host stacks."""
cfg = self._make_multi_host_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_stacks") as mock_get_stacks,
patch("compose_farm.cli.lifecycle.run_on_stacks") as mock_run,
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=[_make_result("multi-host@host1")],
),
patch("compose_farm.cli.lifecycle.remove_stack"),
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
mock_get_stacks.return_value = (["multi-host"], cfg)
down(
stacks=None,
all_stacks=False,
orphaned=False,
host="host1",
config=None,
)
# Verify run_on_stacks was called with filter_host="host1"
mock_run.assert_called_once()
call_kwargs = mock_run.call_args.kwargs
assert call_kwargs.get("filter_host") == "host1"
def test_down_host_filter_skips_state_removal_for_multi_host(self, tmp_path: Path) -> None:
"""--host filter should NOT remove multi-host stacks from state.
When stopping only one instance of a multi-host stack, the stack is still
running on other hosts, so it shouldn't be removed from state.
"""
cfg = self._make_multi_host_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_stacks") as mock_get_stacks,
patch("compose_farm.cli.lifecycle.run_on_stacks"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=[_make_result("multi-host@host1")],
),
patch("compose_farm.cli.lifecycle.remove_stack") as mock_remove,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
mock_get_stacks.return_value = (["multi-host"], cfg)
down(
stacks=None,
all_stacks=False,
orphaned=False,
host="host1",
config=None,
)
# remove_stack should NOT be called for multi-host stacks with host filter
mock_remove.assert_not_called()
def test_down_without_host_filter_removes_from_state(self, tmp_path: Path) -> None:
"""Without --host filter, multi-host stacks SHOULD be removed from state."""
cfg = self._make_multi_host_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_stacks") as mock_get_stacks,
patch("compose_farm.cli.lifecycle.run_on_stacks"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=[
_make_result("multi-host@host1"),
_make_result("multi-host@host2"),
_make_result("multi-host@host3"),
],
),
patch("compose_farm.cli.lifecycle.remove_stack") as mock_remove,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
mock_get_stacks.return_value = (["multi-host"], cfg)
down(
stacks=None,
all_stacks=False,
orphaned=False,
host=None, # No host filter
config=None,
)
# remove_stack SHOULD be called when stopping all instances
mock_remove.assert_called_once_with(cfg, "multi-host")
Reference in New Issue View Git Blame Copy Permalink