mock_runloop.py

"""A tiny in-memory fake of the Runloop client, for offline development.

Implements the subset of the SDK surface used in this exercise, with the same
shapes as the real `runloop_api_client`:

    client.devboxes.create_and_await_running(name=...) -> DevboxView-like
    client.devboxes.execute_sync(id, command=...)      -> ExecutionDetail-like
    client.devboxes.shutdown(id)                       -> DevboxView-like
    client.devboxes.list(status=...)                   -> iterable of views

Failure injection (all optional, via environment variables):
    RLFAN_MOCK_FAIL_RATE   float 0..1, chance each operation raises (default 0)
    RLFAN_MOCK_SEED        int, makes failures/latency deterministic
    RLFAN_MOCK_MAX_LATENCY float seconds, upper bound of simulated latency (default 1.5)

Raised errors mirror the real SDK: `MockAPIConnectionError` and
`MockRateLimitError` both subclass `MockAPIError`, analogous to
`runloop_api_client.APIConnectionError` / `RateLimitError` / `APIError`.
"""

from __future__ import annotations

import os
import random
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Dict, List, Optional


class MockAPIError(Exception):
    """Base class, like runloop_api_client.APIError."""


class MockAPIConnectionError(MockAPIError):
    pass


class MockRateLimitError(MockAPIError):
    pass


@dataclass
class MockDevboxView:
    id: str
    name: str
    status: str  # provisioning | running | failure | shutdown


@dataclass
class MockExecutionDetail:
    devbox_id: str
    exit_status: int
    stdout: str
    stderr: str


@dataclass
class _State:
    devboxes: Dict[str, MockDevboxView] = field(default_factory=dict)
    lock: threading.Lock = field(default_factory=threading.Lock)


class _MockDevboxes:
    def __init__(self, state: _State, rng: random.Random, fail_rate: float, max_latency: float):
        self._state = state
        self._rng = rng
        self._fail_rate = fail_rate
        self._max_latency = max_latency

    # -- internals ---------------------------------------------------------
    def _simulate(self, op: str) -> None:
        time.sleep(self._rng.uniform(0.05, self._max_latency))
        roll = self._rng.random()
        if roll < self._fail_rate:
            if roll < self._fail_rate / 2:
                raise MockRateLimitError(f"mock 429 during {op}")
            raise MockAPIConnectionError(f"mock connection error during {op}")

    # -- public API (mirrors the real SDK subset) --------------------------
    def create_and_await_running(self, name: Optional[str] = None, **_: object) -> MockDevboxView:
        self._simulate("create")
        box = MockDevboxView(
            id=f"dbx_{uuid.uuid4().hex[:10]}",
            name=name or "devbox",
            status="running",
        )
        with self._state.lock:
            self._state.devboxes[box.id] = box
        return box

    def execute_sync(self, id: str, *, command: str, **_: object) -> MockExecutionDetail:
        with self._state.lock:
            box = self._state.devboxes.get(id)
        if box is None or box.status != "running":
            raise MockAPIError(f"devbox {id} is not running")
        self._simulate("execute")
        # `false`-y commands exit nonzero so candidates can test failure paths.
        exit_status = 1 if "false" in command else 0
        return MockExecutionDetail(
            devbox_id=id,
            exit_status=exit_status,
            stdout=f"[mock:{id}] ran: {command}\n" if exit_status == 0 else "",
            stderr="" if exit_status == 0 else f"[mock:{id}] command failed\n",
        )

    def shutdown(self, id: str, **_: object) -> MockDevboxView:
        self._simulate("shutdown")
        with self._state.lock:
            box = self._state.devboxes.get(id)
            if box is None:
                raise MockAPIError(f"no such devbox {id}")
            box.status = "shutdown"
            return box

    def list(self, status: Optional[str] = None, **_: object) -> List[MockDevboxView]:
        with self._state.lock:
            boxes = list(self._state.devboxes.values())
        if status:
            boxes = [b for b in boxes if b.status == status]
        return boxes


class MockRunloop:
    """Drop-in stand-in for `runloop_api_client.Runloop` (exercise subset)."""

    def __init__(self) -> None:
        seed = os.environ.get("RLFAN_MOCK_SEED")
        rng = random.Random(int(seed)) if seed is not None else random.Random()
        fail_rate = float(os.environ.get("RLFAN_MOCK_FAIL_RATE", "0"))
        max_latency = float(os.environ.get("RLFAN_MOCK_MAX_LATENCY", "1.5"))
        self._state = _State()
        self.devboxes = _MockDevboxes(self._state, rng, fail_rate, max_latency)

    # Handy for interviewers: verify nothing was leaked at the end of a run.
    def leaked_devboxes(self) -> List[MockDevboxView]:
        return self.devboxes.list(status="running")