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