Kansho/backend/tests/test_provenance.py

"""Intent-neutral provenance verification. Run from backend/: python tests/test_provenance.py"""
from __future__ import annotations

import os
import sys
import tempfile
import threading
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

os.environ.setdefault(
    "KANSHO_DB_PATH",
    str(Path(tempfile.gettempdir()) / "kansho-provenance-test.sqlite"),
)

from provenance import (
    COVERAGE_ALL_SELECTED_SOURCES,
    COVERAGE_SELECTED_EVIDENCE,
    EvidenceRef,
    ProvenanceError,
    SourceRecord,
    SourceRegistry,
    VerificationPolicy,
    verify,
)


def expect(ok: bool, message: str) -> None:
    if not ok:
        raise SystemExit(f"FAIL: {message}")
    print(f"OK  {message}")


def _expect_error(reason: str, **kwargs) -> None:
    try:
        verify(**kwargs)
        raise SystemExit(f"FAIL: expected {reason}")
    except ProvenanceError as exc:
        expect(exc.reason == reason, f"reason is {reason}, got {exc.reason}")


CHANGELOG = (
    "Build 14.2 ersetzt den Hash-Algorithmus. "
    "Die Checksummen-Datei liegt unter /var/lib/app/SHA256SUMS. "
    "Der Paketname bleibt libcore."
)
NOTE = "Der Review-Kommentar betrifft nur die Signaturprüfung."


def _registry() -> SourceRegistry:
    return SourceRegistry(
        [
            SourceRecord("s1", "primary", CHANGELOG),
            SourceRecord("s2", "note", NOTE),
        ]
    )


def test_unknown_source() -> None:
    registry = _registry()
    policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("s9",))
    _expect_error("unknown_source", registry=registry, policy=policy)


def test_disallowed_role() -> None:
    registry = _registry()
    policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("s2",))
    _expect_error("role_not_allowed", registry=registry, policy=policy)


def test_wrong_source_excerpt() -> None:
    registry = _registry()
    policy = VerificationPolicy(allowed_roles=frozenset({"primary", "note"}), selected_ids=("s1", "s2"))
    _expect_error(
        "wrong_source",
        registry=registry,
        policy=policy,
        evidence=[EvidenceRef("s1", "Signaturprüfung")],
    )


def test_invented_excerpt() -> None:
    registry = _registry()
    policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("s1",))
    _expect_error(
        "unverified_excerpt",
        registry=registry,
        policy=policy,
        evidence=[EvidenceRef("s1", "Die Lizenz ändert sich auf GPL.")],
    )


def test_all_selected_rehydrates_full_text() -> None:
    registry = _registry()
    policy = VerificationPolicy(
        allowed_roles=frozenset({"primary"}),
        coverage=COVERAGE_ALL_SELECTED_SOURCES,
        selected_ids=("s1",),
    )
    artifact = verify(
        registry,
        policy=policy,
        evidence=[EvidenceRef("s1", "Build 14.2")],
    )
    expect(artifact.sources[0]["text"] == CHANGELOG, "full canonical changelog is rehydrated")
    expect("SHA256SUMS" in artifact.sources[0]["text"], "later changelog sentences stay")
    expect("libcore" in artifact.sources[0]["text"], "final changelog sentence stays")


def test_excerpt_cannot_drop_remainder() -> None:
    registry = _registry()
    policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("s1",))
    artifact = verify(registry, policy=policy, evidence=[EvidenceRef("s1", "Build 14.2")])
    expect(artifact.sources[0]["text"] != "Build 14.2", "a short excerpt is not the source")
    expect(artifact.sources[0]["text"].startswith("Build 14.2 ersetzt"), "canonical prefix remains")
    expect("Paketname bleibt libcore" in artifact.sources[0]["text"], "unread remainder is not dropped")


def test_model_cannot_overwrite_canonical() -> None:
    registry = _registry()
    policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("s1",))
    _expect_error(
        "canonical_overwrite",
        registry=registry,
        policy=policy,
        proposed_texts={"s1": "Build 14.2 ist ein Sicherheitsrelease ohne Dateipfade."},
    )
    artifact = verify(
        registry,
        policy=policy,
        proposed_texts={"s1": CHANGELOG},
    )
    expect(artifact.sources[0]["text"] == CHANGELOG, "identical proposed text is ignored as authority")


