Skip to content

HTTP

The HTTP response assertable, the request/message builders, and the framework adapters. See the HTTP guide for examples.

Response

pyssertive.http.FluentResponse

Bases: DebugResponseMixin, CookieAssertionsMixin, HTMLContentAssertionsMixin, JsonContentAssertionsMixin, HeaderAssertionsMixin, HttpStatusAssertionsMixin

Framework-agnostic fluent assertion wrapper for HTTP responses.

Concrete adapter subclasses (e.g. pyssertive.adapters.django.FluentResponse or pyssertive.adapters.httpx.FluentResponse) define their own __init__ with the framework-native response type and override properties when the underlying response shape diverges (e.g. httpx exposes charset_encoding instead of charset).

Every concrete FluentResponse satisfies :class:ResponseProtocol, so developers can write helpers typed against that protocol and have them work uniformly across adapters.

wrapped property

wrapped: Any

status_code property

status_code: int

content property

content: bytes

headers property

headers: Any

cookies property

cookies: Any

charset property

charset: str | None

__getattr__

__getattr__(name: str) -> Any
Source code in src/pyssertive/http/response.py
def __getattr__(self, name: str) -> Any:
    return getattr(self._response, name)

Message builder

pyssertive.http.mcp.MessageBuilder

MessageBuilder(
    request_builder: RequestBuilder[TBuilt],
    *,
    path: str = DEFAULT_PATH,
)

Bases: Generic[TBuilt]

Source code in src/pyssertive/http/mcp.py
def __init__(self, request_builder: RequestBuilder[TBuilt], *, path: str = DEFAULT_PATH) -> None:
    self._rb = request_builder
    self._rb.with_method("POST")
    self._rb.with_path(path)
    self._rb.with_header("Content-Type", "application/json")
    self._rb.with_header("Accept", "application/json, text/event-stream")

    self._method: str | None = None
    self._params: dict[str, Any] | None = None
    self._id: int | str = 1
    self._is_notification: bool = False

DEFAULT_PATH class-attribute instance-attribute

DEFAULT_PATH = '/mcp'

with_id

with_id(msg_id: int | str) -> Self
Source code in src/pyssertive/http/mcp.py
def with_id(self, msg_id: int | str) -> Self:
    if self._is_notification:
        raise ValueError("Notifications cannot have an id")
    self._id = msg_id
    return self

with_auth_token

with_auth_token(token: str) -> Self
Source code in src/pyssertive/http/mcp.py
def with_auth_token(self, token: str) -> Self:
    self._rb.with_header("Authorization", f"Token {token}")
    return self

with_protocol_version

with_protocol_version(version: str) -> Self
Source code in src/pyssertive/http/mcp.py
def with_protocol_version(self, version: str) -> Self:
    self._rb.with_header("MCP-Protocol-Version", version)
    return self

with_session_id

with_session_id(session_id: str) -> Self
Source code in src/pyssertive/http/mcp.py
def with_session_id(self, session_id: str) -> Self:
    self._rb.with_header("MCP-Session-Id", session_id)
    return self

initialize

initialize(
    *,
    protocol: str = "2025-11-25",
    client_name: str = "pyssertive-test",
    client_version: str = "0.0.0",
    capabilities: dict[str, Any] | None = None,
) -> Self
Source code in src/pyssertive/http/mcp.py
def initialize(
    self,
    *,
    protocol: str = "2025-11-25",
    client_name: str = "pyssertive-test",
    client_version: str = "0.0.0",
    capabilities: dict[str, Any] | None = None,
) -> Self:
    self._is_notification = False
    self._method = "initialize"
    self._params = {
        "protocolVersion": protocol,
        "clientInfo": {"name": client_name, "version": client_version},
        "capabilities": capabilities if capabilities is not None else {},
    }
    return self

listing_tools

listing_tools(*, cursor: str | None = None) -> Self
Source code in src/pyssertive/http/mcp.py
def listing_tools(self, *, cursor: str | None = None) -> Self:
    return self._set_listing("tools/list", cursor)

calling_tool

