ontorefactor-governance/tests/test_decision.py

import json
from pathlib import Path

import pytest

from ontorefactor_governance.agents import execute_governance_agent
from ontorefactor_governance.builds import publish_ontology_build, run_ontology_build
from ontorefactor_governance.db import Database
from ontorefactor_governance.decision import collect_project_context, evaluate_retirement_gate, resolve_retirement_target
from ontorefactor_governance.governance_qa import ask_governance_agent
from ontorefactor_governance.llm import LongCatProvider, PROMPT_VERSION
from ontorefactor_governance.service import GovernanceError, create_assertion, create_evidence, seed_challenge_cup_demo, upsert_element


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


@pytest.fixture
def db():
    value = Database("sqlite:///:memory:")
    yield value
    value.close()


def _demo(db):
    seeded = seed_challenge_cup_demo(db, "decision-demo")
    project_id = seeded["project"]["id"]
    definitions = (
        ("01-dcp-user.sql", "ddl"),
        ("02-user-profile-openapi.yaml", "openapi"),
        ("03-dcp-asset-inventory.json", "inventory"),
        ("04-governance-evidence.json", "inventory"),
    )
    sources = [{
        "source_name": name,
        "source_type": source_type,
        "content": (SAMPLES / name).read_text(encoding="utf-8"),
    } for name, source_type in definitions]
    build = run_ontology_build(
        db, "decision-demo", project_id, sources,
        semantic_mode="deterministic", environment="production",
        decision_target="dcp_user.mobile",
    )
    publish_ontology_build(
        db, "decision-demo", project_id, build["build_id"],
        reviewer="test-builder", decision_target="dcp_user.mobile",
    )
    target = resolve_retirement_target(db, "decision-demo", project_id, "dcp_user.mobile")
    return project_id, target["id"]


def _set_metadata(db, project_id, kind, metadata):
    db.execute(
        "UPDATE governance_evidence SET metadata=? WHERE project_id=? AND kind=?",
        (json.dumps(metadata, ensure_ascii=False), project_id, kind),
    )
    db.commit()


def test_demo_decision_is_computed_from_evidence_not_seeded_issues(db):
    project_id, target_id = _demo(db)

    decision = evaluate_retirement_gate(db, "decision-demo", project_id, target_id)

    assert decision["status"] == "BLOCKED"
    assert decision["summary"] == {"passed": 1, "failed": 4, "unknown": 0, "total": 5}
    checks = {item["key"]: item for item in decision["checks"]}
    assert checks["active-consumers"]["actual"] == 3
    assert checks["migration-quality"]["actual"] == 0.9987
    assert checks["compatibility-window"]["actual"] == 0
    assert not db.fetchall("SELECT id FROM governance_issues WHERE project_id=? AND rule_id LIKE 'demo:%'", (project_id,))


def test_fourth_agent_answers_a_natural_language_question_independently(db, monkeypatch):
    monkeypatch.delenv("AGENTPAAS_BASE_URL", raising=False)
    project_id, _ = _demo(db)

    result = ask_governance_agent(
        db, "decision-demo", project_id,
        "请问 dcp_user.mobile 现在是否可以迁移并删除？",
    )

    assert result["agent"] == "governance-decision-agent"
    assert result["decision"]["status"] == "BLOCKED"
    assert result["resolved_target"]["id"]
    assert "目前不可以" in result["answer"]
    run = db.fetchone("SELECT * FROM governance_agent_runs WHERE id=?", (result["run_id"],))
    assert run["source_type"] == "governance-question"
    assert run["decision_id"] == result["decision"]["id"]


def test_fourth_agent_requires_a_published_ontology(db):
    project_id = seed_challenge_cup_demo(db, "unpublished-qa")["project"]["id"]

    with pytest.raises(GovernanceError) as error:
        ask_governance_agent(db, "unpublished-qa", project_id, "dcp_user.mobile 可以删除吗？")

    assert error.value.code == "ONTOLOGY_NOT_PUBLISHED"
    assert error.value.status == 409


def test_retirement_gate_returns_insufficient_when_required_evidence_is_missing(db):
    project_id, target_id = _demo(db)
    db.execute("DELETE FROM governance_evidence WHERE project_id=?", (project_id,))
    db.execute("UPDATE governance_assertions SET review_status='rejected' WHERE project_id=? AND predicate IN ('READS','USES')", (project_id,))
    db.commit()

    decision = evaluate_retirement_gate(db, "decision-demo", project_id, target_id, persist=True)

    assert decision["status"] == "INSUFFICIENT_EVIDENCE"
    assert decision["summary"]["failed"] == 0
    assert decision["summary"]["unknown"] == 4


