diff --git a/fastapi_startkit/src/fastapi_startkit/fastapi/testing/__init__.py b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/__init__.py index a0c62bc7..84fbddd9 100644 --- a/fastapi_startkit/src/fastapi_startkit/fastapi/testing/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/__init__.py @@ -1,3 +1,8 @@ +from fastapi_startkit.fastapi.testing.assertable_json import ( + AssertableJson, + assert_json_structure, +) from fastapi_startkit.fastapi.testing.test_case import HttpTestCase +from fastapi_startkit.fastapi.testing.test_response import TestResponse -__all__ = ["HttpTestCase"] +__all__ = ["HttpTestCase", "TestResponse", "AssertableJson", "assert_json_structure"] diff --git a/fastapi_startkit/src/fastapi_startkit/fastapi/testing/assertable_json.py b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/assertable_json.py new file mode 100644 index 00000000..d9158ca4 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/assertable_json.py @@ -0,0 +1,332 @@ +"""Fluent JSON assertions for HTTP tests. + +Provides :class:`AssertableJson`, a chainable wrapper around a decoded JSON +body (``dict``/``list``). + +Example:: + + response.assert_json(lambda json: ( + json.where("id", 1) + .where("name", "Bedu") + .where("active", lambda v: v is True) + .has("profile", lambda profile: ( + profile.where("email", "a@b.com").etc() + )) + .etc() + )) + +Every ``where``/``has`` call records an interaction with the touched key. +When an assertion scope closes, :meth:`AssertableJson._verify` asserts that +every property of an object was interacted with — unless :meth:`etc` was +called to acknowledge the remaining keys. This catches unexpected or extra +keys in responses. +""" + +from __future__ import annotations + +from typing import Any, Callable, Union + +# A sentinel distinguishing "argument not provided" from an explicit ``None``. +_MISSING = object() + +# Mapping of friendly type names to the Python types that satisfy them. +# ``bool`` is intentionally excluded from the numeric types because in Python +# ``bool`` is a subclass of ``int`` and JSON keeps the two concepts distinct. +_JSON_TYPES: dict[str, Callable[[Any], bool]] = { + "string": lambda v: isinstance(v, str), + "integer": lambda v: isinstance(v, int) and not isinstance(v, bool), + "double": lambda v: isinstance(v, float), + "number": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool), + "boolean": lambda v: isinstance(v, bool), + "array": lambda v: isinstance(v, list), + "object": lambda v: isinstance(v, dict), + "null": lambda v: v is None, +} + + +class AssertableJson: + """A chainable, scope-aware assertion helper for decoded JSON bodies.""" + + def __init__(self, data: Any, path: str = "") -> None: + self._data = data + self._path = path + self._interacted: set[str] = set() + + # ------------------------------------------------------------------ # + # Internal helpers + # ------------------------------------------------------------------ # + def _full(self, key: Any) -> str: + """Return the absolute dot-path to ``key`` for error messages.""" + return f"{self._path}.{key}" if self._path else str(key) + + @staticmethod + def _step(node: Any, seg: str) -> tuple[bool, Any]: + """Descend one segment into a dict key or a list index.""" + if isinstance(node, dict) and seg in node: + return True, node[seg] + if isinstance(node, list) and seg.lstrip("-").isdigit(): + index = int(seg) + if -len(node) <= index < len(node): + return True, node[index] + return False, None + + def _exists(self, key: Any) -> bool: + node = self._data + for seg in str(key).split("."): + found, node = self._step(node, seg) + if not found: + return False + return True + + def _get(self, key: Any) -> Any: + """Resolve a (possibly dotted) ``key`` or raise with the full path.""" + node = self._data + for seg in str(key).split("."): + found, node = self._step(node, seg) + assert found, f"Property [{self._full(key)}] does not exist." + return node + + def _record(self, key: Any) -> None: + """Track that the top-level segment of ``key`` was interacted with.""" + self._interacted.add(str(key).split(".")[0]) + + # ------------------------------------------------------------------ # + # Matching — where family + # ------------------------------------------------------------------ # + def where(self, key: str, expected: Union[Any, Callable[[Any], bool]]) -> "AssertableJson": + """Assert that ``key`` matches ``expected``. + + ``expected`` may be a literal value, or a callable predicate that + receives the actual value and returns a truthy result. + """ + self._record(key) + actual = self._get(key) + if callable(expected): + assert expected(actual), f"Property [{self._full(key)}] was unexpected value [{actual!r}]." + else: + assert actual == expected, ( + f"Property [{self._full(key)}] does not match. Expected [{expected!r}], got [{actual!r}]." + ) + return self + + def where_not(self, key: str, expected: Union[Any, Callable[[Any], bool]]) -> "AssertableJson": + """Assert that ``key`` does NOT match ``expected`` (value or predicate).""" + self._record(key) + actual = self._get(key) + if callable(expected): + assert not expected(actual), ( + f"Property [{self._full(key)}] was not expected to pass the " + f"given predicate but did (value [{actual!r}])." + ) + else: + assert actual != expected, f"Property [{self._full(key)}] was not expected to equal [{expected!r}] but did." + return self + + def where_all(self, bindings: dict) -> "AssertableJson": + """Assert each ``{key: expected}`` pair via :meth:`where`.""" + for key, expected in bindings.items(): + self.where(key, expected) + return self + + def where_all_type(self, bindings: dict) -> "AssertableJson": + """Assert each ``{key: type}`` pair via :meth:`where_type`.""" + for key, expected in bindings.items(): + self.where_type(key, expected) + return self + + def where_type(self, key: str, expected: Union[str, list[str]]) -> "AssertableJson": + """Assert that ``key`` is one of the given JSON type name(s). + + Accepts a single name (``"string"``), a pipe form (``"string|null"``) + or a list (``["string", "null"]``). Supported names: string, integer, + double, number, boolean, array, object, null. + """ + self._record(key) + actual = self._get(key) + if isinstance(expected, str): + names = expected.split("|") + else: + names = list(expected) + for name in names: + if name not in _JSON_TYPES: + raise ValueError(f"Unknown JSON type [{name}] for [{self._full(key)}].") + assert any(_JSON_TYPES[name](actual) for name in names), ( + f"Property [{self._full(key)}] is not of expected type [{'|'.join(names)}]; got value [{actual!r}]." + ) + return self + + # ------------------------------------------------------------------ # + # Presence — has family + # ------------------------------------------------------------------ # + def has( + self, + key: str, + length: Any = _MISSING, + callback: Callable[["AssertableJson"], Any] = _MISSING, + ) -> "AssertableJson": + """Assert that ``key`` exists, optionally its length and/or a scope. + + Supports the shorthands: + + * ``has("key")`` — just assert presence. + * ``has("key", 3)`` — assert it is a sized collection of length 3. + * ``has("key", callback)`` — scope into ``key`` and run nested asserts. + * ``has("key", 3, callback)`` — both length and a nested scope. + """ + self._record(key) + value = self._get(key) + + # ``has("key", callback)`` shorthand: length slot holds the callback. + if callable(length): + callback = length + length = _MISSING + + if length is not _MISSING: + assert isinstance(value, (list, dict, str)), f"Property [{self._full(key)}] is not countable." + assert len(value) == length, ( + f"Property [{self._full(key)}] does not have the expected size. " + f"Expected [{length}], got [{len(value)}]." + ) + + if callback is not _MISSING: + child = AssertableJson(value, self._full(key)) + callback(child) + child._verify() + + return self + + def has_all(self, *keys: str) -> "AssertableJson": + """Assert that every key in ``keys`` exists. Accepts a list or varargs.""" + for key in self._flatten(keys): + assert self._exists(key), f"Property [{self._full(key)}] does not exist." + self._record(key) + return self + + def has_any(self, *keys: str) -> "AssertableJson": + """Assert that at least one key in ``keys`` exists.""" + keys = self._flatten(keys) + for key in keys: + self._record(key) + assert any(self._exists(key) for key in keys), ( + f"None of the expected properties [{', '.join(map(self._full, keys))}] exist." + ) + return self + + def missing(self, key: str) -> "AssertableJson": + """Assert that ``key`` is absent from the current scope.""" + assert not self._exists(key), f"Property [{self._full(key)}] was expected to be missing but exists." + return self + + def missing_all(self, *keys: str) -> "AssertableJson": + """Assert that none of ``keys`` exist.""" + for key in self._flatten(keys): + self.missing(key) + return self + + def count(self, key: str, length: int) -> "AssertableJson": + """Assert that the collection at ``key`` has exactly ``length`` items.""" + return self.has(key, length) + + @staticmethod + def _flatten(keys: tuple) -> tuple: + """Allow both ``has_all("a", "b")`` and ``has_all(["a", "b"])``.""" + if len(keys) == 1 and isinstance(keys[0], (list, tuple)): + return tuple(keys[0]) + return keys + + # ------------------------------------------------------------------ # + # Traversal — first / each + # ------------------------------------------------------------------ # + def first(self, callback: Callable[["AssertableJson"], Any]) -> "AssertableJson": + """Scope into the first child element and run ``callback`` against it.""" + assert self._data, f"Cannot scope into the first element of empty [{self._path or 'root'}]." + if isinstance(self._data, list): + key, value = 0, self._data[0] + else: + key, value = next(iter(self._data.items())) + child = AssertableJson(value, self._full(key)) + callback(child) + child._verify() + self._record(key) + return self + + def each(self, callback: Callable[["AssertableJson"], Any]) -> "AssertableJson": + """Run ``callback`` against every child element in its own scope.""" + if isinstance(self._data, list): + items = list(enumerate(self._data)) + elif isinstance(self._data, dict): + items = list(self._data.items()) + else: + raise AssertionError(f"Property [{self._path or 'root'}] is not iterable for each().") + for key, value in items: + child = AssertableJson(value, self._full(key)) + callback(child) + child._verify() + self._record(key) + return self + + # ------------------------------------------------------------------ # + # Interaction control + # ------------------------------------------------------------------ # + def etc(self) -> "AssertableJson": + """Acknowledge any remaining, un-asserted keys in this scope.""" + if isinstance(self._data, dict): + self._interacted.update(self._data.keys()) + return self + + def _verify(self) -> None: + """Assert every object property was interacted with (or ``etc``'d).""" + if isinstance(self._data, dict): + extra = set(self._data.keys()) - self._interacted + assert not extra, ( + f"Unexpected properties were found at [{self._path or 'root'}]: " + f"{sorted(extra)}. Interact with them or call .etc()." + ) + + # ------------------------------------------------------------------ # + # Phase-2 / TODO — explicit stubs. + # ------------------------------------------------------------------ # + def where_contains(self, key: str, expected: Any) -> "AssertableJson": # pragma: no cover + """TODO(phase-2): assert the collection at ``key`` contains ``expected``.""" + raise NotImplementedError("where_contains is planned for phase 2.") + + def count_between(self, key: str, minimum: int, maximum: int) -> "AssertableJson": # pragma: no cover + """TODO(phase-2): assert the size of ``key`` is within [min, max].""" + raise NotImplementedError("count_between is planned for phase 2.") + + +def assert_json_structure(structure: Any, data: Any, path: str = "root") -> None: + """Assert that ``data`` contains the keys described by ``structure``. + + ``structure`` may be: + + * a list of leaf key names, e.g. ``["name", "sport"]`` — each must exist + on the object ``data``; + * a dict mapping a key to a nested structure, e.g. + ``{"teams": ["name", "sport"]}``; + * a ``"*"`` key whose value is applied to every element of a list, e.g. + ``{"teams": {"*": ["name", "sport"]}}``. + + A leaf may map to ``None`` in dict form to assert presence only. + """ + if isinstance(structure, list): + for value in structure: + if isinstance(value, (list, dict)): + assert_json_structure(value, data, path) + else: + assert isinstance(data, dict) and value in data, f"Missing key [{value}] at [{path}]." + return + + if isinstance(structure, dict): + for key, value in structure.items(): + if key == "*": + assert isinstance(data, list), f"Expected a list at [{path}]." + for index, item in enumerate(data): + assert_json_structure(value, item, f"{path}.{index}") + else: + assert isinstance(data, dict) and key in data, f"Missing key [{key}] at [{path}]." + if value is not None: + assert_json_structure(value, data[key], f"{path}.{key}") + return + + raise TypeError(f"Invalid structure node at [{path}]: {structure!r}") diff --git a/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_case.py b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_case.py index 782b22be..ebd5ea24 100644 --- a/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_case.py +++ b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_case.py @@ -1,5 +1,6 @@ from abc import ABC +from fastapi_startkit.fastapi.testing.test_response import TestResponse from fastapi_startkit.testing import TestCase from httpx import AsyncClient, ASGITransport @@ -18,14 +19,17 @@ async def asyncTearDown(self): await self._client_ctx.__aexit__(None, None, None) await super().asyncTearDown() - async def get(self, url, **kwargs): - return await self.client.get(url, **kwargs) + async def get(self, url, **kwargs) -> TestResponse: + return TestResponse(await self.client.get(url, **kwargs)) - async def post(self, url, **kwargs): - return await self.client.post(url, **kwargs) + async def post(self, url, **kwargs) -> TestResponse: + return TestResponse(await self.client.post(url, **kwargs)) - async def put(self, url, **kwargs): - return await self.client.put(url, **kwargs) + async def put(self, url, **kwargs) -> TestResponse: + return TestResponse(await self.client.put(url, **kwargs)) - async def delete(self, url, **kwargs): - return await self.client.delete(url, **kwargs) + async def patch(self, url, **kwargs) -> TestResponse: + return TestResponse(await self.client.patch(url, **kwargs)) + + async def delete(self, url, **kwargs) -> TestResponse: + return TestResponse(await self.client.delete(url, **kwargs)) diff --git a/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_response.py b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_response.py new file mode 100644 index 00000000..f37e7bd4 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/fastapi/testing/test_response.py @@ -0,0 +1,102 @@ +"""A thin, chainable wrapper around ``httpx.Response`` for HTTP tests. + +:class:`TestResponse` adds assertion helpers +(``assert_status``, ``assert_ok``, ``assert_json``) while transparently +delegating every other attribute/method to the underlying ``httpx.Response`` +via ``__getattr__``. Existing tests that read ``response.status_code`` or call +``response.json()`` continue to work unchanged. +""" + +from __future__ import annotations + +from typing import Any, Callable, Optional + +from httpx import Response + +from fastapi_startkit.fastapi.testing.assertable_json import ( + AssertableJson, + assert_json_structure, +) + + +class TestResponse: + """Wraps an ``httpx.Response`` and adds fluent assertion helpers.""" + + # Prevent pytest from trying to collect this as a test class. + __test__ = False + + def __init__(self, response: Response) -> None: + self._response = response + + @property + def response(self) -> Response: + """Access the underlying ``httpx.Response`` directly.""" + return self._response + + def __getattr__(self, name: str) -> Any: + # Only called when ``name`` is not found on this wrapper, so this + # transparently forwards status_code, headers, json(), text, etc. + return getattr(self._response, name) + + def __repr__(self) -> str: # pragma: no cover - cosmetic + return f"" + + # ------------------------------------------------------------------ # + # Status assertions + # ------------------------------------------------------------------ # + def assert_status(self, code: int) -> "TestResponse": + """Assert the response has the given HTTP status code.""" + actual = self._response.status_code + assert actual == code, f"Expected status code [{code}] but received [{actual}]. Body: {self._response.text}" + return self + + def assert_ok(self) -> "TestResponse": + """Assert the response status code is 200 OK.""" + return self.assert_status(200) + + def assert_created(self) -> "TestResponse": + """Assert the response status code is 201 Created.""" + return self.assert_status(201) + + def assert_no_content(self) -> "TestResponse": + """Assert the response status code is 204 No Content.""" + return self.assert_status(204) + + # ------------------------------------------------------------------ # + # JSON assertions + # ------------------------------------------------------------------ # + def assert_json( + self, + callback: Optional[Callable[[AssertableJson], Any]] = None, + *, + exact: Any = None, + ) -> "TestResponse": + """Assert against the JSON body. + + * ``assert_json(lambda j: j.where(...))`` — run fluent assertions; the + strict interaction model is enforced (unexpected keys fail unless + ``etc()`` is called). + * ``assert_json(exact={...})`` — assert the decoded body equals + ``exact`` exactly. + """ + body = self._response.json() + if callback is not None: + fluent = AssertableJson(body) + callback(fluent) + fluent._verify() + if exact is not None: + assert body == exact, f"JSON body does not match exactly.\nExpected: {exact!r}\nActual: {body!r}" + return self + + def assert_json_structure(self, structure: Any) -> "TestResponse": + """Assert the JSON body has the given key structure. + + Use a ``"*"`` key to apply a nested structure to every element of a + list:: + + response.assert_json_structure({ + "teams": {"*": ["name", "sport"]}, + }) + """ + assert_json_structure(structure, self._response.json()) + return self diff --git a/fastapi_startkit/tests/fastapi/test_assertable_json.py b/fastapi_startkit/tests/fastapi/test_assertable_json.py new file mode 100644 index 00000000..d1cbf1a1 --- /dev/null +++ b/fastapi_startkit/tests/fastapi/test_assertable_json.py @@ -0,0 +1,249 @@ +"""Unit tests for the fluent JSON assertion API (AssertableJson / TestResponse).""" + +import pytest + +from fastapi_startkit.fastapi.testing.assertable_json import ( + AssertableJson, + assert_json_structure, +) + + +def aj(data): + return AssertableJson(data) + + +# --------------------------------------------------------------------------- # +# where / where_not / where_all +# --------------------------------------------------------------------------- # +def test_where_happy_path(): + aj({"id": 1, "name": "Bedu"}).where("id", 1).where("name", "Bedu").etc() + + +def test_where_mismatch_raises_with_full_path(): + with pytest.raises(AssertionError) as exc: + aj({"id": 1}).where("id", 2) + assert "[id]" in str(exc.value) + + +def test_where_missing_property_raises(): + with pytest.raises(AssertionError) as exc: + aj({"id": 1}).where("missing", 1) + assert "[missing] does not exist" in str(exc.value) + + +def test_where_predicate_matcher(): + aj({"age": 30}).where("age", lambda v: v > 18).etc() + + +def test_where_predicate_matcher_failure(): + with pytest.raises(AssertionError): + aj({"age": 10}).where("age", lambda v: v > 18) + + +def test_where_dotted_key(): + data = {"user": {"profile": {"email": "a@b.com"}}} + aj(data).where("user.profile.email", "a@b.com").etc() + + +def test_where_not(): + aj({"role": "user"}).where_not("role", "admin").etc() + with pytest.raises(AssertionError): + aj({"role": "admin"}).where_not("role", "admin") + + +def test_where_all(): + aj({"a": 1, "b": 2}).where_all({"a": 1, "b": 2}).etc() + + +# --------------------------------------------------------------------------- # +# where_type +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize( + "value,type_name", + [ + ("x", "string"), + (5, "integer"), + (1.5, "double"), + (True, "boolean"), + ([], "array"), + ({}, "object"), + (None, "null"), + ], +) +def test_where_type_matches(value, type_name): + aj({"v": value}).where_type("v", type_name).etc() + + +def test_where_type_bool_is_not_integer(): + with pytest.raises(AssertionError): + aj({"v": True}).where_type("v", "integer") + + +def test_where_type_union(): + aj({"v": None}).where_type("v", "string|null").etc() + aj({"v": "x"}).where_type("v", ["string", "null"]).etc() + + +# --------------------------------------------------------------------------- # +# has / has_all / has_any / missing / count +# --------------------------------------------------------------------------- # +def test_has_presence(): + aj({"id": 1}).has("id").etc() + + +def test_has_length(): + aj({"items": [1, 2, 3]}).has("items", 3).etc() + + +def test_has_length_mismatch(): + with pytest.raises(AssertionError): + aj({"items": [1, 2]}).has("items", 3) + + +def test_has_nested_scope(): + data = {"profile": {"email": "a@b.com", "name": "Bedu"}} + aj(data).has( + "profile", + lambda p: p.where("email", "a@b.com").where("name", "Bedu"), + ).etc() + + +def test_has_length_and_scope(): + data = {"rows": [{"id": 1}, {"id": 2}]} + aj(data).has("rows", 2, lambda r: r.first(lambda f: f.where("id", 1).etc())).etc() + + +def test_has_all_and_has_any(): + aj({"a": 1, "b": 2}).has_all("a", "b").etc() + aj({"a": 1, "b": 2}).has_all(["a", "b"]).etc() + aj({"a": 1}).has_any("a", "z").etc() + + +def test_has_any_failure(): + with pytest.raises(AssertionError): + aj({"a": 1}).has_any("x", "y") + + +def test_missing_and_missing_all(): + aj({"a": 1}).missing("b").has("a").etc() + aj({"a": 1}).missing_all("b", "c").has("a").etc() + with pytest.raises(AssertionError): + aj({"a": 1}).missing("a") + + +def test_count(): + aj({"items": [1, 2]}).count("items", 2).etc() + + +# --------------------------------------------------------------------------- # +# first / each +# --------------------------------------------------------------------------- # +def test_first_on_list(): + data = {"users": [{"id": 1}, {"id": 2}]} + aj(data).has("users", lambda u: u.first(lambda f: f.where("id", 1).etc())).etc() + + +def test_each_on_list(): + data = [{"id": 1}, {"id": 2}] + aj(data).each(lambda item: item.has("id")) + + +def test_each_empty_list_ok(): + aj([]).each(lambda item: item.where("id", 1)) + + +# --------------------------------------------------------------------------- # +# etc() and strict interaction verification (_verify) +# --------------------------------------------------------------------------- # +def test_unasserted_prop_fails(): + with pytest.raises(AssertionError) as exc: + fluent = aj({"id": 1, "secret": "leak"}).where("id", 1) + fluent._verify() + assert "secret" in str(exc.value) + + +def test_etc_acknowledges_remaining(): + fluent = aj({"id": 1, "secret": "leak"}).where("id", 1).etc() + fluent._verify() # should not raise + + +def test_verify_nested_scope_enforced(): + data = {"profile": {"email": "a@b.com", "name": "Bedu"}} + with pytest.raises(AssertionError) as exc: + aj(data).has("profile", lambda p: p.where("email", "a@b.com")).etc() + assert "name" in str(exc.value) + + +# --------------------------------------------------------------------------- # +# where_all_type +# --------------------------------------------------------------------------- # +def test_where_all_type(): + data = {"name": "Phoenix Suns", "sport": "basketball"} + aj(data).where_all_type({"name": "string", "sport": "string"}).etc() + + +def test_where_all_type_failure(): + with pytest.raises(AssertionError): + aj({"name": "x", "rank": 1}).where_all_type({"name": "string", "rank": "string"}) + + +# --------------------------------------------------------------------------- # +# Dotted keys that index into lists (e.g. "teams.0") +# --------------------------------------------------------------------------- # +def test_has_dotted_list_index_scope(): + data = {"teams": [{"name": "Phoenix Suns", "sport": "basketball"}]} + aj(data).has("teams", 1).has( + "teams.0", + lambda team: team.where("name", "Phoenix Suns").etc(), + ).etc() + + +def test_where_dotted_list_index(): + data = {"teams": [{"name": "Phoenix Suns"}]} + aj(data).where("teams.0.name", "Phoenix Suns").etc() + + +# --------------------------------------------------------------------------- # +# assert_json_structure +# --------------------------------------------------------------------------- # +def test_structure_simple_list(): + assert_json_structure(["name", "sport"], {"name": "x", "sport": "y"}) + + +def test_structure_missing_key_fails(): + with pytest.raises(AssertionError) as exc: + assert_json_structure(["name", "sport"], {"name": "x"}) + assert "sport" in str(exc.value) + + +def test_structure_wildcard_over_list(): + data = {"teams": [{"name": "a", "sport": "b"}, {"name": "c", "sport": "d"}]} + assert_json_structure({"teams": {"*": ["name", "sport"]}}, data) + + +def test_structure_wildcard_failure(): + data = {"teams": [{"name": "a", "sport": "b"}, {"name": "c"}]} + with pytest.raises(AssertionError) as exc: + assert_json_structure({"teams": {"*": ["name", "sport"]}}, data) + assert "teams.1" in str(exc.value) + + +def test_structure_nested_dict(): + data = {"user": {"profile": {"email": "a@b.com"}}} + assert_json_structure({"user": {"profile": ["email"]}}, data) + + +def test_structure_leaf_none_presence(): + assert_json_structure({"meta": None, "data": ["id"]}, {"meta": 1, "data": {"id": 5}}) + + +# --------------------------------------------------------------------------- # +# Remaining phase-2 stubs +# --------------------------------------------------------------------------- # +def test_phase2_stubs_raise_not_implemented(): + for call in ( + lambda: aj({"a": [1]}).where_contains("a", 1), + lambda: aj({"a": [1]}).count_between("a", 1, 2), + ): + with pytest.raises(NotImplementedError): + call() diff --git a/fastapi_startkit/tests/fastapi/test_test_response.py b/fastapi_startkit/tests/fastapi/test_test_response.py new file mode 100644 index 00000000..95a5f9a4 --- /dev/null +++ b/fastapi_startkit/tests/fastapi/test_test_response.py @@ -0,0 +1,79 @@ +"""Unit tests for the TestResponse httpx wrapper + fluent assert_json.""" + +import json as jsonlib + +import httpx +import pytest + +from fastapi_startkit.fastapi.testing.test_response import TestResponse + + +def make_response(status_code=200, payload=None): + content = jsonlib.dumps(payload if payload is not None else {}) + raw = httpx.Response( + status_code=status_code, + content=content.encode(), + headers={"content-type": "application/json"}, + ) + return TestResponse(raw) + + +def test_passthrough_attributes(): + resp = make_response(200, {"message": "ok"}) + assert resp.status_code == 200 + assert resp.json() == {"message": "ok"} + assert resp.text == '{"message": "ok"}' + + +def test_assert_status_and_ok(): + resp = make_response(200, {"a": 1}) + assert resp.assert_status(200).assert_ok() is resp + + +def test_assert_status_failure_includes_body(): + resp = make_response(500, {"error": "boom"}) + with pytest.raises(AssertionError) as exc: + resp.assert_status(200) + assert "boom" in str(exc.value) + + +def test_assert_created_and_no_content(): + make_response(201, {}).assert_created() + make_response(204, {}).assert_no_content() + + +def test_assert_json_fluent_chain(): + resp = make_response(200, {"id": 1, "name": "Bedu"}) + resp.assert_ok().assert_json(lambda j: j.where("id", 1).where("name", "Bedu").etc()) + + +def test_assert_json_strict_failure_on_extra_key(): + resp = make_response(200, {"id": 1, "secret": "leak"}) + with pytest.raises(AssertionError) as exc: + resp.assert_json(lambda j: j.where("id", 1)) + assert "secret" in str(exc.value) + + +def test_assert_json_exact(): + resp = make_response(200, {"id": 1}) + resp.assert_json(exact={"id": 1}) + with pytest.raises(AssertionError): + resp.assert_json(exact={"id": 2}) + + +def test_assert_json_structure_wildcard(): + payload = {"teams": [{"name": "Suns", "sport": "b"}, {"name": "Cardinals", "sport": "f"}]} + resp = make_response(200, payload) + resp.assert_ok().assert_json_structure({"teams": {"*": ["name", "sport"]}}) + + +def test_assert_json_structure_failure(): + resp = make_response(200, {"teams": [{"name": "Suns"}]}) + with pytest.raises(AssertionError): + resp.assert_json_structure({"teams": {"*": ["name", "sport"]}}) + + +def test_assert_json_dotted_list_index_scope(): + payload = {"teams": [{"name": "Phoenix Suns", "sport": "basketball"}]} + resp = make_response(200, payload) + resp.assert_json(lambda j: j.has("teams", 1).has("teams.0", lambda team: team.where("name", "Phoenix Suns").etc())) diff --git a/fastapi_startkit/uv.lock b/fastapi_startkit/uv.lock index b07acda4..de9262fe 100644 --- a/fastapi_startkit/uv.lock +++ b/fastapi_startkit/uv.lock @@ -593,7 +593,7 @@ wheels = [ [[package]] name = "fastapi-startkit" -version = "0.43.0" +version = "0.44.0" source = { editable = "." } dependencies = [ { name = "cleo" },