ontorefactor-governance/tests/test_api.py

import json

from fastapi.testclient import TestClient

from ontorefactor_governance import api
from ontorefactor_governance.db import Database


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, "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()
    assert payload["overview"]["demo"]["decision"] == "暂缓删除"
    assert payload["overview"]["totals"]["open_issues"] == 4
    db.close()