def test_retirement_gate_returns_allow_only_after_current_facts_pass(db):
    project_id, target_id = _demo(db)
    _set_metadata(db, project_id, "RuntimeObservation", {
        "target": "dcp_user.mobile", "window": "24h", "consumer_count": 0, "active_consumers": [],
    })
    source_rows = db.fetchall("SELECT id,source,metadata FROM governance_evidence WHERE project_id=? AND kind='SourceArtifact'", (project_id,))
    for row in source_rows:
        metadata = json.loads(row["metadata"])
        if "consumer_count" in metadata:
            metadata.update({"consumer_count": 0, "active_consumers": []})
            db.execute("UPDATE governance_evidence SET metadata=? WHERE id=?", (json.dumps(metadata), row["id"]))
    _set_metadata(db, project_id, "QualityReport", {
        "target": "dcp_user.mobile", "checked_rows": 10189432, "unmatched_rows": 0,
        "completeness": 1.0, "threshold": 1.0,
    })
    _set_metadata(db, project_id, "ExpertStatement", {
        "target": "dcp_user.mobile", "approved": ["architecture", "data-owner", "security"],
        "pending": [], "rejected": [],
    })
    _set_metadata(db, project_id, "Contract", {
        "target": "dcp_user.mobile", "required_release_cycles": 2, "completed_release_cycles": 2,
    })
    db.execute("UPDATE governance_assertions SET review_status='rejected' WHERE project_id=? AND predicate IN ('READS','USES')", (project_id,))
    db.commit()

    decision = evaluate_retirement_gate(db, "decision-demo", project_id, target_id, persist=True)

    assert decision["status"] == "ALLOW"
    assert decision["summary"] == {"passed": 5, "failed": 0, "unknown": 0, "total": 5}
    assert not db.fetchall(
        "SELECT id FROM governance_issues WHERE project_id=? AND rule_id LIKE 'decision:retirement:%'",
        (project_id,),
    )


def test_target_scoped_evidence_does_not_leak_between_fields(db):
    project_id, target_id = _demo(db)
    create_evidence(db, "decision-demo", project_id, {
        "kind": "QualityReport", "uri": "evidence://other-field/newer-quality",
        "title": "另一个字段的迁移报告", "source": "quality://other-field/latest",
        "observed_at": "2099-01-01T00:00:00+00:00",
        "metadata": {"target": "dcp_user.email", "completeness": 1.0, "unmatched_rows": 0},
    })

    decision = evaluate_retirement_gate(db, "decision-demo", project_id, target_id)
    quality_check = next(item for item in decision["checks"] if item["key"] == "migration-quality")

    assert quality_check["actual"] == 0.9987
    assert quality_check["state"] == "failed"


def test_repeated_bootstrap_does_not_add_project_data(db):
    project_id, _ = _demo(db)
    before = {
        "builds": db.fetchone("SELECT COUNT(*) n FROM ontology_builds WHERE project_id=?", (project_id,))["n"],
        "elements": db.fetchone("SELECT COUNT(*) n FROM ontology_elements WHERE project_id=?", (project_id,))["n"],
        "evidence": db.fetchone("SELECT COUNT(*) n FROM governance_evidence WHERE project_id=?", (project_id,))["n"],
    }

    seed_challenge_cup_demo(db, "decision-demo")
    after = {
        "builds": db.fetchone("SELECT COUNT(*) n FROM ontology_builds WHERE project_id=?", (project_id,))["n"],
        "elements": db.fetchone("SELECT COUNT(*) n FROM ontology_elements WHERE project_id=?", (project_id,))["n"],
        "evidence": db.fetchone("SELECT COUNT(*) n FROM governance_evidence WHERE project_id=?", (project_id,))["n"],
    }

    assert after == before


def test_pending_generated_assertions_are_used_when_requested(db):
    project_id, target_id = _demo(db)
    db.execute("UPDATE governance_assertions SET review_status='rejected' WHERE project_id=? AND predicate IN ('READS','USES')", (project_id,))
    source = db.fetchone(
        "SELECT id FROM ontology_elements WHERE project_id=? AND kind='BusinessObject' LIMIT 1", (project_id,)
    )
    create_assertion(db, "decision-demo", project_id, {
        "subject_id": source["id"], "predicate": "READS", "object_id": target_id,
        "assertion_type": "inferred", "confidence": .8, "review_status": "pending",
        "generated_by": "longcat:test", "evidence_ids": [],
    })
    for row in db.fetchall("SELECT id,metadata FROM governance_evidence WHERE project_id=?", (project_id,)):
        metadata = json.loads(row["metadata"])
        if "consumer_count" in metadata:
            metadata["consumer_count"] = 0
            db.execute("UPDATE governance_evidence SET metadata=? WHERE id=?", (json.dumps(metadata), row["id"]))
    db.commit()

    using_pending = evaluate_retirement_gate(db, "decision-demo", project_id, target_id, include_pending=True)
    accepted_only = evaluate_retirement_gate(db, "decision-demo", project_id, target_id, include_pending=False)
    pending_check = next(item for item in using_pending["checks"] if item["key"] == "active-consumers")
    accepted_check = next(item for item in accepted_only["checks"] if item["key"] == "active-consumers")

    assert pending_check["actual"] == 1
    assert pending_check["state"] == "failed"
    assert accepted_check["actual"] == 0
    assert accepted_check["state"] == "passed"


