"""Tests for Feishu vc.bot.meeting_invited_v1 event handling."""

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

from gateway.platforms.base import MessageEvent
from plugins.platforms.feishu.feishu_meeting_invite import (
    build_meeting_invite_prompt,
    handle_meeting_invited_event,
    parse_meeting_invited_event,
)


def _user_id(open_id, union_id="on_1", user_id="e65g874e"):
    return {
        "open_id": open_id,
        "union_id": union_id,
        "user_id": user_id,
    }


def _make_payload(event_id="evt_1"):
    return {
        "schema": "2.0",
        "header": {
            "event_id": event_id,
            "event_type": "vc.bot.meeting_invited_v1",
        },
        "event": {
            "meeting": {
                "id": "7646677832873577404",
                "topic": "赵磊的视频会议",
                "meeting_no": "884264377",
                "start_time": "1780384522000",
                "end_time": "1780384522000",
                "host_user": {
                    "id": _user_id("ou_390b35dca44816efc9afa812aaff3a69", "on_host", "e65g874e"),
                    "user_type": 1,
                    "user_role": 2,
                    "user_name": "赵磊",
                },
            },
            "bot": {
                "id": _user_id("ou_4398906db1bc4a2d7ed91b95ffb308d0", "on_bot", ""),
                "user_type": 10,
                "user_role": 0,
                "user_name": "Hermes龙虾",
            },
            "inviter": {
                "id": _user_id(
                    "ou_390b35dca44816efc9afa812aaff3a69",
                    "on_e19a19e6ffafbd54fbb3c4d251d6fa19",
                    "e65g874e",
                ),
                "user_type": 1,
                "user_role": 0,
                "user_name": "赵磊",
            },
            "invite_time": "1780388292",
        },
    }


def _make_payload_with_numeric_inviter_id():
    payload = _make_payload()
    payload["event"]["inviter"]["id"] = "3001"
    return payload


class _Adapter:
    def __init__(self, duplicate=False):
        self.duplicate = duplicate
        self.events = []
        self.dedup_keys = []
        self.profile_requests = []

    def _is_duplicate(self, key):
        self.dedup_keys.append(key)
        return self.duplicate

    def build_source(self, **kwargs):
        return SimpleNamespace(**kwargs)

    async def _resolve_sender_profile(self, sender_id):
        self.profile_requests.append(sender_id)
        return {
            "user_id": getattr(sender_id, "user_id", None) or getattr(sender_id, "open_id", None),
            "user_name": "Resolved Inviter",
            "user_id_alt": getattr(sender_id, "union_id", None),
        }

    async def _handle_message_with_guards(self, event):
        self.events.append(event)


class TestMeetingInviteParsing(unittest.TestCase):
    def test_parse_actual_payload_string_int64_fields(self):
        parsed = parse_meeting_invited_event(_make_payload())

        self.assertIsNotNone(parsed)
        self.assertEqual(parsed.event_id, "evt_1")
        self.assertEqual(parsed.meeting.id, "7646677832873577404")
        self.assertEqual(parsed.meeting.start_time_ms, 1780384522000)
        self.assertEqual(parsed.meeting.end_time_ms, 1780384522000)
        self.assertEqual(parsed.inviter.open_id, "ou_390b35dca44816efc9afa812aaff3a69")
        self.assertEqual(parsed.inviter.user_id, "e65g874e")
        self.assertEqual(parsed.inviter.union_id, "on_e19a19e6ffafbd54fbb3c4d251d6fa19")
        self.assertEqual(parsed.invite_time_s, 1780388292)

    def test_parse_body_content_payload(self):
        payload = _make_payload()
        wrapped = {
            "header": payload["header"],
            "event": {
                "body": {
                    "content": [
                        {
                            "contentType": "application/json",
                            "data": payload["event"],
                        }
                    ]
                }
            },
        }
        parsed = parse_meeting_invited_event(wrapped)

        self.assertIsNotNone(parsed)
        self.assertEqual(parsed.meeting.meeting_no, "884264377")
        self.assertEqual(parsed.inviter.open_id, "ou_390b35dca44816efc9afa812aaff3a69")

    def test_parse_requires_inviter(self):
        payload = _make_payload()
        del payload["event"]["inviter"]

        self.assertIsNone(parse_meeting_invited_event(payload))

    def test_parse_requires_meeting_no(self):
        payload = _make_payload()
        payload["event"]["meeting"]["meeting_no"] = ""

        self.assertIsNone(parse_meeting_invited_event(payload))

    def test_prompt_contains_meeting_and_inviter_context(self):
        parsed = parse_meeting_invited_event(_make_payload())
        prompt = build_meeting_invite_prompt(parsed)

        self.assertIn("You have been invited to join a meeting: 赵磊的视频会议", prompt)
        self.assertIn("Meeting Number: 884264377", prompt)
        self.assertIn("Inviter: 赵磊", prompt)
        self.assertIn("Join the meeting directly.", prompt)
        self.assertIn("You may use lark-cli and the relevant Lark/Feishu meeting skills", prompt)
        self.assertIn("Do not ask the user for confirmation", prompt)
        self.assertIn("If you cannot join the meeting", prompt)
        self.assertNotIn("ou_390b35dca44816efc9afa812aaff3a69", prompt)
        self.assertNotIn("user_id", prompt)
        self.assertNotIn("Use the Meeting Number as the primary credential", prompt)
        self.assertNotIn("meeting_id:", prompt)
        self.assertNotIn("start_time:", prompt)
        self.assertNotIn("end_time:", prompt)
        self.assertNotIn("Invite time:", prompt)


