ontorefactor-governance/tests/test_api.py

import json
from pathlib import Path

from fastapi.testclient import TestClient

from ontorefactor_governance import api
from ontorefactor_governance.db import Database


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


def test_standalone_api_and_web_console(monkeypatch):
    db = Database("sqlite:///:memory:")
    monkeypatch.setattr(api, "get_database", lambda: db)
    client = TestClient(api.app)
    headers = {"X-Tenant-ID": "api-test"}

    assert client.get("/api/health").json()["ok"] is True
    page = client.get("/")
    assert page.status_code == 200
    assert "OntoRefactor" in page.text

    seeded = client.post("/api/v1/governance/bootstrap/tenant-isolation", headers=headers)
    assert seeded.status_code == 200
    project_id = seeded.json()["project"]["id"]

    projects = client.get("/api/v1/governance/projects", headers=headers).json()["projects"]
    assert len(projects) == 1
    assert projects[0]["id"] == project_id

    overview = client.get(f"/api/v1/governance/projects/{project_id}/overview", headers=headers).json()
    assert overview["totals"]["elements"] == 48
    assert overview["totals"]["open_issues"] == 0

    exported = client.get(f"/api/v1/governance/projects/{project_id}/export.jsonld", headers=headers)
    assert exported.status_code == 200
    assert exported.json()["@graph"]

    other_tenant = client.get(f"/api/v1/governance/projects/{project_id}/overview", headers={"X-Tenant-ID": "other"})
    assert other_tenant.status_code == 404
    db.close()


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

    response = client.post(f"/api/v1/governance/projects/{project_id}/agent-runs/analyze", headers=headers, json={
        "source_type": "ddl",
        "source_name": "orders.sql",
        "environment": "development",
        "content": "CREATE TABLE orders (id INTEGER PRIMARY KEY, tenant_id INTEGER NOT NULL);",
    })
    assert response.status_code == 200
    assert response.json()["persisted"]["elements"] == 7
    runs = client.get(f"/api/v1/governance/projects/{project_id}/agent-runs", headers=headers).json()["runs"]
    assert len(runs) == 1
    assert runs[0]["status"] == "completed"
    db.close()


def test_source_file_preview_and_upload_agent_endpoint(monkeypatch):
    db = Database("sqlite:///:memory:")
    monkeypatch.setattr(api, "get_database", lambda: db)
    client = TestClient(api.app)
    headers = {"X-Tenant-ID": "upload-api"}
    project_id = client.post("/api/v1/governance/projects", headers=headers, json={
        "name": "Uploaded governance", "namespace": "urn:api:uploaded"
    }).json()["id"]
    ddl = b"CREATE TABLE iam.customer (id BIGINT PRIMARY KEY, mobile VARCHAR(32));"

    preview = client.post(
        f"/api/v1/governance/projects/{project_id}/sources/preview",
        headers=headers,
        files={"file": ("customer.sql", ddl, "text/plain")},
        data={"source_type": "auto"},
    )
    assert preview.status_code == 200
    assert preview.json()["source_type"] == "ddl"
    assert preview.json()["detected"] == {
        "tables": 1, "columns": 2, "apis": 0, "schemas": 0, "assets": 0,
        "relationships": 0, "evidence": 0, "rules": 0, "sensitive_fields": ["mobile"],
    }
    assert preview.json()["content"].startswith("CREATE TABLE")

    analyzed = client.post(
        f"/api/v1/governance/projects/{project_id}/agent-runs/upload",
        headers=headers,
        files={"file": ("customer.sql", ddl, "text/plain")},
        data={"source_type": "auto", "semantic_mode": "deterministic", "instruction": "识别敏感字段"},
    )
    assert analyzed.status_code == 200
    assert analyzed.json()["status"] == "completed"
    assert analyzed.json()["persisted"]["elements"] > 0
    evidence = client.get(
        f"/api/v1/governance/projects/{project_id}/evidence", headers=headers
    ).json()["evidence"]
    source = next(item for item in evidence if item["kind"] == "SourceArtifact")
    assert source["content"].startswith("CREATE TABLE")
    assert source["metadata"]["upload"]["name"] == "customer.sql"
    db.close()