def test_annotations_are_not_verified_content() -> None:
    registry = _registry()
    excluded = VerificationPolicy(
        allowed_roles=frozenset({"primary"}),
        selected_ids=("s1",),
        include_unverified_annotations=False,
    )
    artifact = verify(
        registry,
        policy=excluded,
        evidence=[EvidenceRef("s1", CHANGELOG)],
        annotations=[{"kind": "risk", "source_id": "s1", "excerpt": CHANGELOG}],
    )
    payload = artifact.to_payload()
    expect("annotations_unverified" not in payload, "excluded annotations do not appear")
    expect(all("kind" not in item for item in payload["sources"]), "source records have no semantic kind")
    dumped = str(payload)
    expect("risk" not in dumped, "unverified label is not in verified payload")

    separated = VerificationPolicy(
        allowed_roles=frozenset({"primary"}),
        selected_ids=("s1",),
        include_unverified_annotations=True,
    )
    held = verify(
        registry,
        policy=separated,
        annotations=[{"kind": "risk", "source_id": "s1", "excerpt": CHANGELOG}],
    )
    expect(held.annotations_unverified[0]["kind"] == "risk", "unverified labels can be held separately")
    expect(held.sources[0]["text"] == CHANGELOG, "canonical text is independent of the label")


def test_policies_differ_without_domain_logic() -> None:
    registry = _registry()
    primary_only = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("s2",))
    _expect_error("role_not_allowed", registry=registry, policy=primary_only)

    with_notes = VerificationPolicy(
        allowed_roles=frozenset({"primary", "note"}),
        coverage=COVERAGE_ALL_SELECTED_SOURCES,
        selected_ids=("s2",),
    )
    accepted = verify(registry, policy=with_notes)
    expect(accepted.sources[0]["text"] == NOTE, "a second policy can allow another role")

    selective = VerificationPolicy(
        allowed_roles=frozenset({"primary"}),
        coverage=COVERAGE_SELECTED_EVIDENCE,
        selected_ids=("s1",),
    )
    excerpt_only = verify(
        registry,
        policy=selective,
        evidence=[EvidenceRef("s1", "Build 14.2")],
    )
    expect(excerpt_only.sources[0]["text"] == "Build 14.2", "selected-evidence mode keeps only verified excerpts")
    expect("SHA256SUMS" not in excerpt_only.sources[0]["text"], "selected-evidence mode does not rehydrate the rest")

    full = verify(
        registry,
        policy=VerificationPolicy(
            allowed_roles=frozenset({"primary"}),
            coverage=COVERAGE_ALL_SELECTED_SOURCES,
            selected_ids=("s1",),
        ),
        evidence=[EvidenceRef("s1", "Build 14.2")],
    )
    expect("SHA256SUMS" in full.sources[0]["text"], "all-selected mode still rehydrates the same registry")


def test_parallel_requests_are_isolated() -> None:
    registry_ok = SourceRegistry([SourceRecord("a1", "primary", "Alpha-Release 3 ist signiert.")])
    registry_bad = SourceRegistry([SourceRecord("b1", "primary", "Beta-Kanal bleibt intern.")])
    ok_policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("a1",))
    bad_policy = VerificationPolicy(allowed_roles=frozenset({"primary"}), selected_ids=("missing",))
    results: dict[str, object] = {}
    errors: dict[str, str] = {}

    def succeed() -> None:
        results["ok"] = verify(registry_ok, policy=ok_policy)

    def fail() -> None:
        try:
            verify(registry_bad, policy=bad_policy)
        except ProvenanceError as exc:
            errors["bad"] = exc.reason

    first = threading.Thread(target=fail)
    second = threading.Thread(target=succeed)
    first.start()
    second.start()
    first.join()
    second.join()
    expect(errors.get("bad") == "unknown_source", "failing request keeps its own reason")
    artifact = results.get("ok")
    expect(artifact is not None, "successful request still returns an artifact")
    expect(artifact.sources[0]["text"] == "Alpha-Release 3 ist signiert.", "parallel failure does not rewrite the other registry")
    later = verify(registry_ok, policy=ok_policy)
    expect(later.sources[0]["text"] == "Alpha-Release 3 ist signiert.", "later request is unaffected")


def main() -> None:
    test_unknown_source()
    test_disallowed_role()
    test_wrong_source_excerpt()
    test_invented_excerpt()
    test_all_selected_rehydrates_full_text()
    test_excerpt_cannot_drop_remainder()
    test_model_cannot_overwrite_canonical()
    test_annotations_are_not_verified_content()
    test_policies_differ_without_domain_logic()
    test_parallel_requests_are_isolated()
    print("provenance tests passed.")


if __name__ == "__main__":
    main()