Spaces:
Runtime error
Runtime error
File size: 2,896 Bytes
46252cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """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())
|