ontorefactor-governance/tests/test_report.py

import hashlib
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from ontorefactor_governance import api, builds
from ontorefactor_governance.builds import publish_ontology_build, run_ontology_build
from ontorefactor_governance.db import Database
from ontorefactor_governance.report import get_report, render_markdown
from ontorefactor_governance.service import (
    GovernanceError, seed_challenge_cup_demo, validate_project,
)


SAMPLES = Path(__file__).parents[1] / "src" / "ontorefactor_governance" / "static" / "samples"


def demo_sources():
    definitions = (
        ("01-dcp-user.sql", "ddl"),
        ("02-user-profile-openapi.yaml", "openapi"),
        ("03-dcp-asset-inventory.json", "inventory"),
        ("04-governance-evidence.json", "inventory"),
    )
    return [
        {"source_name": name, "source_type": source_type, "content": (SAMPLES / name).read_text(encoding="utf-8")}
        for name, source_type in definitions
    ]


def publish_demo(db, tenant, project_id, target="dcp_user.mobile"):
    build = run_ontology_build(
        db, tenant, project_id, demo_sources(),
        semantic_mode="deterministic", instruction="从上传证据构建候选本体",
        environment="production", decision_target=target,
    )
    return publish_ontology_build(
        db, tenant, project_id, build["build_id"],
        reviewer="test-reviewer", accept_semantic_assertions=False, decision_target=target,
    )


def mobile_dossier(report):
    return next(
        item for item in report["refactor_dossiers"]["full"]
        if item["asset"]["uri"].endswith(":dcp_user:mobile")
    )


def test_publish_generates_point_in_time_report():
    db = Database("sqlite:///:memory:")
    tenant = "report-test"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]

    published = publish_demo(db, tenant, project_id)
    assert published["report"]["error"] == ""
    assert published["report"]["id"]

    stored = get_report(db, tenant, project_id, "latest")
    assert stored["version"]["label"] == "v1"
    report = stored["report_json"]
    for section in (
        "format_version", "generator", "header", "assets", "governance_coverage",
        "lineage", "refactor_dossiers", "issues", "review_queue", "diff", "scope",
    ):
        assert section in report, section

    header = report["header"]
    assert header["point_in_time"] is True
    assert header["published_by"] == "test-reviewer"
    assert header["semantic_mode"] == "deterministic"
    expected = {
        name: hashlib.sha256((SAMPLES / name).read_text(encoding="utf-8").encode("utf-8")).hexdigest()
        for name in ("01-dcp-user.sql", "02-user-profile-openapi.yaml",
                     "03-dcp-asset-inventory.json", "04-governance-evidence.json")
    }
    assert {item["name"]: item["sha256"] for item in header["sources"]} == expected

    dossier = mobile_dossier(report)
    assert dossier["gate_status"]["status"] == "BLOCKED"
    assert dossier["gate_status"]["basis"] == "accepted+pending"
    assert published["decision"]["status"] == "BLOCKED"
    db.close()


def test_coverage_gap_matches_validation_findings():
    db = Database("sqlite:///:memory:")
    tenant = "report-coverage"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]
    publish_demo(db, tenant, project_id)

    report = get_report(db, tenant, project_id, "latest")["report_json"]
    findings = validate_project(db, tenant, project_id, persist=False)["findings"]

    owner_gap = {f["target_id"] for f in findings if f["code"] == "PRODUCTION_ASSET_WITHOUT_OWNER"}
    policy_gap = {f["target_id"] for f in findings if f["code"] == "SENSITIVE_FIELD_WITHOUT_POLICY"}
    assert {i["id"] for i in report["governance_coverage"]["OWNED_BY"]["uncovered"]} == owner_gap
    assert {i["id"] for i in report["governance_coverage"]["GOVERNED_BY"]["uncovered"]} == policy_gap
    for predicate in ("OWNED_BY", "GOVERNED_BY", "CLASSIFIED_AS", "VALIDATES"):
        section = report["governance_coverage"][predicate]
        assert section["definition"]
        assert section["eligible"] == section["covered"] + len(section["uncovered"])
    db.close()


