👷 Test: 移除 httpbin 并整理测试 (#2110)

pyproject.toml

@@ -51,6 +51,7 @@ pre-commit = "^3.0.0"
 
 [tool.poetry.group.test.dependencies]
 nonebug = "^0.3.0"
+werkzeug = "^2.3.6"
 pytest-cov = "^4.0.0"
 pytest-xdist = "^3.0.2"
 pytest-asyncio = "^0.21.0"
@@ -68,7 +69,7 @@ fastapi = ["fastapi", "uvicorn"]
 all = ["fastapi", "quart", "aiohttp", "httpx", "websockets", "uvicorn"]
 
 [tool.pytest.ini_options]
-asyncio_mode = "auto"
+asyncio_mode = "strict"
 addopts = "--cov=nonebot --cov-append --cov-report=term-missing"
 filterwarnings = [
     "error",

tests/conftest.py

@@ -1,11 +1,15 @@
 import os
+import threading
 from pathlib import Path
-from typing import TYPE_CHECKING, Set
+from typing import TYPE_CHECKING, Set, Generator
 
 import pytest
 from nonebug import NONEBOT_INIT_KWARGS
+from werkzeug.serving import BaseWSGIServer, make_server
 
 import nonebot
+from nonebot.drivers import URL
+from fake_server import request_handler
 
 os.environ["CONFIG_FROM_ENV"] = '{"test": "test"}'
 os.environ["CONFIG_OVERRIDE"] = "new"
@@ -28,3 +32,20 @@ def load_plugin(nonebug_init: None) -> Set["Plugin"]:
 def load_builtin_plugin(nonebug_init: None) -> Set["Plugin"]:
     # preload builtin plugins
     return nonebot.load_builtin_plugins("echo", "single_session")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def server() -> Generator[BaseWSGIServer, None, None]:
+    server = make_server("127.0.0.1", 0, app=request_handler)
+    thread = threading.Thread(target=server.serve_forever)
+    thread.start()
+    try:
+        yield server
+    finally:
+        server.shutdown()
+        thread.join()
+
+
+@pytest.fixture(scope="session")
+def server_url(server: BaseWSGIServer) -> URL:
+    return URL(f"http://{server.host}:{server.port}")

tests/fake_server.py

@@ -0,0 +1,69 @@
+import json
+import base64
+from typing import Dict, List, Union, TypeVar
+
+from werkzeug import Request, Response
+from werkzeug.datastructures import MultiDict
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+
+def json_safe(string, content_type="application/octet-stream") -> str:
+    try:
+        string = string.decode("utf-8")
+        json.dumps(string)
+        return string
+    except (ValueError, TypeError):
+        return b"".join(
+            [
+                b"data:",
+                content_type.encode("utf-8"),
+                b";base64,",
+                base64.b64encode(string),
+            ]
+        ).decode("utf-8")
+
+
+def flattern(d: "MultiDict[K, V]") -> Dict[K, Union[V, List[V]]]:
+    return {k: v[0] if len(v) == 1 else v for k, v in d.to_dict(flat=False).items()}
+
+
+@Request.application
+def request_handler(request: Request) -> Response:
+    try:
+        _json = json.loads(request.data.decode("utf-8"))
+    except (ValueError, TypeError):
+        _json = None
+
+    return Response(
+        json.dumps(
+            {
+                "url": request.url,
+                "method": request.method,
+                "origin": request.headers.get("X-Forwarded-For", request.remote_addr),
+                "headers": flattern(
+                    MultiDict((k, v) for k, v in request.headers.items())
+                ),
+                "args": flattern(request.args),
+                "form": flattern(request.form),
+                "data": json_safe(request.data),
+                "json": _json,
+                "files": flattern(
+                    MultiDict(
+                        (
+                            k,
+                            json_safe(
+                                v.read(),
+                                request.files[k].content_type
+                                or "application/octet-stream",
+                            ),
+                        )
+                        for k, v in request.files.items()
+                    )
+                ),
+            }
+        ),
+        status=200,
+        content_type="application/json",
+    )

tests/plugins/matcher/matcher_permission.py

@@ -2,6 +2,7 @@
 from nonebot.permission import USER, Permission
 
 default_permission = Permission()
+new_permission = Permission()
 
 test_permission_updater = Matcher.new(permission=default_permission)
 
@@ -14,4 +15,4 @@
 
 @test_custom_updater.permission_updater
 async def _() -> Permission:
-    return default_permission
+    return new_permission