calling_tool(
    name: str, *, arguments: dict[str, Any] | None = None
) -> Self
Source code in src/pyssertive/http/mcp.py
def calling_tool(self, name: str, *, arguments: dict[str, Any] | None = None) -> Self:
    self._is_notification = False
    self._method = "tools/call"
    self._params = {"name": name, "arguments": arguments if arguments is not None else {}}
    return self

listing_resources

listing_resources(*, cursor: str | None = None) -> Self
Source code in src/pyssertive/http/mcp.py
def listing_resources(self, *, cursor: str | None = None) -> Self:
    return self._set_listing("resources/list", cursor)

reading_resource

reading_resource(uri: str) -> Self
Source code in src/pyssertive/http/mcp.py
def reading_resource(self, uri: str) -> Self:
    self._is_notification = False
    self._method = "resources/read"
    self._params = {"uri": uri}
    return self

listing_prompts

listing_prompts(*, cursor: str | None = None) -> Self
Source code in src/pyssertive/http/mcp.py
def listing_prompts(self, *, cursor: str | None = None) -> Self:
    return self._set_listing("prompts/list", cursor)

getting_prompt

getting_prompt(
    name: str, *, arguments: dict[str, Any] | None = None
) -> Self
Source code in src/pyssertive/http/mcp.py
def getting_prompt(self, name: str, *, arguments: dict[str, Any] | None = None) -> Self:
    self._is_notification = False
    self._method = "prompts/get"
    self._params = {"name": name, "arguments": arguments if arguments is not None else {}}
    return self

notifying

notifying(
    method: str, *, params: dict[str, Any] | None = None
) -> Self
Source code in src/pyssertive/http/mcp.py
def notifying(self, method: str, *, params: dict[str, Any] | None = None) -> Self:
    self._is_notification = True
    self._method = method
    self._params = params
    return self

calling

calling(method: str) -> Self
Source code in src/pyssertive/http/mcp.py
def calling(self, method: str) -> Self:
    self._is_notification = False
    self._method = method
    self._params = None
    return self

with_params

with_params(params: dict[str, Any]) -> Self
Source code in src/pyssertive/http/mcp.py
def with_params(self, params: dict[str, Any]) -> Self:
    self._params = params
    return self

build

build() -> TBuilt
Source code in src/pyssertive/http/mcp.py
def build(self) -> TBuilt:
    if self._method is None:
        raise ValueError("Cannot build MCP message: no method set")

    envelope: dict[str, Any] = {"jsonrpc": "2.0"}
    if not self._is_notification:
        envelope["id"] = self._id
    envelope["method"] = self._method

    if self._is_notification:
        if self._params is not None:
            envelope["params"] = self._params
    else:
        envelope["params"] = self._params if self._params is not None else {}

    self._rb.with_body(envelope)
    return self._rb.build()

Django adapter

pyssertive.adapters.django.FluentHttpAssertClient

FluentHttpAssertClient(base_client: Client)

Fluent wrapper for Django's test client.

Example::

client = FluentHttpAssertClient(Client())
client.get('/api/').assert_ok().assert_json()
Source code in src/pyssertive/adapters/django/client.py
def __init__(self, base_client: Client) -> None:
    self._client: Client = base_client

get

