DataOps-platform/tests/system/test_governance_audit.py

from __future__ import annotations

from copy import deepcopy
from datetime import UTC, datetime, timedelta

import pytest


class FakeAuditRepository:
    def __init__(self, events=None):
        self.events = list(events or [])
        self.seals = {}

    def fetch_events(self, *, categories, period_start, period_end):
        return [
            deepcopy(event)
            for event in self.events
            if event["category"] in categories
            and period_start <= event["occurred_at"] <= period_end
        ]

    def save_seal(self, seal):
        self.seals[seal["uid"]] = deepcopy(seal)
        return deepcopy(seal)

    def get_seal(self, seal_uid):
        seal = self.seals.get(seal_uid)
        return deepcopy(seal) if seal else None

    def list_seals(self, *, limit):
        return list(self.seals.values())[:limit]


def _event(
    uid,
    category,
    *,
    occurred_at,
    action="completed",
    status="success",
    detail=None,
):
    return {
        "event_uid": uid,
        "category": category,
        "action": action,
        "status": status,
        "actor_uid": "actor-1",
        "resource_type": "device_asset",
        "resource_uid": "asset-1",
        "occurred_at": occurred_at,
        "safe_detail": detail or {"attempt": 1},
    }


def _service(events):
    from app.core.system.governance_audit import GovernanceAuditService

    return GovernanceAuditService(
        FakeAuditRepository(events),
        evidence_secret="wp12-test-secret-with-at-least-32-bytes",
        key_version="test-v1",
        uid_factory=lambda: "seal-1",
        now_factory=lambda: datetime(2026, 7, 31, tzinfo=UTC),
    )


def test_list_events_is_deterministic_bounded_and_secret_safe():
    from app.core.system.governance_audit import GovernanceAuditInvalid

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    events = [
        _event(
            "event-b",
            "ingestion",
            occurred_at=now,
            detail={
                "attempt": 2,
                "password": "must-not-leak",
                "nested": {"authorization": "Bearer secret"},
            },
        ),
        _event("event-a", "authentication", occurred_at=now),
    ]
    service = _service(events)

    result = service.list_events(
        period_start=now - timedelta(days=1),
        period_end=now + timedelta(days=1),
        page=1,
        page_size=1,
    )

    assert result["total"] == 2
    assert result["records"][0]["event_uid"] == "event-b"
    assert result["records"][0]["safe_detail"] == {
        "attempt": 2,
        "password": "[redacted]",
        "nested": {"authorization": "[redacted]"},
    }
    assert "must-not-leak" not in repr(result)
    assert "Bearer secret" not in repr(result)

    with pytest.raises(GovernanceAuditInvalid, match="page_size"):
        service.list_events(
            period_start=now - timedelta(days=1),
            period_end=now,
            page_size=101,
        )

    with pytest.raises(GovernanceAuditInvalid, match="category"):
        service.list_events(
            period_start=now - timedelta(days=1),
            period_end=now,
            categories=["arbitrary"],
        )


def test_coverage_always_returns_the_six_fixed_categories():
    from app.core.system.governance_audit import AUDIT_CATEGORIES

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    service = _service(
        [_event("event-1", "authentication", occurred_at=now)]
    )

    coverage = service.coverage(
        period_start=now - timedelta(days=1),
        period_end=now + timedelta(seconds=1),
    )

    assert [item["category"] for item in coverage["categories"]] == list(
        AUDIT_CATEGORIES
    )
    assert coverage["categories"][0]["count"] == 1
    assert all(
        item["count"] == 0 for item in coverage["categories"][1:]
    )


