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()