Files
wehub-resource-sync 5296d0e97c
CI / Ban suppressions and legacy annotations (push) Has been cancelled
CI / pytest (push) Has been cancelled
CI / ruff-check (push) Has been cancelled
CI / ruff-format (push) Has been cancelled
CI / ty (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:35:44 +08:00

322 lines
10 KiB
Python

"""Tests for GitHub Models OpenAI-compatible provider."""
from collections.abc import AsyncIterator
from dataclasses import replace
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import httpx
import pytest
from free_claude_code.config.provider_catalog import GITHUB_MODELS_DEFAULT_BASE
from free_claude_code.core.anthropic.models import Message, MessagesRequest
from free_claude_code.core.anthropic.stream_contracts import parse_sse_text
from free_claude_code.providers.base import ProviderConfig
from free_claude_code.providers.github_models import GitHubModelsProvider
from free_claude_code.providers.github_models.client import GITHUB_MODELS_CATALOG_URL
from free_claude_code.providers.model_listing import ModelListResponseError
from tests.providers.support import passthrough_rate_limiter
@pytest.fixture
def github_models_config() -> ProviderConfig:
return ProviderConfig(
api_key="test-github-models-token",
base_url=GITHUB_MODELS_DEFAULT_BASE,
rate_limit=10,
rate_window=60,
enable_thinking=True,
)
@pytest.fixture
def github_models_provider(
github_models_config: ProviderConfig,
) -> GitHubModelsProvider:
return GitHubModelsProvider(
github_models_config, rate_limiter=passthrough_rate_limiter()
)
def _request(model: str = "openai/gpt-4.1") -> MessagesRequest:
return MessagesRequest(
model=model,
max_tokens=100,
messages=[Message(role="user", content="hi")],
)
def _chunk(delta: SimpleNamespace, *, finish_reason: str = "stop") -> SimpleNamespace:
return SimpleNamespace(
choices=[SimpleNamespace(delta=delta, finish_reason=finish_reason)],
usage=SimpleNamespace(completion_tokens=5, prompt_tokens=8),
)
async def _stream(*chunks: SimpleNamespace) -> AsyncIterator[SimpleNamespace]:
for chunk in chunks:
yield chunk
def _catalog_response(payload: object) -> httpx.Response:
return httpx.Response(
200,
json=payload,
request=httpx.Request("GET", GITHUB_MODELS_CATALOG_URL),
)
def test_default_base_url_constant() -> None:
assert GITHUB_MODELS_DEFAULT_BASE == "https://models.github.ai/inference"
def test_init_uses_default_base_url_api_key_and_github_headers(
github_models_config: ProviderConfig,
) -> None:
with patch(
"free_claude_code.providers.openai_chat.provider.AsyncOpenAI"
) as mock_openai:
provider = GitHubModelsProvider(
github_models_config, rate_limiter=passthrough_rate_limiter()
)
assert provider._api_key == "test-github-models-token"
assert provider._base_url == GITHUB_MODELS_DEFAULT_BASE
assert provider._catalog_url == GITHUB_MODELS_CATALOG_URL
assert mock_openai.call_args.kwargs["base_url"] == GITHUB_MODELS_DEFAULT_BASE
assert mock_openai.call_args.kwargs["api_key"] == "test-github-models-token"
assert mock_openai.call_args.kwargs["default_headers"] == {
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2026-03-10",
}
def test_init_strips_trailing_slash(github_models_config: ProviderConfig) -> None:
config = replace(
github_models_config,
base_url=f"{GITHUB_MODELS_DEFAULT_BASE}/",
)
with patch("free_claude_code.providers.openai_chat.provider.AsyncOpenAI"):
provider = GitHubModelsProvider(config, rate_limiter=passthrough_rate_limiter())
assert provider._base_url == GITHUB_MODELS_DEFAULT_BASE
def test_model_list_headers_use_bearer_auth(
github_models_provider: GitHubModelsProvider,
) -> None:
assert github_models_provider._model_list_headers() == {
"Accept": "application/vnd.github+json",
"Authorization": "Bearer test-github-models-token",
"X-GitHub-Api-Version": "2026-03-10",
}
def test_build_request_body_uses_shared_openai_chat_policy(
github_models_provider: GitHubModelsProvider,
) -> None:
request = _request()
body = github_models_provider._build_request_body(request, thinking_enabled=True)
assert body["model"] == "openai/gpt-4.1"
assert body["max_tokens"] == 100
assert "extra_body" not in body
@pytest.mark.asyncio
async def test_lists_stream_tool_capable_models_only(
github_models_provider: GitHubModelsProvider,
) -> None:
with patch.object(
github_models_provider._model_list_client,
"get",
new_callable=AsyncMock,
return_value=_catalog_response(
[
{
"id": "openai/gpt-4.1",
"capabilities": ["streaming", "tool-calling"],
},
{
"id": "openai/text-only",
"capabilities": ["streaming"],
},
{
"id": "openai/no-stream-tools",
"capabilities": ["tool-calling"],
},
]
),
) as mock_get:
assert await github_models_provider.list_model_ids() == frozenset(
{"openai/gpt-4.1"}
)
mock_get.assert_awaited_once_with(
GITHUB_MODELS_CATALOG_URL,
headers={
"Accept": "application/vnd.github+json",
"Authorization": "Bearer test-github-models-token",
"X-GitHub-Api-Version": "2026-03-10",
},
)
@pytest.mark.asyncio
async def test_model_list_rejects_malformed_payload(
github_models_provider: GitHubModelsProvider,
) -> None:
with (
patch.object(
github_models_provider._model_list_client,
"get",
new_callable=AsyncMock,
return_value=_catalog_response({"data": []}),
),
pytest.raises(ModelListResponseError, match="top-level array"),
):
await github_models_provider.list_model_ids()
@pytest.mark.asyncio
async def test_model_list_returns_empty_set_when_no_models_support_streaming_tools(
github_models_provider: GitHubModelsProvider,
) -> None:
with patch.object(
github_models_provider._model_list_client,
"get",
new_callable=AsyncMock,
return_value=_catalog_response(
[
{"id": "openai/text-only", "capabilities": ["streaming"]},
{"id": "openai/non-stream-tool", "capabilities": ["tool-calling"]},
]
),
):
assert await github_models_provider.list_model_ids() == frozenset()
@pytest.mark.asyncio
async def test_stream_response_text(
github_models_provider: GitHubModelsProvider,
) -> None:
delta = SimpleNamespace(
content="Hello from GitHub Models",
reasoning_content=None,
tool_calls=None,
)
with patch.object(
github_models_provider._client.chat.completions,
"create",
new_callable=AsyncMock,
return_value=_stream(_chunk(delta)),
) as mock_create:
events = [
event async for event in github_models_provider.stream_response(_request())
]
parsed = parse_sse_text("".join(events))
assert any(
event.event == "content_block_delta"
and event.data.get("delta", {}).get("text") == "Hello from GitHub Models"
for event in parsed
)
assert mock_create.call_args.kwargs["model"] == "openai/gpt-4.1"
assert mock_create.call_args.kwargs["stream"] is True
@pytest.mark.asyncio
async def test_stream_response_tool_call(
github_models_provider: GitHubModelsProvider,
) -> None:
tool_call = SimpleNamespace(
index=0,
id="call_1",
function=SimpleNamespace(name="echo", arguments='{"value":"x"}'),
)
delta = SimpleNamespace(
content=None, reasoning_content=None, tool_calls=[tool_call]
)
request = MessagesRequest.model_validate(
{
"model": "openai/gpt-4.1",
"messages": [{"role": "user", "content": "Use the tool"}],
"tools": [
{
"name": "echo",
"description": "Echo a value",
"input_schema": {
"type": "object",
"properties": {"value": {"type": "string"}},
"required": ["value"],
},
}
],
}
)
with patch.object(
github_models_provider._client.chat.completions,
"create",
new_callable=AsyncMock,
return_value=_stream(_chunk(delta, finish_reason="tool_calls")),
):
events = [
event async for event in github_models_provider.stream_response(request)
]
parsed = parse_sse_text("".join(events))
assert any(
event.event == "content_block_start"
and event.data.get("content_block", {}).get("type") == "tool_use"
and event.data.get("content_block", {}).get("name") == "echo"
for event in parsed
)
assert any(
event.event == "content_block_delta"
and event.data.get("delta", {}).get("partial_json") == '{"value":"x"}'
for event in parsed
)
@pytest.mark.asyncio
async def test_stream_response_reasoning_content(
github_models_provider: GitHubModelsProvider,
) -> None:
delta = SimpleNamespace(
content=None,
reasoning_content="Thinking via GitHub Models",
tool_calls=None,
)
with patch.object(
github_models_provider._client.chat.completions,
"create",
new_callable=AsyncMock,
return_value=_stream(_chunk(delta)),
):
events = [
event async for event in github_models_provider.stream_response(_request())
]
parsed = parse_sse_text("".join(events))
assert any(
event.event == "content_block_delta"
and event.data.get("delta", {}).get("thinking") == "Thinking via GitHub Models"
for event in parsed
)
@pytest.mark.asyncio
async def test_cleanup(github_models_provider: GitHubModelsProvider) -> None:
github_models_provider._client = AsyncMock()
github_models_provider._model_list_client = AsyncMock()
await github_models_provider.cleanup()
github_models_provider._client.close.assert_called_once()
github_models_provider._model_list_client.aclose.assert_called_once()