alishahryar1--free-claude-code/tests/providers/test_open_router.py

"""Tests for the OpenRouter OpenAI-chat provider."""

from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from free_claude_code.application.errors import InvalidRequestError
from free_claude_code.config.constants import ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS
from free_claude_code.core.anthropic.models import MessagesRequest
from free_claude_code.core.anthropic.stream_contracts import (
    parse_sse_text,
    text_content,
)
from free_claude_code.providers.base import ProviderConfig
from free_claude_code.providers.open_router import OpenRouterProvider
from free_claude_code.providers.openai_chat import OpenAIChatProvider
from tests.providers.request_factory import make_messages_request
from tests.providers.support import passthrough_rate_limiter


class AsyncStream:
    def __init__(self, chunks):
        self._chunks = chunks
        self.closed = False

    def __aiter__(self):
        return self._iter()

    async def _iter(self):
        for chunk in self._chunks:
            yield chunk

    async def aclose(self):
        self.closed = True


def make_request(**overrides):
    return make_messages_request("moonshotai/kimi-k2.6:free", **overrides)


@pytest.fixture
def open_router_provider():
    return OpenRouterProvider(
        ProviderConfig(
            api_key="test_openrouter_key",
            base_url="https://openrouter.ai/api/v1",
            rate_limit=10,
            rate_window=60,
        ),
        rate_limiter=passthrough_rate_limiter(),
    )


def _chunk(
    *,
    content: str | None = None,
    reasoning_content: str | None = None,
    reasoning_details: list[dict] | None = None,
    finish_reason: str | None = None,
):
    delta = SimpleNamespace(
        content=content,
        reasoning_content=reasoning_content,
        tool_calls=None,
    )
    if reasoning_details is not None:
        delta.reasoning_details = reasoning_details
    choice = SimpleNamespace(delta=delta, finish_reason=finish_reason)
    return SimpleNamespace(choices=[choice], usage=None)


def test_init_uses_openai_chat_provider(open_router_provider):
    assert isinstance(open_router_provider, OpenAIChatProvider)
    assert open_router_provider._api_key == "test_openrouter_key"
    assert open_router_provider._base_url == "https://openrouter.ai/api/v1"


def test_build_request_body_uses_openai_chat_shape(open_router_provider):
    body = open_router_provider._build_request_body(make_request())

    assert body["model"] == "moonshotai/kimi-k2.6:free"
    assert body["temperature"] == 0.5
    assert body["messages"] == [
        {"role": "system", "content": "System prompt"},
        {"role": "user", "content": "Hello"},
    ]
    assert body["max_tokens"] == 100
    assert body["extra_body"]["reasoning"] == {"enabled": True}


def test_build_request_body_default_max_tokens(open_router_provider):
    body = open_router_provider._build_request_body(make_request(max_tokens=None))

    assert body["max_tokens"] == ANTHROPIC_DEFAULT_MAX_OUTPUT_TOKENS


def test_openrouter_extra_body_rejects_overriding_reserved_fields(
    open_router_provider,
):
    with pytest.raises(InvalidRequestError, match="model"):
        open_router_provider._build_request_body(
            make_request(extra_body={"model": "hijack"})
        )


def test_openrouter_extra_body_allows_provider_keys(open_router_provider):
    body = open_router_provider._build_request_body(
        make_request(extra_body={"transforms": ["no-web"], "plugins": []}),
        thinking_enabled=False,
    )

    assert body["extra_body"] == {"transforms": ["no-web"], "plugins": []}


def test_build_request_body_omits_reasoning_when_thinking_disabled(
    open_router_provider,
):
    body = open_router_provider._build_request_body(
        make_request(thinking={"type": "disabled"})
    )

    assert "extra_body" not in body


def test_build_request_body_maps_thinking_budget_to_reasoning_max_tokens(
    open_router_provider,
):
    body = open_router_provider._build_request_body(
        make_request(thinking={"type": "enabled", "budget_tokens": 4096})
    )

    assert body["extra_body"]["reasoning"] == {"enabled": True, "max_tokens": 4096}


def test_build_request_body_replays_openrouter_reasoning_details(
    open_router_provider,
):
    detail = {"type": "reasoning.encrypted", "data": "opaque"}
    request = MessagesRequest.model_validate(
        {
            "model": "m",
            "messages": [
                {
                    "role": "assistant",
                    "content": [
                        {
                            "type": "redacted_thinking",
                            "data": '{"type":"reasoning.encrypted","data":"opaque"}',
                        },
                        {"type": "text", "text": "Need a tool."},
                    ],
                },
                {"role": "user", "content": "continue"},
            ],
        }
    )

    body = open_router_provider._build_request_body(request)

    assistant = next(msg for msg in body["messages"] if msg["role"] == "assistant")
    assert assistant["reasoning_details"] == [detail]


@pytest.mark.asyncio
async def test_stream_maps_reasoning_content_and_details(open_router_provider):
    redacted = {"type": "reasoning.encrypted", "data": "opaque"}
    stream = AsyncStream(
        [
            _chunk(reasoning_content="plan "),
            _chunk(reasoning_details=[redacted]),
            _chunk(content="done", finish_reason="stop"),
        ]
    )
    with patch.object(
        open_router_provider._client.chat.completions,
        "create",
        new_callable=AsyncMock,
        return_value=stream,
    ):
        events = [
            event
            async for event in open_router_provider.stream_response(make_request())
        ]

    event_text = "".join(events)
    assert "thinking_delta" in event_text
    assert "plan " in event_text
    assert "redacted_thinking" in event_text
    assert "opaque" in event_text
    assert "done" in text_content(parse_sse_text(event_text))
    assert stream.closed


@pytest.mark.asyncio
async def test_model_infos_filter_tool_models_and_thinking_metadata(
    open_router_provider,
):
    open_router_provider._client.models.list = AsyncMock(
        return_value=SimpleNamespace(
            data=[
                SimpleNamespace(
                    id="tool-model",
                    supported_parameters=["tools", "reasoning"],
                ),
                SimpleNamespace(id="plain-model", supported_parameters=[]),
            ]
        )
    )

    infos = await open_router_provider.list_model_infos()

    assert {(info.model_id, info.supports_thinking) for info in infos} == {
        ("tool-model", True)
    }


@pytest.mark.asyncio
async def test_cleanup_closes_openai_client(open_router_provider):
    open_router_provider._client = MagicMock()
    open_router_provider._client.close = AsyncMock()

    await open_router_provider.cleanup()

    open_router_provider._client.close.assert_awaited_once()