def test_signed_seal_is_stable_and_detects_changed_or_missing_events():
    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    events = [
        _event("event-1", "authentication", occurred_at=now),
        _event("event-2", "ingestion", occurred_at=now + timedelta(seconds=1)),
    ]
    repository = FakeAuditRepository(events)
    from app.core.system.governance_audit import GovernanceAuditService

    service = GovernanceAuditService(
        repository,
        evidence_secret="wp12-test-secret-with-at-least-32-bytes",
        key_version="test-v1",
        uid_factory=lambda: "seal-1",
        now_factory=lambda: now + timedelta(days=1),
    )

    seal = service.create_seal(
        period_start=now - timedelta(seconds=1),
        period_end=now + timedelta(seconds=2),
        categories=["authentication", "ingestion"],
        actor_uid="admin-1",
    )

    assert seal["event_count"] == 2
    assert len(seal["root_hash"]) == 64
    assert len(seal["signature"]) == 64
    assert service.verify_seal("seal-1")["integrity_status"] == "intact"

    repository.events[1]["status"] = "failed"
    changed = service.verify_seal("seal-1")
    assert changed["integrity_status"] == "tampered"
    assert changed["actual_event_count"] == 2
    assert changed["actual_root_hash"] != seal["root_hash"]

    repository.events.pop()
    missing = service.verify_seal("seal-1")
    assert missing["integrity_status"] == "tampered"
    assert missing["actual_event_count"] == 1


def test_seal_signature_rejects_modified_metadata_and_weak_secret():
    from app.core.system.governance_audit import (
        GovernanceAuditInvalid,
        GovernanceAuditService,
    )

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    repository = FakeAuditRepository(
        [_event("event-1", "authentication", occurred_at=now)]
    )
    service = GovernanceAuditService(
        repository,
        evidence_secret="wp12-test-secret-with-at-least-32-bytes",
        key_version="test-v1",
        uid_factory=lambda: "seal-1",
        now_factory=lambda: now + timedelta(days=1),
    )
    service.create_seal(
        period_start=now - timedelta(seconds=1),
        period_end=now + timedelta(seconds=1),
        actor_uid="admin-1",
    )
    repository.seals["seal-1"]["event_count"] = 99

    assert (
        service.verify_seal("seal-1")["integrity_status"]
        == "invalid_signature"
    )

    with pytest.raises(GovernanceAuditInvalid, match="32"):
        GovernanceAuditService(
            repository,
            evidence_secret="short",
            key_version="test-v1",
        )


def test_seal_rejects_invalid_windows_and_unbounded_event_sets():
    from app.core.system.governance_audit import (
        GovernanceAuditInvalid,
        GovernanceAuditService,
    )

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    service = _service([])
    with pytest.raises(GovernanceAuditInvalid, match="period_end"):
        service.create_seal(
            period_start=now,
            period_end=now,
            actor_uid="admin-1",
        )

    class OversizedRepository(FakeAuditRepository):
        def fetch_events(self, **_kwargs):
            event = _event(
                "event-1", "authentication", occurred_at=now
            )
            return [event] * 50_001

    oversized = GovernanceAuditService(
        OversizedRepository(),
        evidence_secret="wp12-test-secret-with-at-least-32-bytes",
        key_version="test-v1",
        now_factory=lambda: now + timedelta(days=1),
    )
    with pytest.raises(GovernanceAuditInvalid, match="50,000"):
        oversized.create_seal(
            period_start=now - timedelta(seconds=1),
            period_end=now + timedelta(seconds=1),
            actor_uid="admin-1",
        )


def test_seal_rejects_a_window_that_ends_in_the_future():
    from app.core.system.governance_audit import (
        GovernanceAuditInvalid,
        GovernanceAuditService,
    )

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    service = GovernanceAuditService(
        FakeAuditRepository(),
        evidence_secret="wp12-test-secret-with-at-least-32-bytes",
        key_version="test-v1",
        now_factory=lambda: now,
    )

    with pytest.raises(GovernanceAuditInvalid, match="future"):
        service.create_seal(
            period_start=now - timedelta(days=1),
            period_end=now + timedelta(seconds=1),
            actor_uid="admin-1",
        )