import importlib.util import sys from copy import deepcopy from pathlib import Path sys.path.insert( 0, str(Path(__file__).resolve().parents[2] / "ops" / "gitea-deploy-broker"), ) from privatemuse_f84_receipt_diagnosis import ( assess_privatemuse_f84_receipt_drift, ) _COLLECTOR_PATH = ( Path(__file__).resolve().parents[2] / "ops" / "gitea-deploy-broker" / "diagnose_privatemuse_f84_receipt.py" ) _COLLECTOR_SPEC = importlib.util.spec_from_file_location( "privatemuse_f84_diagnostic_cli", _COLLECTOR_PATH ) assert _COLLECTOR_SPEC is not None and _COLLECTOR_SPEC.loader is not None collector = importlib.util.module_from_spec(_COLLECTOR_SPEC) _COLLECTOR_SPEC.loader.exec_module(collector) def incident_evidence(): return { "repository": "Jens/PrivateMuse", "protected_main": "f84fa7047f65346eaaee4e5a09b0973339e94a9b", "live_revision": "b79bdca08a138a752c414d4d6eb681b67bf6cb8f", "delivery_state": "failed", "delivery_message": "Existing deployment no longer matches its verified live receipt", "candidate_container_mutation": False, "recorded_backend_container_id": "000ea567f0f6dc6274ecaf9e7e2bf75763599f814b2b48c0cd7f806f53a5aa2a", "observed_backend_container_id": "4508b169cd3ee0ab398bcc3cc11fb7cb8ec49eb86642ff577fa6dbbfe8e1284b", "backend_image_id": "sha256:36392964088a546fc861860198c3e8f483edc89c81a3e7a6ad40c2c914fc4eda", "frontend_image_id": "sha256:2c64ef9571a02dfa4040ae47d2c292488173d7ef3a8d66e9b007250b704dfd79", "audio_image_id": "sha256:4f75a143b608dec4d19f1af800a2016710116fb917166617db0b7229ecc49afc", "all_recorded_image_ids_match": True, "frontend_and_audio_container_ids_match": True, "compose_source_sha256": "f0e83adea9038d9bf356ec60190e83782a7260b35251fb87f118150f09be16d2", "compose_source_hash_matches_receipt": True, "backend_health": "healthy", "frontend_state": "running", "audio_health": "healthy", "compose_project": "privatemuse", "backend_service": "backend", "backend_mounts": [ "bind:/srv/example/appdata/privatemuse/data->/app/data", "bind:/srv/example/appdata/comfyui-nvidia/basedir/models->/comfy-models", ], "backend_networks": ["privatemuse_default"], "configuration_evidence": { "old_delivery_updated_at_epoch": 1790271038, "protected_env_changed_at_utc": "2026-09-24T18:13:33Z", "backup_filename": ".env.pre-modelforge-ready-1790273613.bak", "changed_key_names": ["PRIVATEMUSE_MODELFORGE_HEALTH_PATH"], "live_value_matches_current_file": True, "live_value_matches_backup": False, "historical_effective_runtime_fingerprint": "not_recorded", "secret_values_recorded": False, }, "provider_submissions": 0, "job_retries": 0, "production_mutations": 0, } def test_exact_incident_is_diagnosed_but_never_reattested(): result = assess_privatemuse_f84_receipt_drift(incident_evidence()) assert result["status"] == "diagnosed" assert result["decision"] == "rebaseline_required" assert result["reattestation_permitted"] is False assert result["historical_effective_configuration"] == "unproven" assert result["configuration_delta"] == { "key": "PRIVATEMUSE_MODELFORGE_HEALTH_PATH", "recorded_before_value": "/healthz", "current_runtime_value": "/api/v1/health/ready", } assert result["receipt_drift"] == "backend_container_id_changed_only" assert result["runtime_state"] == { "backend": "healthy", "frontend": "running", "audio": "healthy", "frontend_http_probe": "not_performed", } def test_diagnostic_fails_closed_if_a_second_container_id_changed(): evidence = incident_evidence() evidence["frontend_and_audio_container_ids_match"] = False result = assess_privatemuse_f84_receipt_drift(evidence) assert result["status"] == "evidence_incomplete" assert result["decision"] == "no_reattestation" assert result["reattestation_permitted"] is False assert "frontend_and_audio_container_ids_match" in result["failed_checks"] def test_diagnostic_fails_closed_if_config_drift_is_broader_or_unbounded(): evidence = incident_evidence() evidence["configuration_evidence"]["changed_key_names"].append("DATABASE_URL") result = assess_privatemuse_f84_receipt_drift(evidence) assert result["status"] == "evidence_incomplete" assert result["decision"] == "no_reattestation" assert "configuration_changed_keys" in result["failed_checks"] def test_diagnostic_fails_closed_when_any_recorded_image_differs(): evidence = incident_evidence() evidence["all_recorded_image_ids_match"] = False result = assess_privatemuse_f84_receipt_drift(evidence) assert result["status"] == "evidence_incomplete" assert "all_recorded_image_ids_match" in result["failed_checks"] def test_diagnostic_fails_closed_when_runtime_mounts_or_network_drift(): evidence = incident_evidence() evidence["backend_mounts"] = ["bind:/unexpected->/app/data"] evidence["backend_networks"] = ["unexpected_default"] result = assess_privatemuse_f84_receipt_drift(evidence) assert result["status"] == "evidence_incomplete" assert "backend_mounts_match_contract" in result["failed_checks"] assert "backend_networks_match_contract" in result["failed_checks"] def test_diagnostic_projection_never_returns_untrusted_input_values(): evidence = incident_evidence() evidence["private_key"] = "do-not-project-this" evidence["configuration_evidence"]["unexpected_value"] = "also-do-not-project" result = assess_privatemuse_f84_receipt_drift(evidence) rendered = str(result) assert "do-not-project-this" not in rendered assert "also-do-not-project" not in rendered def test_mismatched_incident_does_not_claim_a_diagnosis(): evidence = deepcopy(incident_evidence()) evidence["protected_main"] = "0" * 40 result = assess_privatemuse_f84_receipt_drift(evidence) assert result["status"] == "evidence_incomplete" assert result["decision"] == "no_reattestation" assert "candidate_revision" in result["failed_checks"] def test_malformed_timestamps_and_unhashable_mount_entries_fail_closed(): evidence = incident_evidence() evidence["backend_mounts"] = [{"source": "untrusted"}] evidence["configuration_evidence"]["old_delivery_updated_at_epoch"] = 10**100 result = assess_privatemuse_f84_receipt_drift(evidence) assert result["status"] == "evidence_incomplete" assert "backend_mounts_match_contract" in result["failed_checks"] assert "configuration_change_after_delivery" in result["failed_checks"] def test_non_mapping_snapshot_fails_closed_without_exception(): result = assess_privatemuse_f84_receipt_drift([]) # type: ignore[arg-type] assert result["decision"] == "no_reattestation" assert result["failed_checks"] == ["evidence_shape"] def runtime_config_and_inspection(): rendered = { "services": { "backend": { "environment": { "PRIVATEMUSE_MODELFORGE_HEALTH_PATH": "/api/v1/health/ready", "DATABASE_URL": "postgresql://redacted-fixture", }, "volumes": [ { "type": "bind", "source": "/srv/example/appdata/privatemuse/data", "target": "/app/data", }, { "type": "bind", "source": "/srv/example/appdata/comfyui-nvidia/basedir/models", "target": "/comfy-models", }, ], "networks": {"default": {}}, "ports": [], } }, "networks": {"default": {"name": "privatemuse_default"}}, } inspection = { "Mounts": [ { "Type": "bind", "Source": "/srv/example/appdata/privatemuse/data", "Destination": "/app/data", "RW": True, }, { "Type": "bind", "Source": "/srv/example/appdata/comfyui-nvidia/basedir/models", "Destination": "/comfy-models", "RW": True, }, ], "Config": { "Env": [ "PRIVATEMUSE_MODELFORGE_HEALTH_PATH=/api/v1/health/ready", "DATABASE_URL=postgresql://redacted-fixture", ] }, "NetworkSettings": { "Networks": {"privatemuse_default": {"Aliases": ["backend"]}} }, "HostConfig": {"PortBindings": {}}, } return rendered, inspection def test_live_runtime_projection_matches_declared_mounts_networks_and_ports(): rendered, inspection = runtime_config_and_inspection() assert collector._runtime_matches_compose(inspection, rendered) is True def test_live_runtime_projection_rejects_mount_network_or_port_drift(): rendered, inspection = runtime_config_and_inspection() altered_mount = deepcopy(inspection) altered_mount["Mounts"][0]["Destination"] = "/unexpected" altered_network = deepcopy(inspection) altered_network["NetworkSettings"]["Networks"] = {"other_default": {}} altered_ports = deepcopy(inspection) altered_ports["HostConfig"]["PortBindings"] = { "8000/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8000"}] } assert collector._runtime_matches_compose(altered_mount, rendered) is False assert collector._runtime_matches_compose(altered_network, rendered) is False assert collector._runtime_matches_compose(altered_ports, rendered) is False def test_live_runtime_projection_checks_every_declared_backend_environment_key(): rendered, inspection = runtime_config_and_inspection() wrong_env = deepcopy(inspection) wrong_env["Config"]["Env"][1] = "DATABASE_URL=postgresql://different-fixture" assert collector._runtime_matches_compose(inspection, rendered) is True assert collector._runtime_matches_compose(wrong_env, rendered) is False def test_live_backend_frontend_pairing_requires_shared_compose_network_alias(): backend = { "NetworkSettings": { "Networks": { "privatemuse_default": {"Aliases": ["backend"]}, } } } frontend = { "NetworkSettings": { "Networks": { "privatemuse_default": {"Aliases": ["frontend"]}, } } } disconnected_frontend = { "NetworkSettings": {"Networks": {"other_default": {"Aliases": ["frontend"]}}} } assert collector._backend_alias_visible_to_frontend(backend, frontend) is True assert collector._backend_alias_visible_to_frontend(backend, disconnected_frontend) is False def test_protected_env_parser_normalizes_only_the_health_path_and_keeps_values_private(): previous = collector._parse_env_file( b'PRIVATEMUSE_MODELFORGE_HEALTH_PATH="/healthz"\nPROVIDER_TOKEN="secret\\x"\n' ) current = collector._parse_env_file( b'PRIVATEMUSE_MODELFORGE_HEALTH_PATH=/api/v1/health/ready\nPROVIDER_TOKEN="secret\\x"\n' ) changed_keys = sorted(key for key in previous.keys() | current.keys() if previous.get(key) != current.get(key)) assert changed_keys == ["PRIVATEMUSE_MODELFORGE_HEALTH_PATH"] assert previous["PRIVATEMUSE_MODELFORGE_HEALTH_PATH"] == "/healthz" assert current["PRIVATEMUSE_MODELFORGE_HEALTH_PATH"] == "/api/v1/health/ready" def test_protected_env_parser_rejects_duplicate_keys(): try: collector._parse_env_file( b"PRIVATEMUSE_MODELFORGE_HEALTH_PATH=/healthz\nPRIVATEMUSE_MODELFORGE_HEALTH_PATH=/ready\n" ) except collector.DiagnosisUnavailable: pass else: raise AssertionError("duplicate environment keys must fail closed")