import pytest from unittest.mock import AsyncMock from app.services.integrations.feishu_integration_service.mixin_receive_prepare import ( FeishuReceivePrepareMixin, ) from app.services.integrations.feishu_integration_service.service import FeishuIntegrationService @pytest.mark.asyncio async def test_handle_message_receive_ignores_bot_sender() -> None: service = FeishuIntegrationService(db=object()) result = await service._handle_message_receive( { "message": {"message_id": "m-bot"}, "sender": {"sender_type": "bot"}, "chat": {"chat_id": "c1"}, } ) assert result["code"] == 0 assert result["msg"] == "Ignore bot/app message" @pytest.mark.asyncio async def test_handle_message_receive_routes_empty_text_branch(monkeypatch) -> None: service = FeishuIntegrationService(db=object()) called: dict[str, object] = {} def _fake_build_receive_context(*, event, state_trace_id): # noqa: ARG001 return { "ingress_source": "feishu", "message": {"message_id": "m-empty", "message_type": "post"}, "sender": {"sender_name": "测试用户"}, "message_id": "m-empty", "sender_type": "user", "raw_message_content": "", "message_content": "", "chat_id": "chat-empty", "mentioned_bot_names": ["PMFlow"], "sender_open_id": "ou-empty", "perf_t0": None, } async def _fake_record_ingress_trace(**kwargs): called["trace"] = kwargs async def _fake_reply_for_empty_text_ingress(**kwargs): called["empty_reply"] = kwargs return True monkeypatch.setattr(service, "_build_receive_context", _fake_build_receive_context) monkeypatch.setattr(service, "_record_ingress_trace", _fake_record_ingress_trace) monkeypatch.setattr(service, "_reply_for_empty_text_ingress", _fake_reply_for_empty_text_ingress) result = await service._handle_message_receive({"type": "im.message.receive_v1"}) assert result["code"] == 0 assert result["msg"] == "No text content" assert called["trace"]["message_id"] == "m-empty" assert called["empty_reply"]["chat_id"] == "chat-empty" def test_classify_message_scene_does_not_infer_execute_from_raw_text_only() -> None: scene = FeishuIntegrationService._classify_message_scene( capture_result={ "status": "answered", "classification": "question", "intent": "question", }, message_content="帮我创建项目和信号源", ) assert scene == "qa" def test_classify_message_scene_uses_bundle_without_planner_classifier(monkeypatch: pytest.MonkeyPatch) -> None: _ = monkeypatch scene = FeishuIntegrationService._classify_message_scene( capture_result={ "semantic_bundle": { "bundle_id": "semantic-qa", "semantic_version": "2026-03-27.workflow-v1", "source_system": "feishu", "primary_workflow": "query_flow", "workflows": [ { "workflow_type": "query_flow", "workflow_action": "answer", "target_scope": "global", "operation_codes": ["query.answer"], "requires_confirmation": False, "confidence": 0.9, "slots": [], "clarification_needed": False, "metadata": {}, } ], "message_gate": "auto_execute", "reply_contract": { "mode": "aggregate_card", "template_key": "workflow_bundle", "summary": "query_flow", }, "diagnostics": {}, } }, message_content="帮我创建项目和信号源", ) assert scene == "qa" @pytest.mark.asyncio async def test_handle_message_receive_orchestrates_capture_and_finalize(monkeypatch) -> None: service = FeishuIntegrationService(db=object()) called: dict[str, object] = {} def _fake_build_receive_context(*, event, state_trace_id): # noqa: ARG001 return { "ingress_source": "long_connection", "message": {"message_id": "m-1", "message_type": "text"}, "sender": {"sender_name": "原始用户名"}, "sender_id": {"open_id": "ou-1"}, "message_id": "m-1", "sender_type": "user", "raw_message_content": "hello", "message_content": "hello", "chat_id": "chat-1", "chat_type": "group", "captured_at": None, "mentioned_bot_names": [], "sender_open_id": "ou-1", "perf_t0": 1.2, "message_attachments": [], "occurred_at_iso": None, } async def _fake_record_ingress_trace(**kwargs): called["trace"] = kwargs async def _fake_resolve_sender_display_name(*, sender, sender_id): # noqa: ARG001 return "解析后用户名" def _fake_normalize_message_attachments_from_context(*, context): # noqa: ARG001 return [] async def _fake_capture_and_persist_from_ingress(**kwargs): called["capture"] = kwargs return { "status": "captured", "classification": "signal", "linked_project_ids": [11], "auto_tasks": [], "ai_trace_id": "trace-1", } async def _fake_finalize_receive_result(**kwargs): called["finalize"] = kwargs return {"code": 0, "msg": "success", "data": {"scene": "execute"}} monkeypatch.setattr(service, "_build_receive_context", _fake_build_receive_context) monkeypatch.setattr(service, "_record_ingress_trace", _fake_record_ingress_trace) monkeypatch.setattr(service, "_resolve_sender_display_name", _fake_resolve_sender_display_name) monkeypatch.setattr(service, "_normalize_message_attachments_from_context", _fake_normalize_message_attachments_from_context) monkeypatch.setattr(service, "_capture_and_persist_from_ingress", _fake_capture_and_persist_from_ingress) monkeypatch.setattr(service, "_finalize_receive_result", _fake_finalize_receive_result) result = await service._handle_message_receive({"type": "im.message.receive_v1"}) assert result["code"] == 0 assert result["data"]["scene"] == "execute" assert called["capture"]["chat_id"] == "chat-1" assert called["capture"]["sender_name"] == "解析后用户名" assert called["finalize"]["chat_type"] == "group" def test_extract_thread_context_reads_top_level_reply_fields() -> None: event = { "thread_id": "th-1", "root_id": "root-1", "parent_id": "reply-1", "message": { "message_id": "msg-1", "content": "{\"text\":\"hello\"}", }, } context = FeishuReceivePrepareMixin._extract_thread_context( event=event, message=event["message"], message_payload={"text": "hello", "attachments": [], "message_type": "text"}, ) assert context == { "thread_id": "th-1", "root_message_id": "root-1", "reply_to_message_id": "reply-1", } def test_extract_thread_context_reads_nested_reply_fields() -> None: event = { "message": { "message_id": "msg-1", "meta": { "conversation": { "thread_id": "th-nested", "root_message_id": "root-nested", "reply_to_message_id": "reply-nested", } }, "content": "{\"text\":\"hello\"}", } } context = FeishuReceivePrepareMixin._extract_thread_context( event=event, message=event["message"], message_payload={"text": "hello", "attachments": [], "message_type": "text"}, ) assert context == { "thread_id": "th-nested", "root_message_id": "root-nested", "reply_to_message_id": "reply-nested", } @pytest.mark.asyncio async def test_finalize_receive_result_group_qa_without_explicit_trigger_does_not_reply(monkeypatch) -> None: service = FeishuIntegrationService(db=object()) called: dict[str, object] = {} async def _fake_safe_reply_text(**kwargs): called["reply_text"] = kwargs async def _fake_reply_with_active_role_prompt(**kwargs): called["reply_llm"] = kwargs async def _fake_record_message_trace(**kwargs): called["trace"] = kwargs monkeypatch.setattr(service, "_safe_reply_text", _fake_safe_reply_text) monkeypatch.setattr(service, "_reply_with_active_role_prompt", _fake_reply_with_active_role_prompt) monkeypatch.setattr(service, "_record_message_trace", _fake_record_message_trace) result = await service._finalize_receive_result( result={ "status": "answered", "classification": "question", "direct_reply_required": True, "direct_reply_text": "这是一条直接回复", "ai_trace_id": "trace-qa-1", }, state_trace_id="state-qa-1", message_id="msg-qa-1", message_content="这个怎么处理", chat_id="chat-qa-1", chat_type="group", sender_open_id="ou-1", sender_name="测试用户", mentioned_bot_names=[], thread_context=None, perf_t0=None, ) assert result["code"] == 0 assert "reply_text" not in called assert "reply_llm" not in called assert called["trace"]["reply_source"] == "none" @pytest.mark.asyncio async def test_finalize_receive_result_group_qa_reply_anchor_allows_direct_reply_without_mention(monkeypatch) -> None: service = FeishuIntegrationService(db=object()) called: dict[str, object] = {} async def _fake_safe_reply_text(**kwargs): called["reply_text"] = kwargs monkeypatch.setattr(service, "_safe_reply_text", _fake_safe_reply_text) monkeypatch.setattr(service, "_reply_with_active_role_prompt", AsyncMock()) monkeypatch.setattr(service, "_record_message_trace", AsyncMock()) result = await service._finalize_receive_result( result={ "status": "answered", "classification": "question", "direct_reply_required": True, "direct_reply_text": "沿用上一条上下文继续答复", "ai_trace_id": "trace-qa-2", }, state_trace_id="state-qa-2", message_id="msg-qa-2", message_content="这条再详细一点", chat_id="chat-qa-2", chat_type="group", sender_open_id="ou-2", sender_name="测试用户", mentioned_bot_names=[], thread_context={"reply_to_message_id": "om-bot-reply-1"}, perf_t0=None, ) assert result["code"] == 0 assert called["reply_text"]["text"] == "沿用上一条上下文继续答复" @pytest.mark.asyncio async def test_finalize_receive_result_group_qa_bound_requirement_thread_allows_direct_reply_without_explicit_trigger( monkeypatch, ) -> None: service = FeishuIntegrationService(db=object()) called: dict[str, object] = {} async def _fake_safe_reply_text(**kwargs): called["reply_text"] = kwargs monkeypatch.setattr(service, "_safe_reply_text", _fake_safe_reply_text) monkeypatch.setattr(service, "_reply_with_active_role_prompt", AsyncMock()) monkeypatch.setattr(service, "_record_message_trace", AsyncMock()) result = await service._finalize_receive_result( result={ "status": "answered", "classification": "question", "direct_reply_required": True, "direct_reply_text": "继续第二轮技术路线收敛", "ai_trace_id": "trace-qa-bound-1", "group_reply_allowed_via_bound_context": True, }, state_trace_id="state-qa-bound-1", message_id="msg-qa-bound-1", message_content="A", chat_id="chat-qa-bound-1", chat_type="group", sender_open_id="ou-bound-1", sender_name="测试用户", mentioned_bot_names=[], thread_context=None, perf_t0=None, ) assert result["code"] == 0 assert called["reply_text"]["text"] == "继续第二轮技术路线收敛"