wehub-resource-sync
4ce4204b6c
CI / Lint (push) Failing after 2s
CI / Build (push) Has been skipped
SDK CI / PHP SDK (push) Failing after 1s
Split PHP SDK / PHP SDK tests (push) Failing after 1s
CI / Test (push) Failing after 1s
CI / Dashboard (push) Failing after 0s
SDK CI / JavaScript SDK (push) Failing after 0s
SDK CI / Python SDK (push) Failing after 2s
SDK CI / Java SDK (push) Failing after 1s
Split PHP SDK / Mirror sdk/php -> rmyndharis/openwa-php (push) Has been skipped
CI / Test (PostgreSQL migrations) (push) Failing after 7m47s
CI / Docker Build (push) Has been skipped
82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
"""Test helpers — an httpx.MockTransport-based recorder.
|
|
|
|
Tests assert on exact method/url/body/headers without any real network or
|
|
global monkey-patching. ``httpx.MockTransport`` is the idiomatic, native way
|
|
to test an httpx-based client.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable
|
|
|
|
import httpx
|
|
|
|
|
|
@dataclass
|
|
class RecordedCall:
|
|
method: str
|
|
url: str
|
|
headers: dict[str, str]
|
|
body: Any
|
|
|
|
|
|
@dataclass
|
|
class MockBackend:
|
|
"""A scripted backend. Raises if a call doesn't match a route."""
|
|
calls: list[RecordedCall] = field(default_factory=list)
|
|
routes: list[tuple[str, str, Callable[[RecordedCall], httpx.Response]]] = field(default_factory=list)
|
|
fallback: Callable[[RecordedCall], httpx.Response] | None = None
|
|
|
|
def on(
|
|
self,
|
|
method: str,
|
|
path_prefix: str,
|
|
status: int = 200,
|
|
body: Any = None,
|
|
) -> "MockBackend":
|
|
def responder(_: RecordedCall) -> httpx.Response:
|
|
if status in (204, 100, 304) or body is None:
|
|
return httpx.Response(status, content=b"")
|
|
return httpx.Response(status, content=json.dumps(body).encode(), headers={"content-type": "application/json"})
|
|
|
|
self.routes.append((method.upper(), path_prefix, responder))
|
|
return self
|
|
|
|
def as_transport(self) -> httpx.MockTransport:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
method = request.method.upper()
|
|
url = str(request.url)
|
|
body: Any = None
|
|
if request.content:
|
|
try:
|
|
body = json.loads(request.content)
|
|
except ValueError:
|
|
body = request.content.decode(errors="replace")
|
|
call = RecordedCall(method=method, url=url, headers=dict(request.headers), body=body)
|
|
self.calls.append(call)
|
|
|
|
# Most-specific (longest) matching prefix wins, so a nested route like
|
|
# "/catalog/products" is preferred over "/catalog" regardless of
|
|
# registration order.
|
|
matches = [(pp, resp) for m, pp, resp in self.routes if m == method and pp in url]
|
|
if matches:
|
|
_, responder = max(matches, key=lambda pair: len(pair[0]))
|
|
return responder(call)
|
|
if self.fallback is not None:
|
|
return self.fallback(call)
|
|
raise AssertionError(f"MockBackend: no route for {method} {url}")
|
|
|
|
return httpx.MockTransport(handler)
|
|
|
|
@property
|
|
def last_call(self) -> RecordedCall:
|
|
return self.calls[-1]
|
|
|
|
|
|
def make_client(backend: MockBackend, base_url: str = "http://localhost:2785", api_key: str = "owa_k1_test"):
|
|
from openwa import OpenWAClient
|
|
|
|
return OpenWAClient(base_url=base_url, api_key=api_key, transport=backend.as_transport())
|