svcforge/tests/integration/test_redis.py

"""Module 10 acceptance: the limiter, the idempotency store, the cache, and the budget.

Six checks, in the spec's order:

1. Rate limit 10/min — the 11th is refused.
2. A check costs exactly ONE Redis command.
3. Idempotency — the same key twice yields the same UUID.
4. Cache — the second read costs one command and no DB query.
5. Redis DOWN = the platform stays UP.
6. The budget metric exists, and `scripts/redis_budget.py` reads it correctly.

**Two tiers, and the split is the budget.** Upstash's free tier is 500K commands/month;
a test suite that hammers it is itself the bug this module is about. So the semantics are
proved against `tests/fakes.py` (free, deterministic, runs on every commit), and only the
things a fake cannot prove — that the Lua is valid Lua, that `KEYS`/`ARGV` are 1-based,
that redis-py bills one EVALSHA, that `decode_responses=True` is set — are proved against
real Upstash under `@pytest.mark.slow`. That tier spends roughly thirty commands per run,
and `pytest -m "not slow"` skips it entirely.

Check 5 needs no server at all: a closed port is a more faithful outage than a mock.
"""

from __future__ import annotations

import importlib.util
import os
import time
from collections.abc import AsyncIterator, Callable
from datetime import UTC, datetime
from pathlib import Path
from types import ModuleType
from typing import Any
from uuid import UUID, uuid4

import pytest
import pytest_asyncio
from prometheus_client import REGISTRY, generate_latest
from redis.asyncio import Redis

from svcforge_core.adapters.redis import (
    IdempotencyStore,
    InstanceCache,
    RateLimiter,
    RateLimitResult,
    make_redis,
)
from svcforge_core.repo.db import DictPool
from svcforge_core.repo.instances import InstanceRepo
from svcforge_core.settings import Settings
from tests.fakes import FakeClock, FakeIdempotencyStore, FakeInstanceCache, FakeRateLimiter
from tests.integration.helpers import build_instance

# Any DSN that parses. These tests never open a Postgres connection through Settings; the
# `pool` fixture owns the real database.
_DUMMY_PG_DSN = "postgresql://unused:unused@127.0.0.1:5432/unused"

# A port nothing listens on. `make_redis` will build a client, every command will get
# ECONNREFUSED, and that is the point of check 5.
_DEAD_REDIS_DSN = "redis://127.0.0.1:1/0"

_T0 = datetime(2026, 7, 17, 12, 0, 0, tzinfo=UTC)


def _metric(op: str) -> float:
    """The budget counter for one op. Absent labels read as 0, not as an error."""
    value = REGISTRY.get_sample_value("svcforge_redis_commands_total", {"op": op})
    return value or 0.0


def _errors(op: str) -> float:
    value = REGISTRY.get_sample_value("svcforge_redis_errors_total", {"op": op})
    return value or 0.0


# --- Fixtures ---------------------------------------------------------------------------


@pytest_asyncio.fixture
async def upstash() -> AsyncIterator[Redis]:
    """The real thing, from `~/.config/svcforge/secrets.env`. Skipped when unset.

    Built through `make_redis` rather than `Redis.from_url` directly, so that
    `decode_responses=True` is covered by these tests instead of being a comment. Forget it
    and every assertion below dies on `bytes != str`, which is the whole reason it is the
    first bug everyone hits.
    """
    dsn = os.getenv("SVCFORGE_REDIS_DSN")
    if not dsn:
        pytest.skip("SVCFORGE_REDIS_DSN unset; real-Upstash checks skipped")
    client = make_redis(Settings(pg_dsn=_DUMMY_PG_DSN, redis_dsn=dsn))  # type: ignore[arg-type]  # pydantic coerces str -> *Dsn
    assert client is not None
    try:
        yield client
    finally:
        await client.aclose()


