"""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, checked_at=self.clock.now(), 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, checked_at=self.clock.now(), ) 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)