ForcePilot/backend/test/unit/channel/channels/feishu/test_feishu.py

from __future__ import annotations

import pytest

from yuxi.channel.channels.feishu.adapter import FeishuAdapter
from yuxi.channel.channels.feishu.config import FeishuConfig
from yuxi.channel.channels.feishu.translator import FeishuTranslator, _extract_attachments, _extract_text
from yuxi.channel.domain.model.shared.channel_type import ChannelType


class TestFeishuTranslator:
    def test_translate_p2p_message(self) -> None:
        raw = {
            "event": {
                "message": {
                    "message_id": "msg-001",
                    "content": '{"text": "hello"}',
                    "chat_type": "p2p",
                },
                "sender": {
                    "sender_id": {"user_id": "ou-123"},
                },
            }
        }
        msg = FeishuTranslator.translate_event(raw)
        assert msg.message_id == "msg-001"
        assert msg.content == "hello"
        assert msg.channel_type == ChannelType.FEISHU
        assert msg.sender.id == "ou-123"
        assert msg.metadata["is_group"] is False

    def test_translate_group_message(self) -> None:
        raw = {
            "event": {
                "message": {
                    "message_id": "msg-002",
                    "content": '{"text": "group hello"}',
                    "chat_type": "group",
                    "chat_id": "oc-456",
                },
                "sender": {
                    "sender_id": {"user_id": "ou-789"},
                },
            }
        }
        msg = FeishuTranslator.translate_event(raw)
        assert msg.metadata["is_group"] is True
        assert msg.metadata["group_id"] == "oc-456"

    def test_translate_missing_fields_defaults(self) -> None:
        raw = {}
        msg = FeishuTranslator.translate_event(raw)
        assert msg.message_id == ""
        assert msg.content == ""

    def test_translate_image_attachment(self) -> None:
        raw = {
            "event": {
                "message": {
                    "message_id": "msg-img",
                    "content": '{"image_key": "img_key_123"}',
                    "message_type": "image",
                    "chat_type": "p2p",
                },
                "sender": {"sender_id": {"user_id": "u1"}},
            }
        }
        msg = FeishuTranslator.translate_event(raw)
        assert len(msg.attachments) == 1
        assert msg.attachments[0].url == "img_key_123"
        assert msg.attachments[0].media_type == "image"

    def test_translate_file_attachment(self) -> None:
        raw = {
            "event": {
                "message": {
                    "message_id": "msg-file",
                    "content": '{"file_key": "fk_123", "file_name": "doc.pdf"}',
                    "message_type": "file",
                    "chat_type": "p2p",
                },
                "sender": {"sender_id": {"user_id": "u1"}},
            }
        }
        msg = FeishuTranslator.translate_event(raw)
        assert len(msg.attachments) == 1
        assert msg.attachments[0].url == "fk_123"
        assert msg.attachments[0].filename == "doc.pdf"
        assert msg.attachments[0].media_type == "file"


class TestExtractText:
    def test_json_text_field(self) -> None:
        message = {"content": '{"text": "hello"}'}
        assert _extract_text(message) == "hello"

    def test_empty_content(self) -> None:
        message = {"content": ""}
        assert _extract_text(message) == ""

    def test_invalid_json_returns_raw(self) -> None:
        message = {"content": "not-json"}
        assert _extract_text(message) == "not-json"

    def test_dict_without_text_returns_raw(self) -> None:
        message = {"content": {"other": "value"}}
        assert _extract_text(message) == {"other": "value"}


class TestExtractAttachments:
    def test_image_type(self) -> None:
        body = {
            "event": {
                "message": {
                    "message_type": "image",
                    "content": '{"image_key": "ik_1"}',
                }
            }
        }
        attachments = _extract_attachments(body)
        assert len(attachments) == 1
        assert attachments[0].media_type == "image"

    def test_file_type(self) -> None:
        body = {
            "event": {
                "message": {
                    "message_type": "file",
                    "content": '{"file_key": "fk_1", "file_name": "a.txt"}',
                }
            }
        }
        attachments = _extract_attachments(body)
        assert len(attachments) == 1
        assert attachments[0].media_type == "file"
        assert attachments[0].filename == "a.txt"

    def test_text_type_no_attachment(self) -> None:
        body = {
            "event": {
                "message": {
                    "message_type": "text",
                    "content": '{"text": "hi"}',
                }
            }
        }
        assert _extract_attachments(body) == []

    def test_invalid_json_no_attachment(self) -> None:
        body = {
            "event": {
                "message": {
                    "message_type": "image",
                    "content": "not-json",
                }
            }
        }
        assert _extract_attachments(body) == []


class TestFeishuConfig:
    def test_defaults(self) -> None:
        config = FeishuConfig()
        assert config.app_id == ""
        assert config.app_secret == ""
        assert config.verification_token == ""
        assert config.ws_enabled is True

    def test_with_values(self) -> None:
        config = FeishuConfig(app_id="id1", app_secret="s1", verification_token="vt1", ws_enabled=False)
        assert config.app_id == "id1"
        assert config.verification_token == "vt1"


class TestFeishuAdapter:
    def test_channel_type(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        assert adapter.channel_type == ChannelType.FEISHU.value

    def test_get_default_config(self) -> None:
        config = FeishuAdapter.get_default_config()
        assert "app_id" in config
        assert "app_secret" in config
        assert "verification_token" in config

    @pytest.mark.asyncio
    async def test_is_healthy_before_open(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        assert await adapter.is_healthy() is False

    @pytest.mark.asyncio
    async def test_is_healthy_after_open(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        await adapter.open()
        assert await adapter.is_healthy() is True

    @pytest.mark.asyncio
    async def test_close(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        await adapter.open()
        await adapter.close()
        assert await adapter.is_healthy() is False

    @pytest.mark.asyncio
    async def test_receive_message(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        raw = {
            "event": {
                "message": {"message_id": "m1", "content": '{"text": "hi"}', "chat_type": "p2p"},
                "sender": {"sender_id": {"user_id": "u1"}},
            }
        }
        msg = await adapter.receive_message(raw)
        assert msg.message_id == "m1"

    @pytest.mark.asyncio
    async def test_send_typing_is_noop(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        await adapter.send_typing("session-1")

    @pytest.mark.asyncio
    async def test_send_media_returns_false(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        result = await adapter.send_media("session-1", url="http://x", media_type="image", metadata={})
        assert result is False

    @pytest.mark.asyncio
    async def test_send_message_without_client_returns_failure(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        result = await adapter.send_message("session-1", "hello", channel_type="feishu", metadata={})
        assert result.success is False

    def test_ws_connection_is_none_when_disabled(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        assert adapter.ws_connection is None

    def test_route_contributor_has_router(self) -> None:
        adapter = FeishuAdapter(app_id="test", app_secret="test", ws_enabled=False)
        contributor = adapter.route_contributor
        assert contributor.router is not None