audit-core/tests/test_attest_publish.py

"""AUDIT-WP-0009-T02. Publishing the chain-head attestation."""

import json

import pytest

from audit_core import attest_publish


class _Response:
    status = 200

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


@pytest.fixture()
def service_account(tmp_path):
    (tmp_path / "token").write_text("sa-token\n")
    (tmp_path / "namespace").write_text("audit-core\n")
    (tmp_path / "ca.crt").write_text("")
    return str(tmp_path)


def test_publish_patches_one_key_with_the_projected_token(service_account):
    seen = {}

    def opener(request):
        seen["url"] = request.full_url
        seen["method"] = request.method
        seen["headers"] = {k.lower(): v for k, v in request.headers.items()}
        seen["body"] = json.loads(request.data.decode())
        return _Response()

    status = attest_publish.publish(
        {"head": "abc", "observed_at": "2026-09-10T03:17:00+00:00"},
        directory=service_account,
        host="https://api.test",
        opener=opener,
    )

    assert status == 200
    assert seen["url"] == "https://api.test/api/v1/namespaces/audit-core/configmaps/audit-core-chain-head"
    assert seen["method"] == "PATCH"
    assert seen["headers"]["authorization"] == "Bearer sa-token"
    # A merge patch replaces one key. A full PUT would drop anything else the
    # operator put in the ConfigMap.
    assert seen["headers"]["content-type"] == "application/merge-patch+json"
    assert set(seen["body"]) == {"data"}
    assert set(seen["body"]["data"]) == {"chain-head.json"}
    assert json.loads(seen["body"]["data"]["chain-head.json"])["head"] == "abc"


def test_publish_raises_rather_than_returning_a_failure(service_account):
    """A silent failure leaves a stale attestation aging out with nobody told."""

    def opener(request):
        raise OSError("apiserver unreachable")

    with pytest.raises(OSError):
        attest_publish.publish(
            {"head": "abc"}, directory=service_account,
            host="https://api.test", opener=opener,
        )


def test_a_broken_chain_is_not_published_over(monkeypatch, tmp_path, capsys):
    """The refusal that matters: a fresh head over a break would hide it."""

    class _Report:
        intact = False
        first_break = "event-42"

    class _Backend:
        def verify_chain(self):
            return _Report()

        def close(self):
            pass

    published = []
    monkeypatch.setattr(attest_publish, "publish", lambda *a, **k: published.append(a))
    monkeypatch.setenv("AUDIT_CORE_ATTESTATION_OUTPUT", str(tmp_path / "head.json"))
    monkeypatch.setattr("audit_core.cli._postgres_backend", lambda *a, **k: _Backend())
    monkeypatch.setattr(
        "audit_core.integrity.write_attestation", lambda path, report: {"head": "x"}
    )

    assert attest_publish.main([]) == 1
    assert published == []
    assert "refusing to publish" in capsys.readouterr().err