SPO / tests /mock /mock_curl_cffi.py
XiangJinYu's picture
add metagpt
fe5c39d verified
raw
history blame
802 Bytes
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()