LedgerLine/tests/conftest.py

"""Shared fixtures for the Ledgerline black-box integration suite.

Per phase-0a-test-suite-spec.md §8. Every environment value (ports, DSN, Redis
URL) comes from harness.config, itself env-driven with the Compose defaults
as fallbacks -- no hardcoded values in test bodies.
"""
import time
from dataclasses import dataclass

import httpx
import psycopg
import pytest
import pytest_asyncio
import redis as redis_lib

from harness.config import GATEWAY_URL, HTTP_SERVICES, POSTGRES_DSN, REDIS_URL, SERVICE_URLS


@pytest.fixture(scope="session")
def dsn():
    return POSTGRES_DSN


@pytest.fixture(scope="session")
def redis_client():
    client = redis_lib.from_url(REDIS_URL, decode_responses=True)
    yield client
    client.close()


@pytest.fixture(scope="session", autouse=True)
def stack_up():
    """Fail fast, with a clear message, if the Compose stack isn't up and ready.

    Every test in this suite implicitly depends on this: none of them are
    meaningful against a partially-started stack, and none of them should hang
    waiting for services that were simply never started.
    """
    deadline = time.monotonic() + 30
    unhealthy = dict(SERVICE_URLS)
    with httpx.Client(timeout=5.0) as client:
        while unhealthy and time.monotonic() < deadline:
            for name, url in list(unhealthy.items()):
                try:
                    resp = client.get(f"{url}/healthz")
                    if resp.status_code == 200:
                        del unhealthy[name]
                except httpx.HTTPError:
                    pass
            if unhealthy:
                time.sleep(1)
    if unhealthy:
        pytest.fail(
            "Stack is not up: " + ", ".join(sorted(unhealthy)) +
            " did not answer /healthz within 30s. Run `make up && make seed` first.",
            pytrace=False,
        )


@pytest.fixture(scope="session")
def seeded(dsn):
    """Fail fast if the seed manifest is missing or empty."""
    try:
        with psycopg.connect(dsn, autocommit=True) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT data FROM seed_manifest LIMIT 1")
                row = cur.fetchone()
    except Exception as exc:
        pytest.fail(f"Seed manifest not readable ({exc}). Run `make seed` first.", pytrace=False)
    if not row or not row[0] or not row[0].get("counts", {}).get("transactions"):
        pytest.fail("Seed manifest is empty or has zero transactions. Run `make seed` first.", pytrace=False)


@pytest.fixture(scope="session")
def manifest(seeded, dsn):
    """The parsed seed manifest (db/seed_manifest.json via the seed_manifest table)."""
    with psycopg.connect(dsn, autocommit=True) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT data FROM seed_manifest LIMIT 1")
            (data,) = cur.fetchone()
    return data


@pytest_asyncio.fixture
async def api():
    """httpx.AsyncClient against the gateway, 15s timeout."""
    async with httpx.AsyncClient(base_url=GATEWAY_URL, timeout=15.0) as client:
        yield client


@dataclass
class ChaosControl:
    client: httpx.AsyncClient

    async def get(self):
        resp = await self.client.get("/api/admin/chaos")
        resp.raise_for_status()
        return resp.json()

    async def set_flag(self, flag, enabled):
        return await self.client.post("/api/admin/chaos", json={"flag": flag, "enabled": enabled})

    async def set_profile(self, profile):
        resp = await self.client.post("/api/admin/chaos/profile", json={"profile": profile})
        resp.raise_for_status()
        return resp.json()


@pytest_asyncio.fixture
async def chaos(api):
    """chaos.set_profile(name), chaos.set_flag(name, bool), chaos.get().

    Teardown always resets the profile to `clean`, even if the test failed --
    design principle #6 in phase-0a-test-suite-spec.md §2: every test that
    enables a chaos flag must restore state.
    """
    control = ChaosControl(api)
    try:
        yield control
    finally:
        await control.set_profile("clean")


@pytest.fixture
def fresh_balances(dsn):
    """fresh_balances(account_id) snapshots a balance now and restores it on
    teardown, so balance assertions don't drift across repeated runs."""
    tracked = {}

    def track(account_id):
        with psycopg.connect(dsn, autocommit=True) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT balance_minor FROM accounts WHERE id = %s", (account_id,))
                row = cur.fetchone()
        if row is None:
            raise ValueError(f"unknown account {account_id}")
        tracked[account_id] = row[0]
        return row[0]

    yield track

    if tracked:
        with psycopg.connect(dsn, autocommit=True) as conn:
            with conn.cursor() as cur:
                for account_id, balance in tracked.items():
                    cur.execute(
                        "UPDATE accounts SET balance_minor = %s WHERE id = %s",
                        (balance, account_id),
                    )