"""Offline tests for Api01 (acme_client.py). Generated by mimicplus; regenerate, don't edit. Run: pytest test_acme_client.py They replay recorded fixtures through an injected transport: no network, no server. """ # ruff: noqa from __future__ import annotations import importlib.util import json import pathlib import sys from typing import Any from urllib.parse import urlsplit import re import pytest _HERE = pathlib.Path(__file__).resolve().parent _spec = importlib.util.spec_from_file_location('acme_client', _HERE / 'acme_client.py') assert _spec is not None and _spec.loader is not None client_mod = importlib.util.module_from_spec(_spec) sys.modules['acme_client'] = client_mod _spec.loader.exec_module(client_mod) CLIENT = client_mod.Api01 FIXTURES: dict[str, dict[str, Any]] = {'get_me': {'status': 200, 'content_type': 'application/json', 'body': {'id': 1, 'username': 'example', 'email': 'example', 'plan': 'example'}, 'synthesized': True}, 'list_products': {'status': 200, 'content_type': 'application/json', 'body': {'items': [{'id': 1, 'in_stock': True, 'name': 'example', 'price': 1.5, 'tags': ['example', 'example']}, {'id': 2, 'in_stock': True, 'name': 'example', 'price': 1.5, 'tags': ['example', 'example']}], 'next_page': 1, 'page': 1, 'per_page': 1, 'total': 1}, 'synthesized': True}, 'get_product': {'status': 200, 'content_type': 'application/json', 'body': {'id': 1, 'in_stock': True, 'name': 'example', 'price': 1.5, 'tags': ['example', 'example']}, 'synthesized': True}, 'create_item': {'status': 201, 'content_type': 'application/json', 'body': {'cart_size': 1, 'item': {'product_id': 1, 'qty': 1}}, 'synthesized': True}, 'search_products': {'status': 200, 'content_type': 'application/json', 'body': {'data': {'searchProducts': {'nodes': [{'id': 'id-0', 'name': 'example', 'price': 1.5}, {'id': 'id-1', 'name': 'example', 'price': 1.5}], 'totalCount': 1}}}, 'synthesized': True}, 'get_order': {'status': 200, 'content_type': 'application/json', 'body': {'data': {'order': {'id': 'example', 'status': 'example', 'total': 1.5, 'items': [1, 1]}}}, 'synthesized': True}, 'add_review': {'status': 200, 'content_type': 'application/json', 'body': {'data': {'addReview': {'id': 'example', 'rating': 1, 'ok': True}}}, 'synthesized': True}} ROUTES: dict[str, list[Any]] = {'get_me': ['GET', '^/api/me/?$', None], 'list_products': ['GET', '^/api/products/?$', None], 'get_product': ['GET', '^/api/products/[^/]+/?$', None], 'create_item': ['POST', '^/api/cart/items/?$', None], 'search_products': ['POST', '^/graphql/?$', 'SearchProducts'], 'get_order': ['POST', '^/graphql/?$', 'GetOrder'], 'add_review': ['POST', '^/graphql/?$', 'AddReview']} AUTH_ROUTES: dict[str, list[Any]] = {'login': ['POST', '^/api/auth/login/?$', None], 'refresh': ['POST', '^/api/auth/refresh/?$', None]} def _graphql_op(req: Any) -> Any: """operationName / persisted-query hash of a GraphQL request (POST body or GET query).""" try: body = json.loads(req.body) if req.body else {} except (ValueError, TypeError): body = {} if not isinstance(body, dict) or not body: from urllib.parse import parse_qsl body = dict(parse_qsl(urlsplit(req.url).query)) if isinstance(body.get("extensions"), str): try: body["extensions"] = json.loads(body["extensions"]) except ValueError: pass ext = body.get("extensions") if isinstance(body.get("extensions"), dict) else {} pq = ext.get("persistedQuery") if isinstance(ext, dict) else None return body.get("operationName") or (pq.get("sha256Hash") if isinstance(pq, dict) else None) def route_of(req: Any, routes: dict[str, list[Any]]) -> Any: path = urlsplit(req.url).path for name, (method, rx, op) in routes.items(): if req.method == method and re.match(rx, path) and (op is None or _graphql_op(req) == op): return name return None class FakeTransport: """Answers each request with the recorded fixture of the endpoint whose route matches.""" def __init__(self) -> None: self.calls: list[Any] = [] def __call__(self, req: Any) -> Any: self.calls.append(req) name = route_of(req, ROUTES) if name is not None: fx = FIXTURES[name] body = fx["body"] raw = body.encode() if isinstance(body, str) else json.dumps(body).encode() return client_mod.RawResponse(fx["status"], {"content-type": fx["content_type"]}, raw, req.url) return client_mod.RawResponse(404, {"content-type": "text/plain"}, b"no fixture", req.url) @pytest.fixture def api() -> Any: client_mod.CircuitBreaker.reset_all() t = FakeTransport() c = CLIENT(transport=t, min_interval=0, respect_robots=False, max_retries=0, auth={}) c.fake = t return c def test_get_me_request_shape(api: Any) -> None: api.get_me(**{}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'GET' assert re.match(ROUTES['get_me'][1], urlsplit(req.url).path) def test_get_me_response_matches_recorded_schema(api: Any) -> None: result = api.get_me(**{}) drift = client_mod.diff_schema(CLIENT.PROBES.get('get_me', {}).get("schema") or {'type': ['object'], 'properties': {'id': {'type': ['integer']}, 'username': {'type': ['string']}, 'email': {'type': ['string']}, 'plan': {'type': ['string']}}, 'required': ['email', 'id', 'plan', 'username']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def test_list_products_request_shape(api: Any) -> None: api.list_products(**{}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'GET' assert re.match(ROUTES['list_products'][1], urlsplit(req.url).path) def test_list_products_response_matches_recorded_schema(api: Any) -> None: result = api.list_products(**{}) drift = client_mod.diff_schema(CLIENT.PROBES.get('list_products', {}).get("schema") or {'type': ['object'], 'properties': {'items': {'type': ['array'], 'items': {'type': ['object'], 'properties': {'id': {'type': ['integer']}, 'in_stock': {'type': ['boolean']}, 'name': {'type': ['string']}, 'price': {'type': ['number']}, 'tags': {'type': ['array'], 'items': {'type': ['string']}}}, 'required': ['id', 'in_stock', 'name', 'price', 'tags']}}, 'next_page': {'type': ['integer', 'null']}, 'page': {'type': ['integer']}, 'per_page': {'type': ['integer']}, 'total': {'type': ['integer']}}, 'required': ['items', 'next_page', 'page', 'per_page', 'total']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def test_list_products_items(api: Any) -> None: result = api.list_products(**{}) items = api.items('list_products', result) assert isinstance(items, list) and items ids = [api.item_id('list_products', it) for it in items] assert len(set(ids)) == len(ids) def test_get_product_request_shape(api: Any) -> None: api.get_product(**{'product_id': 102}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'GET' assert re.match(ROUTES['get_product'][1], urlsplit(req.url).path) def test_get_product_response_matches_recorded_schema(api: Any) -> None: result = api.get_product(**{'product_id': 102}) drift = client_mod.diff_schema(CLIENT.PROBES.get('get_product', {}).get("schema") or {'type': ['object'], 'properties': {'id': {'type': ['integer']}, 'in_stock': {'type': ['boolean']}, 'name': {'type': ['string']}, 'price': {'type': ['number']}, 'tags': {'type': ['array'], 'items': {'type': ['string']}}}, 'required': ['id', 'in_stock', 'name', 'price', 'tags']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def test_create_item_request_shape(api: Any) -> None: api.create_item(**{}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'POST' assert re.match(ROUTES['create_item'][1], urlsplit(req.url).path) assert "json" in req.headers.get("content-type", "") sent = json.loads(req.body) assert set(['product_id', 'qty']) <= set(sent) def test_create_item_response_matches_recorded_schema(api: Any) -> None: result = api.create_item(**{}) drift = client_mod.diff_schema(CLIENT.PROBES.get('create_item', {}).get("schema") or {'type': ['object'], 'properties': {'cart_size': {'type': ['integer']}, 'item': {'type': ['object'], 'properties': {'product_id': {'type': ['integer']}, 'qty': {'type': ['integer']}}, 'required': ['product_id', 'qty']}}, 'required': ['cart_size', 'item']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def test_search_products_request_shape(api: Any) -> None: api.search_products(**{'q': 'mug'}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'POST' assert re.match(ROUTES['search_products'][1], urlsplit(req.url).path) assert "json" in req.headers.get("content-type", "") sent = json.loads(req.body) assert set(['operationName', 'query', 'variables']) <= set(sent) def test_search_products_response_matches_recorded_schema(api: Any) -> None: result = api.search_products(**{'q': 'mug'}) drift = client_mod.diff_schema(CLIENT.PROBES.get('search_products', {}).get("schema") or {'type': ['object'], 'properties': {'data': {'type': ['object'], 'properties': {'searchProducts': {'type': ['object'], 'properties': {'nodes': {'type': ['array'], 'items': {'type': ['object'], 'properties': {'id': {'type': ['string']}, 'name': {'type': ['string']}, 'price': {'type': ['number']}}, 'required': ['id', 'name', 'price']}}, 'totalCount': {'type': ['integer']}}, 'required': ['nodes', 'totalCount']}}, 'required': ['searchProducts']}}, 'required': ['data']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def test_search_products_items(api: Any) -> None: result = api.search_products(**{'q': 'mug'}) items = api.items('search_products', result) assert isinstance(items, list) and items ids = [api.item_id('search_products', it) for it in items] assert len(set(ids)) == len(ids) def test_get_order_request_shape(api: Any) -> None: api.get_order(**{}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'POST' assert re.match(ROUTES['get_order'][1], urlsplit(req.url).path) assert "json" in req.headers.get("content-type", "") sent = json.loads(req.body) assert set(['operationName', 'query', 'variables']) <= set(sent) def test_get_order_response_matches_recorded_schema(api: Any) -> None: result = api.get_order(**{}) drift = client_mod.diff_schema(CLIENT.PROBES.get('get_order', {}).get("schema") or {'type': ['object'], 'properties': {'data': {'type': ['object'], 'properties': {'order': {'type': ['object'], 'properties': {'id': {'type': ['string']}, 'status': {'type': ['string']}, 'total': {'type': ['number']}, 'items': {'type': ['array'], 'items': {'type': ['integer']}}}, 'required': ['id', 'items', 'status', 'total']}}, 'required': ['order']}}, 'required': ['data']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def test_add_review_request_shape(api: Any) -> None: api.add_review(**{'text': 'great'}) assert len(api.fake.calls) == 1 req = api.fake.calls[0] assert req.method == 'POST' assert re.match(ROUTES['add_review'][1], urlsplit(req.url).path) assert "json" in req.headers.get("content-type", "") sent = json.loads(req.body) assert set(['operationName', 'query', 'variables']) <= set(sent) def test_add_review_response_matches_recorded_schema(api: Any) -> None: result = api.add_review(**{'text': 'great'}) drift = client_mod.diff_schema(CLIENT.PROBES.get('add_review', {}).get("schema") or {'type': ['object'], 'properties': {'data': {'type': ['object'], 'properties': {'addReview': {'type': ['object'], 'properties': {'id': {'type': ['string']}, 'rating': {'type': ['integer']}, 'ok': {'type': ['boolean']}}, 'required': ['id', 'ok', 'rating']}}, 'required': ['addReview']}}, 'required': ['data']}, client_mod.infer_schema(result)) assert not [d for d in drift if d["breaking"]], drift def _token_response(n: int) -> dict[str, Any]: flow = CLIENT.AUTH_FLOW body: dict[str, Any] = {} client_mod.set_path(body, flow["token_path"].split("."), f"tok-{n}-not-real") if flow.get("refresh_token_path"): client_mod.set_path(body, flow["refresh_token_path"].split("."), f"rt-{n}-not-real") if flow.get("expires_in_path"): client_mod.set_path(body, flow["expires_in_path"].split("."), 3600) return body class TokenServer(FakeTransport): """Issues tok-1 at login, rejects it once (401), issues tok-2 on refresh.""" def __init__(self) -> None: super().__init__() self.rejected = False def __call__(self, req: Any) -> Any: flow = CLIENT.AUTH_FLOW role = route_of(req, AUTH_ROUTES) ok = {"content-type": "application/json"} if role == flow["login_endpoint"]: self.calls.append(req) return client_mod.RawResponse(200, ok, json.dumps(_token_response(1)).encode(), req.url) if role == flow.get("refresh_endpoint"): self.calls.append(req) assert b"rt-1-not-real" in (req.body or b"") or "rt-1-not-real" in req.url return client_mod.RawResponse(200, ok, json.dumps(_token_response(2)).encode(), req.url) sent = json.dumps(dict(req.headers)) + req.url if "tok-1-not-real" in sent and not self.rejected: self.calls.append(req) self.rejected = True return client_mod.RawResponse(401, ok, b'{"error": "expired"}', req.url) assert "tok-2-not-real" in sent or "tok-1-not-real" in sent, "learned token was not sent" return super().__call__(req) def test_auth_flow_login_use_and_refresh_on_401() -> None: client_mod.CircuitBreaker.reset_all() t = TokenServer() api = CLIENT(transport=t, min_interval=0, respect_robots=False, max_retries=0, auth={}) api.authenticate() api.get_me(**{}) roles = [route_of(c, AUTH_ROUTES) or route_of(c, ROUTES) for c in t.calls] flow = CLIENT.AUTH_FLOW assert roles == [flow["login_endpoint"], 'get_me', flow["refresh_endpoint"], 'get_me'], roles assert "tok-2-not-real" in json.dumps(dict(t.calls[-1].headers)) + t.calls[-1].url def test_selfcheck_offline(api: Any) -> None: rep = api.selfcheck() assert rep["ok"], rep def test_ssrf_guard_blocks_cloud_metadata(api: Any) -> None: with pytest.raises(client_mod.SSRFBlocked): api.request("GET", "http://169.254.169.254/latest/meta-data/") def test_auth_never_sent_to_foreign_hosts(api: Any) -> None: api.auth["authorization"] = "Bearer test-token-not-real" try: api.request("GET", "https://example.org/x") except client_mod.MimicError: pass foreign = [c for c in api.fake.calls if urlsplit(c.url).hostname == "example.org"] assert foreign and "authorization" not in {k.lower() for k in foreign[0].headers}