PMFlow/backend/tests/test_ai_logs_api.py

import os
import unittest
from tempfile import NamedTemporaryFile

from sqlalchemy import and_, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.api.ai_logs import _apply_project_filter
from app.models import AIProcessingLog, Base, IngestEvent, Project, User, WorkItem


class AILogsApiTest(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        tmp = NamedTemporaryFile(prefix="pmflow-ai-logs-test-", suffix=".db", delete=False)
        tmp.close()
        self.db_path = tmp.name
        self.engine = create_async_engine(f"sqlite+aiosqlite:///{self.db_path}", future=True)
        self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False, class_=AsyncSession)
        async with self.engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

        async with self.session_factory() as db:
            user = User(
                tenant_id=1,
                username="boss",
                email="boss@example.com",
                password_hash="hash",
                name="老板",
                role="BOSS",
                assigned_domains=[],
                workload=0,
                status="active",
                is_active=True,
            )
            db.add(user)
            await db.flush()

            project_a = Project(
                tenant_id=1,
                name="项目A",
                title="项目A",
                description="A",
                owner_id=int(user.id),
                status="active",
                domain_id=None,
                assigned_role="pm",
                signal_count=0,
            )
            project_b = Project(
                tenant_id=1,
                name="项目B",
                title="项目B",
                description="B",
                owner_id=int(user.id),
                status="active",
                domain_id=None,
                assigned_role="pm",
                signal_count=0,
            )
            db.add_all([project_a, project_b])
            await db.flush()

            ingest_a = IngestEvent(
                tenant_id=1,
                channel="feishu_message",
                event_key="om-project-a",
                status="done",
                payload={},
                retry_count=0,
                error_message=None,
                processed_at=None,
            )
            ingest_b = IngestEvent(
                tenant_id=1,
                channel="feishu_message",
                event_key="om-project-b",
                status="done",
                payload={},
                retry_count=0,
                error_message=None,
                processed_at=None,
            )
            db.add_all([ingest_a, ingest_b])
            await db.flush()

            db.add_all(
                [
                    WorkItem(
                        tenant_id=1,
                        ingest_event_id=int(ingest_a.id),
                        source_system="feishu",
                        source_message_id="om-project-a",
                        state="succeeded",
                        intent="project.create",
                        decision_level="project",
                        project_id=int(project_a.id),
                    ),
                    WorkItem(
                        tenant_id=1,
                        ingest_event_id=int(ingest_b.id),
                        source_system="feishu",
                        source_message_id="om-project-b",
                        state="succeeded",
                        intent="signal.create",
                        decision_level="project",
                        project_id=int(project_b.id),
                    ),
                ]
            )
            db.add_all(
                [
                    AIProcessingLog(
                        tenant_id=1,
                        feishu_message_id="om-project-a",
                        feishu_item_id=None,
                        raw_message="创建项目A",
                        ai_raw_output="{}",
                        project_score=None,
                        signal_confidence=None,
                        level="ai_pmo",
                        normalized_text="创建项目A",
                        route="ai_pmo.project.create",
                        intent="project.create",
                        executor="ai_pmo_kernel",
                        operation_code="project.create",
                        operation_result={"project_id": int(project_a.id)},
                        reply_source="action_succeeded",
                        routing_decision="ai_pmo.project.create",
                        routing_reason="AI PMO kernel route=ai_pmo.project.create",
                        reasoning_bullets=["work_item_id=1"],
                        processing_status="success",
                        error_message=None,
                        duration_ms=100,
                    ),
                    AIProcessingLog(
                        tenant_id=1,
                        feishu_message_id="om-project-b",
                        feishu_item_id=None,
                        raw_message="创建信号B",
                        ai_raw_output="{}",
                        project_score=None,
                        signal_confidence=None,
                        level="ai_pmo",
                        normalized_text="创建信号B",
                        route="ai_pmo.signal.create",
                        intent="signal.create",
                        executor="ai_pmo_kernel",
                        operation_code="signal.enter",
                        operation_result={"project_id": int(project_b.id)},
                        reply_source="action_succeeded",
                        routing_decision="ai_pmo.signal.create",
                        routing_reason="AI PMO kernel route=ai_pmo.signal.create",
                        reasoning_bullets=["work_item_id=2"],
                        processing_status="success",
                        error_message=None,
                        duration_ms=100,
                    ),
                ]
            )
            await db.commit()

    async def asyncTearDown(self) -> None:
        await self.engine.dispose()
        if getattr(self, "db_path", None):
            try:
                os.remove(self.db_path)
            except FileNotFoundError:
                pass

    async def test_apply_project_filter_includes_ai_pmo_work_item_message_ids(self) -> None:
        async with self.session_factory() as db:
            stmt = select(AIProcessingLog).where(AIProcessingLog.tenant_id == 1).order_by(AIProcessingLog.id.asc())
            rows = (await db.execute(_apply_project_filter(stmt, tenant_id=1, project_id=1))).scalars().all()

        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0].feishu_message_id, "om-project-a")

    async def test_apply_project_filter_keeps_other_project_logs_out(self) -> None:
        async with self.session_factory() as db:
            stmt = select(AIProcessingLog).where(AIProcessingLog.tenant_id == 1).order_by(AIProcessingLog.id.asc())
            rows = (await db.execute(_apply_project_filter(stmt, tenant_id=1, project_id=2))).scalars().all()

        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0].feishu_message_id, "om-project-b")