DataOps-platform/tests/integration/test_governance_audit_postgres.py

from __future__ import annotations

import os
import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import text

pytestmark = pytest.mark.integration


def test_governance_audit_covers_six_sources_and_detects_tampering(monkeypatch):
    database_url = os.environ.get("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")

    monkeypatch.setenv("DATABASE_URL", database_url)
    from app import create_app, db
    from app.core.system.governance_audit import (
        AUDIT_CATEGORIES,
        GovernanceAuditService,
    )
    from app.core.system.governance_audit_repository import (
        SqlAlchemyGovernanceAuditRepository,
    )

    app = create_app()
    app.config.update(TESTING=True)
    suffix = uuid.uuid4().hex[:10]
    ids = {name: str(uuid.uuid4()) for name in (
        "user",
        "source",
        "job",
        "asset_left",
        "asset_right",
        "candidate",
        "review",
        "merge",
        "rollback",
        "ontology",
        "ontology_version",
        "ontology_run",
        "quality_profile",
        "quality_version",
        "quality_run",
        "issue",
        "timeline",
        "query",
        "correlation",
    )}
    now = datetime.now(UTC).replace(microsecond=0)
    start = now - timedelta(minutes=1)
    end = now + timedelta(minutes=1)

    try:
        with app.app_context():
            statements = [
                (
                    """
                    INSERT INTO public.users (
                        id, username, display_name, password_hash, status
                    ) VALUES (
                        CAST(:user AS uuid), :username, :username,
                        'wp12-integration-hash', 'active'
                    )
                    """,
                    {"user": ids["user"], "username": f"wp12-{suffix}"},
                ),
                (
                    """
                    INSERT INTO public.auth_audit_events (
                        user_id, username, event_type, success, detail,
                        created_at
                    ) VALUES (
                        CAST(:user AS uuid), :username, 'login', TRUE,
                        'safe integration fixture', :now
                    )
                    """,
                    {
                        "user": ids["user"],
                        "username": f"wp12-{suffix}",
                        "now": now,
                    },
                ),
                (
                    """
                    INSERT INTO public.ingestion_sources (
                        uid, source_type, name, config, permission_scope,
                        status, created_by
                    ) VALUES (
                        CAST(:source AS uuid), 'database', :name,
                        '{"password":"source-secret"}'::jsonb,
                        '{}'::jsonb, 'active', :user
                    )
                    """,
                    {
                        "source": ids["source"],
                        "name": f"WP12 source {suffix}",
                        "user": ids["user"],
                    },
                ),
                (
                    """
                    INSERT INTO public.ingestion_jobs (
                        uid, source_uid, job_type, status, idempotency_key,
                        parser_version, parameters, statistics, actor_uid,
                        attempt_count, created_at
                    ) VALUES (
                        CAST(:job AS uuid), CAST(:source AS uuid),
                        'database_catalog', 'published', :key, 'wp12-v1',
                        '{}'::jsonb, '{}'::jsonb, :user, 1, :now
                    )
                    """,
                    {
                        "job": ids["job"],
                        "source": ids["source"],
                        "key": suffix.ljust(64, "0"),
                        "user": ids["user"],
                        "now": now,
                    },
                ),
            ]
            for asset_key, marker in (
                ("asset_left", "a"),
                ("asset_right", "b"),
            ):
                statements.append(
                    (
                        """
                        INSERT INTO public.device_assets (
                            uid, asset_type, name, status, current_version,
                            content_hash, attributes, created_by, updated_by
                        ) VALUES (
                            CAST(:uid AS uuid), 'device', :name, 'active', 1,
                            :hash, '{}'::jsonb, :user, :user
                        )
                        """,
                        {
                            "uid": ids[asset_key],
                            "name": f"WP12 asset {marker} {suffix}",
                            "hash": marker * 64,
                            "user": ids["user"],
                        },
                    )
                )
            statements.extend(
                [
                    (
                        """
                        INSERT INTO public.device_entity_match_candidates (
                            uid, left_asset_uid, right_asset_uid,
                            canonical_asset_uid, status, suggestion_source,
                            confidence, explanation, evidence_uids,
                            current_version, created_by
                        ) VALUES (
                            CAST(:candidate AS uuid),
                            CAST(:left_asset AS uuid),
                            CAST(:right_asset AS uuid),
                            CAST(:left_asset AS uuid), 'rolled_back',
                            'manual', 1, '[]'::jsonb, '[]'::jsonb, 2, :user
                        )
                        """,
                        {
                            "candidate": ids["candidate"],
                            "left_asset": ids["asset_left"],
                            "right_asset": ids["asset_right"],
                            "user": ids["user"],
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_entity_match_reviews (
                            uid, candidate_uid, version, decision, reason,
                            actor_uid, created_at
                        ) VALUES (
                            CAST(:review AS uuid), CAST(:candidate AS uuid),
                            1, 'approve', 'safe integration fixture',
                            :user, :now
                        )
                        """,
                        {
                            "review": ids["review"],
                            "candidate": ids["candidate"],
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_entity_merge_events (
                            uid, candidate_uid, canonical_asset_uid,
                            member_asset_uid, review_uid, snapshot,
                            actor_uid, created_at
                        ) VALUES (
                            CAST(:merge AS uuid), CAST(:candidate AS uuid),
                            CAST(:left_asset AS uuid),
                            CAST(:right_asset AS uuid),
                            CAST(:review AS uuid),
                            '{"password":"merge-secret"}'::jsonb,
                            :user, :now
                        )
                        """,
                        {
                            "merge": ids["merge"],
                            "candidate": ids["candidate"],
                            "left_asset": ids["asset_left"],
                            "right_asset": ids["asset_right"],
                            "review": ids["review"],
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_entity_merge_rollbacks (
                            uid, merge_uid, candidate_uid, reason, snapshot,
                            actor_uid, created_at
                        ) VALUES (
                            CAST(:rollback AS uuid), CAST(:merge AS uuid),
                            CAST(:candidate AS uuid), 'safe rollback',
                            '{"password":"rollback-secret"}'::jsonb,
                            :user, :now
                        )
                        """,
                        {
                            "rollback": ids["rollback"],
                            "merge": ids["merge"],
                            "candidate": ids["candidate"],
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.ontologies (
                            uid, code, name, owner_uid, status,
                            draft_revision, created_by
                        ) VALUES (
                            CAST(:ontology AS uuid), :code, :name, :user,
                            'published', 1, :user
                        )
                        """,
                        {
                            "ontology": ids["ontology"],
                            "code": f"WP12-{suffix}",
                            "name": f"WP12 ontology {suffix}",
                            "user": ids["user"],
                        },
                    ),
                    (
                        """
                        INSERT INTO public.ontology_versions (
                            uid, ontology_uid, version, status,
                            graph_document, content_hash, created_by,
                            created_at, published_at
                        ) VALUES (
                            CAST(:version AS uuid), CAST(:ontology AS uuid),
                            1, 'published', '{}'::jsonb, :hash, :user,
                            :now, :now
                        )
                        """,
                        {
                            "version": ids["ontology_version"],
                            "ontology": ids["ontology"],
                            "hash": "c" * 64,
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.ontology_publish_runs (
                            uid, ontology_uid, version_uid, idempotency_key,
                            status, actor_uid, created_at, finished_at
                        ) VALUES (
                            CAST(:run AS uuid), CAST(:ontology AS uuid),
                            CAST(:version AS uuid), :key, 'published',
                            :user, :now, :now
                        )
                        """,
                        {
                            "run": ids["ontology_run"],
                            "ontology": ids["ontology"],
                            "version": ids["ontology_version"],
                            "key": f"wp12-{suffix}",
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_quality_profiles (
                            uid, code, name, created_by
                        ) VALUES (
                            CAST(:profile AS uuid), :code, :name, :user
                        )
                        """,
                        {
                            "profile": ids["quality_profile"],
                            "code": f"wp12-quality-{suffix}",
                            "name": f"WP12 quality {suffix}",
                            "user": ids["user"],
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_quality_profile_versions (
                            uid, profile_uid, version, status, rules,
                            content_hash, created_by, published_by,
                            created_at, published_at
                        ) VALUES (
                            CAST(:version AS uuid), CAST(:profile AS uuid),
                            1, 'published', '[]'::jsonb, :hash,
                            :user, :user, :now, :now
                        )
                        """,
                        {
                            "version": ids["quality_version"],
                            "profile": ids["quality_profile"],
                            "hash": "d" * 64,
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_quality_runs (
                            uid, policy_version_uid, policy_hash, source_uid,
                            status, total_assets, total_violations, score,
                            created_by, created_at
                        ) VALUES (
                            CAST(:run AS uuid), CAST(:version AS uuid), :hash,
                            CAST(:source AS uuid), 'success', 1, 1, 0,
                            :user, :now
                        )
                        """,
                        {
                            "run": ids["quality_run"],
                            "version": ids["quality_version"],
                            "hash": "d" * 64,
                            "source": ids["source"],
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_quality_issues (
                            uid, issue_code, source_violation_uid,
                            source_run_uid, rule_code, severity, priority,
                            asset_uid, field_name, message, evidence,
                            recurrence_key, occurrence_number, status,
                            current_version, created_by, updated_by
                        ) VALUES (
                            CAST(:issue AS uuid), :code,
                            CAST(:violation AS uuid), CAST(:run AS uuid),
                            'WP12-RULE', 'error', 'high',
                            CAST(:asset AS uuid), 'name', 'safe message',
                            '{"password":"issue-secret"}'::jsonb, :key,
                            1, 'closed', 1,
                            CAST(:user AS uuid), CAST(:user AS uuid)
                        )
                        """,
                        {
                            "issue": ids["issue"],
                            "code": f"WP12-{suffix}",
                            "violation": str(uuid.uuid4()),
                            "run": ids["quality_run"],
                            "asset": ids["asset_left"],
                            "key": "e" * 64,
                            "user": ids["user"],
                        },
                    ),
                    (
                        """
                        INSERT INTO public.device_quality_issue_timeline (
                            uid, issue_uid, action, from_status, to_status,
                            actor_uid, note, payload, created_at
                        ) VALUES (
                            CAST(:timeline AS uuid), CAST(:issue AS uuid),
                            'closed', 'pending_review', 'closed',
                            CAST(:user AS uuid), 'must not be selected',
                            '{"password":"timeline-secret"}'::jsonb, :now
                        )
                        """,
                        {
                            "timeline": ids["timeline"],
                            "issue": ids["issue"],
                            "user": ids["user"],
                            "now": now,
                        },
                    ),
                    (
                        """
                        INSERT INTO public.knowledge_query_audits (
                            id, query_hash, user_id, roles,
                            business_domain_uids, mode, retriever_counts,
                            cited_points, degraded_components, correlation_id,
                            latency_ms, created_at
                        ) VALUES (
                            CAST(:query AS uuid), :hash, CAST(:user AS uuid),
                            '["admin"]'::jsonb, '[]'::jsonb, 'hybrid',
                            CAST(:retriever_counts AS jsonb),
                            '["point-1"]'::jsonb,
                            '[]'::jsonb, CAST(:correlation AS uuid), 12, :now
                        )
                        """,
                        {
                            "query": ids["query"],
                            "hash": "f" * 64,
                            "user": ids["user"],
                            "correlation": ids["correlation"],
                            "retriever_counts": '{"canonical":1}',
                            "now": now,
                        },
                    ),
                ]
            )
            for statement, params in statements:
                db.session.execute(text(statement), params)
            db.session.flush()

            service = GovernanceAuditService(
                SqlAlchemyGovernanceAuditRepository(db.session),
                evidence_secret="wp12-integration-secret-with-32-bytes",
                key_version="integration-v1",
                now_factory=lambda: end,
            )
            coverage = service.coverage(
                period_start=start,
                period_end=end,
            )
            assert [item["category"] for item in coverage["categories"]] == list(
                AUDIT_CATEGORIES
            )
            assert all(item["count"] >= 1 for item in coverage["categories"])

            events = service.list_events(
                period_start=start,
                period_end=end,
                page_size=100,
            )
            serialized = repr(events)
            for secret in (
                "source-secret",
                "merge-secret",
                "rollback-secret",
                "issue-secret",
                "timeline-secret",
                "must not be selected",
            ):
                assert secret not in serialized

            seal = service.create_seal(
                period_start=start,
                period_end=end,
                actor_uid=ids["user"],
            )
            assert service.verify_seal(seal["uid"])["integrity_status"] == (
                "intact"
            )

            nested = db.session.begin_nested()
            db.session.execute(
                text(
                    """
                    UPDATE public.auth_audit_events
                    SET success = FALSE
                    WHERE username = :username
                      AND created_at = :now
                    """
                ),
                {"username": f"wp12-{suffix}", "now": now},
            )
            db.session.flush()
            assert service.verify_seal(seal["uid"])["integrity_status"] == (
                "tampered"
            )
            nested.rollback()
            assert service.verify_seal(seal["uid"])["integrity_status"] == (
                "intact"
            )
    finally:
        with app.app_context():
            db.session.rollback()
            db.session.remove()