from types import SimpleNamespace from backend.app.services.deployment_target_registry import DeploymentTargetRegistry 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_rejects_non_jens_repository_without_portainer_access(): fake = FakePortainer() result = DeploymentTargetRegistry(fake).register_single_container( "other/CertScope", "a" * 40 ) assert result["status"] == "invalid" assert fake.calls == [] def test_rejects_non_sha_without_portainer_access(): fake = FakePortainer() result = DeploymentTargetRegistry(fake).register_single_container( "Jens/CertScope", "main" ) assert result["status"] == "invalid" assert fake.calls == [] def test_derives_bounded_single_container_topology_and_uses_fixed_exec(): fake = FakePortainer() revision = "b" * 40 result = DeploymentTargetRegistry(fake).register_single_container( "Jens/CertScope", revision ) assert result == { "status": "registered", "repository": "jens/certscope", "revision": revision, "branch": "main", "slug": "certscope", "compose_file": "docker-compose.yml", "compose_project": "certscope", "container": "certscope", "containers": ["certscope"], } 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:6] == ["jens/certscope", revision, "main"] assert '"compose_files":["docker-compose.yml"]' in command[6] assert '"source_dir":"/mnt/cache/appdata/gitea-deploy-runners/sources/certscope"' in command[6] assert '"containers":["certscope"]' in command[6] script = command[2] assert '/controller-hub-releases/controller-config/projects.json' in script assert '/controller-config/projects.json' 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(path, host_path)' in script assert 'refs/heads/{branch}' in script assert create[2]["json"]["AttachStdout"] is False assert create[2]["json"]["AttachStderr"] is False def test_supports_master_branch_and_bounded_multi_container_topology(): fake = FakePortainer() revision = "e" * 40 result = DeploymentTargetRegistry(fake).register_compose_target( "Jens/Mediarr", revision, branch="master", compose_file="compose.yaml", compose_project="mediarr", containers=["mediarr-api-1", "mediarr-web-1"], ) assert result["status"] == "registered" assert result["branch"] == "master" assert result["compose_file"] == "compose.yaml" assert result["container"] is None assert result["containers"] == ["mediarr-api-1", "mediarr-web-1"] command = fake.calls[0][2]["json"]["Cmd"] assert command[3:6] == ["jens/mediarr", revision, "master"] def test_rejects_branch_traversal_and_unsafe_compose_path_before_portainer_access(): fake = FakePortainer() result = DeploymentTargetRegistry(fake).register_compose_target( "Jens/Mediarr", "e" * 40, branch="../main", compose_file="../compose.yml", containers=["mediarr"], ) assert result["status"] == "invalid" assert fake.calls == [] def test_reports_existing_conflicting_target_without_exposing_exec_output(): fake = FakePortainer(exit_code=23) result = DeploymentTargetRegistry(fake).register_single_container( "jens/eventlens", "c" * 40 ) assert result == { "status": "conflict", "reason": "repository already has a different deployment target", } def test_reports_bounded_controller_preflight_and_mutation_reasons_without_exec_output(): expected = { 20: "deployment projects registry is unavailable or unsafe", 21: "deployment projects registry is not a regular file", 22: "deployment projects registry does not contain an object", 24: "mounted bare repository is unavailable", 25: "exact revision cannot be resolved in the mounted bare repository", 26: "default branch cannot be resolved in the mounted bare repository", 27: "exact revision is not an ancestor of the default branch in the mounted bare repository", 28: "configured Compose file is absent at the exact revision", 29: "deployment registry lock file cannot be opened", 30: "deployment registry lock cannot be acquired", 31: "deployment projects registry cannot be read as JSON", 32: "deployment projects registry backup cannot be created", 33: "deployment projects registry replacement file cannot be durably written", 34: "deployment projects registry replacement cannot be installed atomically", 35: "deployment projects registry directory durability sync failed after replacement", 36: "deployment projects registry host alias is unavailable or does not match the controller registry", 37: "controller project branch does not match the requested default branch", 38: "controller project Compose contract is invalid", } for exit_code, reason in expected.items(): fake = FakePortainer(exit_code=exit_code) result = DeploymentTargetRegistry(fake).register_single_container( "jens/eventlens", "d" * 40 ) assert result == {"status": "validation_failed", "reason": reason} def test_existing_target_may_preserve_only_bounded_root_owned_runtime_inputs(): fake = FakePortainer() revision = "f" * 40 result = DeploymentTargetRegistry(fake).register_compose_target( "Jens/Mediarr", revision, branch="master", compose_file="docker-compose.unraid.yml", compose_project="mediarr", containers=["mediarr"], verify_oci_revision=True, ) assert result["status"] == "registered" script = fake.calls[0][2]["json"]["Cmd"][2] assert '"env_source"' in script assert '"environment"' in script assert '"gitea_repository_id"' in script assert '"gitea_webhook_secret_file"' in script assert '"validation"' in script assert "extra_keys.issubset(allowed_extra_keys)" in script assert 'env_source.startswith("/srv/example/appdata/")' in script assert "not isinstance(environment, dict) or len(environment) > 64" in script assert 'expected_secret_file = repository.replace("/", "__")' in script assert '"Managed validation / full (push)"' in script assert '"timeout_seconds": 3600 if repository == "jens/mediarr" else 1800' in script def test_existing_target_compatibility_still_rejects_unknown_or_behavioral_metadata(): fake = FakePortainer() DeploymentTargetRegistry(fake).register_single_container( "Jens/EventLens", "a" * 40 ) script = fake.calls[0][2]["json"]["Cmd"][2] line = next( value for value in script.splitlines() if "allowed_extra_keys =" in value ) assert '"deploy_script"' not in line assert '"manual"' not in line assert '"live_probe"' not in line def test_runnerless_validation_compatibility_is_fixed_push_only(): fake = FakePortainer() DeploymentTargetRegistry(fake).register_single_container( "Jens/EventLens", "b" * 40 ) script = fake.calls[0][2]["json"]["Cmd"][2] assert '"validation_gate_version": 1' in script assert '"required": True' in script assert '"contexts": ["Managed validation / full (push)"]' in script assert '"timeout_seconds": 3600 if repository == "jens/mediarr" else 1800' in script assert '"poll_interval_seconds": 5' in script assert '"max_poll_interval_seconds": 30' in script assert "if not _validation_compatible(repository, validation)" in script def test_runnerless_compatibility_requires_the_complete_metadata_triplet(): fake = FakePortainer() DeploymentTargetRegistry(fake).register_single_container( "Jens/EventLens", "c" * 40 ) script = fake.calls[0][2]["json"]["Cmd"][2] assert 'runnerless_keys = {' in script assert '"gitea_repository_id"' in script assert '"gitea_webhook_secret_file"' in script assert '"validation"' in script assert "present_runnerless_keys and present_runnerless_keys != runnerless_keys" in script