svcforge/tests/fakes.py

"""Test doubles. The second implementation that makes each Protocol worth having.

These are fakes, not mocks: they have working behaviour (a dict of releases, a monotonic
counter) and are asserted against by their *state*, not by "was this method called with
these arguments". A mock proves your mock does what you told it to.

They live here, outside `svcforge_core`, so that no production import path can reach them.
"""

from __future__ import annotations

import asyncio
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID

from svcforge_core.adapters.clock import Clock
from svcforge_core.adapters.helm import HelmError, ReleaseInfo
from svcforge_core.adapters.redis import RateLimitResult
from svcforge_core.domain.models import CatalogEntry, Instance


class FakeProvisioner:
    """Dict of release -> ReleaseInfo. Same signatures. Optional fail_on / delay for tests.

    This is what makes the worker's tests run in milliseconds with no cluster.
    """

    def __init__(self, *, delay: float = 0.0, fail_on: set[str] | None = None) -> None:
        """`delay`: seconds `install` sleeps, to hold a task in flight (SIGTERM drain tests).

        `fail_on`: substring match against the release name, not equality — release names
        carry a random suffix (`platform-elasticsearch-1a2b3c4d`), so a test that wants "every
        elasticsearch install fails" cannot name the release up front.
        """
        self.delay = delay
        self.fail_on = fail_on or set()

        self.releases: dict[str, ReleaseInfo] = {}
        self.installed: list[str] = []
        self.uninstalled: list[str] = []

        # Concurrency accounting, so a test can assert the worker's semaphore actually caps.
        self.in_flight = 0
        self.max_concurrent = 0

    def _should_fail(self, release: str) -> bool:
        return any(needle in release for needle in self.fail_on)

    async def install(self, release: str, ns: str, entry: CatalogEntry, values: dict[str, Any]) -> None:
        self.in_flight += 1
        self.max_concurrent = max(self.max_concurrent, self.in_flight)
        try:
            if self.delay:
                await asyncio.sleep(self.delay)
            if self._should_fail(release):
                # Same type the real adapter raises, so handlers cannot pass here and fail in prod.
                raise HelmError(f"fake: install of {release} failed on purpose")
            self.releases[release] = ReleaseInfo(
                name=release,
                namespace=ns,
                chart=f"{entry.chart}-{entry.chart_version}",
                status="deployed",
                revision=self.releases[release].revision + 1 if release in self.releases else 1,
                app_version=entry.chart_version,
            )
            # Appended only on success: a failed install must not look installed.
            self.installed.append(release)
        finally:
            self.in_flight -= 1

    async def uninstall(self, release: str, ns: str) -> None:
        """Absent release is not an error: the desired state is already true."""
        self.releases.pop(release, None)
        self.uninstalled.append(release)

    async def list_releases(self) -> list[ReleaseInfo]:
        return list(self.releases.values())


class FakeClock:
    """Time under test control. No sleeping, no monkeypatching the stdlib."""

    def __init__(self, start: datetime) -> None:
        if start.tzinfo is None:
            raise ValueError("FakeClock needs an aware datetime; a naive start defeats the point")
        self._now = start

    def now(self) -> datetime:
        return self._now

    def advance(self, delta: timedelta) -> None:
        """Move forward. Only forward — a clock that goes backwards is a different bug entirely."""
        if delta < timedelta(0):
            raise ValueError("FakeClock cannot go backwards")
        self._now += delta


class FakeNotifier:
    """Records what it was asked to send. Never leaves the process."""

    def __init__(self) -> None:
        self.sent: list[tuple[str, str, dict[str, str]]] = []

    async def send(self, event: str, message: str, fields: dict[str, str] | None = None) -> None:
        self.sent.append((event, message, fields or {}))

    def events(self) -> list[str]:
        return [event for event, _, _ in self.sent]


# --- Redis (Module 10) ------------------------------------------------------------------
#
# Three fakes, and each one is the second implementation that earns its Protocol. They also
# earn their keep against the budget: Upstash's free tier is 500K commands/month, so a test
# that wants a thousand rate-limit checks runs them here and spends nothing. Real Upstash
# is reserved for the handful of `@pytest.mark.slow` tests that prove the wire protocol,
# the Lua, and the command count.
#
# `down=True` is the interesting knob. Every real class degrades internally rather than
# raising, so these degrade the same way — a fake that raises when the real one returns a
# safe default would let a caller ship a `try/except` that production never exercises.


class FakeRateLimiter:
    """In-memory fixed window. Same semantics as the Lua, none of the network.

    Counts `commands` so a test can assert the one-command-per-check budget without a
    server, and takes a Clock so a window rollover is an `advance()` rather than a sleep.
    """

    def __init__(self, limit: int, window_s: int, clock: Clock, *, down: bool = False) -> None:
        self.limit = limit
        self.window_s = window_s
        self.clock = clock
        self.down = down
        self.commands = 0
        self.counts: dict[str, int] = {}

    async def check(self, team: str) -> RateLimitResult:
        window = int(self.clock.now().timestamp()) // self.window_s
        reset_at = datetime.fromtimestamp((window + 1) * self.window_s, tz=UTC)
        self.commands += 1
        if self.down:
            # Fails OPEN, exactly like the real one. A limiter that refused here would make
            # "Redis is down" indistinguishable from "you are over quota".
            return RateLimitResult(
                allowed=True, limit=self.limit, remaining=self.limit, reset_at=reset_at, degraded=True
            )
        key = f"rl:{team}:{window}"
        n = self.counts.get(key, 0) + 1
        self.counts[key] = n
        return RateLimitResult(
            allowed=n <= self.limit,
            limit=self.limit,
            remaining=max(0, self.limit - n),
            reset_at=reset_at,
        )


class FakeIdempotencyStore:
    """A dict with SET-NX semantics. No TTL: no test outlives one."""

    def __init__(self, *, down: bool = False) -> None:
        self.down = down
        self.claims: dict[str, UUID] = {}

    async def claim(self, key: str, instance_id: UUID) -> UUID | None:
        if self.down:
            # Falls through to the DB, where `instances.release_name` is UNIQUE. The real
            # guarantee was never here.
            return None
        existing = self.claims.get(key)
        if existing is not None:
            return existing
        self.claims[key] = instance_id
        return None


class FakeInstanceCache:
    """A dict, plus hit/miss accounting so a test can assert the second read never hits the DB."""

    def __init__(self, *, down: bool = False) -> None:
        self.down = down
        self.entries: dict[UUID, Instance] = {}
        self.hits = 0
        self.misses = 0
        self.invalidations: list[UUID] = []

    async def get(self, instance_id: UUID) -> Instance | None:
        if self.down:
            self.misses += 1
            return None
        inst = self.entries.get(instance_id)
        if inst is None:
            self.misses += 1
            return None
        self.hits += 1
        return inst

    async def put(self, inst: Instance) -> None:
        if self.down:
            return
        self.entries[inst.id] = inst

    async def invalidate(self, instance_id: UUID) -> None:
        self.invalidations.append(instance_id)
        if self.down:
            return
        self.entries.pop(instance_id, None)