conftest.py

"""Shared fixtures and utilities for SDK tests."""
# pyright: reportUnknownVariableType=false

from __future__ import annotations

import asyncio
import threading
from typing import Any, Dict
from dataclasses import field, dataclass
from unittest.mock import Mock, AsyncMock

import httpx
import pytest

from runloop_api_client import Runloop, AsyncRunloop

# Test ID constants
TEST_IDS = {
    "devbox": "dbx_123",
    "execution": "exn_123",
    "snapshot": "snp_123",
    "blueprint": "bpt_123",
    "object": "obj_123",
    "scorer": "sco_123",
    "agent": "agt_123",
    "axon": "axn_123",
    "scenario": "scn_123",
    "scenario_run": "scr_123",
    "benchmark": "bmd_123",
    "benchmark_run": "bmr_123",
    "network_policy": "np_123",
    "gateway_config": "gwc_123",
}

# Test URL constants
TEST_URLS = {
    "upload": "https://upload.example.com/obj_123",
    "download": "https://download.example.com/obj_123",
}

# Timing constants for thread/task synchronization tests
THREAD_STARTUP_DELAY = 0.1  # Time to allow threads/tasks to start
TASK_COMPLETION_SHORT = 0.02  # Brief async operation
TASK_COMPLETION_LONG = 1.0  # Long-running operation for cancellation tests
NUM_CONCURRENT_THREADS = 5  # Number of threads for concurrency tests


# Mock data structures using dataclasses for type safety
@dataclass
class MockDevboxView:
    """Mock DevboxView for testing."""

    id: str = TEST_IDS["devbox"]
    status: str = "running"
    name: str = "test-devbox"


@dataclass
class MockExecutionView:
    """Mock DevboxAsyncExecutionDetailView for testing."""

    execution_id: str = TEST_IDS["execution"]
    devbox_id: str = TEST_IDS["devbox"]
    status: str = "completed"
    exit_status: int = 0
    stdout: str = "output"
    stderr: str = ""
    stdout_truncated: bool = False
    stderr_truncated: bool = False


@dataclass
class MockSnapshotView:
    """Mock DevboxSnapshotView for testing."""

    id: str = TEST_IDS["snapshot"]
    status: str = "completed"
    name: str = "test-snapshot"


@dataclass
class MockBlueprintView:
    """Mock BlueprintView for testing."""

    id: str = TEST_IDS["blueprint"]
    status: str = "built"
    name: str = "test-blueprint"


@dataclass
class MockObjectView:
    """Mock ObjectView for testing."""

    id: str = TEST_IDS["object"]
    upload_url: str = "https://upload.example.com/obj_123"
    name: str = "test-object"


@dataclass
class MockScorerView:
    """Mock ScorerView for testing."""

    id: str = TEST_IDS["scorer"]
    bash_script: str = "echo 'score=1.0'"
    type: str = "test_scorer"


@dataclass
class MockAgentView:
    """Mock AgentView for testing."""

    id: str = TEST_IDS["agent"]
    name: str = "test-agent"
    create_time_ms: int = 1234567890000
    is_public: bool = False
    source: Any = None


@dataclass
class MockAxonView:
    """Mock AxonView for testing."""

    id: str = TEST_IDS["axon"]
    created_at_ms: int = 1234567890000
    name: str = "test-axon"


@dataclass
class MockPublishResultView:
    """Mock PublishResultView for testing."""

    sequence: int = 1
    timestamp_ms: int = 1234567890000


@dataclass
class MockSqlColumnMetaView:
    """Mock SqlColumnMetaView for testing."""

    name: str = "id"
    type: str = "INTEGER"


@dataclass
class MockSqlResultMetaView:
    """Mock SqlResultMetaView for testing."""

    changes: int = 0
    duration_ms: float = 1.5
    rows_read_limit_reached: bool = False


@dataclass
class MockSqlQueryResultView:
    """Mock SqlQueryResultView for testing."""

    columns: list[Any] = field(default_factory=lambda: [MockSqlColumnMetaView()])
    meta: Any = field(default_factory=MockSqlResultMetaView)
    rows: list[Any] = field(default_factory=lambda: [[1, "hello"]])


@dataclass
class MockSqlStepResultView:
    """Mock SqlStepResultView for testing."""

    success: Any = field(default_factory=lambda: MockSqlQueryResultView())
    error: Any = None


