refactor: unify message interactions

2026-07-09 22:43:44 +08:00 · 2026-04-22 15:18:04 +08:00
parent 8c78627647
commit cefb60ba2c
6 changed files with 1657 additions and 1049 deletions
--- a/tests/test_media_interaction.py
+++ b/tests/test_media_interaction.py
@@ -0,0 +1,158 @@
+import sys
+import unittest
+from types import ModuleType
+from unittest.mock import patch
+
+sys.modules.setdefault("qbittorrentapi", ModuleType("qbittorrentapi"))
+setattr(sys.modules["qbittorrentapi"], "TorrentFilesList", list)
+sys.modules.setdefault("transmission_rpc", ModuleType("transmission_rpc"))
+setattr(sys.modules["transmission_rpc"], "File", object)
+sys.modules.setdefault("psutil", ModuleType("psutil"))
+
+from app.chain.interaction import MediaInteractionChain, media_interaction_manager
+from app.chain.message import MessageChain
+from app.core.context import MediaInfo
+from app.core.meta import MetaBase
+from app.schemas.types import MessageChannel
+
+
+class TestMediaInteraction(unittest.TestCase):
+    def tearDown(self):
+        media_interaction_manager.clear()
+
+    @staticmethod
+    def _build_meta(name: str) -> MetaBase:
+        meta = MetaBase(name)
+        meta.name = name
+        meta.begin_season = 1
+        return meta
+
+    def test_message_routes_text_reply_to_media_interaction_before_ai(self):
+        chain = MessageChain()
+        request = media_interaction_manager.create_or_replace(
+            user_id="10001",
+            channel=MessageChannel.Wechat,
+            source="wechat-test",
+            username="tester",
+            action="Search",
+            keyword="星际穿越",
+            title="星际穿越",
+            meta=self._build_meta("星际穿越"),
+            items=[MediaInfo(title="星际穿越", year="2014")],
+        )
+        self.assertIsNotNone(request)
+
+        with patch.object(chain, "_record_user_message"), patch(
+            "app.chain.message.MediaInteractionChain.handle_text_interaction",
+            return_value=True,
+        ) as handle_text, patch.object(chain, "_handle_ai_message") as handle_ai:
+            chain.handle_message(
+                channel=MessageChannel.Wechat,
+                source="wechat-test",
+                userid="10001",
+                username="tester",
+                text="1",
+            )
+
+        handle_text.assert_called_once()
+        handle_ai.assert_not_called()
+
+    def test_callback_routes_to_media_interaction_chain(self):
+        chain = MessageChain()
+        request = media_interaction_manager.create_or_replace(
+            user_id="10001",
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            username="tester",
+            action="Search",
+            keyword="星际穿越",
+            title="星际穿越",
+            meta=self._build_meta("星际穿越"),
+            items=[MediaInfo(title="星际穿越", year="2014")],
+        )
+
+        with patch(
+            "app.chain.message.MediaInteractionChain.handle_callback_interaction",
+            return_value=True,
+        ) as handle_callback:
+            chain._handle_callback(
+                text=f"CALLBACK:media:{request.request_id}:page-next",
+                channel=MessageChannel.Telegram,
+                source="telegram-test",
+                userid="10001",
+                username="tester",
+            )
+
+        handle_callback.assert_called_once()
+
+    def test_media_interaction_starts_search_and_posts_media_list(self):
+        chain = MediaInteractionChain()
+        meta = self._build_meta("星际穿越")
+        medias = [
+            MediaInfo(title="星际穿越", year="2014"),
+            MediaInfo(title="Interstellar", year="2014"),
+        ]
+
+        with patch(
+            "app.chain.interaction.MediaChain.search",
+            return_value=(meta, medias),
+        ), patch.object(chain, "post_medias_message") as post_medias_message:
+            handled = chain.handle_text_interaction(
+                channel=MessageChannel.Telegram,
+                source="telegram-test",
+                userid="10001",
+                username="tester",
+                text="星际穿越",
+            )
+
+        self.assertTrue(handled)
+        post_medias_message.assert_called_once()
+        notification = post_medias_message.call_args.args[0]
+        self.assertTrue(notification.buttons)
+        self.assertTrue(
+            notification.buttons[0][0]["callback_data"].startswith("media:")
+        )
+
+        request = media_interaction_manager.get_by_user("10001")
+        self.assertIsNotNone(request)
+        self.assertEqual(request.action, "Search")
+        self.assertEqual(len(request.items), 2)
+
+    def test_media_interaction_legacy_page_callback_updates_existing_request(self):
+        chain = MediaInteractionChain()
+        request = media_interaction_manager.create_or_replace(
+            user_id="10001",
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            username="tester",
+            action="Search",
+            keyword="星际穿越",
+            title="星际穿越",
+            meta=self._build_meta("星际穿越"),
+            items=[
+                MediaInfo(title=f"资源 {index}", year="2024")
+                for index in range(1, 11)
+            ],
+        )
+
+        with patch.object(chain, "post_medias_message") as post_medias_message:
+            handled = chain.handle_callback_interaction(
+                callback_data="page_n",
+                channel=MessageChannel.Telegram,
+                source="telegram-test",
+                userid="10001",
+                username="tester",
+                original_message_id=123,
+                original_chat_id="456",
+            )
+
+        self.assertTrue(handled)
+        self.assertEqual(request.page, 1)
+        post_medias_message.assert_called_once()
+        notification = post_medias_message.call_args.args[0]
+        self.assertEqual(notification.original_message_id, 123)
+        self.assertEqual(notification.original_chat_id, "456")
+
+
+if __name__ == "__main__":
+    unittest.main()