@samitouri / QOSami_Agents_Zero
Code Commits Issues Pulls Sessions CI/CD
main
main
py 1,333 lines 45.5 KB
Raw
1 from __future__ import annotations
2
3 import asyncio
4 import importlib
5 import json
6 import os
7 import sys
8 import stat
9 import types
10 from pathlib import Path
11
12 import pytest
13 import yaml
14
15 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
16
17 try:
18 import helpers.api # noqa: F401
19 except ModuleNotFoundError as exc:
20 if exc.name != "flask":
21 raise
22 fake_api = types.ModuleType("helpers.api")
23
24 class ApiHandler:
25 def __init__(self, app=None, thread_lock=None):
26 self.app = app
27 self.thread_lock = thread_lock
28
29 class Request:
30 pass
31
32 fake_api.ApiHandler = ApiHandler
33 fake_api.Request = Request
34 sys.modules["helpers.api"] = fake_api
35
36 try:
37 import helpers.extension # noqa: F401
38 except ModuleNotFoundError as exc:
39 if exc.name not in {"regex", "simpleeval"}:
40 raise
41 fake_extension = types.ModuleType("helpers.extension")
42
43 class Extension:
44 def __init__(self, agent=None, **kwargs):
45 self.agent = agent
46 self.kwargs = kwargs
47
48 fake_extension.Extension = Extension
49 sys.modules["helpers.extension"] = fake_extension
50
51 from plugins._oauth.api import status as status_api
52 from plugins._oauth.api import disconnect as disconnect_api
53 from plugins._oauth.api import manual_callback as manual_callback_api
54 from plugins._oauth.api import poll_device_login as poll_device_login_api
55 from plugins._oauth.api import start_device_login as start_device_login_api
56 from plugins._oauth.api import start_login as start_login_api
57 from plugins._oauth.api.models import Models
58 from plugins._oauth.extensions.python._functions.models.get_api_key.end import (
59 _20_oauth_account_dummy_key as oauth_dummy_key,
60 )
61 from plugins._oauth.helpers import state
62 from plugins._oauth.helpers.providers import base as provider_base
63 from plugins._oauth.helpers.providers.base import (
64 CODEX_PROVIDER_ID,
65 DUMMY_API_KEY,
66 GEMINI_API_PROVIDER_ID,
67 GITHUB_COPILOT_PROVIDER_ID,
68 XAI_GROK_PROVIDER_ID,
69 CallbackResult,
70 LoginPollResult,
71 LoginStartResult,
72 ProviderError,
73 provider_data_dir,
74 public_error,
75 write_private_json,
76 )
77 from plugins._oauth.helpers.providers.registry import get_provider, provider_registry
78 from plugins._oauth.helpers.summary import build_oauth_status_summary
79 from plugins._oauth.helpers.usage_plans import usage_plan_catalog
80
81
82 class FakeRequest:
83 headers = {}
84 url_root = "http://localhost:50001/"
85
86
87 def test_registry_exposes_initial_oauth_providers():
88 registry = provider_registry()
89
90 assert list(registry) == [
91 CODEX_PROVIDER_ID,
92 GITHUB_COPILOT_PROVIDER_ID,
93 GEMINI_API_PROVIDER_ID,
94 XAI_GROK_PROVIDER_ID,
95 ]
96 assert registry[CODEX_PROVIDER_ID].metadata().display_name == "Codex/ChatGPT"
97 assert registry[GITHUB_COPILOT_PROVIDER_ID].metadata().model_provider_id == GITHUB_COPILOT_PROVIDER_ID
98 assert registry[GEMINI_API_PROVIDER_ID].metadata().supports_oauth_client_config is True
99 assert registry[XAI_GROK_PROVIDER_ID].metadata().auth_flow == "browser_pkce"
100
101
102 def test_get_provider_rejects_unknown_provider_id():
103 with pytest.raises(KeyError, match="Unknown OAuth provider"):
104 get_provider("missing")
105
106
107 def test_get_provider_coerces_non_string_provider_id():
108 with pytest.raises(KeyError, match="Unknown OAuth provider: 123"):
109 get_provider(123)
110
111
112 @pytest.mark.parametrize("provider_id", [0, False])
113 def test_get_provider_rejects_falsey_non_string_provider_id(provider_id):
114 with pytest.raises(KeyError, match=f"Unknown OAuth provider: {provider_id}"):
115 get_provider(provider_id)
116
117
118 @pytest.mark.parametrize("provider_id", [None, ""])
119 def test_get_provider_defaults_empty_provider_id_to_codex(provider_id):
120 assert get_provider(provider_id).provider_id == CODEX_PROVIDER_ID
121
122
123 def test_state_keeps_login_attempts_provider_scoped():
124 attempt = state.put_attempt(
125 "state-a",
126 "verifier-a",
127 "http://127.0.0.1:56121/callback",
128 provider_id=XAI_GROK_PROVIDER_ID,
129 extra={"nonce": "nonce-a"},
130 )
131
132 loaded = state.get_attempt("state-a")
133
134 assert loaded == attempt
135 assert loaded.provider_id == XAI_GROK_PROVIDER_ID
136 assert loaded.extra == {"nonce": "nonce-a"}
137 assert state.pop_attempt("state-a") == attempt
138 assert state.get_attempt("state-a") is None
139
140
141 def test_state_keeps_device_attempts_provider_scoped():
142 attempt = state.put_device_attempt(
143 "attempt-a",
144 "device-a",
145 "USER-CODE",
146 5,
147 99_999_999_999,
148 provider_id=GITHUB_COPILOT_PROVIDER_ID,
149 extra={"domain": "github.com"},
150 )
151
152 loaded = state.get_device_attempt("attempt-a")
153
154 assert loaded == attempt
155 assert loaded.provider_id == GITHUB_COPILOT_PROVIDER_ID
156 assert loaded.extra == {"domain": "github.com"}
157 assert state.pop_device_attempt("attempt-a") == attempt
158 assert state.get_device_attempt("attempt-a") is None
159
160
161 def test_starter_providers_report_login_flow_values():
162 class FakeProvider:
163 def __init__(self, provider_id: str, flow: str):
164 self.provider_id = provider_id
165 self.flow = flow
166
167 def start_login(self, input=None, request=None):
168 return LoginStartResult(
169 ok=False,
170 provider_id=self.provider_id,
171 flow=self.flow,
172 message="not connected",
173 )
174
175 registry = {
176 GITHUB_COPILOT_PROVIDER_ID: FakeProvider(GITHUB_COPILOT_PROVIDER_ID, "device_code"),
177 XAI_GROK_PROVIDER_ID: FakeProvider(XAI_GROK_PROVIDER_ID, "browser_pkce"),
178 }
179
180 github = registry[GITHUB_COPILOT_PROVIDER_ID].start_login({})
181 xai = registry[XAI_GROK_PROVIDER_ID].start_login({})
182
183 assert github.flow == "device_code"
184 assert xai.flow == "browser_pkce"
185
186
187 def test_result_contract_preserves_account_id_with_account_label():
188 poll = LoginPollResult(
189 ok=True,
190 provider_id=CODEX_PROVIDER_ID,
191 account_label="user@example.com",
192 account_id="acct-1",
193 )
194 callback = CallbackResult(
195 ok=True,
196 provider_id=CODEX_PROVIDER_ID,
197 account_label="user@example.com",
198 account_id="acct-1",
199 )
200
201 assert poll.account_label == "user@example.com"
202 assert poll.account_id == "acct-1"
203 assert callback.account_label == "user@example.com"
204 assert callback.account_id == "acct-1"
205
206
207 def test_provider_data_dir_creates_directory(tmp_path, monkeypatch):
208 fake_files = types.SimpleNamespace(
209 USER_DIR="usr",
210 PLUGINS_DIR="plugins",
211 get_abs_path=lambda *parts: str(tmp_path.joinpath(*parts)),
212 )
213 monkeypatch.setitem(sys.modules, "helpers.files", fake_files)
214
215 path = provider_data_dir("provider-a")
216
217 assert path == tmp_path / "usr" / "plugins" / "_oauth" / "provider-a"
218 assert path.is_dir()
219
220
221 @pytest.mark.parametrize("provider_slug", ["../codex", "nested/codex", "nested\\codex", "", "."])
222 def test_provider_data_dir_rejects_unsafe_slugs(provider_slug):
223 with pytest.raises(ProviderError, match="Invalid OAuth provider storage slug.") as exc_info:
224 provider_data_dir(provider_slug)
225
226 assert exc_info.value.code == "invalid_provider_slug"
227
228
229 def test_write_private_json_does_not_reuse_preexisting_temp_file(tmp_path):
230 path = tmp_path / "auth.json"
231 stale_tmp = tmp_path / "auth.json.tmp"
232 stale_tmp.write_text("stale", encoding="utf-8")
233 stale_tmp.chmod(0o666)
234 stale_inode = stale_tmp.stat().st_ino
235
236 write_private_json(path, {"access_token": "secret"})
237
238 assert stale_tmp.read_text(encoding="utf-8") == "stale"
239 assert stale_tmp.stat().st_ino == stale_inode
240 assert path.read_text(encoding="utf-8").find("secret") > -1
241 assert stat.S_IMODE(path.stat().st_mode) == 0o600
242
243
244 def test_write_private_json_cleans_up_generated_temp_file_on_error(tmp_path, monkeypatch):
245 def fail_dump(*args, **kwargs):
246 raise RuntimeError("write failed")
247
248 monkeypatch.setattr(provider_base.json, "dump", fail_dump)
249
250 path = tmp_path / "auth.json"
251 with pytest.raises(RuntimeError, match="write failed"):
252 write_private_json(path, {"token": "secret"})
253
254 assert list(tmp_path.glob(".auth.json.*.tmp")) == []
255 assert not path.exists()
256
257
258 def test_write_private_json_does_not_follow_preexisting_temp_symlink(tmp_path):
259 leak_target = tmp_path / "leak-target"
260 leak_target.write_text("safe", encoding="utf-8")
261 stale_tmp = tmp_path / "auth.json.tmp"
262 try:
263 stale_tmp.symlink_to(leak_target)
264 except (NotImplementedError, OSError) as exc:
265 pytest.skip(f"symlink creation is not supported: {exc}")
266
267 path = tmp_path / "auth.json"
268 write_private_json(path, {"token": "secret"})
269
270 assert leak_target.read_text(encoding="utf-8") == "safe"
271 assert stale_tmp.is_symlink()
272 assert '"token": "secret"' in path.read_text(encoding="utf-8")
273 assert stat.S_IMODE(path.stat().st_mode) == 0o600
274
275
276 def test_write_private_json_sets_generated_temp_private_before_dump(tmp_path, monkeypatch):
277 observed_modes = []
278 real_dump = provider_base.json.dump
279
280 def inspect_mode_before_dump(data, handle, *args, **kwargs):
281 temp_files = list(tmp_path.glob(".auth.json.*.tmp"))
282 assert len(temp_files) == 1
283 observed_modes.append(stat.S_IMODE(temp_files[0].stat().st_mode))
284 return real_dump(data, handle, *args, **kwargs)
285
286 monkeypatch.setattr(provider_base.json, "dump", inspect_mode_before_dump)
287
288 path = tmp_path / "auth.json"
289 write_private_json(path, {"token": "secret"})
290
291 assert observed_modes == [0o600]
292 assert stat.S_IMODE(path.stat().st_mode) == 0o600
293
294
295 def test_public_error_returns_user_facing_string():
296 assert public_error(RuntimeError("visible")) == "visible"
297 assert public_error(Exception()) == "Exception"
298
299
300 def test_status_api_returns_provider_registry_shape(monkeypatch):
301 class FakeProvider:
302 def __init__(self, provider_id: str):
303 self.provider_id = provider_id
304
305 def status(self):
306 return {"provider_id": self.provider_id, "connected": False}
307
308 fake_registry = {
309 provider_id: FakeProvider(provider_id)
310 for provider_id in [
311 CODEX_PROVIDER_ID,
312 GITHUB_COPILOT_PROVIDER_ID,
313 GEMINI_API_PROVIDER_ID,
314 XAI_GROK_PROVIDER_ID,
315 ]
316 }
317 monkeypatch.setattr(status_api, "provider_registry", lambda: fake_registry)
318 monkeypatch.setattr(status_api, "is_installed", lambda: True)
319
320 response = asyncio.run(status_api.Status(None, None).process({}, FakeRequest()))
321
322 assert response["ok"] is True
323 assert response["routes_installed"] is True
324 assert [provider["provider_id"] for provider in response["providers"]] == [
325 CODEX_PROVIDER_ID,
326 GITHUB_COPILOT_PROVIDER_ID,
327 GEMINI_API_PROVIDER_ID,
328 XAI_GROK_PROVIDER_ID,
329 ]
330 assert set(response["provider_map"]) == {
331 CODEX_PROVIDER_ID,
332 GITHUB_COPILOT_PROVIDER_ID,
333 GEMINI_API_PROVIDER_ID,
334 XAI_GROK_PROVIDER_ID,
335 }
336 assert set(response["usage_plan_catalog"]) >= {
337 CODEX_PROVIDER_ID,
338 GITHUB_COPILOT_PROVIDER_ID,
339 GEMINI_API_PROVIDER_ID,
340 XAI_GROK_PROVIDER_ID,
341 }
342 assert set(response["usage_plan_catalog"]) == {
343 CODEX_PROVIDER_ID,
344 GITHUB_COPILOT_PROVIDER_ID,
345 GEMINI_API_PROVIDER_ID,
346 XAI_GROK_PROVIDER_ID,
347 }
348 assert response["codex"] == response["provider_map"][CODEX_PROVIDER_ID]
349
350
351 def test_oauth_status_summary_adds_accounts_and_usage_windows():
352 class FakeProvider:
353 provider_id = CODEX_PROVIDER_ID
354
355 def status(self):
356 return {
357 "provider_id": CODEX_PROVIDER_ID,
358 "display_name": "Codex/ChatGPT",
359 "short_name": "Codex",
360 "connected": True,
361 "account_label": "user@example.com",
362 "usage": {
363 "available": True,
364 "primary": {"remaining_percent": 91, "label": "5h", "reset_at": 123},
365 "secondary": {"used_percent": 14, "label": "7d", "reset_at": 456},
366 },
367 }
368
369 summary = build_oauth_status_summary(
370 provider_registry=lambda: {CODEX_PROVIDER_ID: FakeProvider()},
371 routes_installed=lambda: True,
372 )
373
374 assert summary["routes_installed"] is True
375 assert summary["connected_count"] == 1
376 assert summary["oauth_accounts"]["connected"][0]["account_label"] == "user@example.com"
377 assert summary["provider_map"][CODEX_PROVIDER_ID]["usage_windows"] == [
378 {"key": "primary", "title": "Session", "label": "5h", "remaining_percent": 91.0, "reset_at": 123},
379 {"key": "secondary", "title": "Week", "label": "7d", "remaining_percent": 86.0, "reset_at": 456},
380 ]
381
382
383 def test_status_api_contains_provider_status_exceptions(monkeypatch):
384 class GoodProvider:
385 provider_id = CODEX_PROVIDER_ID
386
387 def status(self):
388 return {"provider_id": CODEX_PROVIDER_ID, "connected": True}
389
390 class FailingProvider:
391 provider_id = XAI_GROK_PROVIDER_ID
392
393 def metadata(self):
394 return {
395 "provider_id": XAI_GROK_PROVIDER_ID,
396 "display_name": "xAI Grok",
397 "auth_flow": "browser_pkce",
398 }
399
400 def status(self):
401 raise RuntimeError("status failed")
402
403 monkeypatch.setattr(
404 status_api,
405 "provider_registry",
406 lambda: {
407 CODEX_PROVIDER_ID: GoodProvider(),
408 XAI_GROK_PROVIDER_ID: FailingProvider(),
409 },
410 )
411 monkeypatch.setattr(status_api, "is_installed", lambda: True)
412
413 response = asyncio.run(status_api.Status(None, None).process({}, FakeRequest()))
414
415 assert response["ok"] is True
416 assert response["provider_map"][CODEX_PROVIDER_ID]["connected"] is True
417 assert response["provider_map"][XAI_GROK_PROVIDER_ID] == {
418 "provider_id": XAI_GROK_PROVIDER_ID,
419 "display_name": "xAI Grok",
420 "auth_flow": "browser_pkce",
421 "connected": False,
422 "error": "status failed",
423 }
424
425
426 @pytest.mark.parametrize(
427 ("provider_id", "expected_flow"),
428 [
429 (GITHUB_COPILOT_PROVIDER_ID, "device_code"),
430 (GEMINI_API_PROVIDER_ID, "browser_pkce"),
431 (XAI_GROK_PROVIDER_ID, "browser_pkce"),
432 ],
433 )
434 def test_start_login_dispatches_to_selected_starter_provider(monkeypatch, provider_id, expected_flow):
435 class FakeProvider:
436 def __init__(self, provider_id: str):
437 self.provider_id = provider_id
438
439 def start_login(self, input, request):
440 return LoginStartResult(
441 ok=False,
442 provider_id=self.provider_id,
443 flow=expected_flow,
444 message="not connected",
445 )
446
447 monkeypatch.setattr(start_login_api, "get_provider", lambda selected: FakeProvider(selected))
448
449 response = asyncio.run(
450 start_login_api.StartLogin(None, None).process(
451 {"provider_id": provider_id},
452 FakeRequest(),
453 )
454 )
455
456 assert response["ok"] is False
457 assert response["provider_id"] == provider_id
458 assert response["flow"] == expected_flow
459 assert response["message"]
460
461
462 def test_start_login_without_provider_id_uses_legacy_codex_browser_login(monkeypatch):
463 calls = []
464
465 class FakeCodexProvider:
466 def start_browser_login(self, input, request):
467 calls.append(("browser", input, request))
468 return LoginStartResult(
469 ok=True,
470 provider_id=CODEX_PROVIDER_ID,
471 flow="browser_pkce",
472 auth_url="http://auth.example/authorize",
473 redirect_uri="http://localhost/auth/callback",
474 )
475
476 def start_login(self, input, request):
477 calls.append(("device", input, request))
478 return LoginStartResult(ok=True, provider_id=CODEX_PROVIDER_ID, flow="device_code")
479
480 monkeypatch.setattr(start_login_api, "get_provider", lambda provider_id: FakeCodexProvider())
481
482 request = FakeRequest()
483 response = asyncio.run(start_login_api.StartLogin(None, None).process({}, request))
484
485 assert calls == [("browser", {}, request)]
486 assert response["ok"] is True
487 assert response["provider_id"] == CODEX_PROVIDER_ID
488 assert response["flow"] == "browser_pkce"
489 assert response["auth_url"] == "http://auth.example/authorize"
490 assert response["redirect_uri"] == "http://localhost/auth/callback"
491
492
493 def test_start_login_with_blank_provider_id_uses_provider_aware_codex_login(monkeypatch):
494 calls = []
495
496 class FakeCodexProvider:
497 def start_browser_login(self, input, request):
498 calls.append(("browser", input, request))
499 return LoginStartResult(ok=True, provider_id=CODEX_PROVIDER_ID, flow="browser_pkce")
500
501 def start_login(self, input, request):
502 calls.append(("device", input, request))
503 return LoginStartResult(ok=True, provider_id=CODEX_PROVIDER_ID, flow="device_code")
504
505 monkeypatch.setattr(start_login_api, "get_provider", lambda provider_id: FakeCodexProvider())
506
507 request = FakeRequest()
508 response = asyncio.run(
509 start_login_api.StartLogin(None, None).process({"provider_id": ""}, request)
510 )
511
512 assert calls == [("device", {"provider_id": ""}, request)]
513 assert response["ok"] is True
514 assert response["provider_id"] == CODEX_PROVIDER_ID
515 assert response["flow"] == "device_code"
516
517
518 def test_start_login_provider_exception_returns_structured_error(monkeypatch):
519 class FailingProvider:
520 def start_login(self, input, request):
521 raise RuntimeError("login failed")
522
523 monkeypatch.setattr(start_login_api, "get_provider", lambda provider_id: FailingProvider())
524
525 response = asyncio.run(
526 start_login_api.StartLogin(None, None).process(
527 {"provider_id": GITHUB_COPILOT_PROVIDER_ID},
528 FakeRequest(),
529 )
530 )
531
532 assert response == {
533 "ok": False,
534 "provider_id": GITHUB_COPILOT_PROVIDER_ID,
535 "error": "login failed",
536 }
537
538
539 def test_manual_callback_dispatches_to_xai_provider_without_active_attempt():
540 response = asyncio.run(
541 manual_callback_api.ManualCallback(None, None).process(
542 {"provider_id": XAI_GROK_PROVIDER_ID, "callback_url": "http://localhost/callback?code=abc"},
543 FakeRequest(),
544 )
545 )
546
547 assert response["ok"] is False
548 assert response["provider_id"] == XAI_GROK_PROVIDER_ID
549 assert "no active xai grok sign-in attempt" in response["error"].lower()
550
551
552 def test_start_device_login_wrapper_defaults_to_codex_provider(monkeypatch):
553 calls = []
554
555 class FakeProvider:
556 def start_login(self, input, request):
557 calls.append((input, request))
558 return LoginStartResult(
559 ok=True,
560 provider_id=CODEX_PROVIDER_ID,
561 flow="device_code",
562 attempt_id="attempt-1",
563 )
564
565 monkeypatch.setattr(
566 start_device_login_api,
567 "get_provider",
568 lambda provider_id: calls.append(("provider_id", provider_id)) or FakeProvider(),
569 )
570
571 request = FakeRequest()
572 response = asyncio.run(
573 start_device_login_api.StartDeviceLogin(None, None).process(
574 {"ignored_provider_id": XAI_GROK_PROVIDER_ID},
575 request,
576 )
577 )
578
579 assert calls[0] == ("provider_id", CODEX_PROVIDER_ID)
580 assert calls[1][0] == {"ignored_provider_id": XAI_GROK_PROVIDER_ID}
581 assert calls[1][1] is request
582 assert response["ok"] is True
583 assert response["provider_id"] == CODEX_PROVIDER_ID
584 assert response["flow"] == "device_code"
585 assert response["attempt_id"] == "attempt-1"
586
587
588 def test_start_device_login_with_provider_id_calls_github_provider(monkeypatch):
589 calls = []
590
591 class FakeProvider:
592 def start_login(self, input, request):
593 calls.append((input, request))
594 return LoginStartResult(
595 ok=True,
596 provider_id=GITHUB_COPILOT_PROVIDER_ID,
597 flow="device_code",
598 attempt_id="github-attempt-1",
599 verification_url="https://github.com/login/device",
600 user_code="1234-5678",
601 )
602
603 monkeypatch.setattr(
604 start_device_login_api,
605 "get_provider",
606 lambda provider_id: calls.append(("provider_id", provider_id)) or FakeProvider(),
607 )
608
609 request = FakeRequest()
610 payload = {"provider_id": GITHUB_COPILOT_PROVIDER_ID, "enterprise_domain": ""}
611 response = asyncio.run(
612 start_device_login_api.StartDeviceLogin(None, None).process(payload, request)
613 )
614
615 assert calls[0] == ("provider_id", GITHUB_COPILOT_PROVIDER_ID)
616 assert calls[1] == (payload, request)
617 assert response["ok"] is True
618 assert response["provider_id"] == GITHUB_COPILOT_PROVIDER_ID
619 assert response["flow"] == "device_code"
620 assert response["attempt_id"] == "github-attempt-1"
621 assert response["verification_url"] == "https://github.com/login/device"
622 assert response["user_code"] == "1234-5678"
623
624
625 def test_start_device_login_unknown_provider_returns_structured_error():
626 response = asyncio.run(
627 start_device_login_api.StartDeviceLogin(None, None).process(
628 {"provider_id": "missing"},
629 FakeRequest(),
630 )
631 )
632
633 assert response["ok"] is False
634 assert response["provider_id"] == "missing"
635 assert "Unknown OAuth provider" in response["error"]
636
637
638 def test_poll_device_login_wrapper_calls_codex_provider(monkeypatch):
639 calls = []
640
641 class FakeProvider:
642 def poll_login(self, input, request):
643 calls.append((input, request))
644 return LoginPollResult(
645 ok=True,
646 provider_id=CODEX_PROVIDER_ID,
647 completed=True,
648 account_label="user@example.com",
649 account_id="account-1",
650 )
651
652 monkeypatch.setattr(
653 poll_device_login_api,
654 "get_provider",
655 lambda provider_id: calls.append(("provider_id", provider_id)) or FakeProvider(),
656 )
657
658 request = FakeRequest()
659 response = asyncio.run(
660 poll_device_login_api.PollDeviceLogin(None, None).process(
661 {"attempt_id": "attempt-1"},
662 request,
663 )
664 )
665
666 assert calls[0] == ("provider_id", CODEX_PROVIDER_ID)
667 assert calls[1][0] == {"attempt_id": "attempt-1"}
668 assert calls[1][1] is request
669 assert response["ok"] is True
670 assert response["provider_id"] == CODEX_PROVIDER_ID
671 assert response["completed"] is True
672 assert response["account_label"] == "user@example.com"
673 assert response["account_id"] == "account-1"
674
675
676 def test_poll_device_login_with_provider_id_calls_github_provider(monkeypatch):
677 calls = []
678
679 class FakeProvider:
680 def poll_login(self, input, request):
681 calls.append((input, request))
682 return LoginPollResult(
683 ok=True,
684 provider_id=GITHUB_COPILOT_PROVIDER_ID,
685 completed=True,
686 account_label="github.com",
687 )
688
689 monkeypatch.setattr(
690 poll_device_login_api,
691 "get_provider",
692 lambda provider_id: calls.append(("provider_id", provider_id)) or FakeProvider(),
693 )
694
695 request = FakeRequest()
696 payload = {"provider_id": GITHUB_COPILOT_PROVIDER_ID, "attempt_id": "attempt-1"}
697 response = asyncio.run(
698 poll_device_login_api.PollDeviceLogin(None, None).process(payload, request)
699 )
700
701 assert calls[0] == ("provider_id", GITHUB_COPILOT_PROVIDER_ID)
702 assert calls[1] == (payload, request)
703 assert response["ok"] is True
704 assert response["provider_id"] == GITHUB_COPILOT_PROVIDER_ID
705 assert response["completed"] is True
706 assert response["account_label"] == "github.com"
707
708
709 def test_poll_device_login_unknown_provider_returns_structured_error():
710 response = asyncio.run(
711 poll_device_login_api.PollDeviceLogin(None, None).process(
712 {"provider_id": "missing", "attempt_id": "attempt-1"},
713 FakeRequest(),
714 )
715 )
716
717 assert response["ok"] is False
718 assert response["provider_id"] == "missing"
719 assert "Unknown OAuth provider" in response["error"]
720
721
722 @pytest.mark.parametrize(
723 "provider_id",
724 [CODEX_PROVIDER_ID, GITHUB_COPILOT_PROVIDER_ID, GEMINI_API_PROVIDER_ID, XAI_GROK_PROVIDER_ID],
725 )
726 @pytest.mark.parametrize("initial", [None, "None"])
727 def test_oauth_providers_leave_api_key_empty_until_connected(monkeypatch, provider_id, initial):
728 monkeypatch.setattr(oauth_dummy_key, "oauth_provider_is_connected", lambda _provider_id: False)
729 data = {"args": (provider_id,), "kwargs": {}, "result": initial}
730
731 oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data)
732
733 assert data["result"] == initial
734
735
736 @pytest.mark.parametrize(
737 "provider_id",
738 [CODEX_PROVIDER_ID, GITHUB_COPILOT_PROVIDER_ID, GEMINI_API_PROVIDER_ID, XAI_GROK_PROVIDER_ID],
739 )
740 @pytest.mark.parametrize("initial", [None, "None"])
741 def test_oauth_providers_report_dummy_api_key_when_connected(monkeypatch, provider_id, initial):
742 monkeypatch.setattr(oauth_dummy_key, "oauth_provider_is_connected", lambda _provider_id: True)
743 data = {"args": (provider_id,), "kwargs": {}, "result": initial}
744
745 oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data)
746
747 assert data["result"] == DUMMY_API_KEY
748
749
750 @pytest.mark.parametrize(
751 "provider_id",
752 [CODEX_PROVIDER_ID, GITHUB_COPILOT_PROVIDER_ID, GEMINI_API_PROVIDER_ID, XAI_GROK_PROVIDER_ID],
753 )
754 def test_oauth_providers_leave_missing_result_unset_when_disconnected(monkeypatch, provider_id):
755 monkeypatch.setattr(oauth_dummy_key, "oauth_provider_is_connected", lambda _provider_id: False)
756 data = {"args": (provider_id,), "kwargs": {}}
757
758 oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data)
759
760 assert "result" not in data
761
762
763 @pytest.mark.parametrize(
764 "provider_id",
765 [CODEX_PROVIDER_ID, GITHUB_COPILOT_PROVIDER_ID, GEMINI_API_PROVIDER_ID, XAI_GROK_PROVIDER_ID],
766 )
767 def test_oauth_providers_preserve_configured_api_key(provider_id):
768 data = {"args": (provider_id,), "kwargs": {}, "result": "configured"}
769
770 oauth_dummy_key.OAuthAccountDummyKey(agent=None).execute(data=data)
771
772 assert data["result"] == "configured"
773
774
775 def test_model_provider_config_contains_all_oauth_providers():
776 provider_path = Path(__file__).resolve().parents[1] / "plugins/_oauth/conf/model_providers.yaml"
777 provider_config = yaml.safe_load(provider_path.read_text(encoding="utf-8"))
778 chat = provider_config["chat"]
779
780 assert set(chat) == {
781 CODEX_PROVIDER_ID,
782 GITHUB_COPILOT_PROVIDER_ID,
783 GEMINI_API_PROVIDER_ID,
784 XAI_GROK_PROVIDER_ID,
785 }
786 assert "api_key" not in chat[CODEX_PROVIDER_ID]["kwargs"]
787 assert "api_key" not in chat[GITHUB_COPILOT_PROVIDER_ID]["kwargs"]
788 assert "api_key" not in chat[GEMINI_API_PROVIDER_ID]["kwargs"]
789 assert "api_key" not in chat[XAI_GROK_PROVIDER_ID]["kwargs"]
790 assert chat[CODEX_PROVIDER_ID]["kwargs"]["api_base"] == "http://127.0.0.1/oauth/codex/v1"
791 assert (
792 chat[GITHUB_COPILOT_PROVIDER_ID]["kwargs"]["api_base"]
793 == "http://127.0.0.1/oauth/github-copilot/v1"
794 )
795 assert chat[GEMINI_API_PROVIDER_ID]["kwargs"]["api_base"] == "http://127.0.0.1/oauth/gemini-api/v1"
796 assert chat[XAI_GROK_PROVIDER_ID]["kwargs"]["api_base"] == "http://127.0.0.1/oauth/xai-grok/v1"
797 assert "50001" not in json.dumps(provider_config)
798
799
800 def test_oauth_provider_config_marks_oauth_providers_as_oauth_api_key_mode():
801 provider_path = Path(__file__).resolve().parents[1] / "plugins/_oauth/conf/model_providers.yaml"
802 provider_config = yaml.safe_load(provider_path.read_text(encoding="utf-8"))
803 chat = provider_config["chat"]
804
805 assert chat[CODEX_PROVIDER_ID]["api_key_mode"] == "oauth"
806 assert chat[GITHUB_COPILOT_PROVIDER_ID]["api_key_mode"] == "oauth"
807 assert chat[GEMINI_API_PROVIDER_ID]["api_key_mode"] == "oauth"
808 assert chat[XAI_GROK_PROVIDER_ID]["api_key_mode"] == "oauth"
809
810
811 def test_model_config_provider_metadata_stays_oauth_provider_agnostic():
812 metadata_path = Path(__file__).resolve().parents[1] / "plugins/_model_config/provider_metadata.yaml"
813 metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
814
815 assert CODEX_PROVIDER_ID not in metadata["chat"]
816 assert GITHUB_COPILOT_PROVIDER_ID not in metadata["chat"]
817 assert GEMINI_API_PROVIDER_ID not in metadata["chat"]
818 assert XAI_GROK_PROVIDER_ID not in metadata["chat"]
819
820
821 def test_usage_plan_catalog_covers_connectable_subscription_providers_only():
822 catalog = usage_plan_catalog()
823
824 assert set(catalog) == {
825 CODEX_PROVIDER_ID,
826 GITHUB_COPILOT_PROVIDER_ID,
827 GEMINI_API_PROVIDER_ID,
828 XAI_GROK_PROVIDER_ID,
829 }
830
831 assert {plan["id"] for plan in catalog[CODEX_PROVIDER_ID]["plans"]} >= {
832 "free",
833 "go",
834 "plus",
835 "pro",
836 "business",
837 "enterprise_edu",
838 "api_key",
839 }
840 assert {plan["id"] for plan in catalog[GITHUB_COPILOT_PROVIDER_ID]["plans"]} >= {
841 "free",
842 "student",
843 "pro",
844 "pro_plus",
845 "max",
846 "business",
847 "enterprise",
848 }
849 assert {plan["id"] for plan in catalog[GEMINI_API_PROVIDER_ID]["plans"]} >= {
850 "oauth_cloud_project",
851 "api_key",
852 "vertex_ai",
853 }
854 assert catalog[GEMINI_API_PROVIDER_ID]["implemented"] is True
855 assert {plan["id"] for plan in catalog[XAI_GROK_PROVIDER_ID]["plans"]} >= {
856 "free",
857 "supergrok_lite",
858 "supergrok",
859 "supergrok_heavy",
860 "business",
861 "enterprise",
862 "api_credits",
863 }
864
865
866 def test_disconnect_api_returns_provider_result_contract(monkeypatch):
867 class FakeProvider:
868 def __init__(self, provider_id: str):
869 self.provider_id = provider_id
870
871 def disconnect(self):
872 return {"disconnected": True, "removed_auth_files": ["auth.json"]}
873
874 def status(self):
875 return {"provider_id": self.provider_id, "connected": False}
876
877 provider = FakeProvider(GITHUB_COPILOT_PROVIDER_ID)
878 monkeypatch.setattr(disconnect_api, "get_provider", lambda provider_id: provider)
879
880 response = asyncio.run(
881 disconnect_api.Disconnect(None, None).process(
882 {"provider_id": GITHUB_COPILOT_PROVIDER_ID},
883 FakeRequest(),
884 )
885 )
886
887 assert response["ok"] is True
888 assert response["provider_id"] == GITHUB_COPILOT_PROVIDER_ID
889 assert response["result"] == {"disconnected": True, "removed_auth_files": ["auth.json"]}
890 assert response["provider"] == {"provider_id": GITHUB_COPILOT_PROVIDER_ID, "connected": False}
891 assert response["disconnected"] is True
892 assert response["removed_auth_files"] == ["auth.json"]
893
894
895 def test_disconnect_api_keeps_codex_legacy_field(monkeypatch):
896 class FakeProvider:
897 provider_id = CODEX_PROVIDER_ID
898
899 def disconnect(self):
900 return {"disconnected": True}
901
902 def status(self):
903 return {"provider_id": CODEX_PROVIDER_ID, "connected": False}
904
905 provider = FakeProvider()
906 monkeypatch.setattr(disconnect_api, "get_provider", lambda provider_id: provider)
907
908 response = asyncio.run(disconnect_api.Disconnect(None, None).process({}, FakeRequest()))
909
910 assert response["result"] == {"disconnected": True}
911 assert response["provider"] == {"provider_id": CODEX_PROVIDER_ID, "connected": False}
912 assert response["codex"] == response["provider"]
913
914
915 def test_start_login_unknown_provider_returns_structured_error():
916 response = asyncio.run(
917 start_login_api.StartLogin(None, None).process({"provider_id": "missing"}, FakeRequest())
918 )
919
920 assert response["ok"] is False
921 assert response["provider_id"] == "missing"
922 assert "Unknown OAuth provider" in response["error"]
923
924
925 @pytest.mark.parametrize("provider_id", [0, False])
926 @pytest.mark.parametrize(
927 "handler",
928 [
929 start_login_api.StartLogin,
930 Models,
931 disconnect_api.Disconnect,
932 manual_callback_api.ManualCallback,
933 ],
934 )
935 def test_provider_aware_apis_do_not_default_falsey_non_string_provider_ids(handler, provider_id):
936 response = asyncio.run(handler(None, None).process({"provider_id": provider_id}, FakeRequest()))
937
938 assert response["ok"] is False
939 assert response["provider_id"] == str(provider_id)
940 assert f"Unknown OAuth provider: {provider_id}" in response["error"]
941
942
943 def test_disconnect_unknown_provider_returns_structured_error():
944 response = asyncio.run(
945 disconnect_api.Disconnect(None, None).process({"provider_id": "missing"}, FakeRequest())
946 )
947
948 assert response["ok"] is False
949 assert response["provider_id"] == "missing"
950 assert "Unknown OAuth provider" in response["error"]
951
952
953 def test_manual_callback_unknown_provider_returns_structured_error():
954 response = asyncio.run(
955 manual_callback_api.ManualCallback(None, None).process({"provider_id": "missing"}, FakeRequest())
956 )
957
958 assert response["ok"] is False
959 assert response["provider_id"] == "missing"
960 assert "Unknown OAuth provider" in response["error"]
961
962
963 def test_models_api_returns_optional_model_metadata(monkeypatch):
964 class FakeProvider:
965 provider_id = CODEX_PROVIDER_ID
966
967 def model_catalog(self):
968 return [
969 {
970 "slug": "gpt-5.5",
971 "display_name": "GPT-5.5",
972 "description": "Frontier coding model.",
973 }
974 ]
975
976 def models(self):
977 return ["ignored-when-catalog-exists"]
978
979 models_module = sys.modules["plugins._oauth.api.models"]
980 monkeypatch.setattr(models_module, "get_provider", lambda provider_id: FakeProvider())
981
982 response = asyncio.run(Models(None, None).process({"provider_id": CODEX_PROVIDER_ID}, FakeRequest()))
983
984 assert response["ok"] is True
985 assert response["models"] == ["gpt-5.5"]
986 assert response["model_metadata"] == [
987 {
988 "slug": "gpt-5.5",
989 "display_name": "GPT-5.5",
990 "description": "Frontier coding model.",
991 }
992 ]
993
994
995 def test_unknown_provider_id_on_provider_aware_api_returns_structured_error():
996 response = asyncio.run(Models(None, None).process({"provider_id": "missing"}, FakeRequest()))
997
998 assert response["ok"] is False
999 assert response["provider_id"] == "missing"
1000 assert response["models"] == []
1001 assert "Unknown OAuth provider" in response["error"]
1002
1003
1004 def test_register_oauth_routes_adds_codex_routes_and_provider_routes(monkeypatch):
1005 fake_flask = types.ModuleType("flask")
1006
1007 class Response:
1008 def __init__(self, *args, **kwargs):
1009 self.args = args
1010 self.kwargs = kwargs
1011
1012 fake_flask.Response = Response
1013 fake_flask.jsonify = lambda *args, **kwargs: {"args": args, "kwargs": kwargs}
1014 fake_flask.request = types.SimpleNamespace()
1015 fake_flask.stream_with_context = lambda value: value
1016
1017 fake_codex = types.ModuleType("plugins._oauth.helpers.codex")
1018 fake_config = types.ModuleType("plugins._oauth.helpers.config")
1019 fake_config.codex_config = lambda: {
1020 "proxy_base_path": "/oauth/codex",
1021 "callback_path": "/auth/callback",
1022 }
1023
1024 monkeypatch.setitem(sys.modules, "flask", fake_flask)
1025 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.codex", fake_codex)
1026 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.config", fake_config)
1027
1028 module_name = "plugins._oauth.helpers.routes"
1029 previous_routes_module = sys.modules.pop(module_name, None)
1030 try:
1031 routes_module = importlib.import_module(module_name)
1032
1033 class FakeApp:
1034 def __init__(self):
1035 self.view_functions = {}
1036 self.rules = []
1037
1038 def add_url_rule(self, rule, endpoint, view_func, methods):
1039 self.view_functions[endpoint] = view_func
1040 self.rules.append((rule, endpoint, methods))
1041
1042 registered_providers = []
1043
1044 class FakeProvider:
1045 provider_id = "fake_provider"
1046
1047 def register_routes(self, app):
1048 registered_providers.append(app)
1049
1050 fake_provider = FakeProvider()
1051 monkeypatch.setattr(routes_module, "provider_registry", lambda: {"fake_provider": fake_provider})
1052
1053 app = FakeApp()
1054 routes_module.register_oauth_routes(app)
1055 routes_module.register_oauth_routes(app)
1056
1057 assert "oauth_codex_health" in app.view_functions
1058 assert app.rules.count(("/oauth/codex/health", "oauth_codex_health", ["GET"])) == 1
1059 assert registered_providers == [app, app]
1060 finally:
1061 sys.modules.pop(module_name, None)
1062 if previous_routes_module is not None:
1063 sys.modules[module_name] = previous_routes_module
1064
1065
1066 def test_github_copilot_streaming_proxy_streams_successful_upstream(monkeypatch):
1067 fake_flask = types.ModuleType("flask")
1068
1069 class Response:
1070 def __init__(self, *args, **kwargs):
1071 self.args = args
1072 self.kwargs = kwargs
1073
1074 fake_request = types.SimpleNamespace(
1075 method="POST",
1076 host="localhost",
1077 remote_addr="127.0.0.1",
1078 headers={},
1079 args={},
1080 get_json=lambda silent=True: {"stream": True, "model": "gpt-5.2"},
1081 )
1082 fake_flask.Response = Response
1083 fake_flask.jsonify = lambda *args, **kwargs: {"args": args, "kwargs": kwargs}
1084 fake_flask.request = fake_request
1085 fake_flask.stream_with_context = lambda value: value
1086
1087 fake_codex = types.ModuleType("plugins._oauth.helpers.codex")
1088 fake_codex.response_headers = lambda upstream: dict(upstream.headers)
1089 fake_config = types.ModuleType("plugins._oauth.helpers.config")
1090 fake_config.codex_config = lambda: {
1091 "proxy_base_path": "/oauth/codex",
1092 "callback_path": "/auth/callback",
1093 "proxy_token": "",
1094 "require_proxy_token": False,
1095 }
1096
1097 class FakeUpstream:
1098 ok = True
1099 status_code = 200
1100 headers = {}
1101 content = b"not-streamed"
1102
1103 def iter_content(self, chunk_size):
1104 yield b"data: {}\n\n"
1105
1106 fake_requests = types.ModuleType("requests")
1107 fake_requests.post = lambda *args, **kwargs: FakeUpstream()
1108
1109 monkeypatch.setitem(sys.modules, "flask", fake_flask)
1110 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.codex", fake_codex)
1111 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.config", fake_config)
1112 monkeypatch.setitem(sys.modules, "requests", fake_requests)
1113
1114 module_name = "plugins._oauth.helpers.routes"
1115 previous_routes_module = sys.modules.pop(module_name, None)
1116 try:
1117 routes_module = importlib.import_module(module_name)
1118
1119 class FakeProvider:
1120 def ensure_fresh_auth(self):
1121 return {
1122 "access": "fresh-access-token",
1123 "base_url": "https://api.individual.githubcopilot.com",
1124 }
1125
1126 def read_auth(self):
1127 return self.ensure_fresh_auth()
1128
1129 monkeypatch.setattr(routes_module, "get_provider", lambda provider_id: FakeProvider())
1130
1131 response = routes_module.github_copilot_responses()
1132
1133 assert isinstance(response, Response)
1134 assert response.kwargs["headers"]["Content-Type"] == "text/event-stream"
1135 assert response.kwargs["status"] == 200
1136 assert response.args[0] != b"not-streamed"
1137 finally:
1138 sys.modules.pop(module_name, None)
1139 if previous_routes_module is not None:
1140 sys.modules[module_name] = previous_routes_module
1141
1142
1143 def test_github_copilot_proxy_does_not_send_bearer_token_to_malicious_base_url(monkeypatch):
1144 fake_flask = types.ModuleType("flask")
1145
1146 class Response:
1147 def __init__(self, *args, **kwargs):
1148 self.args = args
1149 self.kwargs = kwargs
1150
1151 fake_flask.Response = Response
1152 fake_flask.jsonify = lambda *args, **kwargs: {"args": args, "kwargs": kwargs}
1153 fake_flask.request = types.SimpleNamespace(
1154 method="POST",
1155 host="localhost",
1156 remote_addr="127.0.0.1",
1157 headers={},
1158 args={},
1159 get_json=lambda silent=True: {"stream": False, "model": "gpt-5.2"},
1160 )
1161 fake_flask.stream_with_context = lambda value: value
1162
1163 fake_codex = types.ModuleType("plugins._oauth.helpers.codex")
1164 fake_codex.response_headers = lambda upstream: dict(upstream.headers)
1165 fake_config = types.ModuleType("plugins._oauth.helpers.config")
1166 fake_config.codex_config = lambda: {
1167 "proxy_base_path": "/oauth/codex",
1168 "callback_path": "/auth/callback",
1169 "proxy_token": "",
1170 "require_proxy_token": False,
1171 }
1172
1173 calls = []
1174
1175 class FakeUpstream:
1176 ok = True
1177 status_code = 200
1178 headers = {"Content-Type": "application/json"}
1179 content = b'{"ok":true}'
1180
1181 fake_requests = types.ModuleType("requests")
1182
1183 def fake_post(url, headers, json, stream, timeout):
1184 calls.append((url, headers, json, stream, timeout))
1185 return FakeUpstream()
1186
1187 fake_requests.post = fake_post
1188
1189 monkeypatch.setitem(sys.modules, "flask", fake_flask)
1190 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.codex", fake_codex)
1191 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.config", fake_config)
1192 monkeypatch.setitem(sys.modules, "requests", fake_requests)
1193
1194 module_name = "plugins._oauth.helpers.routes"
1195 previous_routes_module = sys.modules.pop(module_name, None)
1196 try:
1197 routes_module = importlib.import_module(module_name)
1198
1199 class FakeProvider:
1200 def ensure_fresh_auth(self):
1201 return {
1202 "access": "fresh-access-token",
1203 "base_url": "https://evil.example.com/v1",
1204 }
1205
1206 def read_auth(self):
1207 return self.ensure_fresh_auth()
1208
1209 monkeypatch.setattr(routes_module, "get_provider", lambda provider_id: FakeProvider())
1210
1211 response = routes_module.github_copilot_responses()
1212
1213 assert isinstance(response, Response)
1214 assert calls[0][0] == "https://api.individual.githubcopilot.com/responses"
1215 assert calls[0][1]["Authorization"] == "Bearer fresh-access-token"
1216 finally:
1217 sys.modules.pop(module_name, None)
1218 if previous_routes_module is not None:
1219 sys.modules[module_name] = previous_routes_module
1220
1221
1222 def test_proxy_authorization_does_not_trust_host_header(monkeypatch):
1223 fake_flask = types.ModuleType("flask")
1224 fake_flask.Response = object
1225 fake_flask.jsonify = lambda *args, **kwargs: {"args": args, "kwargs": kwargs}
1226 fake_flask.request = types.SimpleNamespace(
1227 host="localhost",
1228 remote_addr="203.0.113.10",
1229 headers={},
1230 args={},
1231 )
1232 fake_flask.stream_with_context = lambda value: value
1233
1234 fake_codex = types.ModuleType("plugins._oauth.helpers.codex")
1235 fake_config = types.ModuleType("plugins._oauth.helpers.config")
1236 fake_config.codex_config = lambda: {
1237 "proxy_base_path": "/oauth/codex",
1238 "callback_path": "/auth/callback",
1239 "proxy_token": "",
1240 "require_proxy_token": False,
1241 }
1242
1243 monkeypatch.setitem(sys.modules, "flask", fake_flask)
1244 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.codex", fake_codex)
1245 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.config", fake_config)
1246
1247 module_name = "plugins._oauth.helpers.routes"
1248 previous_routes_module = sys.modules.pop(module_name, None)
1249 try:
1250 routes_module = importlib.import_module(module_name)
1251
1252 assert routes_module._proxy_authorized() is False
1253 finally:
1254 sys.modules.pop(module_name, None)
1255 if previous_routes_module is not None:
1256 sys.modules[module_name] = previous_routes_module
1257
1258
1259 def test_xai_proxy_does_not_send_bearer_token_to_malicious_base_url(monkeypatch):
1260 fake_flask = types.ModuleType("flask")
1261
1262 class Response:
1263 def __init__(self, *args, **kwargs):
1264 self.args = args
1265 self.kwargs = kwargs
1266
1267 fake_flask.Response = Response
1268 fake_flask.jsonify = lambda *args, **kwargs: {"args": args, "kwargs": kwargs}
1269 fake_flask.request = types.SimpleNamespace(
1270 method="POST",
1271 host="localhost",
1272 remote_addr="127.0.0.1",
1273 headers={},
1274 args={},
1275 get_json=lambda silent=True: {"stream": False, "model": "grok-4.3"},
1276 )
1277 fake_flask.stream_with_context = lambda value: value
1278
1279 fake_codex = types.ModuleType("plugins._oauth.helpers.codex")
1280 fake_codex.response_headers = lambda upstream: dict(upstream.headers)
1281 fake_config = types.ModuleType("plugins._oauth.helpers.config")
1282 fake_config.codex_config = lambda: {
1283 "proxy_base_path": "/oauth/codex",
1284 "callback_path": "/auth/callback",
1285 "proxy_token": "",
1286 "require_proxy_token": False,
1287 }
1288
1289 calls = []
1290
1291 class FakeUpstream:
1292 ok = True
1293 status_code = 200
1294 headers = {"Content-Type": "application/json"}
1295 content = b'{"ok":true}'
1296
1297 fake_requests = types.ModuleType("requests")
1298
1299 def fake_post(url, headers, json, stream, timeout):
1300 calls.append((url, headers, json, stream, timeout))
1301 return FakeUpstream()
1302
1303 fake_requests.post = fake_post
1304
1305 monkeypatch.setitem(sys.modules, "flask", fake_flask)
1306 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.codex", fake_codex)
1307 monkeypatch.setitem(sys.modules, "plugins._oauth.helpers.config", fake_config)
1308 monkeypatch.setitem(sys.modules, "requests", fake_requests)
1309
1310 module_name = "plugins._oauth.helpers.routes"
1311 previous_routes_module = sys.modules.pop(module_name, None)
1312 try:
1313 routes_module = importlib.import_module(module_name)
1314
1315 class FakeProvider:
1316 def ensure_fresh_auth(self):
1317 return {
1318 "access": "access-token",
1319 "refresh": "refresh-token",
1320 "base_url": "https://evil.example/v1",
1321 }
1322
1323 monkeypatch.setattr(routes_module, "get_provider", lambda provider_id: FakeProvider())
1324
1325 response = routes_module.xai_grok_responses()
1326
1327 assert isinstance(response, Response)
1328 assert calls[0][0] == "https://api.x.ai/v1/responses"
1329 assert calls[0][1]["Authorization"] == "Bearer access-token"
1330 finally:
1331 sys.modules.pop(module_name, None)
1332 if previous_routes_module is not None:
1333 sys.modules[module_name] = previous_routes_module