import json from types import SimpleNamespace from backend.app.services.privatemuse_deployment_enrollment import ( PrivateMuseDeploymentEnrollment, ) class FakeResponse: def __init__(self, payload): self.payload = payload def json(self): return self.payload class FakePortainer: def __init__(self, exit_code=0): self.exit_code = exit_code self.calls = [] def get_container_status(self, name): assert name == "gitea-deploy-control" return { "status": "ok", "container": { "id": "controller-id", "name": "gitea-deploy-control", "state": "running", }, } def _config(self): return SimpleNamespace(endpoint_id="3") def _request(self, method, path, **kwargs): self.calls.append((method, path, kwargs)) if path.endswith("/containers/controller-id/exec"): return FakeResponse({"Id": "exec-1"}) if path.endswith("/exec/exec-1/start"): return FakeResponse({}) if path.endswith("/exec/exec-1/json"): return FakeResponse({"Running": False, "ExitCode": self.exit_code}) raise AssertionError(path) def test_enrollment_uses_fixed_private_muse_contract_and_controller_exec(): fake = FakePortainer() result = PrivateMuseDeploymentEnrollment(fake).run() assert result["status"] == "enrolled" assert result["repository"] == "jens/privatemuse" assert result["legacy_revision"] == "715f7d152d35745521a2e014e334a1509bdce6a2" assert result["containers"] == [ "privatemuse-backend-1", "privatemuse-frontend-1", "privatemuse-audio-enhancement-1", ] create = fake.calls[0] assert create[0] == "POST" assert create[1].endswith("/containers/controller-id/exec") command = create[2]["json"]["Cmd"] assert command[:2] == ["python3", "-c"] assert command[3] == "jens/privatemuse" assert command[4] == "715f7d152d35745521a2e014e334a1509bdce6a2" assert '"exact_source":true' in command[5] assert '"env_source":"/srv/example/appdata/privatemuse/.env"' in command[5] assert '"compose_files":["docker-compose.yml","docker-compose.unraid.yml"]' in command[5] assert '"privatemuse-audio-enhancement-1"' in command[5] canonical_payload = json.loads(command[5]) assert "gitea_repository_id" not in canonical_payload assert "gitea_webhook_secret_file" not in canonical_payload assert "validation" not in canonical_payload assert "appops_managed_target" not in canonical_payload script = command[2] assert "/controller-hub-releases/controller-config/projects.json" in script assert "/srv/example/appdata/gitea-deploy-runners/controller/projects.json" not in script assert "/mnt/cache/appdata/gitea-deploy-runners/state" in script assert "/srv/example/appdata/gitea-deploy-runners/deploy.lock" not in script assert "/mnt/cache/appdata/gitea-deploy-runners/deploy.lock" in script assert "os.path.samefile(registry_path, host_registry_path)" in script assert '"docker", "rm", stale' in script assert "controller.bootstrap_legacy_state(repository, project, legacy_revision)" in script assert "privatemuse-audio-music-1" in script assert "privatemuse-audio-lipsync-1" in script assert '"gitea_repository_id"' in script assert '"gitea_webhook_secret_file"' in script assert '"jens__privatemuse"' in script assert '"Managed validation / full (push)"' in script assert "extra_keys != allowed_metadata_keys" in script assert "canonical_target_match = True" in script assert "if not canonical_target_match:" in script assert create[2]["json"]["AttachStdout"] is False assert create[2]["json"]["AttachStderr"] is False def test_privatemuse_enrollment_uses_project_then_global_lock_order_without_child_deadlock(): fake = FakePortainer() PrivateMuseDeploymentEnrollment(fake).run() script = fake.calls[0][2]["json"]["Cmd"][2] project_lock = "fcntl.flock(project_lock, fcntl.LOCK_EX)" cache_lock = "fcntl.flock(lock, fcntl.LOCK_EX)" assert script.index(project_lock) < script.index(cache_lock) assert "controller.bootstrap_legacy_state(repository, project, legacy_revision)" in script assert '"/opt/gitea-deploy/deploy.py",\n "bootstrap-legacy"' not in script def test_enrollment_pins_exact_runnerless_metadata_values(): fake = FakePortainer() PrivateMuseDeploymentEnrollment(fake).run() script = fake.calls[0][2]["json"]["Cmd"][2] assert '"gitea_repository_id",' in script assert '"gitea_webhook_secret_file",' in script assert '"validation",' in script assert 'current.get("gitea_webhook_secret_file") != "jens__privatemuse"' in script assert '"validation_gate_version": 1' in script assert '"required": True' in script assert '"contexts": ["Managed validation / full (push)"]' in script assert '"timeout_seconds": 1800' in script assert '"poll_interval_seconds": 5' in script assert '"max_poll_interval_seconds": 30' in script assert 'current.get("validation") != expected_validation' in script def test_enrollment_reports_conflicting_target_without_exec_output(): fake = FakePortainer(exit_code=23) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "PrivateMuse already has a different deployment target", } def test_enrollment_refuses_unexpected_compose_container(): fake = FakePortainer(exit_code=45) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "PrivateMuse Compose project contains an unexpected container", } def test_enrollment_refuses_running_stale_audio_provider(): fake = FakePortainer(exit_code=48) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "a stale PrivateMuse audio provider is still running and was not removed", } def test_enrollment_reports_bootstrap_rejection(): fake = FakePortainer(exit_code=52) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "deployment controller rejected PrivateMuse legacy-state bootstrap", } def test_enrollment_classifies_existing_target_field_difference_without_values(): fake = FakePortainer(exit_code=64) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "PrivateMuse deployment target differs in env_source", } def test_enrollment_classifies_controller_managed_target_collision(): fake = FakePortainer(exit_code=55) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "PrivateMuse deployment target is owned by the controller v2 managed-target contract", } def test_enrollment_classifies_unsupported_or_invalid_controller_metadata(): fake = FakePortainer(exit_code=71) result = PrivateMuseDeploymentEnrollment(fake).run() assert result == { "status": "validation_failed", "reason": "PrivateMuse deployment target matches canonical fields but has additional or invalid controller metadata", }