@dataclass
class MockSqlBatchResultView:
    """Mock SqlBatchResultView for testing."""

    results: list[Any] = field(default_factory=lambda: [MockSqlStepResultView()])


@dataclass
class MockScenarioView:
    """Mock ScenarioView for testing."""

    id: str = TEST_IDS["scenario"]
    name: str = "test-scenario"
    metadata: Dict[str, str] = field(default_factory=dict)


@dataclass
class MockScenarioRunView:
    """Mock ScenarioRunView for testing."""

    id: str = TEST_IDS["scenario_run"]
    devbox_id: str = TEST_IDS["devbox"]
    scenario_id: str = TEST_IDS["scenario"]
    state: str = "running"
    metadata: Dict[str, str] = field(default_factory=dict)
    scoring_contract_result: object = None


@dataclass
class MockBenchmarkView:
    """Mock BenchmarkView for testing."""

    id: str = TEST_IDS["benchmark"]
    name: str = "test-benchmark"
    metadata: Dict[str, str] = field(default_factory=dict)
    scenario_ids: list[str] = field(default_factory=list)


@dataclass
class MockBenchmarkRunView:
    """Mock BenchmarkRunView for testing."""

    id: str = TEST_IDS["benchmark_run"]
    benchmark_id: str = TEST_IDS["benchmark"]
    state: str = "running"
    metadata: Dict[str, str] = field(default_factory=dict)
    start_time_ms: int = 1234567890000
    duration_ms: int | None = None
    score: float | None = None


@dataclass
class MockEgress:
    """Mock Egress for testing."""

    allow_all: bool = False
    allow_devbox_to_devbox: bool = False
    allowed_hostnames: list[str] = field(default_factory=lambda: ["github.com", "*.npmjs.org"])


@dataclass
class MockNetworkPolicyView:
    """Mock NetworkPolicyView for testing."""

    id: str = TEST_IDS["network_policy"]
    name: str = "test-network-policy"
    description: str | None = "Test network policy description"
    create_time_ms: int = 1234567890000
    update_time_ms: int = 1234567890000
    egress: MockEgress = field(default_factory=MockEgress)


@dataclass
class MockAuthMechanism:
    """Mock AuthMechanism for testing."""

    type: str = "bearer"
    key: str | None = None


@dataclass
class MockGatewayConfigView:
    """Mock GatewayConfigView for testing."""

    id: str = TEST_IDS["gateway_config"]
    name: str = "test-gateway-config"
    endpoint: str = "https://api.example.com"
    description: str | None = "Test gateway config description"
    create_time_ms: int = 1234567890000
    auth_mechanism: MockAuthMechanism = field(default_factory=MockAuthMechanism)


class AsyncIterableMock:
    """A simple async iterable mock for testing paginated responses."""

    def __init__(self, items: list[Any]) -> None:
        self._items = items

    async def __aiter__(self):
        for item in self._items:
            yield item


def create_mock_httpx_client(methods: dict[str, Any] | None = None) -> AsyncMock:
    """
    Create a mock httpx.AsyncClient with proper context manager setup.

    Args:
        methods: Optional dict of method names to AsyncMock return values.
                 Common keys: 'get', 'put'

    Returns:
        Configured AsyncMock for httpx.AsyncClient

    Note: We don't use spec here because we need to manually set context manager
          methods which are not allowed with spec.
    """
    mock_client = AsyncMock()
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)

    if methods:
        for method_name, return_value in methods.items():
            setattr(mock_client, method_name, AsyncMock(return_value=return_value))

    return mock_client


def create_mock_httpx_response(**attrs: Any) -> Mock:
    """
    Create a mock httpx.Response with specified attributes.

    Args:
        **attrs: Attributes to set on the mock response.
                 Common: content, text, encoding

    Returns:
        Mock configured with httpx.Response spec and attributes
    """
    mock_response = Mock(spec=httpx.Response)
    for key, value in attrs.items():
        setattr(mock_response, key, value)
    return mock_response


@pytest.fixture
def mock_client() -> Mock:
    """Create a mock Runloop client."""
    return Mock(spec=Runloop)


@pytest.fixture
def mock_async_client() -> AsyncMock:
    """Create a mock AsyncRunloop client."""
    return AsyncMock(spec=AsyncRunloop)


@pytest.fixture
def devbox_view() -> MockDevboxView:
    """Create a mock DevboxView."""
    return MockDevboxView()


@pytest.fixture
def execution_view() -> MockExecutionView:
    """Create a mock DevboxAsyncExecutionDetailView."""
    return MockExecutionView()


