|
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() |
|
|