def test_dossier_ranking_and_contents():
    db = Database("sqlite:///:memory:")
    tenant = "report-dossier"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]
    publish_demo(db, tenant, project_id)
    report = get_report(db, tenant, project_id, "latest")["report_json"]

    dossiers = report["refactor_dossiers"]
    assert dossiers["high_risk_total"] >= 1
    first = dossiers["full"][0]
    assert first["asset"]["uri"].endswith(":dcp_user:mobile")
    assert set(first["risk_flags"]) == {"sensitive", "governed", "validated", "has-consumers"}

    dossier = mobile_dossier(report)
    assert len(dossier["consumers"]["reads"]) >= 3  # API + 两个任务仍读取 mobile
    assert dossier["owner"] and dossier["policies"]
    assert any(gate.get("dimension") == "completeness" for gate in dossier["quality_gates"])
    assert dossier["evidence_ids"]

    consumers = report["lineage"]["consumers"]
    assert consumers[dossier["asset"]["id"]]["reads"] == len(dossier["consumers"]["reads"])
    db.close()


def test_diff_first_and_second_version():
    db = Database("sqlite:///:memory:")
    tenant = "report-diff"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]

    first = publish_demo(db, tenant, project_id)
    report_v1 = get_report(db, tenant, project_id, first["version"]["id"])["report_json"]
    assert report_v1["diff"]["previous_version"] is None
    assert report_v1["diff"]["elements"]["added_count"] == len(first["version"]["element_ids"])
    assert report_v1["diff"]["elements"]["removed_count"] == 0

    second = publish_demo(db, tenant, project_id)
    report_v2 = get_report(db, tenant, project_id, "latest")["report_json"]
    assert report_v2["header"]["label"] == "v2"
    assert report_v2["diff"]["previous_version"]["label"] == "v1"
    assert report_v2["diff"]["elements"]["added_count"] == 0
    assert report_v2["diff"]["elements"]["removed_count"] == 0
    assert report_v2["diff"]["elements"]["unchanged_count"] == len(first["version"]["element_ids"])
    db.close()


def test_review_queue_lists_pending_inferred_assertions():
    db = Database("sqlite:///:memory:")
    tenant = "report-queue"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]
    publish_demo(db, tenant, project_id)
    report = get_report(db, tenant, project_id, "latest")["report_json"]

    queue = report["review_queue"]
    assert queue["total"] > 0
    assert queue["total"] == len(queue["items"])
    inferred = [item for item in queue["items"] if item["assertion_type"] == "inferred"]
    assert inferred and all(item["generated_by"] for item in inferred)
    db.close()


def test_markdown_rendering_is_deterministic_and_complete():
    db = Database("sqlite:///:memory:")
    tenant = "report-md"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]
    publish_demo(db, tenant, project_id)
    report = get_report(db, tenant, project_id, "latest")["report_json"]

    first, second = render_markdown(report), render_markdown(report)
    assert first == second
    for heading in (
        "# 治理报告 v1", "## 1. 来源与版本", "## 2. 资产清单", "## 3. 治理覆盖度",
        "## 4. 血缘摘要", "## 5. 重构档案", "## 6. 风险与问题", "## 7. 待裁决与版本变更",
        "## 附录：口径与限制",
    ):
        assert heading in first, heading
    assert "BLOCKED" in first
    assert "发布时点" in first

    hostile = {**report, "assets": {**report["assets"], "items": [
        {"id": "x", "kind": "Table", "name": "a|b\nc", "uri": "u", "status": "published",
         "environment": "production", "sensitive": False},
    ], "total": 1, "by_kind": {"Table": 1}, "truncated": False}}
    rendered = render_markdown(hostile)
    assert "a\\|b c" in rendered
    db.close()