@pytest_asyncio.fixture
async def dead_redis() -> AsyncIterator[Redis]:
    """A client pointed at a closed port. No server was harmed, no commands were billed."""
    client = make_redis(Settings(pg_dsn=_DUMMY_PG_DSN, redis_dsn=_DEAD_REDIS_DSN))  # type: ignore[arg-type]  # pydantic coerces str -> *Dsn
    assert client is not None
    try:
        yield client
    finally:
        await client.aclose()


class _CommandCounter:
    """Counts round trips by wrapping `execute_command` on one client instance.

    This counts what Upstash bills, which is the only definition that matters here. The
    metric counter is our own bookkeeping and could be wrong in the same direction as the
    code it measures; this one cannot.
    """

    def __init__(self, r: Redis) -> None:
        self.count = 0
        self._inner: Callable[..., Any] = r.execute_command
        r.execute_command = self._counting  # type: ignore[method-assign]

    async def _counting(self, *args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
        self.count += 1
        return await self._inner(*args, **kwargs)


def _budget_script() -> ModuleType:
    """Import `scripts/redis_budget.py` by path — `scripts/` is not a package, deliberately."""
    path = Path(__file__).resolve().parents[2] / "scripts" / "redis_budget.py"
    spec = importlib.util.spec_from_file_location("redis_budget", path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# --- 1. Rate limit: 10/min, the 11th is refused ------------------------------------------


async def test_eleventh_request_in_the_window_is_refused() -> None:
    """`200 x10` then `429`. The fake, so this runs on every commit for free."""
    limiter = FakeRateLimiter(limit=10, window_s=60, clock=FakeClock(start=_T0))

    results = [await limiter.check("acme") for _ in range(11)]

    assert [r.allowed for r in results] == [True] * 10 + [False]
    assert results[9].remaining == 0
    assert results[10].remaining == 0
    # What the handler puts in `Retry-After` on the 429. Never 0: a client told to retry
    # immediately retries into the same closed window.
    assert results[10].retry_after_s >= 1


async def test_the_window_rolls_over_and_the_caller_is_allowed_again() -> None:
    """A fixed window resets on a boundary, not on a sleep. Hence the injected clock."""
    clock = FakeClock(start=_T0)
    limiter = FakeRateLimiter(limit=2, window_s=60, clock=clock)

    assert (await limiter.check("acme")).allowed
    assert (await limiter.check("acme")).allowed
    assert not (await limiter.check("acme")).allowed

    clock.advance(datetime(2026, 7, 17, 12, 1, 0, tzinfo=UTC) - _T0)
    assert (await limiter.check("acme")).allowed


async def test_teams_do_not_share_a_window() -> None:
    """`rl:{team}:{window}` — a noisy tenant must not refuse a quiet one."""
    limiter = FakeRateLimiter(limit=1, window_s=60, clock=FakeClock(start=_T0))

    assert (await limiter.check("acme")).allowed
    assert not (await limiter.check("acme")).allowed
    assert (await limiter.check("globex")).allowed


@pytest.mark.slow
async def test_real_lua_refuses_the_eleventh(upstash: Redis) -> None:
    """The same assertion against real Upstash. ~11 commands.

    This is what a fake cannot prove: that the script is valid Lua, that `KEYS[1]` and
    `ARGV[1]` are 1-based (0-based indexing would read nil and compare false forever), and
    that `INCR`-then-conditional-`EXPIRE` actually holds a window open.
    """
    team = f"test-{uuid4().hex[:8]}"
    limiter = RateLimiter(upstash, limit=10, window_s=60)

    results = [await limiter.check(team) for _ in range(11)]

    assert [r.allowed for r in results] == [True] * 10 + [False]
    assert not any(r.degraded for r in results)
    assert results[0].remaining == 9

    # The EXPIRE fired on the first INCR only, so the key is not immortal. `Every key gets
    # a TTL` is a rule with no enforcement other than checking.
    window = int(time.time()) // 60
    keys = [f"rl:{team}:{window}", f"rl:{team}:{window - 1}"]
    ttls = [await upstash.ttl(k) for k in keys]
    assert any(0 < ttl <= 60 for ttl in ttls)
    await upstash.delete(*keys)


# --- 2. It costs ONE command per check ---------------------------------------------------


@pytest.mark.slow
async def test_a_check_costs_exactly_one_redis_command(upstash: Redis) -> None:
    """One EVALSHA. Not GET+INCR+EXPIRE, which is three billed commands and a race.

    At 500K/month the difference is material: three commands per request caps the
    platform at 166K requests/month instead of 500K, for a limiter that is also wrong.

    The first call is excluded from the count on purpose. redis-py sends EVALSHA, Upstash
    answers NOSCRIPT because it has never seen the hash, and redis-py replays it as EVAL —
    two commands, once per Redis restart, and irrelevant to the steady state this measures.
    """
    team = f"test-{uuid4().hex[:8]}"
    limiter = RateLimiter(upstash, limit=100, window_s=60)

    await limiter.check(team)  # warm the script cache

    counter = _CommandCounter(upstash)
    before = _metric("ratelimit")
    result = await limiter.check(team)

    assert counter.count == 1, "a rate limit check must be one round trip and one billed command"
    assert _metric("ratelimit") - before == 1, "the budget counter must agree with the wire"
    assert result.allowed

    window = int(time.time()) // 60
    await upstash.delete(f"rl:{team}:{window}", f"rl:{team}:{window - 1}")


# --- 3. Idempotency: the same key twice is one instance ----------------------------------


async def test_same_idempotency_key_returns_the_first_instance_id() -> None:
    """The first caller wins; the second is told who won and must not create anything."""
    store = FakeIdempotencyStore()
    key = str(uuid4())
    first, second = uuid4(), uuid4()

    assert await store.claim(key, first) is None, "None means 'you won, go create it'"
    assert await store.claim(key, second) == first, "the loser gets the winner's id, not its own"


async def test_different_idempotency_keys_do_not_collide() -> None:
    store = FakeIdempotencyStore()
    a, b = uuid4(), uuid4()

    assert await store.claim(str(uuid4()), a) is None
    assert await store.claim(str(uuid4()), b) is None


@pytest.mark.slow
async def test_real_set_nx_ex_claims_once(upstash: Redis) -> None:
    """SET NX EX against Upstash. ~3 commands.

    Also asserts the TTL, because a claim marker without one is a permanent record of a
    request from last March, and 256 MB of those ends by evicting the keys you cared about.
    """
    key = f"test-{uuid4()}"
    first, second = uuid4(), uuid4()
    store = IdempotencyStore(upstash, ttl_s=60)

    assert await store.claim(key, first) is None
    assert await store.claim(key, second) == first

    assert 0 < await upstash.ttl(f"idem:{key}") <= 60
    await upstash.delete(f"idem:{key}")


# --- 4. Cache: the second read costs one command and no DB query -------------------------


class _CountingRepo:
    """Wraps InstanceRepo and counts reads. The DB query count is the assertion."""

    def __init__(self, repo: InstanceRepo) -> None:
        self._repo = repo
        self.gets = 0

    async def get(self, id: UUID, team: str) -> Any:  # noqa: ANN401
        self.gets += 1
        return await self._repo.get(id, team)


async def _read_through(
    instance_id: UUID,
    team: str,
    cache: FakeInstanceCache | InstanceCache,
    repo: _CountingRepo,
) -> Any:  # noqa: ANN401
    """Cache-aside, as `GET /v1/instances/{id}` performs it. Hit = 1 command, miss = 2."""
    cached = await cache.get(instance_id)
    if cached is not None:
        return cached
    inst = await repo.get(instance_id, team)
    if inst is not None:
        await cache.put(inst)
    return inst


async def test_second_read_is_served_from_cache_without_touching_postgres(
    pool: DictPool,
) -> None:
    """Miss, then hit. The DB is read exactly once for two reads."""
    repo = InstanceRepo(pool)
    # `create` returns the row as Postgres stored it. Compare against that, never against
    # the model that went in: `created_at`/`updated_at` are DB defaults, and a timestamptz
    # comes back tagged `Etc/UTC` rather than `timezone.utc`. Same instant, different repr.
    async with pool.connection() as conn:
        inst = await repo.create(conn, build_instance(team="platform"))

    counting = _CountingRepo(repo)
    cache = FakeInstanceCache()

    first = await _read_through(inst.id, inst.team, cache, counting)
    second = await _read_through(inst.id, inst.team, cache, counting)

    assert first == second == inst
    assert counting.gets == 1, "the second read must not reach Postgres"
    assert (cache.misses, cache.hits) == (1, 1)


async def test_invalidate_sends_the_next_read_back_to_postgres() -> None:
    """The worker calls this inside the code path that writes the state, not after it."""
    cache = FakeInstanceCache()
    inst = build_instance()

    await cache.put(inst)
    assert await cache.get(inst.id) == inst

    await cache.invalidate(inst.id)
    assert await cache.get(inst.id) is None


@pytest.mark.slow
async def test_real_cache_hit_costs_one_command_and_no_db_query(pool: DictPool, upstash: Redis) -> None:
    """Real Redis, real Postgres. ~4 commands.

    The round-trip count is the acceptance criterion: a hit that costs two commands is a
    cache that has doubled the bill it was added to reduce.
    """
    repo = InstanceRepo(pool)
    async with pool.connection() as conn:
        inst = await repo.create(conn, build_instance(team="platform"))

    counting = _CountingRepo(repo)
    cache = InstanceCache(upstash, ttl_s=30)

    miss = await _read_through(inst.id, inst.team, cache, counting)
    assert counting.gets == 1

    counter = _CommandCounter(upstash)
    before = _metric("cache_get")
    hit = await _read_through(inst.id, inst.team, cache, counting)

    assert counter.count == 1, "a cache hit is one GET"
    assert _metric("cache_get") - before == 1
    assert counting.gets == 1, "the second read must not reach Postgres"
    # Round-tripped through JSON and back: `decode_responses=True` and the datetime/UUID
    # serialisation both have to be right for this to compare equal.
    assert hit == miss == inst

    assert 0 < await upstash.ttl(f"inst:{inst.id}") <= 30
    await cache.invalidate(inst.id)
    assert await cache.get(inst.id) is None


# --- 5. Redis down = the platform stays up -----------------------------------------------


async def test_rate_limiter_fails_open_when_redis_is_down(dead_redis: Redis) -> None:
    """The load-bearing one. A limiter that fails closed turns a cache outage into an outage.

    `degraded=True` and the error counter are what stop this from being invisible: fail
    open silently and you cannot tell a working limiter from one that has been allowing
    everything for a month.
    """
    limiter = RateLimiter(dead_redis, limit=1, window_s=60)
    before = _errors("ratelimit")

    results = [await limiter.check("acme") for _ in range(3)]

    assert all(r.allowed for r in results), "Redis being down must never refuse legitimate traffic"
    assert all(r.degraded for r in results)
    assert _errors("ratelimit") - before == 3


async def test_idempotency_falls_through_to_the_db_when_redis_is_down(dead_redis: Redis) -> None:
    """None means "create it". Safe only because `instances.release_name` is UNIQUE."""
    store = IdempotencyStore(dead_redis)
    key = str(uuid4())

    assert await store.claim(key, uuid4()) is None
    assert await store.claim(key, uuid4()) is None


async def test_cache_misses_instead_of_raising_when_redis_is_down(dead_redis: Redis) -> None:
    """A miss falls through to Postgres. `put` and `invalidate` swallow it too."""
    cache = InstanceCache(dead_redis)
    inst = build_instance()

    assert await cache.get(inst.id) is None
    await cache.put(inst)  # must not raise
    await cache.invalidate(inst.id)  # must not raise


async def test_platform_serves_reads_from_postgres_with_redis_down(pool: DictPool, dead_redis: Redis) -> None:
    """The acceptance check: `docker compose stop redis` then GET -> 200.

    Every Redis path degrades and the read still returns the row. Nothing here consults
    Redis for readiness, which is the other half of the rule — `/readyz` is Postgres-only,
    so a Redis outage cannot make a single pod unready.
    """
    repo = InstanceRepo(pool)
    async with pool.connection() as conn:
        inst = await repo.create(conn, build_instance(team="platform"))

    limiter = RateLimiter(dead_redis, limit=10, window_s=60)
    cache = InstanceCache(dead_redis)
    store = IdempotencyStore(dead_redis)

    assert (await limiter.check(inst.team)).allowed
    assert await store.claim(str(uuid4()), inst.id) is None
    assert await cache.get(inst.id) is None
    assert await repo.get(inst.id, inst.team) == inst  # the 200


# --- 6. The budget metric exists and is sane ---------------------------------------------


async def test_the_budget_metric_is_exposed_and_labelled_by_op() -> None:
    """`curl -s localhost:8000/metrics | grep svcforge_redis_commands_total`."""
    limiter = FakeRateLimiter(limit=10, window_s=60, clock=FakeClock(start=_T0))
    await limiter.check("acme")  # the fake does not touch the real counter
    RateLimitResult(allowed=True, limit=10, remaining=9, reset_at=_T0)

    text = generate_latest(REGISTRY).decode()

    assert "svcforge_redis_commands_total" in text
    assert "svcforge_redis_errors_total" in text
    # Per-op labels, because "you are over budget" is useless without "on cache_get".
    assert _metric("ratelimit") >= 0


def test_budget_script_projects_a_five_second_poller_over_budget() -> None:
    """The spec's headline number, reproduced: one worker polling every 5s = 518,400/month.

    Exactly the free tier, spent doing nothing. This is the case the script exists to
    catch, so it is the case that is asserted rather than left to a comment.
    """
    budget = _budget_script()
    now = time.time()
    # One hour at one command every five seconds.
    per_op, started_at = budget.collect(
        f'svcforge_redis_commands_total{{op="poll"}} 720.0\nprocess_start_time_seconds {now - 3600}\n'
    )

    assert per_op == {"poll": 720.0}
    assert budget.report(per_op, started_at, budget._FREE_TIER_BUDGET, now) == 1


def test_budget_script_passes_a_request_path_workload() -> None:
    """Request-path volume is bounded by humans, and humans are slow. That is the whole rule."""
    budget = _budget_script()
    now = time.time()
    per_op, started_at = budget.collect(
        f'svcforge_redis_commands_total{{op="ratelimit"}} 100.0\n'
        f'svcforge_redis_commands_total{{op="cache_get"}} 40.0\n'
        f"process_start_time_seconds {now - 3600}\n"
    )

    assert sum(per_op.values()) == 140.0
    assert budget.report(per_op, started_at, budget._FREE_TIER_BUDGET, now) == 0


def test_budget_script_refuses_to_guess_without_the_metric() -> None:
    """No counter means the process never imported the adapter — say so, do not print a 0."""
    budget = _budget_script()

    with pytest.raises(budget.BudgetError, match="not exposed"):
        budget.collect("process_start_time_seconds 1.0\n")

    with pytest.raises(budget.BudgetError, match="process_start_time_seconds"):
        budget.collect('svcforge_redis_commands_total{op="ratelimit"} 5.0\n')


def test_budget_script_refuses_a_non_http_url() -> None:
    """`--url file:///etc/passwd` is not a metrics endpoint."""
    budget = _budget_script()

    with pytest.raises(budget.BudgetError, match="non-http"):
        budget.scrape("file:///etc/passwd")