Files
svcforge/tests/fakes.py
T
Nguyen Minh Phuc 66eb6cb0ee refactor: converge the patterns multiple authors left divergent
The codebase was written by several agents and had the same concept done more
than one way. This makes it read as one voice, with no behaviour change.

Dedup, each to a single canonical form:
  - INSTANCE_COLUMNS: the 13-column instances SELECT list existed as _COLUMNS in
    instances.py and reconcile.py (byte-identical) and inlined a third time in
    the worker. One exported constant now.
  - Settings.runtime_dsn: the three entrypoints each chose between str(pg_dsn)
    and pg_dsn.unicode_string(). One property.
  - yaml_tempfile: helm._ValuesFile and k8s._ManifestFile were the same
    write-yaml-to-a-temp-dir context manager. One helper in adapters/tempyaml.py.
  - services/_runtime.py: sleep_or_stop and install_stop_signals were copied
    between the worker and reconciler loops. One module, so shutdown behaviour
    cannot drift between them.
  - k8s.ensure_namespace used MANAGED_BY_LABEL/VALUE from helm.py instead of a
    hardcoded literal, so the managed-by label has one definition.
  - SvcforgeError is now the root of every svcforge exception (CatalogError,
    IllegalTransition, BadWindow, HandlerError), keeping each stdlib base in the
    MRO, so `except SvcforgeError` means what errors.py says it does.
  - ERROR_MAX_CHARS replaces the repeated `[-2000:]` truncation feeding the same
    error columns.
  - the reconciler reads settings.metrics_port like the worker, dropping its
    duplicate DEFAULT_METRICS_PORT and redundant --metrics-port option; the
    SVCFORGE_METRICS_PORT env override still applies through pydantic.

Two smaller correctness/consistency fixes:
  - RateLimitResult.retry_after_s computed its delta against datetime.now(UTC)
    while the limiter runs on an injectable clock, so it was meaningless under a
    FakeClock and drifted by request latency in production. It now carries a
    checked_at from the same clock as reset_at.
  - handle_provision's notifier.send is wrapped like the reconciler's: a flaky
    webhook after the READY CAS would fail the task, and the retry would hit the
    READY early-return and drop the notification, turning a good provision into a
    failed one.
2026-07-21 01:46:54 +00:00

217 lines
8.1 KiB
Python

"""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)