DataOps-platform/tests/system/test_governance_audit_repository.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta


class FakeRows:
    def __init__(self, rows=None):
        self.rows = list(rows or [])

    def mappings(self):
        return self

    def all(self):
        return self.rows

    def one(self):
        return self.rows[0]

    def one_or_none(self):
        return self.rows[0] if self.rows else None


class RecordingSession:
    def __init__(self, responses=None):
        self.responses = list(responses or [])
        self.calls = []

    def execute(self, statement, params=None):
        self.calls.append((str(statement), dict(params or {})))
        rows = self.responses.pop(0) if self.responses else []
        return FakeRows(rows)

    def flush(self):
        return None


def test_repository_queries_six_authoritative_sources_with_safe_projections():
    from app.core.system.governance_audit import AUDIT_CATEGORIES
    from app.core.system.governance_audit_repository import (
        SqlAlchemyGovernanceAuditRepository,
    )

    session = RecordingSession()
    repository = SqlAlchemyGovernanceAuditRepository(session)
    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)

    assert repository.fetch_events(
        categories=AUDIT_CATEGORIES,
        period_start=now - timedelta(days=1),
        period_end=now,
    ) == []

    sql = "\n".join(statement for statement, _ in session.calls).lower()
    for table in (
        "auth_audit_events",
        "ingestion_jobs",
        "device_entity_match_reviews",
        "device_entity_merge_rollbacks",
        "ontology_publish_runs",
        "device_semantic_code_reviews",
        "device_quality_profile_versions",
        "device_quality_issue_timeline",
        "knowledge_query_audits",
    ):
        assert table in sql
    for forbidden in (
        "select ip_address",
        "select user_agent",
        "last_error",
        "source.config",
        "encrypted_payload",
        "snapshot",
        "timeline.note",
        "timeline.payload",
        "query text",
    ):
        assert forbidden not in sql
    assert all(
        params["period_start"] == now - timedelta(days=1)
        and params["period_end"] == now
        for _, params in session.calls
    )


def test_repository_normalizes_rows_without_exposing_raw_payloads():
    from app.core.system.governance_audit_repository import (
        SqlAlchemyGovernanceAuditRepository,
    )

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    session = RecordingSession(
        [
            [
                {
                    "event_uid": "authentication:7",
                    "category": "authentication",
                    "action": "login",
                    "status": "success",
                    "actor_uid": "admin",
                    "resource_type": "user",
                    "resource_uid": "user-1",
                    "occurred_at": now,
                    "safe_detail": {"username": "admin"},
                }
            ]
        ]
    )
    repository = SqlAlchemyGovernanceAuditRepository(session)

    events = repository.fetch_events(
        categories=["authentication"],
        period_start=now - timedelta(days=1),
        period_end=now + timedelta(days=1),
    )

    assert events == [
        {
            "event_uid": "authentication:7",
            "category": "authentication",
            "action": "login",
            "status": "success",
            "actor_uid": "admin",
            "resource_type": "user",
            "resource_uid": "user-1",
            "occurred_at": now,
            "safe_detail": {"username": "admin"},
        }
    ]


def test_security_snapshot_checks_nested_secret_keys():
    from app.core.system.governance_audit_repository import (
        SqlAlchemyGovernanceAuditRepository,
    )

    session = RecordingSession(
        [
            [{"credential_count": 0, "encrypted_credential_count": 0}],
            [{"plaintext_source_config_count": 0}],
        ]
    )

    SqlAlchemyGovernanceAuditRepository(session).security_snapshot()

    source_sql = session.calls[1][0].lower()
    assert '$.** ? (@.type() == "object").keyvalue()' in source_sql
    assert "like_regex" in source_sql
    assert "password" in source_sql
    assert "authorization" in source_sql


def test_repository_persists_and_reads_append_only_seals():
    from app.core.system.governance_audit_repository import (
        SqlAlchemyGovernanceAuditRepository,
    )

    now = datetime(2026, 7, 30, 8, 0, tzinfo=UTC)
    stored = {
        "uid": "seal-1",
        "period_start": now,
        "period_end": now + timedelta(hours=1),
        "categories": ["authentication"],
        "event_count": 1,
        "root_hash": "a" * 64,
        "signature": "b" * 64,
        "key_version": "v1",
        "sealed_by": "admin-1",
        "created_at": now,
    }
    session = RecordingSession([[stored], [stored], [stored]])
    repository = SqlAlchemyGovernanceAuditRepository(session)

    saved = repository.save_seal(
        {
            **stored,
            "period_start": stored["period_start"].isoformat(),
            "period_end": stored["period_end"].isoformat(),
        }
    )
    loaded = repository.get_seal("seal-1")
    listed = repository.list_seals(limit=10)

    assert saved["uid"] == loaded["uid"] == listed[0]["uid"]
    sql = "\n".join(statement for statement, _ in session.calls).upper()
    assert "INSERT INTO PUBLIC.GOVERNANCE_AUDIT_SEALS" in sql
    assert "UPDATE PUBLIC.GOVERNANCE_AUDIT_SEALS" not in sql
    assert "DELETE FROM PUBLIC.GOVERNANCE_AUDIT_SEALS" not in sql