From a4705369034fd2e2e749eccf5f7e9d5afed4d84f Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:27:41 +0800
Subject: [PATCH 01/12] Add generic OAuth test token helper (#17)

Summary:
Add a reusable pytest helper for refresh-token exchanges against arbitrary OAuth token endpoints.
Validate token responses and surface OAuth error codes without leaking fixture secrets.
Cover successful refreshes, HTTP failures, OAuth errors, and malformed token responses.
---
 tests/helpers/__init__.py             |   1 +
 tests/helpers/oauth_test_token.py     | 121 +++++++++++++++++
 tests/test_oauth_test_token_helper.py | 184 ++++++++++++++++++++++++++
 3 files changed, 306 insertions(+)
 create mode 100644 tests/helpers/__init__.py
 create mode 100644 tests/helpers/oauth_test_token.py
 create mode 100644 tests/test_oauth_test_token_helper.py

diff --git a/tests/helpers/__init__.py b/tests/helpers/__init__.py
new file mode 100644
index 00000000..5a6392cd
--- /dev/null
+++ b/tests/helpers/__init__.py
@@ -0,0 +1 @@
+"""Shared test helpers."""
diff --git a/tests/helpers/oauth_test_token.py b/tests/helpers/oauth_test_token.py
new file mode 100644
index 00000000..662e7bfe
--- /dev/null
+++ b/tests/helpers/oauth_test_token.py
@@ -0,0 +1,121 @@
+"""Generic OAuth token helpers for tests."""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Dict, Optional
+
+
+@dataclass(frozen=True)
+class TestOAuthToken:
+    access_token: str
+    token_type: str
+    expires_in: Optional[int] = None
+    scope: Optional[str] = None
+
+
+def refresh_test_oauth_token(
+    *,
+    token_endpoint: str,
+    client_id: str,
+    client_secret: str,
+    refresh_token: str,
+    timeout: float = 5.0,
+) -> TestOAuthToken:
+    """Refresh a test OAuth token without leaking fixture secrets."""
+
+    body = urllib.parse.urlencode(
+        {
+            "grant_type": "refresh_token",
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "refresh_token": refresh_token,
+        }
+    ).encode("utf-8")
+    request = urllib.request.Request(
+        token_endpoint,
+        data=body,
+        method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            status = response.status
+            payload = response.read()
+    except urllib.error.HTTPError as exc:
+        detail = _oauth_error_from_body(exc.read())
+        suffix = f" ({detail})" if detail else ""
+        raise RuntimeError(
+            f"oauth_test_token_refresh_failed: token endpoint returned HTTP "
+            f"{exc.code}{suffix}"
+        ) from exc
+    except urllib.error.URLError as exc:
+        raise RuntimeError(
+            f"oauth_test_token_refresh_failed: {exc.reason}"
+        ) from exc
+
+    if status < 200 or status >= 300:
+        raise RuntimeError(
+            f"oauth_test_token_refresh_failed: token endpoint returned HTTP {status}"
+        )
+
+    token_response = _parse_json_object(payload)
+    access_token = token_response.get("access_token")
+    token_type = token_response.get("token_type")
+
+    if not isinstance(access_token, str) or not access_token:
+        raise RuntimeError(
+            "oauth_test_token_refresh_failed: missing access_token in token response"
+        )
+    if not isinstance(token_type, str) or not token_type:
+        raise RuntimeError(
+            "oauth_test_token_refresh_failed: missing token_type in token response"
+        )
+
+    expires_in = token_response.get("expires_in")
+    if not isinstance(expires_in, int):
+        expires_in = None
+
+    scope = token_response.get("scope")
+    if not isinstance(scope, str):
+        scope = None
+
+    return TestOAuthToken(
+        access_token=access_token,
+        token_type=token_type,
+        expires_in=expires_in,
+        scope=scope,
+    )
+
+
+def _parse_json_object(payload: bytes) -> Dict[str, object]:
+    try:
+        decoded = json.loads(payload.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError(
+            "oauth_test_token_refresh_failed: invalid JSON token response"
+        ) from exc
+
+    if not isinstance(decoded, dict):
+        raise RuntimeError(
+            "oauth_test_token_refresh_failed: invalid JSON token response"
+        )
+    return decoded
+
+
+def _oauth_error_from_body(payload: bytes) -> Optional[str]:
+    try:
+        decoded = json.loads(payload.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None
+
+    if not isinstance(decoded, dict):
+        return None
+
+    error = decoded.get("error")
+    return error if isinstance(error, str) and error else None
diff --git a/tests/test_oauth_test_token_helper.py b/tests/test_oauth_test_token_helper.py
new file mode 100644
index 00000000..aa87650c
--- /dev/null
+++ b/tests/test_oauth_test_token_helper.py
@@ -0,0 +1,184 @@
+"""Tests for generic test OAuth token helper."""
+
+import json
+import urllib.parse
+from http.server import BaseHTTPRequestHandler
+
+import pytest
+
+from tests.helpers.oauth_test_token import refresh_test_oauth_token
+from tests.test_oauth_discovery import _start_server
+
+
+CLIENT_ID = "test-client"
+CLIENT_SECRET = "test-secret"
+REFRESH_TOKEN = "test-refresh-token"
+ACCESS_TOKEN = "test-access-token"
+
+
+def test_refresh_test_oauth_token_posts_refresh_grant() -> None:
+    captured = {}
+
+    def handle(handler: BaseHTTPRequestHandler) -> None:
+        length = int(handler.headers.get("Content-Length", "0"))
+        body = handler.rfile.read(length).decode("utf-8")
+        captured["content_type"] = handler.headers.get("Content-Type")
+        captured["params"] = urllib.parse.parse_qs(body)
+        _json(
+            handler,
+            200,
+            {
+                "access_token": ACCESS_TOKEN,
+                "token_type": "Bearer",
+                "expires_in": 3600,
+                "scope": "openid profile email",
+            },
+        )
+
+    server = _start_server(handle)
+    try:
+        token = refresh_test_oauth_token(
+            token_endpoint=f"{server.url}/token",
+            client_id=CLIENT_ID,
+            client_secret=CLIENT_SECRET,
+            refresh_token=REFRESH_TOKEN,
+        )
+    finally:
+        server.close()
+
+    assert token.access_token == ACCESS_TOKEN
+    assert token.token_type == "Bearer"
+    assert token.expires_in == 3600
+    assert token.scope == "openid profile email"
+    assert captured["content_type"] == "application/x-www-form-urlencoded"
+    assert captured["params"]["grant_type"] == ["refresh_token"]
+    assert captured["params"]["client_id"] == [CLIENT_ID]
+    assert captured["params"]["client_secret"] == [CLIENT_SECRET]
+    assert captured["params"]["refresh_token"] == [REFRESH_TOKEN]
+
+
+@pytest.mark.parametrize(
+    ("error", "status"),
+    [
+        ("invalid_grant", 400),
+        ("invalid_client", 401),
+        ("unsupported_grant_type", 400),
+    ],
+)
+def test_refresh_test_oauth_token_surfaces_oauth_errors_without_secrets(
+    error: str, status: int
+) -> None:
+    server = _start_server(
+        lambda handler: _json(
+            handler,
+            status,
+            {
+                "error": error,
+                "error_description": (
+                    f"{CLIENT_SECRET} {REFRESH_TOKEN} {ACCESS_TOKEN}"
+                ),
+            },
+        )
+    )
+    try:
+        with pytest.raises(RuntimeError) as exc_info:
+            refresh_test_oauth_token(
+                token_endpoint=f"{server.url}/token",
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                refresh_token=REFRESH_TOKEN,
+            )
+    finally:
+        server.close()
+
+    message = str(exc_info.value)
+    assert error in message
+    assert CLIENT_SECRET not in message
+    assert REFRESH_TOKEN not in message
+    assert ACCESS_TOKEN not in message
+
+
+def test_refresh_test_oauth_token_surfaces_non_oauth_http_error() -> None:
+    server = _start_server(lambda handler: _json(handler, 500, {}))
+    try:
+        with pytest.raises(RuntimeError, match="HTTP 500"):
+            refresh_test_oauth_token(
+                token_endpoint=f"{server.url}/token",
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                refresh_token=REFRESH_TOKEN,
+            )
+    finally:
+        server.close()
+
+
+def test_refresh_test_oauth_token_requires_access_token() -> None:
+    server = _start_server(
+        lambda handler: _json(handler, 200, {"token_type": "Bearer"})
+    )
+    try:
+        with pytest.raises(RuntimeError, match="missing access_token"):
+            refresh_test_oauth_token(
+                token_endpoint=f"{server.url}/token",
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                refresh_token=REFRESH_TOKEN,
+            )
+    finally:
+        server.close()
+
+
+def test_refresh_test_oauth_token_requires_token_type() -> None:
+    server = _start_server(
+        lambda handler: _json(handler, 200, {"access_token": ACCESS_TOKEN})
+    )
+    try:
+        with pytest.raises(RuntimeError, match="missing token_type"):
+            refresh_test_oauth_token(
+                token_endpoint=f"{server.url}/token",
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                refresh_token=REFRESH_TOKEN,
+            )
+    finally:
+        server.close()
+
+
+def test_refresh_test_oauth_token_rejects_invalid_json() -> None:
+    server = _start_server(lambda handler: _raw(handler, 200, b"not json"))
+    try:
+        with pytest.raises(RuntimeError, match="invalid JSON"):
+            refresh_test_oauth_token(
+                token_endpoint=f"{server.url}/token",
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                refresh_token=REFRESH_TOKEN,
+            )
+    finally:
+        server.close()
+
+
+def test_refresh_test_oauth_token_rejects_non_object_json() -> None:
+    server = _start_server(lambda handler: _raw(handler, 200, b"[]"))
+    try:
+        with pytest.raises(RuntimeError, match="invalid JSON"):
+            refresh_test_oauth_token(
+                token_endpoint=f"{server.url}/token",
+                client_id=CLIENT_ID,
+                client_secret=CLIENT_SECRET,
+                refresh_token=REFRESH_TOKEN,
+            )
+    finally:
+        server.close()
+
+
+def _json(handler: BaseHTTPRequestHandler, status: int, body: object) -> None:
+    _raw(handler, status, json.dumps(body).encode("utf-8"))
+
+
+def _raw(handler: BaseHTTPRequestHandler, status: int, body: bytes) -> None:
+    handler.send_response(status)
+    handler.send_header("Content-Type", "application/json")
+    handler.send_header("Content-Length", str(len(body)))
+    handler.end_headers()
+    handler.wfile.write(body)

From e0a4c49aca16a9890fb79210c28eef847506a1d1 Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:30:56 +0800
Subject: [PATCH 02/12] Add custom OAuth test IdP harness (#17)

Summary:
Add a deterministic local OAuth/OIDC IdP test harness with metadata, JWKS, authorize, and token endpoints.
Support fixed fixture credentials and deterministic token, invalid_client, invalid_grant, unsupported_grant_type, and invalid_request responses.
Cover metadata, token exchange, error behavior, and cleanup with focused pytest tests.
---
 tests/helpers/custom_oauth_test_idp.py | 227 +++++++++++++++++++++++++
 tests/test_custom_oauth_test_idp.py    | 186 ++++++++++++++++++++
 2 files changed, 413 insertions(+)
 create mode 100644 tests/helpers/custom_oauth_test_idp.py
 create mode 100644 tests/test_custom_oauth_test_idp.py

diff --git a/tests/helpers/custom_oauth_test_idp.py b/tests/helpers/custom_oauth_test_idp.py
new file mode 100644
index 00000000..4ac2222d
--- /dev/null
+++ b/tests/helpers/custom_oauth_test_idp.py
@@ -0,0 +1,227 @@
+"""Deterministic OAuth/OIDC IdP harness for tests."""
+
+from __future__ import annotations
+
+import json
+import threading
+import urllib.parse
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from types import TracebackType
+from typing import Dict, List, Optional, Type
+
+
+OAUTH_TEST_CLIENT_ID = "test-client"
+OAUTH_TEST_CLIENT_SECRET = "test-secret"
+OAUTH_TEST_REFRESH_TOKEN = "test-refresh-token"
+OAUTH_TEST_ACCESS_TOKEN = "test-access-token"
+
+
+@dataclass(frozen=True)
+class CustomOAuthTestIdpOptions:
+    client_id: str = OAUTH_TEST_CLIENT_ID
+    client_secret: str = OAUTH_TEST_CLIENT_SECRET
+    refresh_token: str = OAUTH_TEST_REFRESH_TOKEN
+    access_token: str = OAUTH_TEST_ACCESS_TOKEN
+    scope: str = "openid profile email"
+    expires_in: int = 3600
+
+
+@dataclass
+class _CustomOAuthTestIdpState:
+    issuer: str
+    client_id: str
+    client_secret: str
+    refresh_token: str
+    access_token: str
+    scope: str
+    expires_in: int
+
+
+class CustomOAuthTestIdp:
+    def __init__(
+        self,
+        server: ThreadingHTTPServer,
+        thread: threading.Thread,
+        state: _CustomOAuthTestIdpState,
+    ) -> None:
+        self._server = server
+        self._thread = thread
+        self._state = state
+        self.issuer = state.issuer
+        self.open_id_configuration_url = (
+            f"{self.issuer}/.well-known/openid-configuration"
+        )
+        self.authorization_server_metadata_url = (
+            f"{self.issuer}/.well-known/oauth-authorization-server"
+        )
+        self.authorization_endpoint = f"{self.issuer}/authorize"
+        self.token_endpoint = f"{self.issuer}/token"
+        self.jwks_url = f"{self.issuer}/jwks"
+
+    def close(self) -> None:
+        self._server.shutdown()
+        self._server.server_close()
+        self._thread.join(timeout=1)
+
+    def __enter__(self) -> "CustomOAuthTestIdp":
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        self.close()
+
+
+def start_custom_oauth_test_idp(
+    options: Optional[CustomOAuthTestIdpOptions] = None,
+) -> CustomOAuthTestIdp:
+    resolved = options or CustomOAuthTestIdpOptions()
+    state = _CustomOAuthTestIdpState(
+        issuer="",
+        client_id=resolved.client_id,
+        client_secret=resolved.client_secret,
+        refresh_token=resolved.refresh_token,
+        access_token=resolved.access_token,
+        scope=resolved.scope,
+        expires_in=resolved.expires_in,
+    )
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, fmt: str, *args: object) -> None:
+            return
+
+        def do_GET(self) -> None:
+            _handle_get(self, state)
+
+        def do_POST(self) -> None:
+            _handle_post(self, state)
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    state.issuer = f"http://127.0.0.1:{server.server_address[1]}"
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return CustomOAuthTestIdp(server, thread, state)
+
+
+def _handle_get(
+    handler: BaseHTTPRequestHandler,
+    state: _CustomOAuthTestIdpState,
+) -> None:
+    parsed = urllib.parse.urlparse(handler.path)
+    if parsed.path in (
+        "/.well-known/openid-configuration",
+        "/.well-known/oauth-authorization-server",
+    ):
+        _json(handler, 200, _authorization_server_metadata(state))
+        return
+
+    if parsed.path == "/jwks":
+        _json(handler, 200, {"keys": []})
+        return
+
+    if parsed.path == "/authorize":
+        _json(
+            handler,
+            200,
+            {
+                "issuer": state.issuer,
+                "message": "custom OAuth test IdP authorization endpoint",
+            },
+        )
+        return
+
+    handler.send_response(404)
+    handler.end_headers()
+
+
+def _handle_post(
+    handler: BaseHTTPRequestHandler,
+    state: _CustomOAuthTestIdpState,
+) -> None:
+    parsed = urllib.parse.urlparse(handler.path)
+    if parsed.path != "/token":
+        handler.send_response(404)
+        handler.end_headers()
+        return
+
+    length = int(handler.headers.get("Content-Length", "0"))
+    body = handler.rfile.read(length).decode("utf-8")
+    params = urllib.parse.parse_qs(body)
+
+    grant_type = _single(params, "grant_type")
+    client_id = _single(params, "client_id")
+    client_secret = _single(params, "client_secret")
+    refresh_token = _single(params, "refresh_token")
+
+    if grant_type is None:
+        _oauth_error(handler, 400, "invalid_request")
+        return
+    if grant_type != "refresh_token":
+        _oauth_error(handler, 400, "unsupported_grant_type")
+        return
+    if client_id is None or client_secret is None or refresh_token is None:
+        _oauth_error(handler, 400, "invalid_request")
+        return
+    if client_id != state.client_id or client_secret != state.client_secret:
+        _oauth_error(handler, 401, "invalid_client")
+        return
+    if refresh_token != state.refresh_token:
+        _oauth_error(handler, 400, "invalid_grant")
+        return
+
+    _json(
+        handler,
+        200,
+        {
+            "access_token": state.access_token,
+            "token_type": "Bearer",
+            "expires_in": state.expires_in,
+            "scope": state.scope,
+        },
+    )
+
+
+def _authorization_server_metadata(
+    state: _CustomOAuthTestIdpState,
+) -> Dict[str, object]:
+    return {
+        "issuer": state.issuer,
+        "authorization_endpoint": f"{state.issuer}/authorize",
+        "token_endpoint": f"{state.issuer}/token",
+        "jwks_uri": f"{state.issuer}/jwks",
+        "registration_endpoint": f"{state.issuer}/register",
+        "scopes_supported": ["openid", "profile", "email"],
+        "response_types_supported": ["code"],
+        "grant_types_supported": ["refresh_token"],
+        "token_endpoint_auth_methods_supported": ["client_secret_post"],
+    }
+
+
+def _single(params: Dict[str, List[str]], name: str) -> Optional[str]:
+    values = params.get(name)
+    return values[0] if values else None
+
+
+def _oauth_error(
+    handler: BaseHTTPRequestHandler,
+    status: int,
+    error: str,
+) -> None:
+    _json(handler, status, {"error": error})
+
+
+def _json(
+    handler: BaseHTTPRequestHandler,
+    status: int,
+    body: object,
+) -> None:
+    data = json.dumps(body).encode("utf-8")
+    handler.send_response(status)
+    handler.send_header("Content-Type", "application/json")
+    handler.send_header("Content-Length", str(len(data)))
+    handler.end_headers()
+    handler.wfile.write(data)
diff --git a/tests/test_custom_oauth_test_idp.py b/tests/test_custom_oauth_test_idp.py
new file mode 100644
index 00000000..6f54e05c
--- /dev/null
+++ b/tests/test_custom_oauth_test_idp.py
@@ -0,0 +1,186 @@
+"""Tests for custom OAuth test IdP harness."""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Dict
+
+import pytest
+
+from tests.helpers.custom_oauth_test_idp import (
+    OAUTH_TEST_ACCESS_TOKEN,
+    OAUTH_TEST_CLIENT_ID,
+    OAUTH_TEST_CLIENT_SECRET,
+    OAUTH_TEST_REFRESH_TOKEN,
+    start_custom_oauth_test_idp,
+)
+from tests.helpers.oauth_test_token import refresh_test_oauth_token
+
+
+def test_custom_oauth_test_idp_serves_metadata() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        openid_metadata = _fetch_json(idp.open_id_configuration_url)
+        oauth_metadata = _fetch_json(idp.authorization_server_metadata_url)
+        jwks = _fetch_json(idp.jwks_url)
+
+        assert openid_metadata["issuer"] == idp.issuer
+        assert openid_metadata["authorization_endpoint"] == idp.authorization_endpoint
+        assert openid_metadata["token_endpoint"] == idp.token_endpoint
+        assert openid_metadata["jwks_uri"] == idp.jwks_url
+        assert openid_metadata["grant_types_supported"] == ["refresh_token"]
+        assert oauth_metadata["issuer"] == idp.issuer
+        assert oauth_metadata["token_endpoint"] == idp.token_endpoint
+        assert jwks == {"keys": []}
+    finally:
+        idp.close()
+
+
+def test_custom_oauth_test_idp_exchanges_fixed_refresh_token() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        token = refresh_test_oauth_token(
+            token_endpoint=idp.token_endpoint,
+            client_id=OAUTH_TEST_CLIENT_ID,
+            client_secret=OAUTH_TEST_CLIENT_SECRET,
+            refresh_token=OAUTH_TEST_REFRESH_TOKEN,
+        )
+    finally:
+        idp.close()
+
+    assert token.access_token == OAUTH_TEST_ACCESS_TOKEN
+    assert token.token_type == "Bearer"
+    assert token.expires_in == 3600
+    assert token.scope == "openid profile email"
+
+
+@pytest.mark.parametrize(
+    ("client_id", "client_secret", "refresh_token", "expected_error"),
+    [
+        (
+            "wrong-client",
+            OAUTH_TEST_CLIENT_SECRET,
+            OAUTH_TEST_REFRESH_TOKEN,
+            "invalid_client",
+        ),
+        (
+            OAUTH_TEST_CLIENT_ID,
+            "wrong-secret",
+            OAUTH_TEST_REFRESH_TOKEN,
+            "invalid_client",
+        ),
+        (
+            OAUTH_TEST_CLIENT_ID,
+            OAUTH_TEST_CLIENT_SECRET,
+            "wrong-refresh-token",
+            "invalid_grant",
+        ),
+    ],
+)
+def test_custom_oauth_test_idp_returns_oauth_errors_deterministically(
+    client_id: str,
+    client_secret: str,
+    refresh_token: str,
+    expected_error: str,
+) -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        with pytest.raises(RuntimeError, match=expected_error):
+            refresh_test_oauth_token(
+                token_endpoint=idp.token_endpoint,
+                client_id=client_id,
+                client_secret=client_secret,
+                refresh_token=refresh_token,
+            )
+    finally:
+        idp.close()
+
+
+def test_custom_oauth_test_idp_rejects_unsupported_grant() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        response = _post_form(
+            idp.token_endpoint,
+            {
+                "grant_type": "client_credentials",
+                "client_id": OAUTH_TEST_CLIENT_ID,
+                "client_secret": OAUTH_TEST_CLIENT_SECRET,
+            },
+        )
+    finally:
+        idp.close()
+
+    assert response.status == 400
+    assert response.body == {"error": "unsupported_grant_type"}
+
+
+def test_custom_oauth_test_idp_rejects_missing_fields() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        response = _post_form(
+            idp.token_endpoint,
+            {
+                "grant_type": "refresh_token",
+                "client_id": OAUTH_TEST_CLIENT_ID,
+            },
+        )
+    finally:
+        idp.close()
+
+    assert response.status == 400
+    assert response.body == {"error": "invalid_request"}
+
+
+def test_custom_oauth_test_idp_authorize_endpoint_is_deterministic() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        body = _fetch_json(idp.authorization_endpoint)
+    finally:
+        idp.close()
+
+    assert body == {
+        "issuer": idp.issuer,
+        "message": "custom OAuth test IdP authorization endpoint",
+    }
+
+
+def test_custom_oauth_test_idp_close_stops_server() -> None:
+    idp = start_custom_oauth_test_idp()
+    idp.close()
+
+    with pytest.raises(urllib.error.URLError):
+        _fetch_json(idp.open_id_configuration_url)
+
+
+class _FormResponse:
+    def __init__(self, status: int, body: object) -> None:
+        self.status = status
+        self.body = body
+
+
+def _fetch_json(url: str) -> object:
+    with urllib.request.urlopen(url, timeout=5) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def _post_form(url: str, params: Dict[str, str]) -> _FormResponse:
+    data = urllib.parse.urlencode(params).encode("utf-8")
+    request = urllib.request.Request(
+        url,
+        data=data,
+        method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+
+    try:
+        with urllib.request.urlopen(request, timeout=5) as response:
+            status = response.status
+            body = json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        status = exc.code
+        body = json.loads(exc.read().decode("utf-8"))
+
+    return _FormResponse(status, body)

From be931106b13cf4fdc94b66c2abbd679fc92fb9be Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:37:26 +0800
Subject: [PATCH 03/12] Add protected MCP endpoint harness (#17)

Summary:
Add a local protected MCP endpoint harness for direct server and gateway URL shapes.
Return OAuth challenges for missing or wrong bearer tokens and expose protected-resource metadata pointing to the custom IdP issuer.
Cover challenge behavior, metadata parity, authenticated access, and cleanup with focused pytest tests.
---
 .../helpers/custom_protected_mcp_endpoints.py | 203 ++++++++++++++++++
 tests/test_custom_protected_mcp_endpoints.py  | 173 +++++++++++++++
 2 files changed, 376 insertions(+)
 create mode 100644 tests/helpers/custom_protected_mcp_endpoints.py
 create mode 100644 tests/test_custom_protected_mcp_endpoints.py

diff --git a/tests/helpers/custom_protected_mcp_endpoints.py b/tests/helpers/custom_protected_mcp_endpoints.py
new file mode 100644
index 00000000..f8f6e322
--- /dev/null
+++ b/tests/helpers/custom_protected_mcp_endpoints.py
@@ -0,0 +1,203 @@
+"""Protected MCP endpoint harness for OAuth auto tests."""
+
+from __future__ import annotations
+
+import json
+import threading
+import urllib.parse
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from types import TracebackType
+from typing import Dict, List, Optional, Type
+
+
+@dataclass(frozen=True)
+class CustomProtectedMcpEndpointsOptions:
+    authorization_server: str
+    access_token: str
+    scopes_supported: Optional[List[str]] = None
+    protected_resource_metadata: Optional[Dict[str, object]] = None
+
+
+@dataclass(frozen=True)
+class CustomProtectedMcpEndpoint:
+    kind: str
+    base_url: str
+    mcp_url: str
+    resource_metadata_url: str
+
+
+@dataclass
+class _EndpointState:
+    kind: str
+    base_url: str
+    authorization_server: str
+    access_token: str
+    scopes_supported: List[str]
+    protected_resource_metadata: Optional[Dict[str, object]]
+
+
+class _RunningEndpoint:
+    def __init__(
+        self,
+        endpoint: CustomProtectedMcpEndpoint,
+        server: ThreadingHTTPServer,
+        thread: threading.Thread,
+    ) -> None:
+        self.endpoint = endpoint
+        self._server = server
+        self._thread = thread
+
+    def close(self) -> None:
+        self._server.shutdown()
+        self._server.server_close()
+        self._thread.join(timeout=1)
+
+
+class CustomProtectedMcpEndpoints:
+    def __init__(
+        self,
+        server: _RunningEndpoint,
+        gateway: _RunningEndpoint,
+    ) -> None:
+        self._server_endpoint = server
+        self._gateway_endpoint = gateway
+        self.server = server.endpoint
+        self.gateway = gateway.endpoint
+
+    def close(self) -> None:
+        self._server_endpoint.close()
+        self._gateway_endpoint.close()
+
+    def __enter__(self) -> "CustomProtectedMcpEndpoints":
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        self.close()
+
+
+def start_custom_protected_mcp_endpoints(
+    options: CustomProtectedMcpEndpointsOptions,
+) -> CustomProtectedMcpEndpoints:
+    server = _start_custom_protected_mcp_endpoint("server", options)
+    try:
+        gateway = _start_custom_protected_mcp_endpoint("gateway", options)
+    except Exception:
+        server.close()
+        raise
+    return CustomProtectedMcpEndpoints(server, gateway)
+
+
+def _start_custom_protected_mcp_endpoint(
+    kind: str,
+    options: CustomProtectedMcpEndpointsOptions,
+) -> _RunningEndpoint:
+    state = _EndpointState(
+        kind=kind,
+        base_url="",
+        authorization_server=options.authorization_server,
+        access_token=options.access_token,
+        scopes_supported=options.scopes_supported or ["openid", "profile", "email"],
+        protected_resource_metadata=options.protected_resource_metadata,
+    )
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, fmt: str, *args: object) -> None:
+            return
+
+        def do_GET(self) -> None:
+            _handle_endpoint_request(self, state)
+
+        def do_POST(self) -> None:
+            _handle_endpoint_request(self, state)
+
+    http_server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    state.base_url = f"http://127.0.0.1:{http_server.server_address[1]}"
+    thread = threading.Thread(target=http_server.serve_forever, daemon=True)
+    thread.start()
+    endpoint = CustomProtectedMcpEndpoint(
+        kind=kind,
+        base_url=state.base_url,
+        mcp_url=f"{state.base_url}/mcp",
+        resource_metadata_url=(
+            f"{state.base_url}/.well-known/oauth-protected-resource/mcp"
+        ),
+    )
+    return _RunningEndpoint(endpoint, http_server, thread)
+
+
+def _handle_endpoint_request(
+    handler: BaseHTTPRequestHandler,
+    state: _EndpointState,
+) -> None:
+    parsed = urllib.parse.urlparse(handler.path)
+
+    if (
+        handler.command == "GET"
+        and parsed.path == "/.well-known/oauth-protected-resource/mcp"
+    ):
+        _json(handler, 200, _protected_resource_metadata(state))
+        return
+
+    if handler.command == "POST" and parsed.path == "/mcp":
+        if not _has_expected_bearer_token(handler, state.access_token):
+            handler.send_response(401)
+            handler.send_header(
+                "WWW-Authenticate",
+                'Bearer realm="mcp", resource_metadata="'
+                f'{state.base_url}/.well-known/oauth-protected-resource/mcp"',
+            )
+            handler.end_headers()
+            return
+
+        _json(
+            handler,
+            200,
+            {
+                "jsonrpc": "2.0",
+                "id": "custom-protected-mcp-response",
+                "result": {
+                    "endpoint": state.kind,
+                    "authenticated": True,
+                },
+            },
+        )
+        return
+
+    handler.send_response(404)
+    handler.end_headers()
+
+
+def _protected_resource_metadata(state: _EndpointState) -> Dict[str, object]:
+    if state.protected_resource_metadata is not None:
+        return state.protected_resource_metadata
+    return {
+        "resource": f"{state.base_url}/mcp",
+        "authorization_servers": [state.authorization_server],
+        "scopes_supported": state.scopes_supported,
+    }
+
+
+def _has_expected_bearer_token(
+    handler: BaseHTTPRequestHandler,
+    access_token: str,
+) -> bool:
+    return handler.headers.get("Authorization") == f"Bearer {access_token}"
+
+
+def _json(
+    handler: BaseHTTPRequestHandler,
+    status: int,
+    body: object,
+) -> None:
+    data = json.dumps(body).encode("utf-8")
+    handler.send_response(status)
+    handler.send_header("Content-Type", "application/json")
+    handler.send_header("Content-Length", str(len(data)))
+    handler.end_headers()
+    handler.wfile.write(data)
diff --git a/tests/test_custom_protected_mcp_endpoints.py b/tests/test_custom_protected_mcp_endpoints.py
new file mode 100644
index 00000000..ad286be5
--- /dev/null
+++ b/tests/test_custom_protected_mcp_endpoints.py
@@ -0,0 +1,173 @@
+"""Tests for custom protected MCP endpoint harness."""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+
+import pytest
+
+from tests.helpers.custom_oauth_test_idp import (
+    OAUTH_TEST_ACCESS_TOKEN,
+    start_custom_oauth_test_idp,
+)
+from tests.helpers.custom_protected_mcp_endpoints import (
+    CustomProtectedMcpEndpoint,
+    CustomProtectedMcpEndpointsOptions,
+    start_custom_protected_mcp_endpoints,
+)
+
+
+def test_server_and_gateway_endpoints_advertise_oauth_protection() -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    try:
+        for endpoint in (endpoints.server, endpoints.gateway):
+            response = _post_json(endpoint.mcp_url, {"jsonrpc": "2.0", "id": "probe"})
+            assert response.status == 401
+            assert response.headers["WWW-Authenticate"] == (
+                'Bearer realm="mcp", resource_metadata="'
+                f'{endpoint.resource_metadata_url}"'
+            )
+    finally:
+        endpoints.close()
+        idp.close()
+
+
+def test_server_and_gateway_metadata_points_to_custom_idp_issuer() -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    try:
+        for endpoint in (endpoints.server, endpoints.gateway):
+            metadata = _fetch_json(endpoint.resource_metadata_url)
+            assert metadata == {
+                "resource": endpoint.mcp_url,
+                "authorization_servers": [idp.issuer],
+                "scopes_supported": ["openid", "profile", "email"],
+            }
+    finally:
+        endpoints.close()
+        idp.close()
+
+
+def test_server_and_gateway_endpoints_reject_wrong_bearer_token() -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    try:
+        for endpoint in (endpoints.server, endpoints.gateway):
+            response = _post_json(
+                endpoint.mcp_url,
+                {"jsonrpc": "2.0", "id": "wrong-token"},
+                access_token="wrong-token",
+            )
+            assert response.status == 401
+    finally:
+        endpoints.close()
+        idp.close()
+
+
+def test_server_and_gateway_endpoints_accept_deterministic_bearer_token() -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    try:
+        for endpoint in (endpoints.server, endpoints.gateway):
+            body = _post_authenticated(endpoint)
+            assert body["result"] == {
+                "endpoint": endpoint.kind,
+                "authenticated": True,
+            }
+    finally:
+        endpoints.close()
+        idp.close()
+
+
+def test_close_stops_both_local_endpoints() -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    server_metadata_url = endpoints.server.resource_metadata_url
+    gateway_metadata_url = endpoints.gateway.resource_metadata_url
+    endpoints.close()
+    idp.close()
+
+    with pytest.raises(urllib.error.URLError):
+        _fetch_json(server_metadata_url)
+    with pytest.raises(urllib.error.URLError):
+        _fetch_json(gateway_metadata_url)
+
+
+def _post_authenticated(endpoint: CustomProtectedMcpEndpoint) -> object:
+    response = _post_json(
+        endpoint.mcp_url,
+        {"jsonrpc": "2.0", "id": endpoint.kind},
+        access_token=OAUTH_TEST_ACCESS_TOKEN,
+    )
+    assert response.status == 200
+    return response.body
+
+
+class _Response:
+    def __init__(
+        self,
+        status: int,
+        headers: dict,
+        body: object = None,
+    ) -> None:
+        self.status = status
+        self.headers = headers
+        self.body = body
+
+
+def _fetch_json(url: str) -> object:
+    with urllib.request.urlopen(url, timeout=5) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def _post_json(
+    url: str,
+    body: object,
+    access_token: str = "",
+) -> _Response:
+    data = json.dumps(body).encode("utf-8")
+    headers = {"Content-Type": "application/json"}
+    if access_token:
+        headers["Authorization"] = f"Bearer {access_token}"
+    request = urllib.request.Request(
+        url,
+        data=data,
+        method="POST",
+        headers=headers,
+    )
+
+    try:
+        with urllib.request.urlopen(request, timeout=5) as response:
+            response_body = response.read()
+            parsed_body = json.loads(response_body.decode("utf-8"))
+            return _Response(response.status, dict(response.headers), parsed_body)
+    except urllib.error.HTTPError as exc:
+        return _Response(exc.code, dict(exc.headers))

From fbc6244c0fccdf296ae3cef5a1325a395f9e2bbb Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:38:31 +0800
Subject: [PATCH 04/12] Add direct custom IdP OAuth auto test (#17)

Summary:
Add a stable direct MCP endpoint OAuth auto test through GopherAgent.create_with_url.
Mock the native FFI boundary and assert the refreshed deterministic access_token reaches agent_create_by_url runtime options.
Add custom IdP dynamic registration support required by the existing OAuth resolver path.
---
 tests/helpers/custom_oauth_test_idp.py |  24 ++++-
 tests/test_oauth_auto_custom_idp.py    | 132 +++++++++++++++++++++++++
 2 files changed, 153 insertions(+), 3 deletions(-)
 create mode 100644 tests/test_oauth_auto_custom_idp.py

diff --git a/tests/helpers/custom_oauth_test_idp.py b/tests/helpers/custom_oauth_test_idp.py
index 4ac2222d..b583cb97 100644
--- a/tests/helpers/custom_oauth_test_idp.py
+++ b/tests/helpers/custom_oauth_test_idp.py
@@ -143,11 +143,29 @@ def _handle_post(
     state: _CustomOAuthTestIdpState,
 ) -> None:
     parsed = urllib.parse.urlparse(handler.path)
-    if parsed.path != "/token":
-        handler.send_response(404)
-        handler.end_headers()
+    if parsed.path == "/register":
+        _json(
+            handler,
+            201,
+            {
+                "client_id": state.client_id,
+                "client_secret": state.client_secret,
+            },
+        )
+        return
+
+    if parsed.path == "/token":
+        _handle_token_request(handler, state)
         return
 
+    handler.send_response(404)
+    handler.end_headers()
+
+
+def _handle_token_request(
+    handler: BaseHTTPRequestHandler,
+    state: _CustomOAuthTestIdpState,
+) -> None:
     length = int(handler.headers.get("Content-Length", "0"))
     body = handler.rfile.read(length).decode("utf-8")
     params = urllib.parse.parse_qs(body)
diff --git a/tests/test_oauth_auto_custom_idp.py b/tests/test_oauth_auto_custom_idp.py
new file mode 100644
index 00000000..1d4320a0
--- /dev/null
+++ b/tests/test_oauth_auto_custom_idp.py
@@ -0,0 +1,132 @@
+"""OAuth auto verification with custom IdP."""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+import gopher_mcp_python.agent as agent_module
+import gopher_mcp_python.oauth_resolver as oauth_resolver
+from gopher_mcp_python import GopherAgent
+from gopher_mcp_python.runtime_options import GopherAgentTokenRecord
+from tests.helpers.custom_oauth_test_idp import (
+    OAUTH_TEST_ACCESS_TOKEN,
+    OAUTH_TEST_CLIENT_SECRET,
+    OAUTH_TEST_REFRESH_TOKEN,
+    start_custom_oauth_test_idp,
+)
+from tests.helpers.custom_protected_mcp_endpoints import (
+    CustomProtectedMcpEndpoint,
+    CustomProtectedMcpEndpointsOptions,
+    start_custom_protected_mcp_endpoints,
+)
+
+
+PROVIDER = "AnthropicProvider"
+MODEL = "test-model"
+
+
+class _FakeLibrary:
+    def __init__(self) -> None:
+        self.calls: List[tuple] = []
+
+    def agent_create_by_url(self, provider, model, url, runtime_options=None):
+        self.calls.append(("url", provider, model, url, runtime_options))
+        return 4001
+
+    def agent_release(self, handle):
+        self.calls.append(("release", handle))
+
+    def get_last_error_message(self):
+        return None
+
+    def clear_error(self):
+        self.calls.append(("clear_error",))
+
+
+class _RefreshTokenStore:
+    def __init__(self) -> None:
+        self.tokens: Dict[str, GopherAgentTokenRecord] = {}
+        self.set_calls: List[tuple] = []
+        self.delete_calls: List[str] = []
+
+    async def get(self, key: str) -> Optional[GopherAgentTokenRecord]:
+        if key not in self.tokens:
+            return GopherAgentTokenRecord(
+                access_token="expired-access-token",
+                refresh_token=OAUTH_TEST_REFRESH_TOKEN,
+                token_type="Bearer",
+                expires_at=0,
+            )
+        return self.tokens[key]
+
+    async def set(self, key: str, token: GopherAgentTokenRecord) -> None:
+        self.set_calls.append((key, token))
+        self.tokens[key] = token
+
+    async def delete(self, key: str) -> None:
+        self.delete_calls.append(key)
+        self.tokens.pop(key, None)
+
+
+def test_injects_refreshed_token_for_direct_mcp_server_endpoint(
+    monkeypatch,
+    capsys,
+) -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    try:
+        _expect_refreshed_token_injected_for_endpoint(
+            monkeypatch,
+            endpoint=endpoints.server,
+        )
+    finally:
+        endpoints.close()
+        idp.close()
+        oauth_resolver.set_oauth_resolver_hooks_for_test()
+        oauth_resolver.set_oauth_url_runtime_options_resolver_for_test()
+
+    captured = capsys.readouterr()
+    assert OAUTH_TEST_CLIENT_SECRET not in captured.out
+    assert OAUTH_TEST_CLIENT_SECRET not in captured.err
+    assert OAUTH_TEST_REFRESH_TOKEN not in captured.out
+    assert OAUTH_TEST_REFRESH_TOKEN not in captured.err
+    assert OAUTH_TEST_ACCESS_TOKEN not in captured.out
+    assert OAUTH_TEST_ACCESS_TOKEN not in captured.err
+
+
+def _expect_refreshed_token_injected_for_endpoint(
+    monkeypatch,
+    endpoint: CustomProtectedMcpEndpoint,
+) -> None:
+    fake = _FakeLibrary()
+    token_store = _RefreshTokenStore()
+    monkeypatch.setattr(agent_module, "_initialized", True)
+    monkeypatch.setattr(
+        agent_module.GopherOrchLibrary,
+        "get_instance",
+        staticmethod(lambda: fake),
+    )
+
+    agent = GopherAgent.create_with_url(
+        PROVIDER,
+        MODEL,
+        endpoint.mcp_url,
+        {
+            "oauth": {
+                "token_store": token_store,
+            },
+        },
+    )
+
+    call = fake.calls[0]
+    assert call[:4] == ("url", PROVIDER, MODEL, endpoint.mcp_url)
+    assert call[4].access_token == OAUTH_TEST_ACCESS_TOKEN
+    assert token_store.set_calls
+    assert token_store.set_calls[0][1].access_token == OAUTH_TEST_ACCESS_TOKEN
+    assert token_store.set_calls[0][1].token_type == "Bearer"
+    agent.dispose()

From 53edd1e403fc0d0e3ae608d864949b080ecb0dc8 Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:38:55 +0800
Subject: [PATCH 05/12] Add gateway custom IdP OAuth auto test (#17)

Summary:
Extend the custom IdP OAuth auto test to cover the MCP gateway endpoint shape.
Reuse the direct endpoint setup and assertions so gateway and server coverage stay parallel.
Verify GopherAgent.create_with_url injects the deterministic access_token for gateway URLs.
---
 tests/test_oauth_auto_custom_idp.py | 25 ++++++++++++++++++++++++-
 1 file changed, 24 insertions(+), 1 deletion(-)

diff --git a/tests/test_oauth_auto_custom_idp.py b/tests/test_oauth_auto_custom_idp.py
index 1d4320a0..d135cec5 100644
--- a/tests/test_oauth_auto_custom_idp.py
+++ b/tests/test_oauth_auto_custom_idp.py
@@ -71,6 +71,29 @@ async def delete(self, key: str) -> None:
 def test_injects_refreshed_token_for_direct_mcp_server_endpoint(
     monkeypatch,
     capsys,
+) -> None:
+    _expect_refreshed_token_injected_for_endpoint_name(
+        monkeypatch,
+        capsys,
+        endpoint_name="server",
+    )
+
+
+def test_injects_refreshed_token_for_mcp_gateway_endpoint(
+    monkeypatch,
+    capsys,
+) -> None:
+    _expect_refreshed_token_injected_for_endpoint_name(
+        monkeypatch,
+        capsys,
+        endpoint_name="gateway",
+    )
+
+
+def _expect_refreshed_token_injected_for_endpoint_name(
+    monkeypatch,
+    capsys,
+    endpoint_name: str,
 ) -> None:
     idp = start_custom_oauth_test_idp()
     endpoints = start_custom_protected_mcp_endpoints(
@@ -82,7 +105,7 @@ def test_injects_refreshed_token_for_direct_mcp_server_endpoint(
     try:
         _expect_refreshed_token_injected_for_endpoint(
             monkeypatch,
-            endpoint=endpoints.server,
+            endpoint=getattr(endpoints, endpoint_name),
         )
     finally:
         endpoints.close()

From 9cedce014cfb49f70073ecd318a5d0685055029e Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:41:03 +0800
Subject: [PATCH 06/12] Add custom IdP OAuth failure tests (#17)

Summary:
Add deterministic failure-mode coverage for custom IdP OAuth auto verification.
Cover invalid_grant, invalid_client, unsupported grant type, invalid protected-resource metadata, wrong bearer token rejection, and refresh fallback cleanup.
Assert failure messages stay free of fixture secrets.
---
 tests/test_oauth_auto_custom_idp_failures.py | 285 +++++++++++++++++++
 1 file changed, 285 insertions(+)
 create mode 100644 tests/test_oauth_auto_custom_idp_failures.py

diff --git a/tests/test_oauth_auto_custom_idp_failures.py b/tests/test_oauth_auto_custom_idp_failures.py
new file mode 100644
index 00000000..fe3afdb1
--- /dev/null
+++ b/tests/test_oauth_auto_custom_idp_failures.py
@@ -0,0 +1,285 @@
+"""Failure-mode tests for custom IdP OAuth auto verification."""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Dict, List, Optional
+
+import pytest
+
+import gopher_mcp_python.agent as agent_module
+import gopher_mcp_python.oauth_resolver as oauth_resolver
+from gopher_mcp_python import GopherAgent
+from gopher_mcp_python.runtime_options import GopherAgentTokenRecord
+from tests.helpers.custom_oauth_test_idp import (
+    OAUTH_TEST_ACCESS_TOKEN,
+    OAUTH_TEST_CLIENT_ID,
+    OAUTH_TEST_CLIENT_SECRET,
+    OAUTH_TEST_REFRESH_TOKEN,
+    start_custom_oauth_test_idp,
+)
+from tests.helpers.custom_protected_mcp_endpoints import (
+    CustomProtectedMcpEndpointsOptions,
+    start_custom_protected_mcp_endpoints,
+)
+from tests.helpers.oauth_test_token import refresh_test_oauth_token
+
+
+PROVIDER = "AnthropicProvider"
+MODEL = "test-model"
+FIXTURE_SECRETS = [
+    OAUTH_TEST_CLIENT_SECRET,
+    OAUTH_TEST_REFRESH_TOKEN,
+    OAUTH_TEST_ACCESS_TOKEN,
+]
+
+
+class _FakeLibrary:
+    def __init__(self) -> None:
+        self.calls: List[tuple] = []
+
+    def agent_create_by_url(self, provider, model, url, runtime_options=None):
+        self.calls.append(("url", provider, model, url, runtime_options))
+        return 5001
+
+    def agent_release(self, handle):
+        self.calls.append(("release", handle))
+
+    def get_last_error_message(self):
+        return None
+
+    def clear_error(self):
+        self.calls.append(("clear_error",))
+
+
+class _RefreshTokenStore:
+    def __init__(self, refresh_token: str) -> None:
+        self.refresh_token = refresh_token
+        self.deleted_keys: List[str] = []
+
+    async def get(self, key: str) -> Optional[GopherAgentTokenRecord]:
+        return GopherAgentTokenRecord(
+            access_token="expired-access-token",
+            refresh_token=self.refresh_token,
+            token_type="Bearer",
+            expires_at=0,
+        )
+
+    async def set(self, key: str, token: GopherAgentTokenRecord) -> None:
+        return None
+
+    async def delete(self, key: str) -> None:
+        self.deleted_keys.append(key)
+
+
+def test_wrong_refresh_token_returns_secret_safe_invalid_grant_failure() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        _expect_failure_without_fixture_secrets(
+            lambda: refresh_test_oauth_token(
+                token_endpoint=idp.token_endpoint,
+                client_id=OAUTH_TEST_CLIENT_ID,
+                client_secret=OAUTH_TEST_CLIENT_SECRET,
+                refresh_token="wrong-refresh-token",
+            ),
+            "invalid_grant",
+        )
+    finally:
+        idp.close()
+
+
+def test_wrong_client_credentials_return_secret_safe_invalid_client_failure() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        _expect_failure_without_fixture_secrets(
+            lambda: refresh_test_oauth_token(
+                token_endpoint=idp.token_endpoint,
+                client_id="wrong-client",
+                client_secret=OAUTH_TEST_CLIENT_SECRET,
+                refresh_token=OAUTH_TEST_REFRESH_TOKEN,
+            ),
+            "invalid_client",
+        )
+    finally:
+        idp.close()
+
+
+def test_unsupported_grant_type_returns_clear_oauth_failure() -> None:
+    idp = start_custom_oauth_test_idp()
+    try:
+        response = _post_form(
+            idp.token_endpoint,
+            {
+                "grant_type": "client_credentials",
+                "client_id": OAUTH_TEST_CLIENT_ID,
+                "client_secret": OAUTH_TEST_CLIENT_SECRET,
+            },
+        )
+    finally:
+        idp.close()
+
+    assert response.status == 400
+    assert response.body == {"error": "unsupported_grant_type"}
+
+
+def test_missing_protected_resource_metadata_fields_fail_clearly(monkeypatch) -> None:
+    fake = _install_fake_library(monkeypatch)
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+            protected_resource_metadata={"resource": "missing-authorization-servers"},
+        )
+    )
+    try:
+        _expect_failure_without_fixture_secrets(
+            lambda: GopherAgent.create_with_url(
+                PROVIDER,
+                MODEL,
+                endpoints.server.mcp_url,
+            ),
+            "authorization_servers",
+        )
+    finally:
+        endpoints.close()
+        idp.close()
+        oauth_resolver.set_oauth_resolver_hooks_for_test()
+        oauth_resolver.set_oauth_url_runtime_options_resolver_for_test()
+
+    assert fake.calls == []
+
+
+def test_wrong_bearer_token_is_rejected_by_protected_endpoint() -> None:
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    try:
+        response = _post_json(
+            endpoints.server.mcp_url,
+            {"jsonrpc": "2.0", "id": "wrong-token"},
+            access_token="wrong-token",
+        )
+    finally:
+        endpoints.close()
+        idp.close()
+
+    assert response.status == 401
+    assert response.headers["WWW-Authenticate"].endswith(
+        f'{endpoints.server.resource_metadata_url}"'
+    )
+
+
+def test_sdk_refresh_failure_stays_secret_safe_before_fallback_failure(
+    monkeypatch,
+) -> None:
+    _install_fake_library(monkeypatch)
+    idp = start_custom_oauth_test_idp()
+    endpoints = start_custom_protected_mcp_endpoints(
+        CustomProtectedMcpEndpointsOptions(
+            authorization_server=idp.issuer,
+            access_token=OAUTH_TEST_ACCESS_TOKEN,
+        )
+    )
+    token_store = _RefreshTokenStore("wrong-refresh-token")
+
+    def open_authorization_url(url, open_browser=None):
+        raise RuntimeError("authorization fallback disabled for failure test")
+
+    monkeypatch.setattr(oauth_resolver, "open_authorization_url", open_authorization_url)
+
+    try:
+        _expect_failure_without_fixture_secrets(
+            lambda: GopherAgent.create_with_url(
+                PROVIDER,
+                MODEL,
+                endpoints.server.mcp_url,
+                {
+                    "oauth": {
+                        "token_store": token_store,
+                    },
+                },
+            ),
+            "authorization fallback disabled for failure test",
+        )
+    finally:
+        endpoints.close()
+        idp.close()
+        oauth_resolver.set_oauth_resolver_hooks_for_test()
+        oauth_resolver.set_oauth_url_runtime_options_resolver_for_test()
+
+    assert token_store.deleted_keys
+
+
+def _install_fake_library(monkeypatch) -> _FakeLibrary:
+    fake = _FakeLibrary()
+    monkeypatch.setattr(agent_module, "_initialized", True)
+    monkeypatch.setattr(
+        agent_module.GopherOrchLibrary,
+        "get_instance",
+        staticmethod(lambda: fake),
+    )
+    return fake
+
+
+def _expect_failure_without_fixture_secrets(call, expected_message: str) -> None:
+    with pytest.raises(Exception) as exc_info:
+        call()
+
+    message = str(exc_info.value)
+    assert expected_message in message
+    for secret in FIXTURE_SECRETS:
+        assert secret not in message
+
+
+class _Response:
+    def __init__(self, status: int, headers: dict, body: object = None) -> None:
+        self.status = status
+        self.headers = headers
+        self.body = body
+
+
+def _post_form(url: str, params: Dict[str, str]) -> _Response:
+    data = urllib.parse.urlencode(params).encode("utf-8")
+    request = urllib.request.Request(
+        url,
+        data=data,
+        method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    return _send_request(request)
+
+
+def _post_json(url: str, body: object, access_token: str = "") -> _Response:
+    headers = {"Content-Type": "application/json"}
+    if access_token:
+        headers["Authorization"] = f"Bearer {access_token}"
+    request = urllib.request.Request(
+        url,
+        data=json.dumps(body).encode("utf-8"),
+        method="POST",
+        headers=headers,
+    )
+    return _send_request(request)
+
+
+def _send_request(request: urllib.request.Request) -> _Response:
+    try:
+        with urllib.request.urlopen(request, timeout=5) as response:
+            body = response.read()
+            parsed_body = json.loads(body.decode("utf-8")) if body else None
+            return _Response(response.status, dict(response.headers), parsed_body)
+    except urllib.error.HTTPError as exc:
+        try:
+            body = exc.read()
+        except ConnectionError:
+            body = b""
+        parsed_body = json.loads(body.decode("utf-8")) if body else None
+        return _Response(exc.code, dict(exc.headers), parsed_body)

From e2fad1ac358b80f9d012394fba06a294953886e4 Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:43:36 +0800
Subject: [PATCH 07/12] Add custom IdP OAuth pytest script (#17)

Summary:
Add a focused script for the custom IdP OAuth auto pytest suite.
Select python, python3, or PYTHON so the command works across local shells and CI.
Keep the command limited to local deterministic tests without Gmail, hosted services, or secrets.
---
 scripts/test-oauth-custom-idp.sh | 25 +++++++++++++++++++++++++
 1 file changed, 25 insertions(+)
 create mode 100755 scripts/test-oauth-custom-idp.sh

diff --git a/scripts/test-oauth-custom-idp.sh b/scripts/test-oauth-custom-idp.sh
new file mode 100755
index 00000000..b816b239
--- /dev/null
+++ b/scripts/test-oauth-custom-idp.sh
@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+
+cd "${REPO_ROOT}"
+
+PYTHON_BIN="${PYTHON:-}"
+if [ -z "${PYTHON_BIN}" ]; then
+  if command -v python >/dev/null 2>&1; then
+    PYTHON_BIN="python"
+  else
+    PYTHON_BIN="python3"
+  fi
+fi
+
+"${PYTHON_BIN}" -m pytest \
+  tests/test_oauth_auto_custom_idp.py \
+  tests/test_oauth_auto_custom_idp_failures.py \
+  tests/test_custom_oauth_test_idp.py \
+  tests/test_custom_protected_mcp_endpoints.py \
+  tests/test_oauth_test_token_helper.py \
+  "$@"

From c67a5e9c75bb7c8a7a9668f5e2ae1f9548676f33 Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:44:01 +0800
Subject: [PATCH 08/12] Run custom IdP OAuth tests in CI (#17)

Summary:
Add a dedicated OAuth verification workflow for pull requests and manual runs.
Run the focused custom IdP pytest suite on Ubuntu and macOS.
Keep the CI path independent from hosted Gopher endpoints, Gmail, and GitHub Secrets.
---
 .github/workflows/oauth-verify.yml | 34 ++++++++++++++++++++++++++++++
 1 file changed, 34 insertions(+)
 create mode 100644 .github/workflows/oauth-verify.yml

diff --git a/.github/workflows/oauth-verify.yml b/.github/workflows/oauth-verify.yml
new file mode 100644
index 00000000..9ea61039
--- /dev/null
+++ b/.github/workflows/oauth-verify.yml
@@ -0,0 +1,34 @@
+name: OAuth Verify
+
+on:
+  pull_request:
+    branches: [main]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  custom-idp-oauth:
+    name: Custom IdP OAuth (${{ matrix.os }})
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-14]
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+
+      - name: Install package
+        run: python -m pip install -e ".[dev]"
+
+      - name: Run custom IdP OAuth verification
+        run: scripts/test-oauth-custom-idp.sh

From cb871afeaedf8c844e4824686b2f18a9fc22054f Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:44:46 +0800
Subject: [PATCH 09/12] Document custom IdP OAuth verification (#17)

Summary:
Document the deterministic custom IdP OAuth auto verification path.
Explain direct server and gateway endpoint coverage with local fixture credentials.
Clarify that Gmail and real Gopher endpoint checks remain optional smoke tests.
---
 README.md                     |  4 +++
 docs/oauth-auto-custom-idp.md | 65 +++++++++++++++++++++++++++++++++++
 2 files changed, 69 insertions(+)
 create mode 100644 docs/oauth-auto-custom-idp.md

diff --git a/README.md b/README.md
index e392010f..0e343abe 100644
--- a/README.md
+++ b/README.md
@@ -190,6 +190,10 @@ Result class with status and metadata.
 pytest
 ```
 
+For deterministic OAuth auto verification with a local custom IdP and local MCP
+server/gateway endpoints, see
+[`docs/oauth-auto-custom-idp.md`](docs/oauth-auto-custom-idp.md).
+
 ### Code Formatting
 
 This project uses Black for code formatting and Ruff for linting.
diff --git a/docs/oauth-auto-custom-idp.md b/docs/oauth-auto-custom-idp.md
new file mode 100644
index 00000000..fd866834
--- /dev/null
+++ b/docs/oauth-auto-custom-idp.md
@@ -0,0 +1,65 @@
+# OAuth Auto Verification With Custom IdP
+
+The stable OAuth auto verification path uses local test fixtures instead of
+Gmail, hosted Gopher services, or real OAuth provider credentials. It verifies
+the Python SDK behavior that matters for automatic OAuth:
+
+- discovering OAuth protection from an MCP endpoint
+- reading protected resource metadata
+- using OAuth authorization server metadata
+- refreshing a cached token through the token endpoint
+- injecting the refreshed bearer token into `GopherAgent.create_with_url`
+  runtime options before the native FFI call
+
+The tests cover both endpoint shapes used by deployments:
+
+- direct MCP server endpoint
+- MCP gateway endpoint
+
+Fixture credentials such as `test-client`, `test-secret`, and
+`test-refresh-token` are local test data. They are not GitHub Secrets, and the
+suite asserts that fixture secrets do not appear in captured output or errors.
+
+## Local Command
+
+Run the deterministic custom IdP suite with:
+
+```bash
+scripts/test-oauth-custom-idp.sh
+```
+
+The script runs:
+
+```bash
+python -m pytest \
+  tests/test_oauth_auto_custom_idp.py \
+  tests/test_oauth_auto_custom_idp_failures.py \
+  tests/test_custom_oauth_test_idp.py \
+  tests/test_custom_protected_mcp_endpoints.py \
+  tests/test_oauth_test_token_helper.py
+```
+
+Extra pytest arguments can be passed through:
+
+```bash
+scripts/test-oauth-custom-idp.sh -q
+```
+
+## CI Coverage
+
+`.github/workflows/oauth-verify.yml` runs the same suite on pull requests and
+manual dispatch. It installs the package with development dependencies and does
+not require hosted endpoints, Gmail accounts, OAuth client secrets, refresh
+tokens, LLM provider keys, or other real credentials.
+
+## Live Smoke Tests
+
+Gmail or real Gopher endpoint verification remains useful as an optional smoke
+test because it proves compatibility with external provider policy, hosted
+gateway configuration, and real account consent. Those checks are operationally
+different from SDK correctness tests: they depend on provider availability,
+account security rules, valid refresh tokens, and live service configuration.
+
+Keep live smoke tests manual, scheduled, or otherwise separate from the stable
+pull-request gate. The API example workflow and docs live under
+`examples/api/`.

From cbbe63e6853b96c04ec21abf769ce1f23c51944d Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:45:16 +0800
Subject: [PATCH 10/12] Document optional native OAuth follow-up (#17)

Summary:
Document the proposed native end-to-end OAuth verification shape.
Keep the native path separate from the stable pull-request gate until deterministic.
Link the follow-up from the custom IdP OAuth verification guide.
---
 docs/oauth-auto-custom-idp.md       |  3 ++
 docs/oauth-auto-native-follow-up.md | 45 +++++++++++++++++++++++++++++
 2 files changed, 48 insertions(+)
 create mode 100644 docs/oauth-auto-native-follow-up.md

diff --git a/docs/oauth-auto-custom-idp.md b/docs/oauth-auto-custom-idp.md
index fd866834..87186145 100644
--- a/docs/oauth-auto-custom-idp.md
+++ b/docs/oauth-auto-custom-idp.md
@@ -63,3 +63,6 @@ account security rules, valid refresh tokens, and live service configuration.
 Keep live smoke tests manual, scheduled, or otherwise separate from the stable
 pull-request gate. The API example workflow and docs live under
 `examples/api/`.
+
+For a possible full native end-to-end extension, see
+[`oauth-auto-native-follow-up.md`](oauth-auto-native-follow-up.md).
diff --git a/docs/oauth-auto-native-follow-up.md b/docs/oauth-auto-native-follow-up.md
new file mode 100644
index 00000000..f9cb6396
--- /dev/null
+++ b/docs/oauth-auto-native-follow-up.md
@@ -0,0 +1,45 @@
+# Optional Native OAuth E2E Follow-Up
+
+The custom IdP OAuth verification suite is the stable pull-request signal. It
+starts local OAuth and protected MCP endpoint harnesses, lets
+`GopherAgent.create_with_url` resolve OAuth automatically, and asserts that the
+native FFI boundary receives runtime options containing the refreshed bearer
+token.
+
+A full native end-to-end test can be added later, but it should stay separate
+from the stable PR gate unless it is fully deterministic.
+
+## Proposed Shape
+
+Reuse the same model as `docs/oauth-auto-custom-idp.md`:
+
+- local custom OAuth/OIDC IdP
+- local protected MCP server endpoint
+- local protected MCP gateway endpoint
+- refresh-token-backed OAuth setup
+
+Add one deterministic protected MCP tool:
+
+```text
+tool: whoami
+input: {}
+output: { "subject": "test-user@example.test" }
+```
+
+Then create a real `GopherAgent` through `GopherAgent.create_with_url` and run a
+query that must call the protected `whoami` tool. The test should assert the
+authenticated result, not just successful agent creation.
+
+## When To Enable
+
+Keep this path manual, scheduled, or non-blocking until these dependencies are
+controlled:
+
+- native package availability for each target platform
+- deterministic LLM/provider behavior or a reliable test provider
+- stable local MCP tool execution behavior
+- clear runtime bounds suitable for CI
+
+Until those are in place, the focused custom IdP tests remain the correct PR
+gate because they verify the SDK OAuth flow without external service or native
+runtime flake.

From 122cbd22cf7ae4fe0f1fa1c6d754b61143ade6d7 Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 00:45:57 +0800
Subject: [PATCH 11/12] Make Gmail OAuth smoke tests optional (#17)

Summary:
Remove pull-request triggering from the live SDK example workflow.
Keep live example verification available through schedule, branch push, and manual dispatch.
Document that the custom IdP workflow is the stable pull-request OAuth gate.
---
 .github/workflows/verify-examples.yml | 9 +++------
 docs/oauth-auto-custom-idp.md         | 6 ++++--
 2 files changed, 7 insertions(+), 8 deletions(-)

diff --git a/.github/workflows/verify-examples.yml b/.github/workflows/verify-examples.yml
index 4f815295..09de1e77 100644
--- a/.github/workflows/verify-examples.yml
+++ b/.github/workflows/verify-examples.yml
@@ -1,10 +1,10 @@
 name: Verify SDK Examples
 
 on:
-  pull_request:
-    branches: [main]
   push:
     branches: [iml_verify_auto]
+  schedule:
+    - cron: '17 10 * * 1'
   workflow_dispatch:
     inputs:
       mode:
@@ -66,9 +66,7 @@ jobs:
           python -m venv native-preflight
           source native-preflight/bin/activate
           python -m pip install --upgrade pip
-          if [ "${{ github.event_name }}" = "pull_request" ]; then
-            python -m pip install -e . "gopher-mcp-python-native-${{ matrix.platform }}"
-          elif [ "$VERIFY_PYPI_VERSION" = "latest" ]; then
+          if [ "$VERIFY_PYPI_VERSION" = "latest" ]; then
             python -m pip install gopher-mcp-python gopher-mcp-python-native-${{ matrix.platform }}
           else
             python -m pip install "gopher-mcp-python==${VERIFY_PYPI_VERSION}" "gopher-mcp-python-native-${{ matrix.platform }}==${VERIFY_PYPI_VERSION}"
@@ -181,7 +179,6 @@ jobs:
           ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
           GOPHER_API_KEY: ${{ secrets.GOPHER_API_KEY }}
           GOPHER_MCP_URL: ${{ secrets.GOPHER_MCP_URL }}
-          SDK_INSTALL_SPEC: ${{ github.event_name == 'pull_request' && github.workspace || '' }}
         run: |
           unset GOPHER_SDK_TEST
           VERIFY_LIVE_PROMPT="Get my mail profile" \
diff --git a/docs/oauth-auto-custom-idp.md b/docs/oauth-auto-custom-idp.md
index 87186145..2da60e94 100644
--- a/docs/oauth-auto-custom-idp.md
+++ b/docs/oauth-auto-custom-idp.md
@@ -61,8 +61,10 @@ different from SDK correctness tests: they depend on provider availability,
 account security rules, valid refresh tokens, and live service configuration.
 
 Keep live smoke tests manual, scheduled, or otherwise separate from the stable
-pull-request gate. The API example workflow and docs live under
-`examples/api/`.
+pull-request gate. In this repository, `.github/workflows/oauth-verify.yml` is
+the pull-request OAuth gate, while `.github/workflows/verify-examples.yml` is
+reserved for scheduled, branch-triggered, or manually dispatched example smoke
+verification. The API example workflow and docs live under `examples/api/`.
 
 For a possible full native end-to-end extension, see
 [`oauth-auto-native-follow-up.md`](oauth-auto-native-follow-up.md).

From 24f078efeb65d364f4c8452bb2e16487fcabcbc9 Mon Sep 17 00:00:00 2001
From: RahulHere <RahulHere@someday.on>
Date: Thu, 20 Aug 2026 08:14:32 +0800
Subject: [PATCH 12/12] Update OAuth verification workflow tests (#17)

Summary:
Assert the OAuth verify workflow is the pull-request SDK OAuth gate.
Assert the live example workflow remains optional for smoke verification.
Include the branch push trigger for OAuth verification on the feature branch.
---
 .github/workflows/oauth-verify.yml   |  2 ++
 tests/test_linux_native_packaging.py | 30 ++++++++++++++++++----------
 2 files changed, 22 insertions(+), 10 deletions(-)

diff --git a/.github/workflows/oauth-verify.yml b/.github/workflows/oauth-verify.yml
index 9ea61039..39cb31ea 100644
--- a/.github/workflows/oauth-verify.yml
+++ b/.github/workflows/oauth-verify.yml
@@ -1,6 +1,8 @@
 name: OAuth Verify
 
 on:
+  push:
+    branches: [main, feature/sdk-oauth-verify]
   pull_request:
     branches: [main]
   workflow_dispatch:
diff --git a/tests/test_linux_native_packaging.py b/tests/test_linux_native_packaging.py
index cb429464..235f63bc 100644
--- a/tests/test_linux_native_packaging.py
+++ b/tests/test_linux_native_packaging.py
@@ -23,6 +23,10 @@ def _verify_examples_workflow() -> str:
     return (ROOT / ".github" / "workflows" / "verify-examples.yml").read_text()
 
 
+def _oauth_verify_workflow() -> str:
+    return (ROOT / ".github" / "workflows" / "oauth-verify.yml").read_text()
+
+
 def _verify_examples_script() -> str:
     return (ROOT / "scripts" / "verify-examples.sh").read_text()
 
@@ -46,22 +50,28 @@ def test_linux_x64_uses_digest_pinned_ubuntu_builder_image() -> None:
     assert re.search(r"\subuntu:20\.04\s", build_script) is None
 
 
-def test_verify_examples_prs_install_checked_out_sdk() -> None:
+def test_oauth_verify_prs_install_checked_out_sdk() -> None:
+    workflow = _oauth_verify_workflow()
+
+    assert "pull_request:" in workflow
+    assert "branches: [main]" in workflow
+    assert 'python -m pip install -e ".[dev]"' in workflow
+    assert "scripts/test-oauth-custom-idp.sh" in workflow
+
+
+def test_verify_examples_workflow_stays_optional_for_live_smoke() -> None:
     workflow = _verify_examples_workflow()
 
     assert (
         "VERIFY_EXAMPLES_MODE: ${{ github.event_name == 'workflow_dispatch' "
         "&& inputs.mode || 'auto' }}"
     ) in workflow
-    assert 'if [ "${{ github.event_name }}" = "pull_request" ]; then' in workflow
-    assert (
-        'python -m pip install -e . "gopher-mcp-python-native-${{ matrix.platform }}"'
-        in workflow
-    )
-    assert (
-        "SDK_INSTALL_SPEC: ${{ github.event_name == 'pull_request' && "
-        "github.workspace || '' }}"
-    ) in workflow
+    assert "pull_request:" not in workflow
+    assert "schedule:" in workflow
+    assert "workflow_dispatch:" in workflow
+    assert "branches: [iml_verify_auto]" in workflow
+    assert 'if [ "${{ github.event_name }}" = "pull_request" ]; then' not in workflow
+    assert "SDK_INSTALL_SPEC:" not in workflow
 
 
 def test_verify_examples_workflow_bounds_pr_cost_and_runtime() -> None: