HttpPlugin

HttpPlugin

HttpPlugin(verifier, require_response=True)

Bases: BasePlugin

HTTP interception plugin. Requires python-tripwire[http] extra.

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

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

def __init__(self, verifier: "StrictVerifier", require_response: bool = True) -> None:
    super().__init__(verifier)
    self._mock_queue: list[HttpMockEntry] = []
    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._tripwire_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.tripwire.http].

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

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

load_config

load_config(config)

Apply [tool.tripwire.http] configuration.

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

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

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

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

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

    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.tripwire.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='', raised=_ABSENT, require_response=None)

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

When raised is provided, the assertion is always terminal (error interactions have no response to chain). Returns None.

When require_response is False, this method is terminal: it asserts only the four request fields and returns None.

When require_response is True, this method returns an HttpAssertionBuilder.

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

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

    When ``raised`` is provided, the assertion is always terminal (error
    interactions have no response to chain). Returns ``None``.

    When ``require_response`` is False, this method is terminal:
    it asserts only the four request fields and returns ``None``.

    When ``require_response`` is True, this method returns an
    ``HttpAssertionBuilder``.
    """
    if raised is not _ABSENT:
        # Error assertion: always request-only (no response to assert)
        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,
                raised=raised,
            )
        finally:
            self._asserting_request_only = False
        return None

    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/tripwire/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,
        )
    )

mock_error

mock_error(method, url, *, raises, params=None, required=True)

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

When the interceptor matches this mock, the interaction is recorded with request fields + raised, then the exception is re-raised into the code under test.

The error config is appended to the unified mock queue alongside HttpMockConfig entries, preserving FIFO ordering for mixed success/error sequences.

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

def mock_error(
    self,
    method: str,
    url: str,
    *,
    raises: BaseException,
    params: dict[str, str] | None = None,
    required: bool = True,
) -> None:
    """Register a mock error for the given method + URL pair.

    When the interceptor matches this mock, the interaction is recorded
    with request fields + raised, then the exception is re-raised into
    the code under test.

    The error config is appended to the unified mock queue alongside
    HttpMockConfig entries, preserving FIFO ordering for mixed
    success/error sequences.
    """
    self._mock_queue.append(
        HttpErrorConfig(
            method=method.upper(),
            url=url,
            params=params,
            raises=raises,
            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/tripwire/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))

check_conflicts

check_conflicts()

Verify httpx sync/async transports and requests adapter have not been patched by a third party.

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

def check_conflicts(self) -> None:
    """Verify httpx sync/async transports and requests adapter have not been patched by a
    third party."""
    current_httpx_sync = httpx.HTTPTransport.handle_request
    if (
        current_httpx_sync is not _HTTPX_ORIGINAL_HANDLE
        and current_httpx_sync is not _tripwire_httpx_handle
    ):
        patcher = _identify_patcher(current_httpx_sync)
        raise ConflictError(
            target="httpx.HTTPTransport.handle_request",
            patcher=patcher,
        )

    current_httpx_async = httpx.AsyncHTTPTransport.handle_async_request
    if (
        current_httpx_async is not _HTTPX_ORIGINAL_ASYNC_HANDLE
        and current_httpx_async is not _tripwire_httpx_async_handle
    ):
        patcher = _identify_patcher(current_httpx_async)
        raise ConflictError(
            target="httpx.AsyncHTTPTransport.handle_async_request",
            patcher=patcher,
        )

    current_requests = requests.adapters.HTTPAdapter.send
    if (
        current_requests is not _REQUESTS_ORIGINAL_SEND
        and current_requests is not _tripwire_requests_send
    ):
        patcher = _identify_patcher(current_requests)
        raise ConflictError(
            target="requests.adapters.HTTPAdapter.send",
            patcher=patcher,
        )

    if _AIOHTTP_AVAILABLE:
        current_aiohttp = aiohttp.ClientSession._request
        if (
            current_aiohttp is not _AIOHTTP_ORIGINAL_REQUEST
            and current_aiohttp is not _tripwire_aiohttp_request
        ):
            patcher = _identify_patcher(current_aiohttp)
            raise ConflictError(
                target="aiohttp.ClientSession._request",
                patcher=patcher,
            )

assertable_fields

assertable_fields(interaction)

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

Error interactions (raised in details): request fields + raised. Request-only mode: four request fields. Full mode: all seven fields.

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

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

    Error interactions (raised in details): request fields + raised.
    Request-only mode: four request fields.
    Full mode: all seven fields.
    """
    if "raised" in interaction.details:
        return frozenset({"method", "url", "request_headers", "request_body", "raised"})
    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",
        }
    )