ontorefactor-governance/tests/test_builds.py

import json
from pathlib import Path

from ontorefactor_governance.builds import (
    list_ontology_versions,
    publish_ontology_build,
    run_ontology_build,
)
from ontorefactor_governance.db import Database
from ontorefactor_governance.llm import LongCatProvider, PROMPT_VERSION
from ontorefactor_governance.service import graph, overview, seed_challenge_cup_demo


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 test_empty_demo_build_review_publish_and_decide(monkeypatch):
    monkeypatch.delenv("AGENTPAAS_BASE_URL", raising=False)
    db = Database("sqlite:///:memory:")
    seeded = seed_challenge_cup_demo(db, "build-test")
    project_id = seeded["project"]["id"]

    assert seeded["overview"]["totals"]["elements"] == 0
    assert seeded["overview"]["totals"]["assertions"] == 0
    assert seeded["overview"]["demo"]["decision_code"] == "NOT_BUILT"

    build = run_ontology_build(
        db, "build-test", project_id, demo_sources(),
        semantic_mode="deterministic",
        instruction="从上传证据构建候选本体",
        environment="production",
        decision_target="dcp_user.mobile",
    )

    assert build["status"] == "review_required"
    assert build["agents"] == [
        "asset-construction-agent", "semantic-fusion-agent", "ontology-validation-agent",
    ]
    assert build["diff"]["new_elements"] > 0
    assert build["diff"]["new_assertions"] > 0
    assert build["diff"]["candidate_rules"] == 5
    assert build["candidate_validation"]["valid"] is True
    assert all(item["status"] == "candidate" for item in db.fetchall(
        "SELECT status FROM ontology_elements WHERE project_id=?", (project_id,)
    ))
    assert all(item["review_status"] == "pending" for item in db.fetchall(
        "SELECT review_status FROM governance_assertions WHERE project_id=?", (project_id,)
    ))
    assert not db.fetchall("SELECT id FROM governance_rules WHERE project_id=?", (project_id,))

    published = publish_ontology_build(
        db, "build-test", project_id, build["build_id"],
        reviewer="test-reviewer",
        accept_semantic_assertions=False,
        decision_target="dcp_user.mobile",
    )

    assert published["version"]["label"] == "v1"
    assert published["decision"]["status"] == "BLOCKED"
    assert published["decision"]["summary"] == {"passed": 1, "failed": 4, "unknown": 0, "total": 5}
    assert published["published"]["accepted_assertions"] > 0
    assert published["published"]["pending_assertions"] > 0
    assert len(list_ontology_versions(db, "build-test", project_id)) == 1
    assert overview(db, "build-test", project_id)["demo"]["decision_code"] == "BLOCKED"
    db.close()


def test_second_build_reports_matches_instead_of_reseeding():
    db = Database("sqlite:///:memory:")
    project_id = seed_challenge_cup_demo(db, "repeat-build")["project"]["id"]
    first = run_ontology_build(
        db, "repeat-build", project_id, demo_sources(), semantic_mode="deterministic",
    )
    publish_ontology_build(db, "repeat-build", project_id, first["build_id"], reviewer="reviewer")

    second = run_ontology_build(
        db, "repeat-build", project_id, demo_sources(), semantic_mode="deterministic",
    )

    assert second["diff"]["new_elements"] == 0
    assert second["diff"]["matched_elements"] == second["summary"]["elements"]
    assert second["diff"]["matched_assertions"] > 0
    db.close()


def test_different_input_constructs_a_different_ontology():
    db = Database("sqlite:///:memory:")
    project_id = seed_challenge_cup_demo(db, "custom-build")["project"]["id"]

    result = run_ontology_build(db, "custom-build", project_id, [{
        "source_name": "factory.sql",
        "source_type": "ddl",
        "content": "CREATE TABLE mes.machine_alarm (alarm_id BIGINT PRIMARY KEY, machine_id BIGINT NOT NULL, occurred_at TIMESTAMP);",
    }], semantic_mode="deterministic")

    uris = {item["uri"] for item in result["diff"]["preview_elements"]}
    assert any("machine_alarm" in uri for uri in uris)
    assert not any("dcp_user" in uri for uri in uris)
    assert result["diff"]["candidate_rules"] == 0
    db.close()


