CI: 添加测试用例 (#97)

* 🎨 improve tests structure

* ✅ add utils test

* ✅ add matrix for bytes key

* ✅ add bot tests

* ✅ add missing model field

* ✅ update bot test

tests/conftest.py

@@ -21,13 +21,14 @@
 
 
 def pytest_configure(config: pytest.Config) -> None:
-    with (Path(__file__).parent / "bots.json").open("r") as f:
+    with (Path(__file__).parent.joinpath("data", "bots.json")).open("r") as f:
         feishu_bots = json.load(f)
 
     config.stash[NONEBOT_INIT_KWARGS] = {
         "driver": "~fastapi+~httpx",
         "log_level": "DEBUG",
         "feishu_bots": feishu_bots,
+        "nickname": ["bot"],
     }
 
 

tests/data/bots.json

@@ -0,0 +1,12 @@
+[
+  {
+    "app_id": "app_id_connection",
+    "app_secret": "app_secret_connection",
+    "verification_token": "verification_token_connection"
+  },
+  {
+    "app_id": "app_id_tome",
+    "app_secret": "app_secret_tome",
+    "verification_token": "verification_token_tome"
+  }
+]

tests/data/bots_info.json

@@ -0,0 +1,11 @@
+{
+  "code": 0,
+  "msg": "ok",
+  "bot": {
+    "activate_status": 2,
+    "app_name": "app_name",
+    "avatar_url": "https://s1-imfile.feishucdn.com/static-resource/v1/da5xxxx14b16113",
+    "ip_white_list": [],
+    "open_id": "open_id"
+  }
+}

tests/data/events.json

@@ -0,0 +1,78 @@
+[
+  {
+    "_model": "GroupMessageEvent",
+    "schema": "2.0",
+    "header": {
+      "event_id": "0a67****6a9953c",
+      "token": "CAkC****1Fk5J",
+      "create_time": "1708251664941",
+      "event_type": "im.message.receive_v1",
+      "tenant_key": "tenant_key",
+      "app_id": "app_id_tome"
+    },
+    "event": {
+      "message": {
+        "chat_id": "oc_b5fc2808***5970",
+        "chat_type": "group",
+        "content": "{\"text\":\"bot /say 12345\"}",
+        "create_time": "1708251664735",
+        "message_id": "om_b9c9da5***7d74a",
+        "message_type": "text",
+        "update_time": "1708251664735"
+      },
+      "sender": {
+        "sender_id": {
+          "open_id": "ou_23e65b***4fd28",
+          "union_id": "on_30c91***e04",
+          "user_id": "92***53"
+        },
+        "sender_type": "user",
+        "tenant_key": "123456789"
+      }
+    }
+  },
+  {
+    "_model": "GroupMessageEvent",
+    "schema": "2.0",
+    "header": {
+      "event_id": "746*****d335b08",
+      "token": "CAkC****k5J",
+      "create_time": "1708399533117",
+      "event_type": "im.message.receive_v1",
+      "tenant_key": "tenant_key",
+      "app_id": "app_id_tome"
+    },
+    "event": {
+      "message": {
+        "chat_id": "chat_id",
+        "chat_type": "group",
+        "content": "{\"text\":\"@_user_1 /say 34567\"}",
+        "create_time": "1708399532591",
+        "mentions": [
+          {
+            "id": {
+              "open_id": "open_id",
+              "union_id": "union_id",
+              "user_id": "user_id"
+            },
+            "key": "@_user_1",
+            "name": "bot",
+            "tenant_key": "tenant_key"
+          }
+        ],
+        "message_id": "message_id",
+        "message_type": "text",
+        "update_time": "1708399532591"
+      },
+      "sender": {
+        "sender_id": {
+          "open_id": "sender_open_id",
+          "union_id": "sender_union_id",
+          "user_id": "sender_user_id"
+        },
+        "sender_type": "user",
+        "tenant_key": "13cb691f67cf575d"
+      }
+    }
+  }
+]

tests/data/payloads.json

@@ -0,0 +1,7 @@
+[
+  {
+    "challenge": "3axxxx99-c46f-4x99-8xxe-7268xxx4b5",
+    "token": "CAxxxFk5J",
+    "type": "url_verification"
+  }
+]

tests/fake_server.py

@@ -1,5 +1,6 @@
 import json
 import base64
+from pathlib import Path
 from typing import Dict, List, Union, TypeVar
 
 from werkzeug import Request, Response
@@ -84,20 +85,11 @@ def handle_get_tenant_access_token():
 
 
 def handle_get_bot_info():
+    with open(Path(__file__).parent.joinpath("data", "bots_info.json")) as f:
+        bots_info = json.load(f)
+
     return Response(
-        json.dumps(
-            {
-                "code": 0,
-                "msg": "ok",
-                "bot": {
-                    "activate_status": 2,
-                    "app_name": "name",
-                    "avatar_url": "https://s1-imfile.feishucdn.com/static-resource/v1/da5xxxx14b16113",
-                    "ip_white_list": [],
-                    "open_id": "open_id",
-                },
-            }
-        ),
+        json.dumps(bots_info),
         status=200,
         content_type="application/json",
     )

tests/test_feishu_bot.py

@@ -0,0 +1,96 @@
+import json
+from pathlib import Path
+
+import pytest
+from nonebug import App
+from nonebot.compat import type_validate_python
+
+import nonebot
+from nonebot.adapters.feishu import Adapter
+from nonebot.adapters.feishu.models import BotInfo
+from nonebot.adapters.feishu.config import BotConfig
+from nonebot.adapters.feishu.event import GroupMessageEvent
+from nonebot.adapters.feishu.bot import Bot, _check_at_me, _check_reply, _check_nickname
+
+with (Path(__file__).parent.joinpath("data", "bots.json")).open("r") as f:
+    feishu_bots = json.load(f)
+
+test_bot = feishu_bots[1]
+bot_id = test_bot["app_id"]
+
+with (Path(__file__).parent.joinpath("data", "bots_info.json")).open("r") as f:
+    bots_info = json.load(f)
+
+with (Path(__file__).parent.joinpath("data", "events.json")).open(
+    "r", encoding="utf8"
+) as f:
+    test_events = json.load(f)
+
+
+@pytest.mark.asyncio
+async def test_check_tome(server_url: str):
+    event_data = test_events[0]
+    event_data.pop("_model", None)
+    event = Adapter.json_to_event(event_data)
+    assert isinstance(event, GroupMessageEvent)
+
+    adapter = nonebot.get_adapter(Adapter)
+    adapter.feishu_config.feishu_api_base = str(server_url)  # type: ignore
+    bot = Bot(
+        adapter,
+        bot_id,
+        bot_config=type_validate_python(BotConfig, test_bot),
+        bot_info=type_validate_python(BotInfo, bots_info["bot"]),
+    )
+    _check_at_me(bot, event)
+    assert not event.to_me
+
+    event_data = test_events[1]
+    event_data.pop("_model", None)
+    event = Adapter.json_to_event(event_data)
+    _check_at_me(bot, event)
+    assert event.to_me
+
+
+@pytest.mark.asyncio
+async def test_check_reply(server_url: str):
+    event_data = test_events[0]
+    event_data.pop("_model", None)
+    event = Adapter.json_to_event(event_data)
+    assert isinstance(event, GroupMessageEvent)
+
+    adapter = nonebot.get_adapter(Adapter)
+    adapter.feishu_config.feishu_api_base = str(server_url)  # type: ignore
+    bot = Bot(
+        adapter,
+        bot_id,
+        bot_config=type_validate_python(BotConfig, test_bot),
+        bot_info=type_validate_python(BotInfo, bots_info["bot"]),
+    )
+    await _check_reply(bot, event)
+    assert not event.reply
+
+
+@pytest.mark.asyncio
+async def test_check_nickname(app: App, server_url: str):
+    event_data = test_events[0]
+    event_data.pop("_model", None)
+    event = Adapter.json_to_event(event_data)
+    assert isinstance(event, GroupMessageEvent)
+
+    adapter = nonebot.get_adapter(Adapter)
+    adapter.feishu_config.feishu_api_base = str(server_url)  # type: ignore
+    bot = Bot(
+        adapter,
+        bot_id,
+        bot_config=type_validate_python(BotConfig, test_bot),
+        bot_info=type_validate_python(BotInfo, bots_info["bot"]),
+    )
+    _check_nickname(bot, event)
+    assert event.to_me
+
+    event_data = test_events[1]
+    event_data.pop("_model", None)
+    event = Adapter.json_to_event(event_data)
+    _check_nickname(bot, event)
+    assert not event.to_me

tests/test_feishu_connection.py

@@ -8,26 +8,25 @@
 import nonebot
 from nonebot.adapters.feishu import Adapter
 
-with (Path(__file__).parent / "bots.json").open("r") as f:
+with (Path(__file__).parent.joinpath("data", "bots.json")).open("r") as f:
     feishu_bots = json.load(f)
 
-test_bot = feishu_bots[0]
-bot_id = test_bot["app_id"]
+bot_id = feishu_bots[0]["app_id"]
 
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("endpoints", [f"/feishu/{bot_id}"])
 async def test_http(app: App, endpoints: str, server_url: URL):
-    with (Path(__file__).parent / "events.json").open("r") as f:
-        test_events = json.load(f)
+    with (Path(__file__).parent.joinpath("data", "payloads.json")).open("r") as f:
+        test_payloads = json.load(f)
 
     adapter = nonebot.get_adapter(Adapter)
-    adapter.feishu_config.feishu_api_base = str(server_url)
+    adapter.feishu_config.feishu_api_base = str(server_url)  # type: ignore
 
     async with app.test_server() as ctx:
         client = ctx.get_client()
-        event = test_events[0]
-        resp = await client.post(endpoints, json=event)
+        payload = test_payloads[0]
+        resp = await client.post(endpoints, json=payload)
         assert resp.status_code == 200
         bots = nonebot.get_bots()
         assert bot_id in bots

tests/test_feishu_event.py

@@ -0,0 +1,18 @@
+import json
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+
+from nonebot.adapters.feishu import Adapter
+
+
+@pytest.mark.asyncio
+async def test_event():
+    with (Path(__file__).parent.joinpath("data", "events.json")).open("r") as f:
+        test_events: List[Dict[str, Any]] = json.load(f)
+
+    for event_data in test_events:
+        model_name: str = event_data.pop("_model", None)
+        event = Adapter.json_to_event(event_data)
+        assert model_name == event.__class__.__name__

tests/test_feishu_utils.py

@@ -0,0 +1,11 @@
+import pytest
+
+from nonebot.adapters.feishu.utils import AESCipher
+
+
+@pytest.mark.parametrize("key", ["test key", b"test key"])
+def test_aes(key: str):
+    encrypt = "P37w+VZImNgPEO1RBhJ6RtKl7n6zymIbEG1pReEzghk="
+    cipher = AESCipher(key)
+
+    assert cipher.decrypt_string(encrypt) == "hello world"