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

"""Tests for Hugging Face Inference Providers."""

from dataclasses import replace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from free_claude_code.config.provider_catalog import HUGGINGFACE_DEFAULT_BASE
from free_claude_code.core.anthropic import ReasoningReplayMode
from free_claude_code.providers.base import ProviderConfig
from tests.providers.request_factory import make_messages_request
from tests.providers.support import passthrough_rate_limiter, profiled_provider


def make_request(**overrides):
    return make_messages_request("openai/gpt-oss-120b:fastest", **overrides)


@pytest.fixture
def huggingface_config():
    return ProviderConfig(
        api_key="test_hf_key",
        base_url=HUGGINGFACE_DEFAULT_BASE,
        rate_limit=10,
        rate_window=60,
        enable_thinking=True,
    )


@pytest.fixture
def huggingface_provider(huggingface_config):
    return profiled_provider(
        "huggingface", huggingface_config, rate_limiter=passthrough_rate_limiter()
    )


def test_default_base_url_constant():
    assert HUGGINGFACE_DEFAULT_BASE == "https://router.huggingface.co/v1"


def test_init_uses_default_base_url_and_api_key(huggingface_config):
    with patch(
        "free_claude_code.providers.openai_chat.provider.AsyncOpenAI"
    ) as mock_openai:
        provider = profiled_provider(
            "huggingface", huggingface_config, rate_limiter=passthrough_rate_limiter()
        )

    assert provider._api_key == "test_hf_key"
    assert provider._base_url == HUGGINGFACE_DEFAULT_BASE
    mock_openai.assert_called_once()


def test_init_strips_trailing_slash(huggingface_config):
    config = replace(huggingface_config, base_url=f"{HUGGINGFACE_DEFAULT_BASE}/")

    with patch("free_claude_code.providers.openai_chat.provider.AsyncOpenAI"):
        provider = profiled_provider(
            "huggingface", config, rate_limiter=passthrough_rate_limiter()
        )

    assert provider._base_url == HUGGINGFACE_DEFAULT_BASE


def test_build_request_body_keeps_max_tokens(huggingface_provider):
    with patch(
        "free_claude_code.providers.openai_chat.request_policy.build_base_request_body"
    ) as mock_convert:
        mock_convert.return_value = {
            "model": "openai/gpt-oss-120b:fastest",
            "messages": [{"role": "user", "name": "alice", "content": "hi"}],
            "max_tokens": 42,
        }

        body = huggingface_provider._build_request_body(make_request())

    mock_convert.assert_called_once()
    assert (
        mock_convert.call_args.kwargs["reasoning_replay"]
        is ReasoningReplayMode.DISABLED
    )
    assert body["messages"][0].get("name") == "alice"
    assert body["max_tokens"] == 42
    assert "max_completion_tokens" not in body


def test_build_request_body_preserves_caller_extra_body(huggingface_provider):
    extra_body = {"provider": "auto", "routing": {"bill_to": "my-org"}}
    req = make_request(extra_body=extra_body)

    body = huggingface_provider._build_request_body(req)

    assert body["extra_body"] == extra_body
    assert body["extra_body"] is not extra_body
    assert body["extra_body"]["routing"] is not extra_body["routing"]


def test_build_request_body_does_not_replay_prior_thinking_blocks(
    huggingface_provider,
):
    req = make_request(
        system=None,
        messages=[
            {
                "role": "assistant",
                "content": [
                    {"type": "thinking", "thinking": "hidden prior thought"},
                    {"type": "text", "text": "visible answer"},
                ],
            }
        ],
    )

    body = huggingface_provider._build_request_body(req)

    assert body["messages"] == [{"role": "assistant", "content": "visible answer"}]
    assert "reasoning_content" not in body["messages"][0]
    assert "hidden prior thought" not in str(body)


def test_build_request_body_does_not_replay_top_level_reasoning_content(
    huggingface_provider,
):
    req = make_request(
        system=None,
        messages=[
            {
                "role": "assistant",
                "content": "visible answer",
                "reasoning_content": "hidden prior reasoning",
            }
        ],
    )

    body = huggingface_provider._build_request_body(req)

    assert body["messages"] == [{"role": "assistant", "content": "visible answer"}]
    assert "hidden prior reasoning" not in str(body)


@pytest.mark.asyncio
async def test_stream_response_text(huggingface_provider):
    mock_chunk = MagicMock()
    mock_chunk.choices = [
        MagicMock(
            delta=MagicMock(
                content="Hello from Hugging Face",
                reasoning_content=None,
                tool_calls=None,
            ),
            finish_reason="stop",
        )
    ]
    mock_chunk.usage = MagicMock(completion_tokens=5, prompt_tokens=10)

    async def mock_stream():
        yield mock_chunk

    with patch.object(
        huggingface_provider._client.chat.completions, "create", new_callable=AsyncMock
    ) as mock_create:
        mock_create.return_value = mock_stream()

        events = [
            event
            async for event in huggingface_provider.stream_response(make_request())
        ]

    assert any(
        '"text_delta"' in event and "Hello from Hugging Face" in event
        for event in events
    )


@pytest.mark.asyncio
async def test_stream_response_reasoning_content(huggingface_provider):
    mock_chunk = MagicMock()
    mock_chunk.choices = [
        MagicMock(
            delta=MagicMock(
                content=None,
                reasoning_content="Thinking via router",
                tool_calls=None,
            ),
            finish_reason="stop",
        )
    ]
    mock_chunk.usage = MagicMock(completion_tokens=2, prompt_tokens=10)

    async def mock_stream():
        yield mock_chunk

    with patch.object(
        huggingface_provider._client.chat.completions, "create", new_callable=AsyncMock
    ) as mock_create:
        mock_create.return_value = mock_stream()

        events = [
            event
            async for event in huggingface_provider.stream_response(make_request())
        ]

    assert any(
        '"thinking_delta"' in event and "Thinking via router" in event
        for event in events
    )


@pytest.mark.asyncio
async def test_cleanup(huggingface_provider):
    huggingface_provider._client = AsyncMock()

    await huggingface_provider.cleanup()

    huggingface_provider._client.close.assert_called_once()