101 lines
2.5 KiB
Python
101 lines
2.5 KiB
Python
"""
|
|
Shared test fixtures and config.
|
|
|
|
Fixtures defined here available to all tests
|
|
"""
|
|
|
|
from statistics import mode
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from anthropic.types import ContentBlock, Message, TextBlock, Usage
|
|
|
|
from agent.config import Settings
|
|
|
|
|
|
@pytest.fixture
|
|
def settings():
|
|
"""provide test settings don't load from .env"""
|
|
return Settings(
|
|
anthropic_api_key="test-key-12345",
|
|
model="claude-test-model",
|
|
max_tokens=100,
|
|
safedir="./test-workspace",
|
|
)
|
|
|
|
|
|
# tests/conftest.py - replace mock_anthropic_client fixture
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_anthropic_client():
|
|
"""Mock Anthropic client with streaming support."""
|
|
mock_client = AsyncMock()
|
|
|
|
# Create mock stream
|
|
mock_stream = AsyncMock()
|
|
|
|
# Mock text_stream - simple async generator
|
|
async def fake_text():
|
|
yield "42"
|
|
|
|
mock_stream.text_stream = fake_text()
|
|
|
|
# Mock get_final_message
|
|
mock_stream.get_final_message = AsyncMock(
|
|
return_value=Message(
|
|
id="msg_test123",
|
|
type="message",
|
|
role="assistant",
|
|
content=[TextBlock(type="text", text="42")],
|
|
model="claude-test-model",
|
|
stop_reason="end_turn",
|
|
usage=Usage(input_tokens=10, output_tokens=1),
|
|
)
|
|
)
|
|
|
|
# Mock context manager
|
|
mock_stream.__aenter__ = AsyncMock(return_value=mock_stream)
|
|
mock_stream.__aexit__ = AsyncMock()
|
|
|
|
# Wire up
|
|
mock_client.messages.stream = MagicMock(return_value=mock_stream)
|
|
|
|
return mock_client
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_history():
|
|
"""sample conversation history for testing"""
|
|
return [
|
|
{"role": "user", "content": "Hello"},
|
|
{"role": "assistance", "content": "Hi there!"},
|
|
{"role": "user", "content": "What's 2+2?"},
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_sandbox():
|
|
"""Mock sandbox for unit tests."""
|
|
sandbox = MagicMock()
|
|
sandbox.run = AsyncMock(return_value="mock output\n")
|
|
return sandbox
|
|
|
|
|
|
@pytest.fixture
|
|
def crashed_sandbox():
|
|
"""Mock sandbox that simulates a crash."""
|
|
sandbox = MagicMock()
|
|
sandbox.run = AsyncMock(side_effect=RuntimeError("Container crashed unexpectedly"))
|
|
return sandbox
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_podman_client():
|
|
"""Mock Podman client for unit tests."""
|
|
with patch("sandbox.session.podman.PodmanClient") as mock:
|
|
mock_container = MagicMock()
|
|
mock_container.exec_run.return_value = (0, b"mock output\n")
|
|
mock.return_value.containers.run.return_value = mock_container
|
|
yield mock, mock_container
|