HttpPlugin

HttpPlugin

HttpPlugin(verifier, require_response=False)

Bases: BasePlugin

HTTP interception plugin. Requires bigfoot[http] extra.

Patches httpx sync/async transports, requests HTTPAdapter, urllib openers, aiohttp ClientSession (if installed), and asyncio.BaseEventLoop.run_in_executor at the class level. Uses reference counting so nested sandboxes work correctly.

Source code in src/bigfoot/plugins/http.py

def __init__(self, verifier: "StrictVerifier", require_response: bool = False) -> None:
    super().__init__(verifier)
    self._mock_queue: list[HttpMockConfig] = []
    self._sentinel = HttpRequestSentinel(self)
    self._pass_through_rules: list[tuple[str, str]] = []
    self._asserting_request_only: bool = False
    self._require_response: bool = require_response
    self.load_config(
        self.verifier._bigfoot_config.get(self.config_key() or "", {})
    )

request property

request

Sentinel used as source argument in verifier.assert_interaction().

config_key classmethod

config_key()

Return 'http', mapping this plugin to [tool.bigfoot.http].

Source code in src/bigfoot/plugins/http.py

@classmethod
def config_key(cls) -> str | None:
    """Return 'http', mapping this plugin to [tool.bigfoot.http]."""
    return "http"

load_config

load_config(config)

Apply [tool.bigfoot.http] configuration.

Recognized keys: require_response (bool): When True, assert_request() returns an HttpAssertionBuilder requiring .assert_response() to complete the assertion. Default False.

Unknown keys are silently ignored for forward-compatibility. Raises TypeError for require_response with a non-bool value.

Source code in src/bigfoot/plugins/http.py

def load_config(self, config: dict[str, Any]) -> None:
    """Apply [tool.bigfoot.http] configuration.

    Recognized keys:
        require_response (bool): When True, assert_request() returns an
            HttpAssertionBuilder requiring .assert_response() to complete
            the assertion. Default False.

    Unknown keys are silently ignored for forward-compatibility.
    Raises TypeError for require_response with a non-bool value.
    """
    if "require_response" in config:
        val = config["require_response"]
        if not isinstance(val, bool):
            raise TypeError(
                f"[tool.bigfoot.http] require_response must be a bool, "
                f"got {type(val).__name__}"
            )
        self._require_response = val

assert_request

assert_request(method, url, headers=None, body='', require_response=None)

Assert an HTTP request interaction, optionally requiring a chained response assertion.

When require_response is False (the default, or when the instance was constructed with require_response=False), this method is terminal: it asserts only the four request fields and returns None.

When require_response is True (either via the per-call argument or the instance default), this method returns an HttpAssertionBuilder. The caller must chain .assert_response() on the builder to complete the assertion with all seven fields.

Source code in src/bigfoot/plugins/http.py

def assert_request(
    self,
    method: str,
    url: str,
    headers: dict[str, Any] | None = None,
    body: str = "",
    require_response: bool | None = None,
) -> "HttpAssertionBuilder | None":
    """Assert an HTTP request interaction, optionally requiring a chained response assertion.

    When ``require_response`` is False (the default, or when the instance was
    constructed with ``require_response=False``), this method is terminal: it
    asserts only the four request fields and returns ``None``.

    When ``require_response`` is True (either via the per-call argument or the
    instance default), this method returns an ``HttpAssertionBuilder``.  The
    caller must chain ``.assert_response()`` on the builder to complete the
    assertion with all seven fields.
    """
    effective = require_response if require_response is not None else self._require_response
    if not effective:
        self._asserting_request_only = True
        try:
            self.verifier.assert_interaction(
                self._sentinel,
                method=method,
                url=url,
                request_headers=headers if headers is not None else {},
                request_body=body,
            )
        finally:
            self._asserting_request_only = False
        return None
    return HttpAssertionBuilder(
        verifier=self.verifier,
        sentinel=self._sentinel,
        plugin=self,
        method=method,
        url=url,
        headers=headers if headers is not None else {},
        body=body,
    )

mock_response

mock_response(method, url, *, json=None, body=None, status=200, headers=None, params=None, required=True)

Register a mock response for the given method + URL pair.

Source code in src/bigfoot/plugins/http.py

def mock_response(
    self,
    method: str,
    url: str,
    *,
    json: object = None,
    body: str | bytes | None = None,
    status: int = 200,
    headers: dict[str, str] | None = None,
    params: dict[str, str] | None = None,
    required: bool = True,
) -> None:
    """Register a mock response for the given method + URL pair."""
    if json is not None and body is not None:
        raise ValueError("json and body are mutually exclusive")

    response_headers: dict[str, str] = headers or {}
    if json is not None:
        response_body = json_module.dumps(json).encode("utf-8")
        response_headers.setdefault("content-type", "application/json")
    elif body is not None:
        response_body = body.encode("utf-8") if isinstance(body, str) else body
    else:
        response_body = b""

    self._mock_queue.append(
        HttpMockConfig(
            method=method.upper(),
            url=url,
            params=params,
            response_status=status,
            response_headers=response_headers,
            response_body=response_body,
            required=required,
        )
    )

pass_through

pass_through(method, url)

Register a permanent pass-through rule for the given method + URL.

Requests matching this rule are forwarded to the real backend instead of raising UnmockedInteractionError. The interaction is still recorded on the timeline and must be asserted.

The URL must match exactly (scheme, host, path). Query parameters are not considered for pass-through rule matching.

Source code in src/bigfoot/plugins/http.py

def pass_through(self, method: str, url: str) -> None:
    """Register a permanent pass-through rule for the given method + URL.

    Requests matching this rule are forwarded to the real backend instead
    of raising UnmockedInteractionError. The interaction is still recorded
    on the timeline and must be asserted.

    The URL must match exactly (scheme, host, path). Query parameters are
    not considered for pass-through rule matching.
    """
    self._pass_through_rules.append((method.upper(), url))

activate

activate()

Reference-counted class-level patch installation.

Source code in src/bigfoot/plugins/http.py

def activate(self) -> None:
    """Reference-counted class-level patch installation."""
    with HttpPlugin._install_lock:
        if HttpPlugin._install_count == 0:
            self._check_conflicts()
            self._install_patches()
        HttpPlugin._install_count += 1

assertable_fields

assertable_fields(interaction)

Return the field names required in **expected when asserting an HTTP interaction.

When _asserting_request_only is True (set by the terminal path of assert_request()), only the four request fields are required. Otherwise all seven fields are required.

Source code in src/bigfoot/plugins/http.py

def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
    """Return the field names required in **expected when asserting an HTTP interaction.

    When ``_asserting_request_only`` is True (set by the terminal path of
    ``assert_request()``), only the four request fields are required.
    Otherwise all seven fields are required.
    """
    if self._asserting_request_only:
        return frozenset({"method", "url", "request_headers", "request_body"})
    return frozenset(
        {
            "method", "url", "request_headers", "request_body",
            "status", "response_headers", "response_body",
        }
    )