get(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def get(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.get(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

post

post(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def post(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.post(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

put

put(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def put(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.put(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

patch

patch(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def patch(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.patch(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

delete

delete(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def delete(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.delete(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

head

head(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def head(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.head(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

options

options(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def options(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.options(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

trace

trace(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/django/client.py
def trace(self, *args: Any, **kwargs: Any) -> FluentResponse:
    response = self._client.trace(*args, **kwargs)
    return FluentResponse(response)  # type: ignore[arg-type]

__getattr__

__getattr__(name: str) -> Any
Source code in src/pyssertive/adapters/django/client.py
def __getattr__(self, name: str) -> Any:
    return getattr(self._client, name)

pyssertive.adapters.django.DjangoRequestBuilder

DjangoRequestBuilder(
    rf: RequestFactory | None = None,
    method: str = "GET",
    base_url: str | None = None,
    path: str = "/",
    data: dict[str, Any] | None = None,
)

Bases: BaseRequestBuilder[HttpRequest]

Source code in src/pyssertive/adapters/django/request.py
def __init__(
    self,
    rf: RequestFactory | None = None,
    method: str = "GET",
    base_url: str | None = None,
    path: str = "/",
    data: dict[str, Any] | None = None,
) -> None:
    super().__init__(method=method, base_url=base_url, path=path, data=data)
    self.rf = rf or RequestFactory()
    self._user: AbstractBaseUser | None = None
    self._meta: dict[str, Any] = {}
    self._custom_properties: dict[str, Any] = {}

rf instance-attribute

rf = rf or RequestFactory()

with_user

with_user(user: AbstractBaseUser) -> Self
Source code in src/pyssertive/adapters/django/request.py
def with_user(self, user: AbstractBaseUser) -> Self:
    self._user = user
    return self

with_meta

with_meta(key: str, value: Any) -> Self
Source code in src/pyssertive/adapters/django/request.py
def with_meta(self, key: str, value: Any) -> Self:
    self._meta[key] = str(value)
    return self

with_property

with_property(name: str, value: Any) -> Self
Source code in src/pyssertive/adapters/django/request.py
def with_property(self, name: str, value: Any) -> Self:
    self._custom_properties[name] = value
    return self

build

build() -> HttpRequest
Source code in src/pyssertive/adapters/django/request.py
def build(self) -> HttpRequest:
    import json
    from urllib.parse import urlencode

    method_map: dict[str, Callable[..., HttpRequest]] = {
        "GET": self.rf.get,
        "POST": self.rf.post,
        "PUT": self.rf.put,
        "PATCH": self.rf.patch,
        "DELETE": self.rf.delete,
        "HEAD": self.rf.head,
        "OPTIONS": self.rf.options,
    }
    if self._method not in method_map:
        raise ValueError(f"Unsupported HTTP method: {self._method}")

    path = self._path
    if self._query:
        qs = urlencode(self._query, doseq=True)
        path = f"{path}{'&' if '?' in path else '?'}{qs}"

    content_type = self._headers.get("Content-Type") or self._headers.get("content-type")
    body: Any = self._body if self._body is not None else {}
    rf_call = method_map[self._method]
    if content_type and "application/json" in content_type and isinstance(body, dict):
        body = json.dumps(body).encode("utf-8")
        request: HttpRequest = rf_call(path, body, content_type=content_type, headers=self._headers)
    else:
        request = rf_call(path, body, headers=self._headers)

    if self._user:
        request.user = self._user  # type: ignore[assignment]
    for key, value in self._cookies.items():
        request.COOKIES[key] = value
    for key, value in self._meta.items():
        request.META[key] = value
    for key, value in self._custom_properties.items():
        setattr(request, key, value)

    return request

Database helpers

pyssertive.adapters.django.db

assert_queryset_equal

assert_queryset_equal(
    actual: QuerySet | list[Any] | Iterator[Any],
    expected: Collection,
    *,
    transform: Callable[[Any], Any] = repr,
    ordered: bool = True,
) -> None
Source code in src/pyssertive/adapters/django/db.py
def assert_queryset_equal(
    actual: QuerySet | list[Any] | Iterator[Any],
    expected: Collection,
    *,
    transform: Callable[[Any], Any] = repr,
    ordered: bool = True,
) -> None:
    TransactionTestCase().assertQuerySetEqual(actual, expected, transform=transform, ordered=ordered)

assert_num_queries

assert_num_queries(
    expected: int,
) -> Generator[None, Any, None]
Source code in src/pyssertive/adapters/django/db.py
@contextmanager
def assert_num_queries(expected: int) -> Generator[None, Any, None]:
    with CaptureQueriesContext(connection) as ctx:
        yield
        actual = len(ctx)
        assert actual == expected, f"Expected {expected} queries, but got {actual}.\nQueries:\n" + "\n".join(
            q["sql"] for q in ctx.captured_queries
        )

assert_model_exists

assert_model_exists(model: type[Model], **filters) -> None
Source code in src/pyssertive/adapters/django/db.py
def assert_model_exists(model: type[Model], **filters) -> None:
    assert model.objects.filter(**filters).exists(), f"{model.__name__} does not exist with filters: {filters}"  # type: ignore[attr-defined]

assert_model_not_exists

assert_model_not_exists(
    model: type[Model], **filters
) -> None
Source code in src/pyssertive/adapters/django/db.py
def assert_model_not_exists(model: type[Model], **filters) -> None:
    assert not model.objects.filter(**filters).exists(), f"{model.__name__} unexpectedly exists with filters: {filters}"  # type: ignore[attr-defined]

assert_model_count

assert_model_count(
    model: type[Model], expected: int, **filters
) -> None
Source code in src/pyssertive/adapters/django/db.py
def assert_model_count(model: type[Model], expected: int, **filters) -> None:
    actual = model.objects.filter(**filters).count()  # type: ignore[attr-defined]
    assert actual == expected, f"Expected {expected} records for {model.__name__}, got {actual}"

assert_model_soft_deleted

assert_model_soft_deleted(
    model: type[Model], **filters
) -> None

Assumes model uses soft-deletion with a deleted_at datetime field.

Source code in src/pyssertive/adapters/django/db.py
def assert_model_soft_deleted(model: type[Model], **filters) -> None:
    """Assumes model uses soft-deletion with a `deleted_at` datetime field."""
    obj = model.objects.filter(**filters).first()  # type: ignore[attr-defined]
    assert obj is not None, f"{model.__name__} does not exist with filters: {filters}"
    assert hasattr(obj, "deleted_at"), f"{model.__name__} has no 'deleted_at' field"
    assert obj.deleted_at is not None, f"{model.__name__} is not soft deleted"

httpx adapter

pyssertive.adapters.httpx.FluentHttpAssertClient

FluentHttpAssertClient(base_client: Client)
Source code in src/pyssertive/adapters/httpx/client.py
def __init__(self, base_client: httpx.Client) -> None:
    self._client: httpx.Client = base_client

get

get(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def get(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.get(*args, **kwargs))

post

post(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def post(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.post(*args, **kwargs))

put

put(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def put(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.put(*args, **kwargs))

patch

patch(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def patch(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.patch(*args, **kwargs))

delete

delete(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def delete(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.delete(*args, **kwargs))

head

head(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def head(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.head(*args, **kwargs))

options

options(*args: Any, **kwargs: Any) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def options(self, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.options(*args, **kwargs))

request

request(
    method: str, *args: Any, **kwargs: Any
) -> FluentResponse
Source code in src/pyssertive/adapters/httpx/client.py
def request(self, method: str, *args: Any, **kwargs: Any) -> FluentResponse:
    return FluentResponse(self._client.request(method, *args, **kwargs))

__getattr__

__getattr__(name: str) -> Any
Source code in src/pyssertive/adapters/httpx/client.py
def __getattr__(self, name: str) -> Any:
    return getattr(self._client, name)

pyssertive.adapters.httpx.HttpxRequestBuilder

HttpxRequestBuilder(
    method: str = "GET",
    base_url: str | None = None,
    path: str = "/",
    data: dict[str, Any] | None = None,
)

Bases: BaseRequestBuilder[Request]

Source code in src/pyssertive/http/request.py
def __init__(
    self,
    method: str = "GET",
    base_url: str | None = None,
    path: str = "/",
    data: dict[str, Any] | None = None,
) -> None:
    self._method: str = method.upper()
    self._headers: dict[str, str] = {}
    self._cookies: dict[str, str] = {}
    self._query: dict[str, Any] = {}
    self._body: Any = data
    self._base_url: str | None = base_url
    self._path: str = "/"
    self.with_path(path)

build

build() -> Request
Source code in src/pyssertive/adapters/httpx/request.py
def build(self) -> httpx.Request:
    kwargs: dict[str, Any] = {
        "method": self._method,
        "url": self._path,
        "headers": self._headers or None,
        "params": self._query or None,
        "cookies": self._cookies or None,
    }
    if self._body is not None:
        kwargs["json"] = self._body
    return httpx.Request(**kwargs)