mock search engine api

tests/mock/mock_aiohttp.py

@@ -0,0 +1,41 @@
+import json
+from typing import Callable
+
+from aiohttp.client import ClientSession
+
+origin_request = ClientSession.request
+
+
+class MockAioResponse:
+    check_funcs: dict[tuple[str, str], Callable[[dict], str]] = {}
+    rsp_cache: dict[str, str] = {}
+    name = "aiohttp"
+
+    def __init__(self, session, method, url, **kwargs) -> None:
+        fn = self.check_funcs.get((method, url))
+        self.key = f"{self.name}-{method}-{url}-{fn(kwargs) if fn else json.dumps(kwargs, sort_keys=True)}"
+        self.mng = self.response = None
+        if self.key not in self.rsp_cache:
+            self.mng = origin_request(session, method, url, **kwargs)
+
+    async def __aenter__(self):
+        if self.response:
+            await self.response.__aenter__()
+        elif self.mng:
+            self.response = await self.mng.__aenter__()
+        return self
+
+    async def __aexit__(self, *args, **kwargs):
+        if self.response:
+            await self.response.__aexit__(*args, **kwargs)
+            self.response = None
+        elif self.mng:
+            await self.mng.__aexit__(*args, **kwargs)
+            self.mng = None
+
+    async def json(self, *args, **kwargs):
+        if self.key in self.rsp_cache:
+            return self.rsp_cache[self.key]
+        data = await self.response.json(*args, **kwargs)
+        self.rsp_cache[self.key] = data
+        return data

tests/mock/mock_curl_cffi.py

@@ -0,0 +1,22 @@
+import json
+from typing import Callable
+
+from curl_cffi import requests
+
+origin_request = requests.Session.request
+
+
+class MockCurlCffiResponse(requests.Response):
+    check_funcs: dict[tuple[str, str], Callable[[dict], str]] = {}
+    rsp_cache: dict[str, str] = {}
+    name = "curl-cffi"
+
+    def __init__(self, session, method, url, **kwargs) -> None:
+        super().__init__()
+        fn = self.check_funcs.get((method, url))
+        self.key = f"{self.name}-{method}-{url}-{fn(kwargs) if fn else json.dumps(kwargs, sort_keys=True)}"
+        self.response = None
+        if self.key not in self.rsp_cache:
+            response = origin_request(session, method, url, **kwargs)
+            self.rsp_cache[self.key] = response.content.decode()
+        self.content = self.rsp_cache[self.key].encode()

tests/mock/mock_httplib2.py

@@ -0,0 +1,29 @@
+import json
+from typing import Callable
+from urllib.parse import parse_qsl, urlparse
+
+import httplib2
+
+origin_request = httplib2.Http.request
+
+
+class MockHttplib2Response(httplib2.Response):
+    check_funcs: dict[tuple[str, str], Callable[[dict], str]] = {}
+    rsp_cache: dict[str, str] = {}
+    name = "httplib2"
+
+    def __init__(self, http, uri, method="GET", **kwargs) -> None:
+        url = uri.split("?")[0]
+        result = urlparse(uri)
+        params = dict(parse_qsl(result.query))
+        fn = self.check_funcs.get((method, uri))
+        new_kwargs = {"params": params}
+        key = f"{self.name}-{method}-{url}-{fn(new_kwargs) if fn else json.dumps(new_kwargs)}"
+        if key not in self.rsp_cache:
+            _, self.content = origin_request(http, uri, method, **kwargs)
+            self.rsp_cache[key] = self.content.decode()
+        self.content = self.rsp_cache[key]
+
+    def __iter__(self):
+        yield self
+        yield self.content.encode()