From 007102d8e1d161144cbb237ae462de20a861fc06 Mon Sep 17 00:00:00 2001 From: Devesh36 <142524747+Devesh36@users.noreply.github.com> Date: Tue, 29 Sep 2026 16:38:39 +0530 Subject: [PATCH] test: cover status lifecycle fixture states --- tests/unit/system/operator/test_inventory.py | 71 ++++++++++++++++++++ tests/unit/system/operator/test_units.py | 61 +++++++++++++++++ 2 files changed, 132 insertions(+) diff --git a/tests/unit/system/operator/test_inventory.py b/tests/unit/system/operator/test_inventory.py index 9584713..89ec961 100644 --- a/tests/unit/system/operator/test_inventory.py +++ b/tests/unit/system/operator/test_inventory.py @@ -95,6 +95,77 @@ def test_status_is_graceful_before_init(tmp_path: Path) -> None: assert report.warnings == () +def test_status_reports_initialized_unbound_without_mutating_fixture(tmp_path: Path) -> None: + paths = _paths(tmp_path) + paths.config.parent.mkdir() + paths.config.write_text("", encoding="utf-8") + before = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*"))) + + report = _service( + tmp_path, + receipt=_Receipt(exists=True), + catalog=OwnedUnitCatalog(receipt_present=False), + ).inspect() + + after = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*"))) + assert report.summary is SystemSummary.UNBOUND + assert report.items[0].state is ObservationState.PRESENT + assert report.items[1].state is ObservationState.ABSENT + assert after == before + + +def test_status_reports_bound_but_inactive_units_without_mutating_fixture( + tmp_path: Path, +) -> None: + paths = _paths(tmp_path) + paths.config.parent.mkdir() + paths.config.write_text("", encoding="utf-8") + before = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*"))) + + report = _service( + tmp_path, + receipt=_Receipt(exists=True), + catalog=OwnedUnitCatalog( + receipt_present=True, + generation="bound-generation", + units=( + OwnedUnit( + name="lychd-vessel.service", + sources=("/units/lychd-vessel.container",), + state=ObservationState.INACTIVE, + unit_file_state="disabled", + ), + ), + ), + ).inspect() + + after = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*"))) + assert report.summary is SystemSummary.STOPPED + assert report.items[0].state is ObservationState.PRESENT + assert report.items[1].state is ObservationState.PRESENT + assert report.items[2].state is ObservationState.INACTIVE + assert "1 owned" in report.items[2].detail + assert after == before + + +def test_status_reports_removal_as_uninitialized_without_mutating_fixture( + tmp_path: Path, +) -> None: + before = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*"))) + + report = _service( + tmp_path, + receipt=_Receipt(exists=False), + catalog=OwnedUnitCatalog(receipt_present=False), + ).inspect() + + after = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob("*"))) + assert report.summary is SystemSummary.NOT_INITIALIZED + assert report.items[0].state is ObservationState.ABSENT + assert report.items[1].state is ObservationState.ABSENT + assert after == before + + def test_corrupt_receipt_is_degraded_not_a_crash(tmp_path: Path) -> None: paths = _paths(tmp_path) paths.config.parent.mkdir() diff --git a/tests/unit/system/operator/test_units.py b/tests/unit/system/operator/test_units.py index 886ad99..51fec72 100644 --- a/tests/unit/system/operator/test_units.py +++ b/tests/unit/system/operator/test_units.py @@ -135,3 +135,64 @@ def test_deactivating_vessel_remains_active_for_authority_arbitration( ) assert service.inspect().units[0].state is ObservationState.ACTIVE + + +def test_missing_owned_unit_is_reported_with_its_load_state(tmp_path: Path) -> None: + source = tmp_path / "lychd-vessel.container" + runner = _Runner( + [ + ProcessResult( + argv=(), + returncode=0, + stdout="LoadState=not-found\nActiveState=inactive\nUnitFileState=\n", + ) + ] + ) + service = OwnedUnitInventoryService( + cast( + "ScribeService", + _Scribe( + OwnedBindings( + receipt_present=True, + generation="generation", + quadlet_sources=(source,), + runtime_units=("lychd-vessel.service",), + ) + ), + ), + runner, + systemctl_bin="/usr/bin/systemctl", + ) + + unit = service.inspect().units[0] + + assert unit.state is ObservationState.INACTIVE + assert unit.detail == "load state: not-found" + assert runner.calls + + +def test_unavailable_systemctl_keeps_owned_unit_state_unknown(tmp_path: Path) -> None: + source = tmp_path / "lychd-vessel.container" + runner = _Runner([]) + service = OwnedUnitInventoryService( + cast( + "ScribeService", + _Scribe( + OwnedBindings( + receipt_present=True, + generation="generation", + quadlet_sources=(source,), + runtime_units=("lychd-vessel.service",), + ) + ), + ), + runner, + systemctl_bin=None, + ) + + catalog = service.inspect() + + assert len(catalog.units) == 1 + assert catalog.units[0].state is ObservationState.UNKNOWN + assert catalog.units[0].detail == "systemctl is unavailable" + assert runner.calls == []