wehub-resource-sync
245 lines
8.5 KiB
Python
245 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from openharness.api.client import ApiMessageRequest, ApiMessageCompleteEvent, ApiTextDeltaEvent
|
|
from openharness.api.codex_client import (
|
|
CodexApiClient,
|
|
_convert_messages_to_codex,
|
|
_format_codex_stream_error,
|
|
_resolve_codex_url,
|
|
)
|
|
from openharness.engine.messages import ConversationMessage, ImageBlock, TextBlock, ToolResultBlock, ToolUseBlock
|
|
|
|
|
|
class _FakeStreamResponse:
|
|
def __init__(self, *, status_code: int = 200, lines: list[str] | None = None, body: str = "") -> None:
|
|
self.status_code = status_code
|
|
self._lines = lines or []
|
|
self._body = body.encode("utf-8")
|
|
|
|
async def __aenter__(self) -> "_FakeStreamResponse":
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
async def aread(self) -> bytes:
|
|
return self._body
|
|
|
|
async def aiter_lines(self):
|
|
for line in self._lines:
|
|
yield line
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
def __init__(self, response: _FakeStreamResponse, sink: dict[str, Any]) -> None:
|
|
self._response = response
|
|
self._sink = sink
|
|
|
|
async def __aenter__(self) -> "_FakeAsyncClient":
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def stream(self, method: str, url: str, *, headers: dict[str, str], json: dict[str, Any]):
|
|
self._sink["method"] = method
|
|
self._sink["url"] = url
|
|
self._sink["headers"] = headers
|
|
self._sink["json"] = json
|
|
return self._response
|
|
|
|
|
|
def _b64url(data: dict[str, object]) -> str:
|
|
raw = json.dumps(data, separators=(",", ":")).encode("utf-8")
|
|
import base64
|
|
|
|
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _fake_codex_token() -> str:
|
|
payload = {"https://api.openai.com/auth": {"chatgpt_account_id": "acct_test"}}
|
|
return f"{_b64url({'alg': 'none', 'typ': 'JWT'})}.{_b64url(payload)}.sig"
|
|
|
|
|
|
def test_convert_messages_to_codex():
|
|
messages = [
|
|
ConversationMessage.from_user_text("Inspect file"),
|
|
ConversationMessage(
|
|
role="assistant",
|
|
content=[
|
|
TextBlock(text="I'll inspect it."),
|
|
ToolUseBlock(id="call_123", name="read_file", input={"path": "README.md"}),
|
|
],
|
|
),
|
|
ConversationMessage(
|
|
role="user",
|
|
content=[ToolResultBlock(tool_use_id="call_123", content="hello", is_error=False)],
|
|
),
|
|
]
|
|
|
|
converted = _convert_messages_to_codex(messages)
|
|
|
|
assert converted[0] == {
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": "Inspect file"}],
|
|
}
|
|
assert converted[1]["type"] == "message"
|
|
assert converted[1]["role"] == "assistant"
|
|
assert converted[2]["type"] == "function_call"
|
|
assert converted[2]["call_id"] == "call_123"
|
|
assert json.loads(converted[2]["arguments"]) == {"path": "README.md"}
|
|
assert converted[3] == {
|
|
"type": "function_call_output",
|
|
"call_id": "call_123",
|
|
"output": "hello",
|
|
}
|
|
|
|
|
|
def test_convert_user_message_with_tool_result_before_text_to_codex():
|
|
messages = [
|
|
ConversationMessage(
|
|
role="user",
|
|
content=[
|
|
ToolResultBlock(tool_use_id="call_123", content="done", is_error=False),
|
|
TextBlock(text="next request"),
|
|
],
|
|
)
|
|
]
|
|
|
|
converted = _convert_messages_to_codex(messages)
|
|
|
|
assert converted == [
|
|
{"type": "function_call_output", "call_id": "call_123", "output": "done"},
|
|
{"role": "user", "content": [{"type": "input_text", "text": "next request"}]},
|
|
]
|
|
|
|
|
|
|
|
def test_convert_multimodal_user_message_to_codex():
|
|
messages = [
|
|
ConversationMessage(
|
|
role="user",
|
|
content=[
|
|
TextBlock(text="What is in this image?"),
|
|
ImageBlock(media_type="image/png", data="YWJj", source_path="/tmp/example.png"),
|
|
],
|
|
)
|
|
]
|
|
|
|
converted = _convert_messages_to_codex(messages)
|
|
|
|
assert converted == [{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "input_text", "text": "What is in this image?"},
|
|
{"type": "input_image", "image_url": "data:image/png;base64,YWJj"},
|
|
],
|
|
}]
|
|
|
|
|
|
def test_resolve_codex_url_ignores_unrelated_base_url():
|
|
assert _resolve_codex_url("https://api.moonshot.cn/anthropic") == "https://chatgpt.com/backend-api/codex/responses"
|
|
|
|
|
|
def test_format_codex_stream_error_includes_code_and_request_id():
|
|
message = _format_codex_stream_error(
|
|
{
|
|
"type": "error",
|
|
"message": "Upstream overloaded",
|
|
"code": "overloaded",
|
|
"request_id": "req_123",
|
|
},
|
|
fallback="Codex error",
|
|
)
|
|
assert message == "Upstream overloaded (code=overloaded) [request_id=req_123]"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_codex_client_streams_text(monkeypatch):
|
|
sink: dict[str, Any] = {}
|
|
response = _FakeStreamResponse(
|
|
lines=[
|
|
'event: response.output_item.added',
|
|
'data: {"type":"response.output_item.added","item":{"id":"msg_1","type":"message","content":[],"role":"assistant"}}',
|
|
"",
|
|
'event: response.output_text.delta',
|
|
'data: {"type":"response.output_text.delta","delta":"CODE"}',
|
|
"",
|
|
'event: response.output_text.delta',
|
|
'data: {"type":"response.output_text.delta","delta":"X_OK"}',
|
|
"",
|
|
'event: response.output_item.done',
|
|
'data: {"type":"response.output_item.done","item":{"id":"msg_1","type":"message","content":[{"type":"output_text","text":"CODEX_OK","annotations":[]}]}}',
|
|
"",
|
|
'event: response.completed',
|
|
'data: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":12,"output_tokens":3}}}',
|
|
"",
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
"openharness.api.codex_client.httpx.AsyncClient",
|
|
lambda *args, **kwargs: _FakeAsyncClient(response, sink),
|
|
)
|
|
|
|
client = CodexApiClient(_fake_codex_token())
|
|
request = ApiMessageRequest(
|
|
model="gpt-5.5",
|
|
messages=[ConversationMessage.from_user_text("hi")],
|
|
system_prompt="Be helpful.",
|
|
effort="xhigh",
|
|
)
|
|
events = [event async for event in client.stream_message(request)]
|
|
|
|
assert [event.text for event in events if isinstance(event, ApiTextDeltaEvent)] == ["CODE", "X_OK"]
|
|
complete = next(event for event in events if isinstance(event, ApiMessageCompleteEvent))
|
|
assert complete.message.text == "CODEX_OK"
|
|
assert complete.usage.input_tokens == 12
|
|
assert complete.usage.output_tokens == 3
|
|
assert sink["url"].endswith("/codex/responses")
|
|
assert sink["json"]["instructions"] == "Be helpful."
|
|
assert sink["json"]["model"] == "gpt-5.5"
|
|
assert sink["json"]["reasoning"] == {"effort": "xhigh"}
|
|
assert sink["headers"]["OpenAI-Beta"] == "responses=experimental"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_codex_client_emits_tool_use(monkeypatch):
|
|
sink: dict[str, Any] = {}
|
|
response = _FakeStreamResponse(
|
|
lines=[
|
|
'data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","arguments":"","call_id":"call_abc","name":"glob"}}',
|
|
"",
|
|
'data: {"type":"response.output_item.done","item":{"id":"fc_1","type":"function_call","arguments":"{\\"pattern\\":\\"src/**/*.py\\"}","call_id":"call_abc","name":"glob"}}',
|
|
"",
|
|
'data: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":7,"output_tokens":2}}}',
|
|
"",
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
"openharness.api.codex_client.httpx.AsyncClient",
|
|
lambda *args, **kwargs: _FakeAsyncClient(response, sink),
|
|
)
|
|
|
|
client = CodexApiClient(_fake_codex_token())
|
|
request = ApiMessageRequest(
|
|
model="gpt-5.4",
|
|
messages=[ConversationMessage.from_user_text("glob")],
|
|
system_prompt="Use tools.",
|
|
tools=[{"name": "glob", "description": "find files", "input_schema": {"type": "object"}}],
|
|
)
|
|
events = [event async for event in client.stream_message(request)]
|
|
|
|
complete = next(event for event in events if isinstance(event, ApiMessageCompleteEvent))
|
|
assert complete.stop_reason == "tool_use"
|
|
assert len(complete.message.tool_uses) == 1
|
|
tool_use = complete.message.tool_uses[0]
|
|
assert tool_use.id == "call_abc"
|
|
assert tool_use.name == "glob"
|
|
assert tool_use.input == {"pattern": "src/**/*.py"}
|
|
assert sink["json"]["tools"][0]["name"] == "glob"
|