from __future__ import annotations import subprocess import unittest from types import SimpleNamespace from unittest.mock import patch from app.core.command_runner import CommandError from app.core.installer import APT_NONINTERACTIVE_ENV, AptInstaller from app.core.manifest_validator import ManifestValidator from app.core.service_manager import ServiceManager from app.core.task_runner import InstalledComponentVerificationError, TaskRunner def apt_manifest(component: dict | None = None) -> dict: return { "schemaVersion": "1.0", "appId": "postgresql", "appName": "PostgreSQL", "version": "16", "components": [ component or { "componentId": "postgresql", "type": "apt", "packageName": "postgresql", "version": "16", } ], } class FakeCommandRunner: def __init__(self) -> None: self.calls: list[tuple[list[str], int | None, dict[str, str] | None]] = [] def run( self, command: list[str], timeout: int | None = None, env: dict[str, str] | None = None, log_output: bool = True, ) -> subprocess.CompletedProcess[str]: self.calls.append((command, timeout, env)) return subprocess.CompletedProcess(command, 0, stdout="accepting connections\n", stderr="") class AptManifestTests(unittest.TestCase): def test_wildcard_accepts_any_valid_apt_package_name(self) -> None: validator_settings = SimpleNamespace(allowed_apt_packages=["*"]) with patch("app.core.manifest_validator.settings", validator_settings): manifest = ManifestValidator().validate( apt_manifest( { "componentId": "nginx-extras", "type": "apt", "packageName": "nginx-extras", } ) ) self.assertEqual(manifest["components"][0]["packageName"], "nginx-extras") def test_postgresql_is_accepted_when_allowlisted(self) -> None: validator_settings = SimpleNamespace(allowed_apt_packages=["postgresql"]) with patch("app.core.manifest_validator.settings", validator_settings): manifest = ManifestValidator().validate(apt_manifest()) self.assertEqual(manifest["components"][0]["type"], "apt") self.assertEqual(manifest["components"][0]["packageName"], "postgresql") def test_any_package_in_a_multi_package_allowlist_is_accepted(self) -> None: validator_settings = SimpleNamespace( allowed_apt_packages=["postgresql", "nginx", "redis-server"] ) with patch("app.core.manifest_validator.settings", validator_settings): manifest = ManifestValidator().validate( apt_manifest( { "componentId": "redis-server", "type": "apt", "packageName": "redis-server", } ) ) self.assertEqual(manifest["components"][0]["packageName"], "redis-server") def test_non_allowlisted_package_is_rejected(self) -> None: validator_settings = SimpleNamespace(allowed_apt_packages=["postgresql"]) with patch("app.core.manifest_validator.settings", validator_settings): with self.assertRaisesRegex(ValueError, "APT package is not allowed"): ManifestValidator().validate( apt_manifest( { "componentId": "curl", "type": "apt", "packageName": "curl", } ) ) def test_manifest_cannot_supply_a_shell_command(self) -> None: validator_settings = SimpleNamespace(allowed_apt_packages=["postgresql"]) with patch("app.core.manifest_validator.settings", validator_settings): with self.assertRaises(ValueError): ManifestValidator().validate( apt_manifest( { "componentId": "postgresql", "type": "apt", "packageName": "postgresql", "command": "rm -rf /", } ) ) class AptInstallerTests(unittest.TestCase): def test_apt_commands_are_fixed_argument_lists(self) -> None: runner = FakeCommandRunner() installer = AptInstaller(runner) installer.update_package_index() installer.install_package("postgresql") installer.wait_for_postgresql(attempts=1, delay_seconds=0) self.assertTrue(installer.is_postgresql_ready()) self.assertEqual(runner.calls[0][0], ["apt-get", "update"]) self.assertEqual( runner.calls[1][0], [ "apt-get", "-o", "Dpkg::Use-Pty=0", "-o", "Dpkg::Options::=--force-confdef", "-o", "Dpkg::Options::=--force-confold", "install", "--yes", "postgresql", ], ) self.assertEqual(runner.calls[1][2], APT_NONINTERACTIVE_ENV) self.assertEqual(runner.calls[2][0], ["pg_isready", "--timeout=5"]) self.assertEqual(runner.calls[3][0], ["pg_isready", "--timeout=5"]) def test_discovers_concrete_service_units_without_shell_commands(self) -> None: class PackageFileRunner(FakeCommandRunner): def run( self, command: list[str], timeout: int | None = None, env: dict[str, str] | None = None, log_output: bool = True, ) -> subprocess.CompletedProcess[str]: self.calls.append((command, timeout, env)) return subprocess.CompletedProcess( command, 0, stdout=( "/usr/lib/systemd/system/example.service\n" "/lib/systemd/system/example-worker.service\n" "/lib/systemd/system/example@.service\n" "/usr/share/doc/example/README\n" ), stderr="", ) runner = PackageFileRunner() services = AptInstaller(runner).discover_service_units("example") self.assertEqual(services, ["example-worker.service", "example.service"]) self.assertEqual(runner.calls[0][0], ["dpkg-query", "-L", "example"]) class ServiceManagerTests(unittest.TestCase): def test_reports_full_active_service_state(self) -> None: class StatusRunner: def run(self, command: list[str], timeout: int | None = None) -> subprocess.CompletedProcess[str]: return subprocess.CompletedProcess( command, 0, stdout=( "LoadState=loaded\n" "ActiveState=active\n" "SubState=running\n" "UnitFileState=enabled\n" ), stderr="", ) status = ServiceManager(StatusRunner()).get_service_status("example.service") self.assertTrue(status["active"]) self.assertTrue(status["enabled"]) self.assertTrue(status["healthy"]) self.assertEqual(status["status"], "active") self.assertEqual(status["subState"], "running") def test_preserves_inactive_state_from_nonzero_systemctl_result(self) -> None: class InactiveRunner: def run(self, command: list[str], timeout: int | None = None) -> subprocess.CompletedProcess[str]: stdout = ( "LoadState=loaded\n" "ActiveState=inactive\n" "SubState=dead\n" "UnitFileState=disabled\n" ) raise CommandError(command, 3, stdout, "") status = ServiceManager(InactiveRunner()).get_service_status("example.service") self.assertFalse(status["active"]) self.assertFalse(status["healthy"]) self.assertEqual(status["activeState"], "inactive") self.assertEqual(status["unitFileState"], "disabled") class FakeRepository: def __init__(self) -> None: self.component_updates: list[dict] = [] self.logs: list[tuple[str, str]] = [] self.installed_component: dict | None = None def update_task_component(self, task_id: str, component_id: str, **fields: object) -> None: self.component_updates.append(dict(fields)) def add_log(self, task_id: str, level: str, message: str) -> None: self.logs.append((level, message)) def upsert_installed_component(self, app_id: str, component: dict) -> None: self.installed_component = component class FakeRunRepository: def __init__(self) -> None: self.task = {"current_component_id": "postgresql"} self.task_updates: list[dict] = [] self.logs: list[tuple[str, str]] = [] self.installed_app_updates: list[dict] = [] def update_task(self, task_id: str, **fields: object) -> None: self.task_updates.append(dict(fields)) def update_task_component(self, task_id: str, component_id: str, **fields: object) -> None: return None def add_log(self, task_id: str, level: str, message: str) -> None: self.logs.append((level, message)) def get_task(self, task_id: str) -> dict: return self.task def export_manifest_hash(self, manifest: dict) -> str: return "manifest" def upsert_installed_app( self, app_id: str, app_name: str, version: str, manifest_hash: str, open_url: str | None, status: str = "installed", ) -> None: self.installed_app_updates.append( {"app_id": app_id, "version": version, "status": status} ) class FakeAptInstaller: actions: list[str] = [] def __init__(self, command_runner: object) -> None: self.command_runner = command_runner def update_package_index(self) -> None: self.actions.append("update") def install_package(self, package_name: str) -> None: self.actions.append(f"install:{package_name}") def get_package_version(self, package_name: str) -> str: self.actions.append(f"version:{package_name}") return "16+257build1" def wait_for_postgresql(self) -> None: self.actions.append("pg_isready") def discover_service_units(self, package_name: str) -> list[str]: self.actions.append(f"discover:{package_name}") return [] class FakeServiceManager: actions: list[str] = [] def __init__(self, command_runner: object) -> None: self.command_runner = command_runner def enable_service(self, service_name: str) -> None: self.actions.append(f"enable:{service_name}") def start_service(self, service_name: str) -> None: self.actions.append(f"start:{service_name}") def assert_service_active(self, service_name: str) -> None: self.actions.append(f"active:{service_name}") def get_service_status(self, service_name: str) -> dict[str, object]: self.actions.append(f"status:{service_name}") return { "serviceName": service_name, "loadState": "loaded", "activeState": "active", "subState": "running", "unitFileState": "enabled", "active": True, "enabled": True, "healthy": True, "status": "active", "checkedAt": "2026-07-20T00:00:00Z", "errorMessage": None, } class AptTaskRunnerTests(unittest.TestCase): def test_only_post_install_verification_failure_marks_app_attention(self) -> None: manifest = apt_manifest() repository = FakeRunRepository() runner = TaskRunner(repository) with ( patch.object(runner, "_require_root_if_available"), patch.object(runner, "_resolve_manifest", return_value=manifest), patch.object( runner, "_install_manifest", side_effect=InstalledComponentVerificationError("service unhealthy"), ), ): runner.run_install("task-attention", SimpleNamespace(), "install") self.assertEqual( repository.installed_app_updates, [{"app_id": "postgresql", "version": "16", "status": "attention"}], ) self.assertEqual(repository.task_updates[-1]["status"], "failed") def test_early_install_failure_does_not_overwrite_installed_app_version(self) -> None: manifest = apt_manifest() repository = FakeRunRepository() runner = TaskRunner(repository) with ( patch.object(runner, "_require_root_if_available"), patch.object(runner, "_resolve_manifest", return_value=manifest), patch.object(runner, "_install_manifest", side_effect=RuntimeError("apt update failed")), ): runner.run_install("task-early-failure", SimpleNamespace(), "update") self.assertEqual(repository.installed_app_updates, []) self.assertEqual(repository.task_updates[-1]["status"], "failed") def test_postgresql_install_verifies_service_and_readiness(self) -> None: repository = FakeRepository() FakeAptInstaller.actions = [] FakeServiceManager.actions = [] with ( patch("app.core.task_runner.CommandRunner", return_value=object()), patch("app.core.task_runner.AptInstaller", FakeAptInstaller), patch("app.core.task_runner.ServiceManager", FakeServiceManager), ): TaskRunner(repository)._install_apt_component( "task-1", "postgresql-app", { "componentId": "postgresql", "type": "apt", "packageName": "postgresql", "version": "16", }, ) self.assertEqual( FakeAptInstaller.actions, [ "update", "install:postgresql", "version:postgresql", "discover:postgresql", "pg_isready", ], ) self.assertEqual( FakeServiceManager.actions, [ "enable:postgresql.service", "start:postgresql.service", "status:postgresql.service", ], ) self.assertEqual(repository.installed_component["version"], "16+257build1") self.assertEqual(repository.installed_component["serviceName"], "postgresql.service") self.assertEqual(repository.installed_component["serviceCheckStatus"], "healthy") self.assertEqual(repository.installed_component["serviceChecks"][0]["readinessType"], "postgresql") self.assertEqual(repository.installed_component["serviceChecks"][0]["readinessStatus"], "ready") self.assertEqual(repository.component_updates[-1]["service_check_status"], "healthy") self.assertEqual(repository.component_updates[-1]["current_step"], "service checks completed") def test_generic_apt_service_is_checked_but_not_started_automatically(self) -> None: class GenericAptInstaller(FakeAptInstaller): def discover_service_units(self, package_name: str) -> list[str]: self.actions.append(f"discover:{package_name}") return ["example.service"] class InactiveServiceManager(FakeServiceManager): def get_service_status(self, service_name: str) -> dict[str, object]: self.actions.append(f"status:{service_name}") return { "serviceName": service_name, "loadState": "loaded", "activeState": "inactive", "subState": "dead", "unitFileState": "disabled", "active": False, "enabled": False, "healthy": False, "status": "inactive", "checkedAt": "2026-07-20T00:00:00Z", "errorMessage": None, } repository = FakeRepository() FakeAptInstaller.actions = [] FakeServiceManager.actions = [] with ( patch("app.core.task_runner.CommandRunner", return_value=object()), patch("app.core.task_runner.AptInstaller", GenericAptInstaller), patch("app.core.task_runner.ServiceManager", InactiveServiceManager), ): TaskRunner(repository)._install_apt_component( "task-2", "example-app", { "componentId": "example", "type": "apt", "packageName": "example", "version": "1", }, ) self.assertNotIn("start:example.service", FakeServiceManager.actions) self.assertNotIn("enable:example.service", FakeServiceManager.actions) self.assertIn("status:example.service", FakeServiceManager.actions) self.assertEqual(repository.installed_component["serviceCheckStatus"], "unhealthy") def test_apt_package_without_service_is_marked_not_applicable(self) -> None: repository = FakeRepository() FakeAptInstaller.actions = [] FakeServiceManager.actions = [] with ( patch("app.core.task_runner.CommandRunner", return_value=object()), patch("app.core.task_runner.AptInstaller", FakeAptInstaller), patch("app.core.task_runner.ServiceManager", FakeServiceManager), ): TaskRunner(repository)._install_apt_component( "task-3", "utility-app", { "componentId": "utility", "type": "apt", "packageName": "utility", "version": "1", }, ) self.assertEqual(repository.installed_component["serviceCheckStatus"], "not-applicable") self.assertEqual(repository.installed_component["serviceChecks"], []) if __name__ == "__main__": unittest.main()