model-selector/tests/test_providers.py

from __future__ import annotations

import asyncio
import tempfile
from dataclasses import asdict
from pathlib import Path
import unittest

from model_selector.cache import SnapshotCache
from model_selector.providers import ProviderModel, derive_features, provider_model_to_config_entry
from model_selector.service import ModelDashboardService
from model_selector.config import OpenClawConfigRepository


class FakeAdapter:
    def __init__(self, provider_key: str, models: list[dict]) -> None:
        self.provider_key = provider_key
        self.models = models

    async def fetch(self):
        from model_selector.config import provider_label
        from model_selector.providers import ProviderModel

        return [
            ProviderModel(
                provider_key=self.provider_key,
                provider_label=provider_label(self.provider_key),
                model_id=item["id"],
                name=item.get("name", item["id"]),
                source="live",
                reasoning=item.get("reasoning"),
                context_window=item.get("context_window"),
                max_tokens=item.get("max_tokens"),
                pricing_prompt=item.get("pricing_prompt"),
                pricing_completion=item.get("pricing_completion"),
                pricing_request=item.get("pricing_request"),
                input_modalities=tuple(item.get("input_modalities", ("text",))),
                output_modalities=tuple(item.get("output_modalities", ("text",))),
                supported_parameters=tuple(item.get("supported_parameters", ())),
                active=item.get("active"),
                features=tuple(item.get("features", ())),
                raw=item,
                default_parameters=item.get("default_parameters"),
            )
            for item in self.models
        ]


class ProviderTests(unittest.IsolatedAsyncioTestCase):
    def test_feature_derivation_understands_free_and_structured(self) -> None:
        raw = {
            "id": "openai/gpt-oss-20b",
            "pricing": {"prompt": "0", "completion": "0", "request": "0"},
            "supported_parameters": ["response_format", "tools"],
            "architecture": {"input_modalities": ["text"], "output_modalities": ["text"]},
        }
        features = derive_features("openrouter", raw)
        self.assertIn("free", features)
        self.assertIn("structured output", features)
        self.assertIn("tool use", features)

    def test_openrouter_reasoning_surfaces_in_config_entry(self) -> None:
        model = ProviderModel(
            provider_key="openrouter",
            provider_label="OpenRouter",
            model_id="~anthropic/claude-sonnet-latest",
            name="Claude Sonnet",
            source="live",
            reasoning=True,
            context_window=200000,
            max_tokens=64000,
            pricing_prompt=None,
            pricing_completion=None,
            pricing_request=None,
            input_modalities=("text", "image", "file"),
            output_modalities=("text",),
            supported_parameters=("reasoning", "response_format"),
            active=None,
            features=("reasoning",),
            raw={"id": "~anthropic/claude-sonnet-latest"},
            default_parameters={"reasoning": {"effort": "high"}},
        )

        entry = provider_model_to_config_entry(model)

        self.assertTrue(entry["reasoning"])
        self.assertEqual(entry["thinking"], {"effort": "high"})
        self.assertEqual(entry["input"], ["text", "image"])

    async def test_refresh_writes_snapshot_json(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            config_path = Path(tmp) / "openclaw.json"
            cache_dir = Path(tmp) / "cache"
            config_path.write_text(
                """
{
  "models": {
    "providers": {
      "openai": {"models": [{"id": "gpt-5.4", "name": "GPT-5.4", "reasoning": true, "input": ["text"], "contextWindow": 100, "maxTokens": 10}]},
      "groq-cloud": {"models": [{"id": "openai/gpt-oss-20b", "name": "GPT OSS 20B", "reasoning": true, "input": ["text"], "contextWindow": 100, "maxTokens": 10}]},
      "openrouter": {"models": [{"id": "openai/gpt-oss-120b", "name": "GPT OSS 120B", "reasoning": true, "input": ["text"], "contextWindow": 100, "maxTokens": 10}]}
    }
  },
  "agents": {"defaults": {"models": {}}}
}
""".strip(),
                encoding="utf-8",
            )

            repo = OpenClawConfigRepository(config_path)
            service = ModelDashboardService(
                repo,
                SnapshotCache(cache_dir),
                adapters={
                    "openai": FakeAdapter("openai", [{"id": "gpt-5.4", "name": "GPT-5.4", "reasoning": True}]),
                    "groq-cloud": FakeAdapter("groq-cloud", [{"id": "openai/gpt-oss-20b", "name": "GPT OSS 20B", "reasoning": True}]),
                    "openrouter": FakeAdapter("openrouter", [{"id": "openai/gpt-oss-120b", "name": "GPT OSS 120B"}]),
                },
            )

            states = await service.refresh_all()

            self.assertIn("openai", states)
            self.assertTrue((cache_dir / "openai.json").exists())
            self.assertTrue((cache_dir / "groq-cloud.json").exists())
            self.assertTrue((cache_dir / "openrouter.json").exists())