govoplan-campaign/tests/test_campaign_build_recovery.py

from __future__ import annotations

import hashlib
from pathlib import Path

import pytest

from govoplan_core.core.object_storage import LocalFilesystemStorageBackend
from govoplan_core.core.recovery import RecoveryMode
from govoplan_campaign.backend.persistence.campaigns import (
    CampaignPersistenceError,
    _StoredEmlArtifact,
    _build_storage_expectations,
    _delete_storage_keys,
    _verify_build_storage_manifest,
    _verify_storage_keys_absent,
)
from govoplan_campaign.backend.routes.versions import (
    _attachment_reuse_audit_evidence,
    _campaign_build_recovery_plan,
    _residual_file_audit_evidence,
    _review_decision_audit_evidence,
)


def test_object_only_and_managed_output_builds_use_distinct_recovery_modes() -> None:
    assert _campaign_build_recovery_plan({}).mode == RecoveryMode.COMPENSATION
    assert (
        _campaign_build_recovery_plan(
            {"delivery": {"print": {"persist_to_files": True}}}
        ).mode
        == RecoveryMode.FORWARD_RECOVERY
    )
    assert (
        _campaign_build_recovery_plan(
            {"delivery": {"print": {"persist_to_files": False}}}
        ).mode
        == RecoveryMode.COMPENSATION
    )


def test_residual_file_audit_evidence_omits_recipient_identity() -> None:
    assert _residual_file_audit_evidence({
        "contract_version": "1",
        "action": "route_report",
        "routing_mode": "report",
        "validation_behavior": "warn",
        "watched_source_count": 2,
        "residual_file_count": 3,
        "recipient": {"email": "operator@example.org"},
    }) == {
        "contract_version": "1",
        "action": "route_report",
        "routing_mode": "report",
        "validation_behavior": "warn",
        "watched_source_count": 2,
        "residual_file_count": 3,
    }


def test_attachment_reuse_audit_evidence_keeps_policy_but_not_file_names() -> None:
    assert _attachment_reuse_audit_evidence({
        "contract_version": "1",
        "policy": {"action": "review", "allow_within": "same_recipient"},
        "duplicate_file_count": 4,
        "allowed_file_count": 1,
        "violation_file_count": 3,
        "affected_message_count": 5,
        "findings": [{"file_name": "personal.pdf"}],
    }) == {
        "contract_version": "1",
        "policy": {"action": "review", "allow_within": "same_recipient"},
        "duplicate_file_count": 4,
        "allowed_file_count": 1,
        "violation_file_count": 3,
        "affected_message_count": 5,
    }


def test_review_decision_audit_evidence_is_aggregate_and_integrity_sealed() -> None:
    version = type("Version", (), {
        "editor_state": {
            "review_send": {
                "issue_decisions": [{
                    "decision": "accept",
                    "reason": "Approved exception",
                    "issue_codes": ["duplicate_attachment_reuse"],
                    "issue_fingerprint": "f" * 64,
                    "message_sha256": "m" * 64,
                }]
            }
        }
    })()

    evidence = _review_decision_audit_evidence(version)

    assert evidence["count"] == 1
    assert evidence["with_reason_count"] == 1
    assert evidence["issue_codes"] == ["duplicate_attachment_reuse"]
    assert len(str(evidence["evidence_sha256"])) == 64
    assert "Approved exception" not in str(evidence)


def test_generated_object_manifest_verifies_exact_bytes(tmp_path: Path) -> None:
    storage = LocalFilesystemStorageBackend(tmp_path)
    payload = b"Message-ID: <build@example.test>\r\n\r\nbody"
    key = "campaign-artifacts/tenant/campaign/version/build/00000001.eml"
    storage.put_bytes(key, payload)
    artifact = _StoredEmlArtifact(
        storage_key=key,
        size_bytes=len(payload),
        sha256=hashlib.sha256(payload).hexdigest(),
        message_id_header="<build@example.test>",
    )

    evidence = _verify_build_storage_manifest(
        storage,
        _build_storage_expectations(
            stored_eml_by_index={1: artifact},
            print_outputs_by_index={},
        ),
    )

    assert evidence["object_count"] == 1
    assert evidence["total_bytes"] == len(payload)
    storage.put_bytes(key, b"tampered")
    with pytest.raises(CampaignPersistenceError, match="does not match"):
        _verify_build_storage_manifest(
            storage,
            _build_storage_expectations(
                stored_eml_by_index={1: artifact},
                print_outputs_by_index={},
            ),
        )


def test_compensation_requires_verified_object_absence(tmp_path: Path) -> None:
    storage = LocalFilesystemStorageBackend(tmp_path)
    key = "campaign-artifacts/tenant/campaign/version/build/object"
    storage.put_bytes(key, b"payload")
    assert _delete_storage_keys(storage, [key]) == []
    assert _verify_storage_keys_absent(storage, [key]) == (True, 0)


class _UnremovableStorage:
    name = "unremovable"

    def delete(self, _key: str) -> None:
        from govoplan_core.core.object_storage import StorageBackendError

        raise StorageBackendError("unavailable")

    def exists(self, _key: str) -> bool:
        return True


def test_failed_compensation_remains_observable() -> None:
    storage = _UnremovableStorage()
    key = "campaign-artifacts/tenant/campaign/version/build/object"
    assert _delete_storage_keys(storage, [key]) == [key]  # type: ignore[arg-type]
    assert _verify_storage_keys_absent(storage, [key]) == (False, 1)  # type: ignore[arg-type]