mock-provider

name	mock-provider
description	Generate mock adapters for external services in tests. Use when: (1) Testing code that calls OpenAI/Anthropic, (2) Mocking storage providers, (3) Testing without external dependencies, (4) Simulating error conditions. Creates deterministic mocks with configurable responses and async support. NOT for integration tests with real services.

Mock Provider Generator

Generate mock implementations of adapter interfaces for testing.

Quick Reference

/mock-provider llm         # Mock LLM adapter
/mock-provider storage     # Mock storage adapter
/mock-provider search      # Mock search adapter
/mock-provider all         # All mock providers

Generated Structure

tests/
└── mocks/
    ├── __init__.py
    ├── llm.py            # Mock LLM provider
    ├── storage.py        # Mock storage provider
    ├── search.py         # Mock search provider
    └── fixtures.py       # Pytest fixtures

Implementation

1. Mock LLM Provider

# tests/mocks/llm.py
from typing import AsyncIterator
from adapters.llm.base import LLMAdapter, LLMResponse, Message
from dataclasses import dataclass, field

@dataclass
class MockLLMAdapter:
    """Mock LLM adapter for testing."""

    # Configurable responses
    default_response: str = "This is a mock response."
    responses: dict[str, str] = field(default_factory=dict)
    token_count: int = 10
    should_fail: bool = False
    error_message: str = "Mock error"
    stream_delay: float = 0.01

    # Call tracking
    calls: list[dict] = field(default_factory=list)

    async def chat(
        self,
        messages: list[Message],
        model: str | None = None,
        temperature: float = 0.7,
        max_tokens: int | None = None,
    ) -> LLMResponse:
        self.calls.append({
            "method": "chat",
            "messages": messages,
            "model": model,
            "temperature": temperature,
        })

        if self.should_fail:
            raise Exception(self.error_message)

        # Check for specific response based on last message
        last_content = messages[-1].content if messages else ""
        response_text = self.responses.get(last_content, self.default_response)

        return LLMResponse(
            content=response_text,
            model=model or "mock-model",
            usage={
                "prompt_tokens": self.token_count,
                "completion_tokens": len(response_text.split()),
                "total_tokens": self.token_count + len(response_text.split()),
            },
        )

    async def stream(
        self,
        messages: list[Message],
        model: str | None = None,
        temperature: float = 0.7,
    ) -> AsyncIterator[str]:
        self.calls.append({
            "method": "stream",
            "messages": messages,
            "model": model,
        })

        if self.should_fail:
            raise Exception(self.error_message)

        last_content = messages[-1].content if messages else ""
        response_text = self.responses.get(last_content, self.default_response)

        import asyncio
        for word in response_text.split():
            await asyncio.sleep(self.stream_delay)
            yield word + " "

    def count_tokens(self, text: str) -> int:
        return len(text.split())

    def reset(self) -> None:
        """Reset call history."""
        self.calls.clear()

    def assert_called_with(self, content: str) -> None:
        """Assert that a message containing content was sent."""
        for call in self.calls:
            for msg in call.get("messages", []):
                if content in msg.content:
                    return
        raise AssertionError(f"No call with content '{content}' found")

2. Mock Storage Provider

# tests/mocks/storage.py
from dataclasses import dataclass, field
from io import BytesIO

@dataclass
class MockStorageAdapter:
    """Mock storage adapter for testing."""

    files: dict[str, bytes] = field(default_factory=dict)
    should_fail: bool = False
    calls: list[dict] = field(default_factory=list)

    async def upload(
        self,
        key: str,
        data: bytes | BytesIO,
        content_type: str = "application/octet-stream",
    ) -> str:
        self.calls.append({"method": "upload", "key": key})

        if self.should_fail:
            raise Exception("Upload failed")

        if isinstance(data, BytesIO):
            data = data.read()
        self.files[key] = data
        return f"mock://storage/{key}"

    async def download(self, key: str) -> bytes:
        self.calls.append({"method": "download", "key": key})

        if self.should_fail:
            raise Exception("Download failed")

        if key not in self.files:
            raise FileNotFoundError(f"Key not found: {key}")
        return self.files[key]

    async def delete(self, key: str) -> None:
        self.calls.append({"method": "delete", "key": key})
        self.files.pop(key, None)

    async def exists(self, key: str) -> bool:
        return key in self.files

    async def list_keys(self, prefix: str = "") -> list[str]:
        return [k for k in self.files.keys() if k.startswith(prefix)]

    def reset(self) -> None:
        self.files.clear()
        self.calls.clear()

3. Pytest Fixtures

# tests/mocks/fixtures.py
import pytest
from tests.mocks.llm import MockLLMAdapter
from tests.mocks.storage import MockStorageAdapter

@pytest.fixture
def mock_llm():
    """Fixture for mock LLM adapter."""
    adapter = MockLLMAdapter()
    yield adapter
    adapter.reset()

@pytest.fixture
def mock_llm_with_responses():
    """Fixture for mock LLM with custom responses."""
    def _create(**responses):
        return MockLLMAdapter(responses=responses)
    return _create

@pytest.fixture
def mock_storage():
    """Fixture for mock storage adapter."""
    adapter = MockStorageAdapter()
    yield adapter
    adapter.reset()

@pytest.fixture
def mock_llm_error():
    """Fixture for mock LLM that fails."""
    return MockLLMAdapter(should_fail=True, error_message="API rate limited")

Usage in Tests

import pytest
from services.chat_service import ChatService

@pytest.mark.asyncio
async def test_chat_response(mock_llm):
    mock_llm.default_response = "Hello, I'm an AI assistant."

    service = ChatService(llm=mock_llm)
    response = await service.chat("Hi there!")

    assert "AI assistant" in response.content
    assert len(mock_llm.calls) == 1
    mock_llm.assert_called_with("Hi there!")

@pytest.mark.asyncio
async def test_chat_error_handling(mock_llm_error):
    service = ChatService(llm=mock_llm_error)

    with pytest.raises(Exception, match="rate limited"):
        await service.chat("Hi")

@pytest.mark.asyncio
async def test_file_upload(mock_storage):
    await mock_storage.upload("test.txt", b"Hello")

    assert await mock_storage.exists("test.txt")
    assert await mock_storage.download("test.txt") == b"Hello"