PMFlow/backend/tests/test_inbox_queue_view.py

import unittest
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock

from app.models.feishu_item import FeishuItem
from app.services.feishu_signal_pipeline import FOLLOWUP_STATUS_RE, FeishuSignalPipelineService


class InboxQueueViewTest(unittest.TestCase):
    def test_has_processed_message_ignores_ingress_trace_only(self) -> None:
        service = FeishuSignalPipelineService(db=SimpleNamespace(), tenant_id=1)  # type: ignore[arg-type]

        async def fake_execute(stmt):
            sql = str(stmt)
            if "FROM feishu_items" in sql:
                return SimpleNamespace(scalar_one_or_none=lambda: None)
            self.assertIn("ai_processing_logs.level IS NULL OR ai_processing_logs.level !=", sql)
            self.assertIn(
                "ai_processing_logs.routing_decision IS NULL OR ai_processing_logs.routing_decision !=",
                sql,
            )
            return SimpleNamespace(scalar_one_or_none=lambda: None)

        service.db.execute = AsyncMock(side_effect=fake_execute)  # type: ignore[attr-defined]

        exists = asyncio.run(service._has_processed_message("om_ingress_only"))

        self.assertFalse(exists)

    def test_dedupe_queue_items_aggregates_occurrence_count(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]
        first = FeishuItem(
            tenant_id=1,
            raw_message="接入 GitHub 信号源",
            level="signal",
            card_type="signal",
            route_status="pending",
            dedupe_key="signal:3:group:external_integration",
            occurrence_count=2,
        )
        second = FeishuItem(
            tenant_id=1,
            raw_message="接入 GitHub 信号源",
            level="signal",
            card_type="signal",
            route_status="pending",
            dedupe_key="signal:3:group:external_integration",
            occurrence_count=1,
        )

        rows = service._dedupe_queue_items([first, second])

        self.assertEqual(len(rows), 1)
        self.assertEqual(int(rows[0].occurrence_count), 3)

    def test_resolve_message_class_falls_back_from_route_status(self) -> None:
        pending_item = FeishuItem(
            tenant_id=1,
            raw_message="同步 GitHub issue",
            level="signal",
            card_type="signal",
            route_status="pending",
        )
        auto_done_item = FeishuItem(
            tenant_id=1,
            raw_message="同步 GitHub issue",
            level="signal",
            card_type="signal",
            route_status="auto_processed",
        )

        self.assertEqual(FeishuSignalPipelineService._resolve_message_class(pending_item), "action_required")
        self.assertEqual(FeishuSignalPipelineService._resolve_message_class(auto_done_item), "auto_done")

    def test_resolve_message_class_supports_new_status_machine(self) -> None:
        queued_item = FeishuItem(
            tenant_id=1,
            raw_message="新增一个项目",
            level="project",
            card_type="new_project_candidate",
            route_status="queued_for_decision",
        )
        undone_item = FeishuItem(
            tenant_id=1,
            raw_message="新增一个项目",
            level="command",
            card_type="project_create",
            route_status="undone",
        )

        self.assertEqual(FeishuSignalPipelineService._resolve_message_class(queued_item), "action_required")
        self.assertEqual(FeishuSignalPipelineService._resolve_message_class(undone_item), "auto_done")

    def test_extract_project_intent_fields_requires_title_description_and_scope(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]

        fields = service._extract_project_intent_fields("新增一个项目，名称叫飞书协同中台，用于统一多群消息分发，范围是 PMFlow 主数据入库。")

        self.assertTrue(fields["ready"])
        self.assertEqual(fields["title"], "飞书协同中台")
        self.assertIn("统一多群消息分发", fields["description"])
        self.assertIn("PMFlow 主数据入库", fields["project_scope"])

    def test_resolve_intent_gate_recognizes_explicit_project_create(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]

        gate = asyncio.run(
            service._resolve_intent_gate(
                message="不对，是新增进 PMFlow 的项目列表里面，名称叫飞书协同中台，用于统一多群消息分发，范围是 PMFlow 主数据入库。",
                inferred_project_id=None,
            )
        )

        self.assertEqual(gate["intent"], "create_project")
        self.assertEqual(gate["canonical_action"], "project.create")

    def test_resolve_intent_gate_uses_context_prefill_for_project_create(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]
        service._maybe_prefill_project_fields_from_context = AsyncMock(  # type: ignore[method-assign]
            return_value={
                "title": "医药十五五方向研究",
                "description": "基于最近一轮医药方向问答整理投资与立项建议。",
                "project_scope": "围绕创新药、医疗器械、公共卫生与老龄化方向做机会梳理。",
                "missing_fields": [],
                "ready": True,
            }
        )

        gate = asyncio.run(
            service._resolve_intent_gate(
                message="汇总以上内容给我创建个项目在项目列表",
                inferred_project_id=None,
                source_chat_id="oc_test",
                source_sender_open_id="ou_test",
                sender="林宇航",
            )
        )

        self.assertEqual(gate["intent"], "create_project")
        self.assertEqual(gate["fields"]["title"], "医药十五五方向研究")

    def test_resolve_intent_gate_uses_embedded_message_context_for_project_create(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]

        gate = asyncio.run(
            service._resolve_intent_gate(
                message=(
                    '目前"十五五"规划（2026-2030）还在制定中，但从已有信号来看，医药行业这几个方向值得关注：'
                    " 创新药会是核心 中医药传承创新 应对老龄化 供应链安全 出海国际化"
                    " 根据这个汇总一下，给我创建一个项目在项目列表"
                ),
                inferred_project_id=None,
            )
        )

        self.assertEqual(gate["intent"], "create_project")
        self.assertEqual(gate["fields"]["title"], "十五五医药方向研究")
        self.assertIn("创新药会是核心", gate["fields"]["description"])

    def test_resolve_intent_gate_blocks_project_create_when_recent_context_is_failure_text(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]
        service._load_recent_chat_turns = AsyncMock(  # type: ignore[method-assign]
            return_value=[
                {
                    "question": "目前十五五规划里，医药行业值得关注哪些方向？",
                    "answer": "我暂时无法生成可靠回答（模型服务异常）。为避免误导，本次不直接给结论；请稍后重试。",
                }
            ]
        )
        service._load_recent_chat_reply_fallback_text = AsyncMock(return_value="")  # type: ignore[method-assign]

        gate = asyncio.run(
            service._resolve_intent_gate(
                message="根据上面汇总一下，给我创建一个项目在项目列表",
                inferred_project_id=None,
                source_chat_id="oc_test",
                source_sender_open_id="ou_test",
                sender="林宇航",
            )
        )

        self.assertEqual(gate["intent"], "clarification_required")
        self.assertIn("title", gate["fields"]["missing_fields"])

    def test_followup_continue_accepts_project_reference_message(self) -> None:
        self.assertTrue(
            FeishuSignalPipelineService._looks_like_continue_followup("你应该继续创建我刚才要求的项目")
        )
        self.assertFalse(
            FeishuSignalPipelineService._looks_like_continue_followup("继续创建一个项目，名称叫飞书协同中台")
        )

    def test_followup_status_regex_matches_creation_question(self) -> None:
        self.assertTrue(bool(FOLLOWUP_STATUS_RE.search("是否已经创建")))
        self.assertTrue(bool(FOLLOWUP_STATUS_RE.search("创建了吗")))

    def test_select_followup_status_target_prefers_project_create(self) -> None:
        source_item = FeishuItem(
            tenant_id=1,
            raw_message="新增一个Web搜索信号源",
            level="command",
            card_type="command",
            route_status="auto_done",
            result_payload={"operation_code": "source.connect"},
        )
        project_item = FeishuItem(
            tenant_id=1,
            raw_message="给我创建一个项目在项目列表",
            level="command",
            card_type="project_create",
            route_status="auto_done",
            result_payload={"operation_code": "project.create"},
        )

        target = FeishuSignalPipelineService._select_followup_status_target(
            message="是否已经创建",
            recent_items=[source_item, project_item],
        )

        self.assertIs(target, project_item)

    def test_resolve_project_candidate_fields_blocks_placeholder_project(self) -> None:
        service = FeishuSignalPipelineService(db=None, tenant_id=1)  # type: ignore[arg-type]
        item = FeishuItem(
            tenant_id=1,
            raw_message="给我创建一个项目在项目列表",
            level="project",
            card_type="new_project_candidate",
            route_status="queued_for_decision",
            result_payload={
                "operation_code": "project.create",
                "operation_payload": {
                    "title": "",
                    "description": "",
                },
                "project_scope": "",
            },
            blocking_reason="未执行：当前建项目请求还缺少 项目名称。",
        )

        fields = service._resolve_project_candidate_fields(item=item, suggestion=None)

        self.assertFalse(fields["ready"])
        self.assertIn("title", fields["missing_fields"])

    def test_create_project_gate_missing_fields_only_returns_clarification(self) -> None:
        service = FeishuSignalPipelineService(db=SimpleNamespace(), tenant_id=1)  # type: ignore[arg-type]
        service._resolve_command_operator_user = AsyncMock(return_value={"id": 1})  # type: ignore[method-assign]
        service._can_execute_high_value_action = lambda _operator: True  # type: ignore[method-assign]
        service._queue_project_intent_item = AsyncMock()  # type: ignore[method-assign]
        service._write_processing_log = AsyncMock()  # type: ignore[method-assign]
        service._execute_controlled_operation = AsyncMock(  # type: ignore[method-assign]
            side_effect=AssertionError("executor should not run")
        )

        result = asyncio.run(
            service._handle_create_project_gate(
                message="汇总以上内容给我创建个项目在项目列表",
                sender="林宇航",
                source_chat_id="oc_test",
                source_sender_open_id="ou_test",
                feishu_message_id="om_test",
                captured_at=None,
                intent_gate={
                    "intent": "clarification_required",
                    "canonical_action": "project.create",
                    "fields": {
                        "title": "",
                        "description": "汇总以上内容给我创建个项目在项目列表",
                        "project_scope": "",
                        "missing_fields": ["title"],
                        "ready": False,
                    },
                    "clarification_template": "未执行：当前建项目请求还缺少 项目名称。",
                },
                process_started_at=0.0,
            )
        )

        self.assertEqual(result["intent"], "clarification_required")
        self.assertEqual(result["direct_reply_text"], "未执行：当前建项目请求还缺少 项目名称。")
        self.assertEqual(result["results"], [])
        service._queue_project_intent_item.assert_not_awaited()
        service._execute_controlled_operation.assert_not_awaited()

    def test_create_task_gate_missing_fields_only_returns_clarification(self) -> None:
        service = FeishuSignalPipelineService(db=SimpleNamespace(), tenant_id=1)  # type: ignore[arg-type]
        service._resolve_command_operator_user = AsyncMock(return_value={"id": 1})  # type: ignore[method-assign]
        service._can_execute_high_value_action = lambda _operator: True  # type: ignore[method-assign]
        service._queue_command_intent_item = AsyncMock()  # type: ignore[method-assign]
        service._write_processing_log = AsyncMock()  # type: ignore[method-assign]
        service._execute_controlled_operation = AsyncMock(  # type: ignore[method-assign]
            side_effect=AssertionError("executor should not run")
        )

        result = asyncio.run(
            service._handle_create_task_gate(
                message="帮我建个任务",
                sender="林宇航",
                source_chat_id="oc_test",
                source_sender_open_id="ou_test",
                feishu_message_id="om_test",
                captured_at=None,
                intent_gate={
                    "intent": "clarification_required",
                    "canonical_action": "task.create",
                    "fields": {
                        "project_id": None,
                        "title": "",
                        "description": "",
                        "missing_fields": ["project_id", "title"],
                        "ready": False,
                    },
                    "clarification_template": "未执行：当前建任务请求还缺少 归属项目、任务标题。",
                },
                process_started_at=0.0,
            )
        )

        self.assertEqual(result["intent"], "clarification_required")
        self.assertEqual(result["direct_reply_text"], "未执行：当前建任务请求还缺少 归属项目、任务标题。")
        self.assertEqual(result["results"], [])
        service._queue_command_intent_item.assert_not_awaited()
        service._execute_controlled_operation.assert_not_awaited()

    def test_should_hide_pending_item_from_inbox_for_query_and_missing_fields(self) -> None:
        lookup_item = FeishuItem(
            tenant_id=1,
            raw_message="现在有哪些项目",
            level="signal",
            card_type="signal",
            route_status="pending",
        )
        clarification_item = FeishuItem(
            tenant_id=1,
            raw_message="汇总以上内容给我创建个项目在项目列表",
            level="project",
            card_type="new_project_candidate",
            route_status="queued_for_decision",
            decision_reason="missing_required_fields",
        )
        status_item = FeishuItem(
            tenant_id=1,
            raw_message="你好 长连接时都已通",
            level="signal",
            card_type="signal",
            route_status="pending",
        )

        self.assertTrue(FeishuSignalPipelineService._should_hide_pending_item_from_inbox(lookup_item))
        self.assertTrue(FeishuSignalPipelineService._should_hide_pending_item_from_inbox(clarification_item))
        self.assertTrue(FeishuSignalPipelineService._should_hide_pending_item_from_inbox(status_item))

    def test_build_project_preview_prefers_structured_title_and_detail(self) -> None:
        title = FeishuSignalPipelineService._build_project_preview_title(
            title="",
            description="基于最近一轮医药方向问答整理投资与立项建议。",
            project_scope="围绕创新药、医疗器械、公共卫生与老龄化方向做机会梳理。",
        )
        summary = FeishuSignalPipelineService._build_project_preview_summary(
            description="基于最近一轮医药方向问答整理投资与立项建议。",
            project_scope="围绕创新药、医疗器械、公共卫生与老龄化方向做机会梳理。",
            blocking_reason="未执行：当前建项目请求还缺少 项目名称。",
            fallback_text="汇总以上内容给我创建个项目在项目列表",
        )

        self.assertEqual(title, "围绕创新药")
        self.assertIn("投资与立项建议", summary)