je_web_runner.api.networking

Façade: API mock / contract / GraphQL / mock services / HAR replay.

exception je_web_runner.api.networking.ApiMockError

Bases: WebRunnerException

Raised when route configuration is invalid.

exception je_web_runner.api.networking.ContractError

Bases: WebRunnerException

Raised when validation fails.

class je_web_runner.api.networking.GraphQLClient(endpoint: 'str', headers: 'Optional[Dict[str, str]]' = None, timeout: 'float' = 10.0)

Bases: object

endpoint: str

execute(query: str, variables: Dict[str, Any] | None = None, operation_name: str | None = None) → Dict[str, Any]

headers: Dict[str, str] | None = None

introspect() → Dict[str, Any]

timeout: float = 10.0

exception je_web_runner.api.networking.GraphQLError

Bases: WebRunnerException

Raised when the GraphQL response contains errors or transport fails.

class je_web_runner.api.networking.HarEntry(method: 'str', path: 'str', status: 'int', headers: 'Dict[str, str]'=<factory>, body: 'str' = '', body_is_base64: 'bool' = False)

Bases: object

body: str = ''

body_is_base64: bool = False

headers: Dict[str, str]

method: str

path: str

status: int

exception je_web_runner.api.networking.HarReplayError

Bases: WebRunnerException

Raised when the HAR file is invalid or the server can’t bind.

class je_web_runner.api.networking.HarReplayServer(entries: List[HarEntry], host: str = '127.0.0.1', port: int = 0, not_found_status: int = 404)

Bases: object

In-process HTTP server that replays HAR responses.

calls: List[Tuple[str, str]]

find(method: str, path: str) → HarEntry | None

start() → str

stop() → None

class je_web_runner.api.networking.MockOAuthServer(host: str = '127.0.0.1', port: int = 0)

Bases: object

Issues a fake bearer token from POST /token.

start() → str

stop() → None

class je_web_runner.api.networking.MockResponse(status: int = 200, body: Any = '', headers: Dict[str, str]=<factory>, content_type: str = 'application/json')

Bases: object

Static response payload.

body: Any = ''

content_type: str = 'application/json'

headers: Dict[str, str]

status: int = 200

to_payload() → Dict[str, Any]

class je_web_runner.api.networking.MockRoute(method: str, url_pattern: str, response: MockResponse, times: int | None = None, times_seen: int = 0)

Bases: object

Single matcher + response.

consume() → bool

matches(method: str, url: str) → bool

method: str

response: MockResponse

times: int | None = None

times_seen: int = 0

url_pattern: str

class je_web_runner.api.networking.MockRouter

Bases: object

Ordered list of MockRoute.

add(method: str, url_pattern: str, body: Any = '', status: int = 200, headers: Dict[str, str] | None = None, content_type: str = 'application/json', times: int | None = None) → MockRoute

attach_to_page(page: Any) → None: Wire the router onto a Playwright page via page.route('**/*').

calls() → List[Tuple[str, str]]

match(method: str, url: str) → MockRoute | None

class je_web_runner.api.networking.MockS3Storage(buckets: Dict[str, ~typing.Dict[str, bytes]]=<factory>)

Bases: object

In-memory key-value store mimicking S3 put_object / get_object.

buckets: Dict[str, Dict[str, bytes]]

create_bucket(name: str) → None

get_object(bucket: str, key: str) → bytes

list_objects(bucket: str) → List[str]

put_object(bucket: str, key: str, body: bytes) → None

exception je_web_runner.api.networking.MockServiceError

Bases: WebRunnerException

Raised when a mock service fails to start or respond.

class je_web_runner.api.networking.MockSmtpServer(host: str = '127.0.0.1', port: int = 0)

Bases: object

Capture-only SMTP server. Bind to 127.0.0.1 and call start().

captured: List[str]

start() → int

stop() → None

class je_web_runner.api.networking.SchemaResult(valid: 'bool', errors: 'List[str]' = <factory>)

Bases: object

errors: List[str]

valid: bool

je_web_runner.api.networking.assert_valid(response_body: Any, schema: Dict[str, Any]) → None: Convenience: raise on validation failure.

je_web_runner.api.networking.extract_field(payload: Dict[str, Any], path: str) → Any: 用 a.b.c[0].d 形式的路徑從 GraphQL 回應中取值 Pluck a value out of payload['data'] using a dotted path with optional [index] accessors.

je_web_runner.api.networking.introspect_types(payload: Dict[str, Any]) → List[str]: Return the list of type names from an introspection payload.

je_web_runner.api.networking.load_har(source: str | Path) → List[HarEntry]: Read a HAR file and return its entries projected to HarEntry.

je_web_runner.api.networking.register_route(method: str, url_pattern: str, body: str | Dict[str, Any] | List[Any] = '', status: int = 200, times: int | None = None) → MockRoute: Register a route on the module-level singleton.

je_web_runner.api.networking.reset_global_router() → None: Clear all registered routes and call history.

je_web_runner.api.networking.validate_against_openapi(response_body: Any, openapi_doc: Dict[str, Any], path: str, method: str, status: int) → SchemaResult: 從 OpenAPI 3 文件抓出對應 schema 後驗證 Look up paths[path][method].responses[status].content."application/json".schema in an OpenAPI 3 document, then validate the response body.

je_web_runner.api.networking.validate_response(response_body: Any, schema: Dict[str, Any]) → SchemaResult: Validate response_body against schema.