class TestMeetingInviteHandler(unittest.TestCase):
    def _run(self, coro):
        return asyncio.run(coro)

    def test_routes_as_synthetic_message_to_inviter_open_id(self):
        adapter = _Adapter()

        self._run(handle_meeting_invited_event(adapter, _make_payload()))

        self.assertEqual(adapter.dedup_keys, ["vc_invite:evt_1"])
        self.assertEqual(len(adapter.events), 1)
        event = adapter.events[0]
        self.assertIsInstance(event, MessageEvent)
        self.assertEqual(event.source.chat_id, "ou_390b35dca44816efc9afa812aaff3a69")
        self.assertEqual(event.source.chat_type, "dm")
        self.assertEqual(event.source.user_id, "e65g874e")
        self.assertEqual(event.source.user_name, "Resolved Inviter")
        self.assertEqual(event.source.chat_name, "Resolved Inviter")
        self.assertEqual(event.source.user_id_alt, "on_e19a19e6ffafbd54fbb3c4d251d6fa19")
        self.assertEqual(len(adapter.profile_requests), 1)
        self.assertEqual(adapter.profile_requests[0].open_id, "ou_390b35dca44816efc9afa812aaff3a69")
        self.assertEqual(adapter.profile_requests[0].user_id, "e65g874e")
        self.assertEqual(adapter.profile_requests[0].union_id, "on_e19a19e6ffafbd54fbb3c4d251d6fa19")
        self.assertIsNone(event.message_id)
        self.assertIn("You have been invited to join a meeting: 赵磊的视频会议", event.text)
        self.assertNotIn("{'open_id'", event.text)

    def test_duplicate_event_is_dropped(self):
        adapter = _Adapter(duplicate=True)

        self._run(handle_meeting_invited_event(adapter, _make_payload()))

        self.assertEqual(adapter.dedup_keys, ["vc_invite:evt_1"])
        self.assertEqual(adapter.events, [])

    def test_inviter_without_open_id_is_dropped(self):
        payload = _make_payload_with_numeric_inviter_id()
        adapter = _Adapter()

        self._run(handle_meeting_invited_event(adapter, payload))

        self.assertEqual(adapter.events, [])


class TestMeetingInviteSendRouting(unittest.TestCase):
    def _run(self, coro):
        return asyncio.run(coro)

    def test_feishu_user_id_prefix_sends_with_user_id_receive_type(self):
        from gateway.config import PlatformConfig
        from plugins.platforms.feishu.adapter import FeishuAdapter

        created_requests = []

        class _Message:
            @staticmethod
            def create(request):
                created_requests.append(request)
                return SimpleNamespace(success=lambda: True, data=SimpleNamespace(message_id="om_1"))

        adapter = FeishuAdapter(PlatformConfig())
        adapter._client = SimpleNamespace(
            im=SimpleNamespace(v1=SimpleNamespace(message=SimpleNamespace(create=_Message.create)))
        )

        with patch.object(
            FeishuAdapter,
            "_build_create_message_body",
            staticmethod(lambda **kwargs: SimpleNamespace(**kwargs)),
        ), patch.object(
            FeishuAdapter,
            "_build_create_message_request",
            staticmethod(lambda receive_id_type, request_body: SimpleNamespace(
                receive_id_type=receive_id_type,
                request_body=request_body,
            )),
        ):
            self._run(adapter._send_raw_message(
                chat_id="feishu_user_id:3001",
                msg_type="text",
                payload='{"text":"ok"}',
                reply_to=None,
                metadata=None,
            ))

        self.assertEqual(created_requests[0].receive_id_type, "user_id")
        self.assertEqual(created_requests[0].request_body.receive_id, "3001")


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