190 lines
7.3 KiB
Python
190 lines
7.3 KiB
Python
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",
|
|
}
|