mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-05-31 04:55:58 +00:00
4a836940f3
Convert Chat Completions image_url content parts into Responses API input_image parts instead of normalizing multimodal messages down to text. Keep text-only content lists as plain text and add OAuth bridge regression tests for image passthrough.
293 lines
9.2 KiB
Python
293 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from plugins._oauth.helpers import codex
|
|
from plugins._oauth.extensions.python._functions.models.get_api_key.end._20_codex_account_dummy_key import (
|
|
CodexAccountDummyKey,
|
|
)
|
|
|
|
|
|
def test_generate_pkce_produces_urlsafe_verifier_and_challenge():
|
|
pair = codex.generate_pkce()
|
|
|
|
assert 43 <= len(pair.verifier) <= 128
|
|
assert pair.verifier
|
|
assert pair.challenge
|
|
assert "=" not in pair.verifier
|
|
assert "=" not in pair.challenge
|
|
|
|
|
|
def test_build_authorize_url_uses_existing_a0_origin_callback(monkeypatch):
|
|
monkeypatch.setattr(
|
|
codex,
|
|
"codex_config",
|
|
lambda: {
|
|
"issuer": "https://auth.openai.com",
|
|
"client_id": "app_EMoamEEZ73f0CkXaXp7hrann",
|
|
"scopes": [
|
|
"openid",
|
|
"profile",
|
|
"email",
|
|
"offline_access",
|
|
"api.connectors.read",
|
|
"api.connectors.invoke",
|
|
],
|
|
"forced_workspace_id": "",
|
|
},
|
|
)
|
|
pair = codex.PkcePair(verifier="verifier", challenge="challenge")
|
|
auth_url = codex.build_authorize_url(
|
|
"http://localhost:50001/auth/callback",
|
|
"state",
|
|
pair,
|
|
)
|
|
|
|
assert auth_url.startswith("https://auth.openai.com/oauth/authorize?")
|
|
assert "redirect_uri=http%3A%2F%2Flocalhost%3A50001%2Fauth%2Fcallback" in auth_url
|
|
assert "code_challenge=challenge" in auth_url
|
|
assert "originator=codex_cli_rs" in auth_url
|
|
|
|
|
|
def test_chat_messages_to_response_body_extracts_instructions():
|
|
body = codex.chat_messages_to_response_body(
|
|
{
|
|
"model": "gpt-5.2",
|
|
"messages": [
|
|
{"role": "system", "content": "Be precise."},
|
|
{"role": "user", "content": "Hello"},
|
|
],
|
|
"temperature": 0.2,
|
|
"reasoning_effort": "high",
|
|
}
|
|
)
|
|
|
|
assert body["model"] == "gpt-5.2"
|
|
assert body["instructions"] == "Be precise."
|
|
assert body["input"] == [{"role": "user", "content": "Hello"}]
|
|
assert body["temperature"] == 0.2
|
|
assert body["reasoning"] == {"effort": "high"}
|
|
|
|
|
|
def test_chat_messages_to_response_body_preserves_image_parts_for_responses():
|
|
data_url = "data:image/png;base64,abcd"
|
|
|
|
body = codex.chat_messages_to_response_body(
|
|
{
|
|
"model": "gpt-5.5",
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": "Inspect this screenshot."},
|
|
{"type": "image_url", "image_url": {"url": data_url}},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
assert body["input"] == [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "input_text", "text": "Inspect this screenshot."},
|
|
{"type": "input_image", "image_url": data_url, "detail": "auto"},
|
|
],
|
|
}
|
|
]
|
|
|
|
|
|
def test_chat_messages_to_response_body_keeps_text_only_lists_as_text():
|
|
body = codex.chat_messages_to_response_body(
|
|
{
|
|
"model": "gpt-5.5",
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": "first"},
|
|
{"type": "text", "text": "second"},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
assert body["input"] == [{"role": "user", "content": "first\nsecond"}]
|
|
|
|
|
|
def test_response_text_reads_output_text_or_output_blocks():
|
|
assert codex.response_text({"output_text": "direct"}) == "direct"
|
|
|
|
assert (
|
|
codex.response_text(
|
|
{
|
|
"output": [
|
|
{
|
|
"content": [
|
|
{"type": "output_text", "text": "a"},
|
|
{"type": "output_text", "text": "b"},
|
|
]
|
|
}
|
|
]
|
|
}
|
|
)
|
|
== "ab"
|
|
)
|
|
|
|
|
|
def test_parse_sse_block_joins_data_lines():
|
|
event = codex.parse_sse_block(
|
|
'event: response.completed\ndata: {"response":\ndata: {"id":"r"}}\n'
|
|
)
|
|
|
|
assert event["event"] == "response.completed"
|
|
assert json.loads(event["data"]) == {"response": {"id": "r"}}
|
|
|
|
|
|
def test_extract_sse_text_deltas_reads_chat_completion_chunks():
|
|
deltas = codex.extract_sse_text_deltas(
|
|
{
|
|
"id": "chatcmpl_test",
|
|
"choices": [
|
|
{"delta": {"role": "assistant"}},
|
|
{"delta": {"content": "Hel"}},
|
|
{"delta": {"content": "lo"}},
|
|
],
|
|
}
|
|
)
|
|
|
|
assert deltas == ["Hel", "lo"]
|
|
|
|
|
|
def test_collect_completed_response_falls_back_to_text_deltas():
|
|
class FakeResponse:
|
|
encoding = "utf-8"
|
|
|
|
def iter_content(self, chunk_size=8192, decode_unicode=True):
|
|
del chunk_size, decode_unicode
|
|
yield b'data: {"choices":[{"delta":{"content":"Hel"}}]}\n\n'
|
|
yield b'data: {"choices":[{"delta":{"content":"lo"}}]}\n\n'
|
|
yield b'event: response.completed\ndata: {"response":{"output":[]}}\n\n'
|
|
yield b"data: [DONE]\n\n"
|
|
|
|
assert codex.collect_completed_response(FakeResponse()) == {"output": [], "output_text": "Hello"}
|
|
|
|
|
|
def test_normalize_usage_payload_reads_codex_windows():
|
|
usage = codex.normalize_usage_payload(
|
|
{
|
|
"plan_type": "plus",
|
|
"rate_limit": {
|
|
"primary_window": {
|
|
"used_percent": 39,
|
|
"reset_at": 1_738_300_000,
|
|
"limit_window_seconds": 18_000,
|
|
},
|
|
"secondary_window": {
|
|
"used_percent": 15,
|
|
"reset_at": 1_738_900_000,
|
|
"limit_window_seconds": 604_800,
|
|
},
|
|
},
|
|
"credits": {"has_credits": True, "unlimited": False, "balance": 5.39},
|
|
}
|
|
)
|
|
|
|
assert usage["available"] is True
|
|
assert usage["plan_type"] == "plus"
|
|
assert usage["primary"]["used_percent"] == 39
|
|
assert usage["primary"]["remaining_percent"] == 61
|
|
assert usage["primary"]["label"] == "5h"
|
|
assert usage["secondary"]["used_percent"] == 15
|
|
assert usage["secondary"]["label"] == "7d"
|
|
assert usage["credits"]["balance"] == 5.39
|
|
|
|
|
|
def test_normalize_usage_payload_accepts_zero_percent_headers():
|
|
usage = codex.normalize_usage_payload(
|
|
{},
|
|
{
|
|
"x-codex-primary-used-percent": "0",
|
|
"x-codex-primary-window-minutes": "300",
|
|
},
|
|
)
|
|
|
|
assert usage["available"] is True
|
|
assert usage["primary"]["used_percent"] == 0
|
|
assert usage["primary"]["remaining_percent"] == 100
|
|
assert usage["primary"]["label"] == "5h"
|
|
|
|
|
|
def test_disconnect_auth_clears_chatgpt_tokens_and_preserves_api_key(tmp_path, monkeypatch):
|
|
private_auth = tmp_path / "private-auth.json"
|
|
shared_auth = tmp_path / "shared-auth.json"
|
|
private_auth.write_text(
|
|
json.dumps(
|
|
{
|
|
"auth_mode": "chatgpt",
|
|
"OPENAI_API_KEY": None,
|
|
"tokens": {
|
|
"access_token": "access",
|
|
"refresh_token": "refresh",
|
|
"id_token": "id",
|
|
"account_id": "account",
|
|
},
|
|
"last_refresh": "2026-01-01T00:00:00Z",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
shared_auth.write_text(
|
|
json.dumps(
|
|
{
|
|
"auth_mode": "chatgpt",
|
|
"OPENAI_API_KEY": "sk-keep",
|
|
"tokens": {"access_token": "access", "account_id": "account"},
|
|
"last_refresh": "2026-01-01T00:00:00Z",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(codex, "resolve_auth_file_candidates", lambda: [private_auth, shared_auth])
|
|
|
|
result = codex.disconnect_auth()
|
|
|
|
assert result["disconnected"] is True
|
|
assert str(private_auth) in result["removed_auth_files"]
|
|
assert not private_auth.exists()
|
|
preserved = json.loads(shared_auth.read_text(encoding="utf-8"))
|
|
assert preserved == {"OPENAI_API_KEY": "sk-keep"}
|
|
|
|
|
|
def test_provider_config_uses_container_local_agent_zero_origin():
|
|
provider_path = Path(__file__).resolve().parents[1] / "plugins/_oauth/conf/model_providers.yaml"
|
|
provider_config = yaml.safe_load(provider_path.read_text(encoding="utf-8"))
|
|
codex_provider = provider_config["chat"]["codex_oauth"]
|
|
|
|
assert codex_provider["name"] == "Codex/ChatGPT Account"
|
|
assert codex_provider["models_list"]["endpoint_url"] == "/models"
|
|
assert codex_provider["kwargs"]["api_base"] == "http://127.0.0.1/oauth/codex/v1"
|
|
assert "50001" not in json.dumps(codex_provider)
|
|
|
|
|
|
def test_codex_provider_reports_dummy_api_key_when_missing():
|
|
data = {"args": ("codex_oauth",), "kwargs": {}, "result": "None"}
|
|
|
|
CodexAccountDummyKey(agent=None).execute(data=data)
|
|
|
|
assert data["result"] == "oauth"
|
|
|
|
|
|
def test_codex_provider_preserves_configured_api_key():
|
|
data = {"args": ("codex_oauth",), "kwargs": {}, "result": "configured"}
|
|
|
|
CodexAccountDummyKey(agent=None).execute(data=data)
|
|
|
|
assert data["result"] == "configured"
|