def test_build_uses_stable_m3_and_assembles_only_required_m2_types():
    db = Database("sqlite:///:memory:")
    project_id = seed_challenge_cup_demo(db, "catalog-build")["project"]["id"]

    result = run_ontology_build(db, "catalog-build", project_id, [{
        "source_name": "orders.sql",
        "source_type": "ddl",
        "content": "CREATE TABLE sales.orders (order_id BIGINT PRIMARY KEY, amount DECIMAL(12,2));",
    }], semantic_mode="deterministic")

    elements = db.fetchall(
        "SELECT id,layer,profile,name,source FROM ontology_elements WHERE project_id=?",
        (project_id,),
    )
    assert {item["name"] for item in elements if item["layer"] == "M3"} == {
        "MetaElement", "MetaEntityType", "MetaRelationType", "MetaAttributeType",
        "MetaConstraintType", "MetaAssertionType", "MetaProfile", "MetaView",
    }
    m2_names = {item["name"] for item in elements if item["layer"] == "M2"}
    assert m2_names == {"Table", "Column", "BusinessObject"}
    assert "Owner" not in m2_names
    assert all(item["source"] == "ontology/m3/meta-core.ttl" for item in elements if item["layer"] == "M3")
    assert all(item["source"].startswith("ontology/m2/") for item in elements if item["layer"] == "M2")

    adjacent_types = db.fetchall(
        "SELECT s.layer subject_layer,o.layer object_layer,a.generated_by "
        "FROM governance_assertions a JOIN ontology_elements s ON s.id=a.subject_id "
        "JOIN ontology_elements o ON o.id=a.object_id "
        "WHERE a.project_id=? AND a.predicate='INSTANCE_OF' AND s.layer IN ('M1','M2')",
        (project_id,),
    )
    assert adjacent_types
    assert all(
        (item["subject_layer"], item["object_layer"]) in {("M1", "M2"), ("M2", "M3")}
        for item in adjacent_types
    )
    assert result["summary"]["elements"] == len(elements)
    db.close()


def test_graph_returns_node_audit_and_evidence_details():
    db = Database("sqlite:///:memory:")
    project_id = seed_challenge_cup_demo(db, "graph-detail")["project"]["id"]
    run_ontology_build(db, "graph-detail", project_id, [{
        "source_name": "customer.sql",
        "source_type": "ddl",
        "content": "CREATE TABLE crm.customer (customer_id BIGINT PRIMARY KEY, mobile VARCHAR(32));",
    }], semantic_mode="deterministic")

    data = graph(db, "graph-detail", project_id)
    physical = next(item for item in data["nodes"] if item["kind"] == "PhysicalTable")
    meta = next(item for item in data["nodes"] if item["name"] == "MetaEntityType")
    assert physical["source_files"]
    assert physical["evidence_ids"]
    assert physical["confidence"] == 1.0
    assert physical["review_status"] == "pending"
    assert data["evidence"]
    assert meta["source"] == "ontology/m3/meta-core.ttl"
    assert data["semantics"]["node_status"]["candidate"] == "dashed"
    assert data["semantics"]["edge_status"]["accepted"] == "solid"
    db.close()


def test_semantic_agent_receives_current_upload_evidence(monkeypatch):
    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": 40, "output_tokens": 10, "latency_ms": 5,
            "request_hash": "request", "response_hash": "response",
        }
        return json.dumps({
            "business_objects": [], "owners": [], "classifications": [],
            "quality_rules": [], "relationships": [], "issues": [],
        })

    monkeypatch.setenv("LONGCAT_API_KEY", "test-key")
    monkeypatch.setattr(LongCatProvider, "chat", fake_chat)
    db = Database("sqlite:///:memory:")
    project_id = seed_challenge_cup_demo(db, "llm-build")["project"]["id"]

    result = run_ontology_build(
        db, "llm-build", project_id, demo_sources(), semantic_mode="llm",
        decision_target="dcp_user.mobile",
    )

    assert result["semantic_mode"] == "llm"
    assert "手机号明文与密文双跑比对" in captured["prompt"]
    assert "consumer_count" in captured["prompt"]
    assert "生产消费者必须归零" in captured["prompt"]
    db.close()