@pytest.fixture
def snapshot_view() -> MockSnapshotView:
    """Create a mock DevboxSnapshotView."""
    return MockSnapshotView()


@pytest.fixture
def blueprint_view() -> MockBlueprintView:
    """Create a mock BlueprintView."""
    return MockBlueprintView()


@pytest.fixture
def object_view() -> MockObjectView:
    """Create a mock ObjectView."""
    return MockObjectView()


@pytest.fixture
def scorer_view() -> MockScorerView:
    """Create a mock ScorerView."""
    return MockScorerView()


@pytest.fixture
def agent_view() -> MockAgentView:
    """Create a mock AgentView."""
    return MockAgentView()


@pytest.fixture
def axon_view() -> MockAxonView:
    """Create a mock AxonView."""
    return MockAxonView()


@pytest.fixture
def scenario_view() -> MockScenarioView:
    """Create a mock ScenarioView."""
    return MockScenarioView()


@pytest.fixture
def scenario_run_view() -> MockScenarioRunView:
    """Create a mock ScenarioRunView."""
    return MockScenarioRunView()


@pytest.fixture
def benchmark_view() -> MockBenchmarkView:
    """Create a mock BenchmarkView."""
    return MockBenchmarkView()


@pytest.fixture
def benchmark_run_view() -> MockBenchmarkRunView:
    """Create a mock BenchmarkRunView."""
    return MockBenchmarkRunView()


@pytest.fixture
def network_policy_view() -> MockNetworkPolicyView:
    """Create a mock NetworkPolicyView."""
    return MockNetworkPolicyView()


@pytest.fixture
def gateway_config_view() -> MockGatewayConfigView:
    """Create a mock GatewayConfigView."""
    return MockGatewayConfigView()


@pytest.fixture
def mock_httpx_response() -> Mock:
    """Create a mock httpx.Response."""
    response = Mock(spec=httpx.Response)
    response.status_code = 200
    response.content = b"test content"
    response.text = "test content"
    response.encoding = "utf-8"
    response.raise_for_status = Mock()
    return response


@pytest.fixture
def mock_stream() -> Mock:
    """Create a mock Stream for testing.

    Note: We don't use spec here because we need to manually set context manager
          and iterator methods which are not allowed with spec.
    """
    stream = Mock()
    stream.__iter__ = Mock(return_value=iter([]))
    stream.__enter__ = Mock(return_value=stream)
    stream.__exit__ = Mock(return_value=None)
    stream.close = Mock()
    return stream


@pytest.fixture
def mock_async_stream() -> AsyncMock:
    """Create a mock AsyncStream for testing.

    Note: We don't use spec here because we need to manually set context manager
          and async iterator methods which are not allowed with spec.
    """

    async def async_iter():
        return
        yield  # Make this a generator

    stream = AsyncMock()
    stream.__aiter__ = Mock(return_value=async_iter())
    stream.__aenter__ = AsyncMock(return_value=stream)
    stream.__aexit__ = AsyncMock(return_value=None)
    stream.close = AsyncMock()
    return stream


@pytest.fixture
async def async_task_cleanup():
    """
    Fixture to ensure async tasks are properly cleaned up after tests.

    Usage:
        async def test_something(async_task_cleanup):
            task = asyncio.create_task(some_coroutine())
            async_task_cleanup.append(task)
            # Task will be automatically cancelled and awaited on teardown

    Yields:
        List to append tasks to for automatic cleanup
    """
    tasks: list[asyncio.Task[Any]] = []
    yield tasks
    # Cleanup: cancel all tasks and wait for them to finish
    for task in tasks:
        if not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass


@pytest.fixture
def thread_cleanup():
    """
    Fixture to ensure threads are properly cleaned up after tests.

    Usage:
        def test_something(thread_cleanup):
            threads, stop_events = thread_cleanup
            stop_event = threading.Event()
            thread = threading.Thread(target=worker, args=(stop_event,))
            thread.start()
            threads.append(thread)
            stop_events.append(stop_event)
            # Thread will be automatically stopped and joined on teardown

    Yields:
        Tuple of (threads list, stop_events list) for automatic cleanup
    """
    threads: list[threading.Thread] = []
    stop_events: list[threading.Event] = []
    yield threads, stop_events
    # Cleanup: signal all threads to stop and wait for them
    for event in stop_events:
        event.set()
    for thread in threads:
        thread.join(timeout=2.0)