Function _make_mock_transport

python/openshell/sandbox_test.py:532–572

Build an httpx.MockTransport that serves OIDC discovery + token refresh from an in-memory script. `refresh_responses` is consumed in order across successive POSTs to the token endpoint (which lets tests assert refresh-token rotation semantics across multiple refreshes).

(
    *,
    discovery: dict | None = None,
    refresh_responses: list[dict] | None = None,
    discovery_status: int = 200,
    refresh_status: int = 200,
    seen_refresh: list[Any] | None = None,
    seen_discovery: list[Any] | None = None,
)

def _make_mock_transport(
    *,
    discovery: dict | None = None,
    refresh_responses: list[dict] | None = None,
    discovery_status: int = 200,
    refresh_status: int = 200,
    seen_refresh: list[Any] | None = None,
    seen_discovery: list[Any] | None = None,
):
    """Build an httpx.MockTransport that serves OIDC discovery + token
    refresh from an in-memory script.

    `refresh_responses` is consumed in order across successive POSTs to
    the token endpoint (which lets tests assert refresh-token rotation
    semantics across multiple refreshes).
    """
    import httpx as _httpx

    refresh_iter = iter(
        refresh_responses or [{"access_token": "refreshed-jwt", "expires_in": 3600}]
    )

    def handler(request: _httpx.Request) -> _httpx.Response:
        if request.url.path.endswith("/.well-known/openid-configuration"):
            if seen_discovery is not None:
                seen_discovery.append(str(request.url))
            body = discovery or {
                "issuer": DEFAULT_ISSUER,
                "token_endpoint": DEFAULT_TOKEN_ENDPOINT,
            }
            return _httpx.Response(discovery_status, json=body)
        # Anything else is a refresh exchange.
        if seen_refresh is not None:
            seen_refresh.append((str(request.url), bytes(request.content)))
        try:
            body = next(refresh_iter)
        except StopIteration:
            return _httpx.Response(500, json={"error": "test_script_exhausted"})
        return _httpx.Response(refresh_status, json=body)

    return _httpx.MockTransport(handler)


def _install_mock_transport(refresher: Any, transport: Any) -> None: