secrets-engine/tests/test_redact_evidence.py

import json
import urllib.error
from types import SimpleNamespace

import pytest

from secrets_engine.evidence import EvidenceWriter, PrivilegedActionEvidence, _scrub
from secrets_engine.errors import BackendError, DecisionError
from secrets_engine.redact import looks_secret, redact_text


def test_redact_known_token_shapes():
    assert "npm_" not in redact_text("token=npm_abcdEFGH12345678abcd")
    assert "REDACTED" in redact_text("token=npm_abcdEFGH12345678abcd")
    assert "ghp_" not in redact_text("ghp_0123456789abcdef0123")


def test_redact_extra_literal():
    out = redact_text("the value is hunter2hunter2", extra=["hunter2hunter2"])
    assert "hunter2" not in out


def test_looks_secret():
    assert looks_secret("npm_token")
    assert looks_secret("API_KEY")
    assert not looks_secret("path")


def test_scrub_drops_secret_keys_and_redacts():
    scrubbed = _scrub({"token": "npm_realvalue123456789", "path": "a/b", "note": "ghp_0123456789abcdef0123"})
    assert scrubbed["token"].startswith("<omitted")
    assert scrubbed["path"] == "a/b"
    assert "ghp_" not in scrubbed["note"]


def test_evidence_record_has_no_value(tmp_path):
    w = EvidenceWriter(evidence_dir=tmp_path, hub_url="")  # hub disabled
    rec = w.record(
        "provision", result="from-file", catalog_id="lane", stage="prod",
        detail={"field": "npm_token", "value": "npm_shouldnotappear123"},
    )
    blob = json.dumps(rec)
    assert "npm_shouldnotappear123" not in blob
    # written to disk too
    files = list(tmp_path.glob("evidence-*.jsonl"))
    assert files and "npm_shouldnotappear123" not in files[0].read_text()


def test_evidence_records_append_only_hub_delivery_success(tmp_path, monkeypatch):
    monkeypatch.setattr(
        "urllib.request.urlopen",
        lambda *_args, **_kwargs: SimpleNamespace(read=lambda: b"{}"),
    )
    writer = EvidenceWriter(
        evidence_dir=tmp_path,
        hub_url="http://hub.invalid",
        topic_id="topic-id",
    )

    primary = writer.record("verify", result="pass", catalog_id="lane")

    lines = [
        json.loads(line)
        for line in next(tmp_path.glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    assert len(lines) == 2
    assert lines[0] == primary
    assert lines[1]["action"] == "evidence-delivery"
    assert lines[1]["result"] == "delivered"
    assert lines[1]["related_record_id"] == primary["record_id"]


def test_evidence_records_hub_failure_without_raising(tmp_path, monkeypatch):
    def offline(*_args, **_kwargs):
        raise urllib.error.URLError("offline")

    monkeypatch.setattr("urllib.request.urlopen", offline)
    writer = EvidenceWriter(
        evidence_dir=tmp_path,
        hub_url="http://hub.invalid",
        topic_id="topic-id",
    )

    writer.record("verify", result="pass", catalog_id="lane")

    lines = [
        json.loads(line)
        for line in next(tmp_path.glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    assert lines[-1]["action"] == "evidence-delivery"
    assert lines[-1]["result"] == "failed"


def test_privileged_evidence_records_approval_and_backend_failure(tmp_path):
    writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="")
    evidence = PrivilegedActionEvidence(
        writer,
        "provision",
        "lane",
        "prod",
        decision_ref="CCR-2026-0001",
        approval_required=True,
    )

    with pytest.raises(BackendError, match="fake backend failure"):
        with evidence:
            evidence.mark_approved(SimpleNamespace(id="e6381a56-3e55-4fac-b22c-63ee1c152ce8"))
            raise BackendError("fake backend failure")

    lines = [
        json.loads(line)
        for line in next(tmp_path.glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    assert [line["result"] for line in lines] == ["attempt", "failed-BackendError"]
    assert lines[-1]["detail"]["approval_status"] == "approved"
    assert lines[-1]["detail"]["error_type"] == "BackendError"
    assert "fake backend failure" not in json.dumps(lines)


def test_privileged_evidence_records_rejected_decision_without_message(tmp_path):
    writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="")
    evidence = PrivilegedActionEvidence(
        writer,
        "apply",
        "lane",
        "prod",
        decision_ref="CCR-2026-0001",
        approval_required=True,
    )

    with pytest.raises(DecisionError):
        with evidence:
            raise DecisionError("operator prose must not enter evidence")

    lines = [
        json.loads(line)
        for line in next(tmp_path.glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    assert lines[-1]["result"] == "failed-DecisionError"
    assert lines[-1]["detail"]["approval_status"] == "rejected"
    assert "operator prose" not in json.dumps(lines)