def test_report_failure_does_not_block_publish(monkeypatch):
    def boom(*args, **kwargs):
        raise RuntimeError("synthetic report failure")

    db = Database("sqlite:///:memory:")
    tenant = "report-fail"
    project_id = seed_challenge_cup_demo(db, tenant)["project"]["id"]
    build = run_ontology_build(
        db, tenant, project_id, demo_sources(), semantic_mode="deterministic",
    )
    monkeypatch.setattr(builds, "build_governance_report", boom)
    published = publish_ontology_build(db, tenant, project_id, build["build_id"], reviewer="test-reviewer")

    assert published["version"]["label"] == "v1"
    assert published["report"]["id"] is None
    assert "synthetic report failure" in published["report"]["error"]
    issues = db.fetchall(
        "SELECT rule_id FROM governance_issues WHERE tenant_id=? AND project_id=? AND rule_id='report:generation-failed'",
        (tenant, project_id),
    )
    assert len(issues) == 1
    with pytest.raises(GovernanceError) as caught:
        get_report(db, tenant, project_id, "latest")
    assert caught.value.code == "REPORT_NOT_FOUND"
    assert caught.value.status == 404

    # 自愈：报告生成器恢复后，重复发布走幂等路径不补报告，保持只读
    monkeypatch.undo()
    republished = publish_ontology_build(db, tenant, project_id, build["build_id"], reviewer="test-reviewer")
    assert republished["idempotent"] is True
    assert republished["report"] is None
    db.close()


def test_report_api_endpoint(monkeypatch):
    db = Database("sqlite:///:memory:")
    monkeypatch.setattr(api, "get_database", lambda: db)
    client = TestClient(api.app)
    headers = {"X-Tenant-ID": "report-api"}
    project_id = client.post("/api/v1/governance/projects", headers=headers, json={
        "name": "Report API", "namespace": "urn:report:api",
    }).json()["id"]

    build = client.post(
        f"/api/v1/governance/projects/{project_id}/ontology-builds", headers=headers,
        json={"sources": [
            {"source_type": item["source_type"], "source_name": item["source_name"], "content": item["content"]}
            for item in demo_sources()
        ], "environment": "production", "decision_target": "dcp_user.mobile",
            "semantic_mode": "deterministic"},
    )
    assert build.status_code == 200, build.text
    build_id = build.json()["build_id"]
    published = client.post(
        f"/api/v1/governance/projects/{project_id}/ontology-builds/{build_id}/publish",
        headers=headers, json={"reviewer": "api-reviewer", "decision_target": "dcp_user.mobile"},
    )
    assert published.status_code == 200, published.text
    report_id = published.json()["report"]["id"]
    assert report_id

    versions = client.get(f"/api/v1/governance/projects/{project_id}/ontology-versions", headers=headers).json()
    version_id = versions["versions"][0]["id"]
    assert versions["versions"][0]["report_id"] == report_id

    as_json = client.get(
        f"/api/v1/governance/projects/{project_id}/ontology-versions/latest/report",
        headers=headers, params={"format": "json"},
    )
    assert as_json.status_code == 200
    for section in ("header", "assets", "governance_coverage", "refactor_dossiers", "review_queue"):
        assert section in as_json.json(), section

    by_id = client.get(
        f"/api/v1/governance/projects/{project_id}/ontology-versions/{version_id}/report", headers=headers,
    )
    assert by_id.status_code == 200
    assert by_id.json() == as_json.json()

    as_md = client.get(
        f"/api/v1/governance/projects/{project_id}/ontology-versions/latest/report",
        headers=headers, params={"format": "md"},
    )
    assert as_md.status_code == 200
    assert as_md.headers["content-type"].startswith("text/markdown")
    assert "# 治理报告 v1" in as_md.text
    assert "governance-report.md" in as_md.headers["content-disposition"]

    assert client.get(
        f"/api/v1/governance/projects/{project_id}/ontology-versions/latest/report",
        headers={"X-Tenant-ID": "other"},
    ).status_code == 404
    assert client.get(
        f"/api/v1/governance/projects/{project_id}/ontology-versions/ov_missing/report", headers=headers,
    ).status_code == 404
    assert client.get(
        f"/api/v1/governance/projects/{project_id}/ontology-versions/latest/report",
        headers=headers, params={"format": "xml"},
    ).status_code == 422
    db.close()