def test_generated_assertion_on_equivalent_imported_field_is_used(db):
    project_id, target_id = _demo(db)
    imported_target = upsert_element(db, "decision-demo", project_id, {
        "layer": "M1", "profile": "software", "kind": "Column", "name": "DcpUserMobileColumn",
        "uri": "urn:ontorefactor:column:iam:dcp_user:mobile", "attributes": {"sensitive": True},
    })
    imported_cipher = upsert_element(db, "decision-demo", project_id, {
        "layer": "M1", "profile": "software", "kind": "Column", "name": "DcpUserMobileCipherColumn",
        "uri": "urn:ontorefactor:column:iam:dcp_user:mobile_cipher", "attributes": {"sensitive": True},
    })
    generated = create_assertion(db, "decision-demo", project_id, {
        "subject_id": imported_target["id"], "predicate": "MIRRORS_TO", "object_id": imported_cipher["id"],
        "assertion_type": "inferred", "confidence": .8, "review_status": "pending",
        "generated_by": "longcat:test", "evidence_ids": [],
    })

    decision = evaluate_retirement_gate(db, "decision-demo", project_id, target_id, include_pending=True)
    security_check = next(item for item in decision["checks"] if item["key"] == "encrypted-replacement")

    assert imported_target["id"] in decision["resolved_target_ids"]
    assert generated["id"] in security_check["assertion_ids"]


def test_agent_reads_bounded_project_context_before_generating_assertions(db):
    project_id, _ = _demo(db)
    context = collect_project_context(db, "decision-demo", project_id, target_hint="dcp_user.mobile")
    assert context["scope"]["included_evidence"] == context["scope"]["total_evidence"] == 9
    assert context["scope"]["included_assets"] <= 80

    result = execute_governance_agent(db, "decision-demo", project_id, {
        "source_type": "ddl", "source_name": "live.sql", "semantic_mode": "deterministic",
        "decision_target": "dcp_user.mobile",
        "content": "CREATE TABLE iam.dcp_user (id BIGINT PRIMARY KEY, mobile VARCHAR(32));",
    })

    assert result["decision"]["status"] == "BLOCKED"
    assert result["project_context"]["scope"]["total_evidence"] == 9
    run = db.fetchone("SELECT context_hash,context_scope,decision_id FROM governance_agent_runs WHERE id=?", (result["run_id"],))
    assert run["context_hash"]
    assert run["decision_id"] == result["decision"]["id"]


def test_project_context_redacts_credentials_before_model_use(db):
    project_id, _ = _demo(db)
    create_evidence(db, "decision-demo", project_id, {
        "kind": "SourceArtifact", "uri": "evidence://secret/redaction-test",
        "title": "连接器配置", "source": "connector://test",
        "content": "Authorization: Bearer abcdefghijklmnopqrstuvwxyz and ak_1234567890abcdefghijkl",
        "metadata": {"apiKey": "ak_should_not_escape_123456", "credential": "private"},
    })

    context = collect_project_context(db, "decision-demo", project_id, target_hint="dcp_user.mobile")
    serialized = json.dumps(context, ensure_ascii=False)

    assert "ak_should_not_escape" not in serialized
    assert "abcdefghijklmnopqrstuvwxyz" not in serialized
    assert "[REDACTED]" in serialized


def test_longcat_can_generate_cross_source_assertion_from_project_evidence(db, monkeypatch):
    project_id, target_id = _demo(db)
    module = db.fetchone(
        "SELECT id FROM ontology_elements WHERE project_id=? AND kind='Module' LIMIT 1", (project_id,)
    )
    captured = {}

    def fake_chat(provider, messages):
        captured["prompt"] = messages[-1]["content"]
        provider.last_metrics = {
            "provider": "longcat", "model": "LongCat-2.0", "prompt_version": PROMPT_VERSION,
            "input_tokens": 50, "output_tokens": 20, "latency_ms": 8,
            "request_hash": "request", "response_hash": "response",
        }
        return json.dumps({
            "business_objects": [], "owners": [], "classifications": [], "quality_rules": [], "issues": [],
            "relationships": [{
                "subject_asset_id": f"existing:{module['id']}", "predicate": "USES",
                "object_asset_id": f"existing:{target_id}", "confidence": .8,
                "reason": "项目证据显示模块仍使用手机号字段",
            }],
        })

    monkeypatch.setenv("LONGCAT_API_KEY", "test-key")
    monkeypatch.setattr(LongCatProvider, "chat", fake_chat)
    result = execute_governance_agent(db, "decision-demo", project_id, {
        "source_type": "ddl", "source_name": "change.sql", "semantic_mode": "llm",
        "decision_target": "dcp_user.mobile", "instruction": "判断字段能否退役",
        "content": "CREATE TABLE change_request (id BIGINT PRIMARY KEY);",
    })

    assert "手机号明文与密文双跑比对" in captured["prompt"]
    generated = db.fetchone(
        "SELECT * FROM governance_assertions WHERE project_id=? AND subject_id=? AND predicate='USES' AND object_id=?",
        (project_id, module["id"], target_id),
    )
    assert generated["review_status"] == "pending"
    assert generated["generated_by"].startswith("longcat:")
    active_check = next(item for item in result["decision"]["checks"] if item["key"] == "active-consumers")
    assert generated["id"] in active_check["assertion_ids"]