Class HeadlessOAuth

tests/interaction/auth/_harness.py:129–177

Completes the authorize step in-process by following the redirect through the bridge. `redirect_handler` GETs the authorize URL on the bound client (with `auth=None` so the request does not re-enter the locked auth flow), parses `code` and `state` from the 302 `Location`, and stashes th

class HeadlessOAuth:
    """Completes the authorize step in-process by following the redirect through the bridge.

    `redirect_handler` GETs the authorize URL on the bound client (with `auth=None` so the
    request does not re-enter the locked auth flow), parses `code` and `state` from the 302
    `Location`, and stashes them; `callback_handler` returns the stashed pair. Tests inspect
    `authorize_url` to assert what the SDK put on the authorize request.

    `state_override`: when set, `callback_handler` returns this value as the state instead of
    the one parsed from the redirect, so tests can drive the state-mismatch path.

    `iss_override`: when set, `callback_handler` returns this value as the RFC 9207 issuer
    instead of the one parsed from the redirect, so tests can drive the iss-mismatch path.
    """

    def __init__(self, *, state_override: str | None = None, iss_override: str | None = None) -> None:
        self.authorize_url: str | None = None
        self.authorize_urls: list[str] = []
        self.error: str | None = None
        self._state_override = state_override
        self._iss_override = iss_override
        self._http: httpx.AsyncClient | None = None
        self._code: str = ""
        self._state: str | None = None
        self._iss: str | None = None

    def bind(self, http_client: httpx.AsyncClient) -> None:
        self._http = http_client

    async def redirect_handler(self, authorization_url: str) -> None:
        assert self._http is not None
        self.authorize_url = authorization_url
        self.authorize_urls.append(authorization_url)
        # auth=None is load-bearing: without it the GET re-enters OAuthClientProvider.async_auth_flow
        # through its context lock and the flow deadlocks.
        response = await self._http.get(authorization_url, follow_redirects=False, auth=None)
        assert response.status_code == 302, f"authorize endpoint returned {response.status_code}: {response.text}"
        params = parse_qs(urlsplit(response.headers["location"]).query)
        self._code = params.get("code", [""])[0]
        self._state = params.get("state", [None])[0]
        self._iss = params.get("iss", [None])[0]
        self.error = params.get("error", [None])[0]

    async def callback_handler(self) -> AuthorizationCodeResult:
        return AuthorizationCodeResult(
            code=self._code,
            state=self._state_override if self._state_override is not None else self._state,
            iss=self._iss_override if self._iss_override is not None else self._iss,
        )


def auth_settings(