def test_llm_status_is_safe_and_forced_mode_requires_key(monkeypatch):
    monkeypatch.delenv("LONGCAT_API_KEY", raising=False)
    monkeypatch.setenv("ONTOREFACTOR_LLM_MODE", "auto")
    db = Database("sqlite:///:memory:")
    monkeypatch.setattr(api, "get_database", lambda: db)
    client = TestClient(api.app)
    headers = {"X-Tenant-ID": "llm-api"}

    status = client.get("/api/v1/governance/llm/status", headers=headers)
    assert status.status_code == 200
    assert status.json()["configured"] is False
    assert "api_key" not in json.dumps(status.json()).lower()

    project_id = client.post("/api/v1/governance/projects", headers=headers, json={
        "name": "LLM project", "namespace": "urn:api:llm"
    }).json()["id"]
    response = client.post(f"/api/v1/governance/projects/{project_id}/agent-runs/analyze", headers=headers, json={
        "source_type": "ddl", "semantic_mode": "llm",
        "content": "CREATE TABLE customer (id INTEGER PRIMARY KEY);",
    })
    assert response.status_code == 503
    assert response.json()["detail"]["code"] == "LLM_NOT_CONFIGURED"
    db.close()


def test_llm_connection_failure_is_a_structured_api_error(monkeypatch):
    def fail_connection():
        from ontorefactor_governance.service import GovernanceError
        raise GovernanceError("LongCat 鉴权失败，请检查 LONGCAT_API_KEY", code="LLM_AUTH_FAILED", status=502)

    monkeypatch.setattr(api, "test_llm_connection", fail_connection)
    client = TestClient(api.app)
    response = client.post("/api/v1/governance/llm/test", headers={"X-Tenant-ID": "llm-api"})

    assert response.status_code == 502
    assert response.json()["detail"]["code"] == "LLM_AUTH_FAILED"


def test_challenge_cup_demo_bootstrap_endpoint(monkeypatch):
    db = Database("sqlite:///:memory:")
    monkeypatch.setattr(api, "get_database", lambda: db)
    client = TestClient(api.app)
    headers = {"X-Tenant-ID": "challenge-demo"}

    response = client.post("/api/v1/governance/bootstrap/challenge-cup", headers=headers)

    assert response.status_code == 200
    payload = response.json()
    project_id = payload["project"]["id"]
    assert payload["overview"]["demo"]["decision"] == "尚未构建本体"
    assert payload["overview"]["totals"]["elements"] == 0
    assert payload["overview"]["totals"]["assertions"] == 0
    assert payload["overview"]["totals"]["evidence"] == 0

    names = [
        "01-dcp-user.sql", "02-user-profile-openapi.yaml",
        "03-dcp-asset-inventory.json", "04-governance-evidence.json",
    ]
    files = [("files", (name, (SAMPLES / name).read_bytes(), "application/octet-stream")) for name in names]
    built = client.post(
        f"/api/v1/governance/projects/{project_id}/ontology-builds/upload",
        headers=headers,
        files=files,
        data={"semantic_mode": "deterministic", "environment": "production", "decision_target": "dcp_user.mobile"},
    )
    assert built.status_code == 200
    build = built.json()
    assert build["status"] == "review_required"
    assert build["diff"]["new_elements"] > 0

    published = client.post(
        f"/api/v1/governance/projects/{project_id}/ontology-builds/{build['build_id']}/publish",
        headers=headers,
        json={"reviewer": "api-test", "accept_semantic_assertions": False, "decision_target": "dcp_user.mobile"},
    )
    assert published.status_code == 200
    assert published.json()["decision"]["status"] == "BLOCKED"
    assert published.json()["version"]["label"] == "v1"

    evaluated = client.post(
        f"/api/v1/governance/projects/{project_id}/decisions/retirement",
        headers=headers,
        json={"target": "dcp_user.mobile", "include_pending_assertions": True, "persist": True},
    )
    assert evaluated.status_code == 200
    assert evaluated.json()["status"] == "BLOCKED"
    assert evaluated.json()["summary"] == {"passed": 1, "failed": 4, "unknown": 0, "total": 5}

    answered = client.post(
        f"/api/v1/governance/projects/{project_id}/governance-agent/ask",
        headers=headers,
        json={"question": "dcp_user.mobile 是否可以迁移并删除？", "persist": True},
    )
    assert answered.status_code == 200
    assert answered.json()["agent"] == "governance-decision-agent"
    assert answered.json()["decision"]["status"] == "BLOCKED"
    assert answered.json()["resolved_target"]["id"]

    decisions = client.get(
        f"/api/v1/governance/projects/{project_id}/decisions", headers=headers
    ).json()["decisions"]
    assert len(decisions) >= 1
    assert decisions[0]["checks"]
    db.close()