PMFlow/backend/tests/test_semantic_fact_admin_api.py

from __future__ import annotations

from types import SimpleNamespace

import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy import JSON, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.api.semantic_facts_api import routes as semantic_fact_routes
from app.dependencies.auth import get_current_user, get_tenant_id
from app.core.database import get_db
from app.models.base import Base
from app.models.ingest_event import IngestEvent
from app.models.knowledge_base_doc import KnowledgeBaseDoc
from app.models.project import Project
from app.models.semantic_fact import SemanticFact
from app.models.semantic_fact_extraction_run import SemanticFactExtractionRun
from app.models.signal import Signal
from app.models.task import Task
from app.models.truth_node import TruthNode
from app.models.user import User


def _use_sqlite_json_types(*tables) -> None:
    for table in tables:
        for column in table.columns:
            if column.type.__class__.__name__ in {"JSONB", "ARRAY", "Vector"}:
                column.type = JSON()


@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    tables = [
        User.__table__,
        Project.__table__,
        Signal.__table__,
        TruthNode.__table__,
        Task.__table__,
        KnowledgeBaseDoc.__table__,
        SemanticFact.__table__,
        SemanticFactExtractionRun.__table__,
        IngestEvent.__table__,
    ]
    _use_sqlite_json_types(*tables)
    async with engine.begin() as conn:
        await conn.run_sync(lambda sync_conn: Base.metadata.create_all(sync_conn, tables=tables))
    maker = async_sessionmaker(engine, expire_on_commit=False)
    async with maker() as session:
        yield session
    await engine.dispose()


def _build_app(db_session, *, role: str) -> FastAPI:
    app = FastAPI()
    app.include_router(semantic_fact_routes.router)

    async def _fake_get_db():
        yield db_session

    app.dependency_overrides[get_db] = _fake_get_db
    app.dependency_overrides[get_tenant_id] = lambda: 1
    app.dependency_overrides[get_current_user] = lambda: SimpleNamespace(
        id=7,
        tenant_id=1,
        role=role,
        username="tester",
        email="tester@example.com",
        name="Tester",
    )
    return app


@pytest.mark.asyncio
async def test_semantic_fact_admin_api_requires_manager_role(db_session) -> None:
    app = _build_app(db_session, role="BUILDER")
    transport = ASGITransport(app=app)

    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/semantic-facts/admin/overview")

    assert response.status_code == 403


@pytest.mark.asyncio
async def test_semantic_fact_admin_api_exposes_overview_objects_detail_and_actions(db_session) -> None:
    project = Project(
        tenant_id=1,
        name="invoice-helper",
        title="开票助手",
        description="自动生成发票",
        owner_id=None,
        status="active",
        assigned_role="backend",
    )
    db_session.add(project)
    await db_session.flush()
    db_session.add(
        SemanticFact(
            tenant_id=1,
            entity_type="project",
            entity_id=int(project.id),
            fact_namespace="alias",
            fact_key="aliases",
            fact_value_text="开票系统",
            fact_value_norm="invoice_system",
            value_type="keyword",
            confidence=0.95,
            evidence_source_type="project",
            evidence_source_id=int(project.id),
            evidence_excerpt="自动生成发票",
            extract_version="semantic-facts-v1",
            freshness_at=None,
            is_active=True,
        )
    )
    db_session.add(
        SemanticFactExtractionRun(
            tenant_id=1,
            entity_type="project",
            entity_id=int(project.id),
            trigger_type="manual_retry",
            status="completed",
            input_fingerprint="fp-1",
            facts_written=1,
            failure_reason=None,
            extract_version="semantic-facts-v1",
            started_at="2026-04-15T10:00:00+00:00",
            finished_at="2026-04-15T10:00:02+00:00",
        )
    )
    await db_session.commit()

    app = _build_app(db_session, role="REVIEWER")
    transport = ASGITransport(app=app)

    async with AsyncClient(transport=transport, base_url="http://test") as client:
        overview_response = await client.get("/semantic-facts/admin/overview")
        objects_response = await client.get("/semantic-facts/admin/objects", params={"entity_type": "project", "q": "开票"})
        detail_response = await client.get(f"/semantic-facts/admin/objects/project/{int(project.id)}")
        retry_response = await client.post(f"/semantic-facts/admin/objects/project/{int(project.id)}/reextract")
        retry_event = (
            await db_session.execute(
                select(IngestEvent).where(
                    IngestEvent.tenant_id == 1,
                    IngestEvent.channel == "semantic_fact_extract",
                    IngestEvent.event_key == f"project:{int(project.id)}",
                )
            )
        ).scalar_one()
        retry_trigger_type = retry_event.payload["trigger_type"]
        backfill_response = await client.post(
            "/semantic-facts/admin/backfill",
            json={"entity_types": ["project"], "only_missing": False},
        )

    assert overview_response.status_code == 200
    assert overview_response.json()["code"] == 0
    assert overview_response.json()["data"]["queue"]["pending"] >= 1

    assert objects_response.status_code == 200
    assert objects_response.json()["code"] == 0
    assert objects_response.json()["data"]["items"][0]["display_name"] == "开票助手"

    assert detail_response.status_code == 200
    assert detail_response.json()["code"] == 0
    assert detail_response.json()["data"]["facts"][0]["fact_value_text"] == "开票系统"

    assert retry_response.status_code == 200
    assert retry_response.json()["code"] == 0
    assert retry_trigger_type == "manual_retry"

    assert backfill_response.status_code == 200
    assert backfill_response.json()["code"] == 0
    assert backfill_response.json()["data"]["enqueued"] >= 1