PMFlow/backend/tests/test_sstc_signal_state_machine.py

from __future__ import annotations

import unittest
from datetime import datetime
from types import SimpleNamespace

from fastapi import HTTPException

from app.api.sstc import (
    _build_signal_created_at_range,
    _collect_locked_signal_conflicts,
    _extract_source_attachments,
    _match_signal_filters,
    _normalize_signal_status_code,
)
from app.services.feishu_integration import FeishuIntegrationService


class SignalStateMachineHelpersTest(unittest.TestCase):
    def test_normalize_signal_status_code_prefers_linked_state(self) -> None:
        status_code = _normalize_signal_status_code(
            linked_state="linked",
            signal_status="待处理",
            processed=False,
        )
        self.assertEqual(status_code, "linked")

    def test_normalize_signal_status_code_uses_closed_for_processed(self) -> None:
        status_code = _normalize_signal_status_code(
            linked_state=None,
            signal_status="",
            processed=True,
        )
        self.assertEqual(status_code, "closed")

    def test_collect_locked_signal_conflicts(self) -> None:
        charters = [
            SimpleNamespace(
                id=101,
                title="宪章A",
                content={"signal_ids": [1, 2]},
                signal_id=None,
            ),
            SimpleNamespace(
                id=102,
                title="宪章B",
                content={"signal_ids": [3]},
                signal_id=5,
            ),
        ]
        conflict_ids, conflict_map = _collect_locked_signal_conflicts(
            charters=charters,  # type: ignore[arg-type]
            candidate_signal_ids=[2, 5, 9],
        )
        self.assertEqual(conflict_ids, [2, 5])
        self.assertEqual(conflict_map[2][0]["charter_id"], 101)
        self.assertEqual(conflict_map[5][0]["charter_id"], 102)

    def test_build_signal_created_at_range_rejects_invalid_order(self) -> None:
        with self.assertRaises(HTTPException):
            _build_signal_created_at_range(
                created_from="2026-03-11",
                created_to="2026-03-10",
            )

    def test_match_signal_filters_keyword_and_status(self) -> None:
        row = {
            "id": "88",
            "title": "新增飞书信号源",
            "content": "支持附件同步",
            "proposer_name": "老板",
            "charter_name": "信号治理宪章",
            "source": "Feishu",
            "priority": "P1",
            "status_code": "pending",
        }
        matched = _match_signal_filters(
            row=row,
            status_filter="pending",
            keyword="附件",
            source="Feishu",
            priority="p1",
            created_from_utc=datetime(2026, 3, 1, 0, 0, 0),
            created_to_utc=datetime(2026, 3, 20, 0, 0, 0),
            created_at_dt=datetime(2026, 3, 10, 12, 0, 0),
        )
        self.assertTrue(matched)

    def test_extract_source_attachments_normalizes_and_deduplicates(self) -> None:
        attachments = _extract_source_attachments(
            {
                "attachments": [
                    {"attachment_type": "image", "image_key": "img_a", "file_name": "截图A"},
                    {"attachment_type": "img", "image_key": "img_a", "file_name": "截图A重复"},
                    {"attachment_type": "file", "file_key": "file_b", "file_name": "需求文档.pdf"},
                ]
            }
        )
        self.assertEqual(len(attachments), 2)
        self.assertEqual(attachments[0]["attachment_type"], "image")
        self.assertEqual(attachments[1]["file_name"], "需求文档.pdf")


class FeishuMessagePayloadParserTest(unittest.TestCase):
    def setUp(self) -> None:
        self.service = FeishuIntegrationService(db=None)  # type: ignore[arg-type]

    def test_extract_message_payload_for_image_message(self) -> None:
        payload = self.service._extract_message_payload(
            {
                "message": {
                    "message_type": "image",
                    "content": '{"image_key":"img_001"}',
                }
            }
        )
        self.assertEqual(payload["message_type"], "image")
        self.assertEqual(len(payload["attachments"]), 1)
        self.assertIn("附件消息", str(payload["text"] or ""))

    def test_extract_message_payload_for_file_message(self) -> None:
        payload = self.service._extract_message_payload(
            {
                "message": {
                    "message_type": "file",
                    "content": '{"file_key":"file_001","file_name":"立项材料.docx","file_size":1024}',
                }
            }
        )
        self.assertEqual(len(payload["attachments"]), 1)
        attachment = payload["attachments"][0]
        self.assertEqual(attachment["attachment_type"], "file")
        self.assertEqual(attachment["file_name"], "立项材料.docx")

    def test_extract_message_payload_for_post_with_text_and_image(self) -> None:
        payload = self.service._extract_message_payload(
            {
                "message": {
                    "message_type": "post",
                    "content": {
                        "post": {
                            "zh_cn": {
                                "content": [
                                    [
                                        {"tag": "text", "text": "请评估这张图的风险点"},
                                        {"tag": "img", "image_key": "img_post_1"},
                                    ]
                                ]
                            }
                        }
                    },
                }
            }
        )
        self.assertEqual(payload["message_type"], "post")
        self.assertEqual(len(payload["attachments"]), 1)
        self.assertIn("请评估这张图的风险点", str(payload["text"] or ""))


if __name__ == "__main__":
    unittest.main()