DataOps-platform/tests/system/test_governance_audit_api.py

from __future__ import annotations


class FakeGovernanceAuditService:
    def __init__(self):
        self.calls = []

    def coverage(self, **filters):
        self.calls.append(("coverage", filters))
        return {
            "period_start": filters["period_start"].isoformat(),
            "period_end": filters["period_end"].isoformat(),
            "categories": [
                {
                    "category": "authentication",
                    "count": 1,
                    "available": True,
                    "latest_at": "2026-07-30T00:00:00Z",
                }
            ],
        }

    def list_events(self, **filters):
        self.calls.append(("events", filters))
        return {
            "records": [
                {
                    "event_uid": "authentication:1",
                    "category": "authentication",
                    "action": "login",
                    "status": "success",
                    "safe_detail": {"username": "admin"},
                }
            ],
            "page": int(filters["page"]),
            "page_size": int(filters["page_size"]),
            "total": 1,
        }

    def create_seal(self, **payload):
        self.calls.append(("create_seal", payload))
        return {
            "uid": "seal-1",
            "event_count": 1,
            "root_hash": "a" * 64,
            "signature": "b" * 64,
        }

    def list_seals(self, *, limit):
        self.calls.append(("list_seals", {"limit": limit}))
        return [{"uid": "seal-1", "event_count": 1}]

    def verify_seal(self, seal_uid):
        self.calls.append(("verify_seal", {"seal_uid": seal_uid}))
        return {"uid": seal_uid, "integrity_status": "intact"}


def _client(monkeypatch, *, role="admin"):
    from app import create_app
    from app.api.system import governance_audit

    service = FakeGovernanceAuditService()
    identity = {
        "id": "00000000-0000-7000-8000-000000000111",
        "roles": [role],
    }
    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: identity,
    )
    monkeypatch.setattr(
        governance_audit,
        "get_governance_audit_service",
        lambda: service,
    )
    monkeypatch.setattr(
        governance_audit,
        "get_security_checks",
        lambda: {
            "seal_ready": True,
            "assurance": "签名封存用于检测篡改，不等于阻止数据库管理员修改",
            "checks": [{"code": "safe_error_boundary", "status": "passed"}],
        },
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_admin_reads_security_coverage_events_and_seals(monkeypatch):
    http, service = _client(monkeypatch)
    window = (
        "period_start=2026-07-01T00:00:00Z&"
        "period_end=2026-07-30T00:00:00Z"
    )

    security = http.get(
        "/api/system/governance-audit/security-checks"
    )
    coverage = http.get(
        f"/api/system/governance-audit/coverage?{window}"
    )
    events = http.get(
        f"/api/system/governance-audit/events?{window}"
        "&category=authentication&page=1&page_size=20"
    )
    seals = http.get("/api/system/governance-audit/seals?limit=25")

    assert security.status_code == 200
    assert security.get_json()["data"]["seal_ready"] is True
    assert coverage.status_code == 200
    assert events.status_code == 200
    assert events.get_json()["data"]["records"][0]["safe_detail"] == {
        "username": "admin"
    }
    assert seals.status_code == 200
    assert [call[0] for call in service.calls] == [
        "coverage",
        "events",
        "list_seals",
    ]


def test_seal_creation_uses_server_identity_and_verification_is_explicit(
    monkeypatch,
):
    http, service = _client(monkeypatch)

    created = http.post(
        "/api/system/governance-audit/seals",
        json={
            "period_start": "2026-07-01T00:00:00Z",
            "period_end": "2026-07-30T00:00:00Z",
            "categories": ["authentication", "ingestion"],
            "sealed_by": "attacker-controlled",
        },
    )
    verified = http.post(
        "/api/system/governance-audit/seals/seal-1/verify"
    )

    assert created.status_code == 200
    assert service.calls[0][1]["actor_uid"] == (
        "00000000-0000-7000-8000-000000000111"
    )
    assert "sealed_by" not in service.calls[0][1]
    assert verified.status_code == 200
    assert verified.get_json()["data"]["integrity_status"] == "intact"


def test_viewer_and_editor_cannot_read_or_seal(monkeypatch):
    for role in ("viewer", "editor"):
        http, _service = _client(monkeypatch, role=role)
        assert http.get(
            "/api/system/governance-audit/events"
        ).status_code == 403
        assert http.post(
            "/api/system/governance-audit/seals",
            json={},
        ).status_code == 403


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

    http, service = _client(monkeypatch)

    def invalid(**_kwargs):
        raise GovernanceAuditInvalid("page_size must be between 1 and 100")

    service.list_events = invalid
    response = http.get(
        "/api/system/governance-audit/events"
        "?period_start=2026-07-01T00:00:00Z"
        "&period_end=2026-07-30T00:00:00Z&page_size=101"
    )
    assert response.status_code == 400
    assert response.get_json()["error"]["code"] == (
        "GOVERNANCE_AUDIT_INVALID"
    )

    def unavailable(**_kwargs):
        raise RuntimeError(
            "postgresql://admin:database-secret@db/dataops "
            "authorization=Bearer-secret"
        )

    service.list_events = unavailable
    failed = http.get(
        "/api/system/governance-audit/events"
        "?period_start=2026-07-01T00:00:00Z"
        "&period_end=2026-07-30T00:00:00Z"
    )
    payload = failed.get_json()
    assert failed.status_code == 503
    assert payload["message"] == "审计与运行证据暂不可用"
    assert "database-secret" not in repr(payload)
    assert "Bearer-secret" not in repr(payload)


def test_production_without_dedicated_key_blocks_seal_and_verify(monkeypatch):
    from flask import Flask

    from app.api.system import governance_audit

    production = Flask(__name__)
    production.config.update(
        FLASK_ENV="production",
        TESTING=False,
    )
    development = Flask(__name__)
    development.config.update(
        FLASK_ENV="development",
        TESTING=False,
    )

    with production.app_context():
        assert governance_audit._sealing_allowed(False) is False
        assert governance_audit._sealing_allowed(True) is True
    with development.app_context():
        assert governance_audit._sealing_allowed(False) is True

    http, service = _client(monkeypatch)
    monkeypatch.setattr(
        governance_audit,
        "_sealing_allowed",
        lambda _dedicated_ready: False,
    )
    created = http.post(
        "/api/system/governance-audit/seals",
        json={
            "period_start": "2026-07-01T00:00:00Z",
            "period_end": "2026-07-30T00:00:00Z",
        },
    )
    verified = http.post(
        "/api/system/governance-audit/seals/seal-1/verify"
    )

    assert created.status_code == 503
    assert verified.status_code == 503
    assert service.calls == []