scripts/test_feed_client.py

#!/usr/bin/env python3
"""
scripts/test_feed_client.py — the Redis pool feed, against a real server.

    venv/bin/python scripts/test_feed_client.py

Starts a throwaway redis-server on a spare port when one is available and
drives modules/feed_client.py through the real contract. When no
redis-server binary exists (a bare CI image), the server-backed section is
skipped and the failure-mode section still runs in full — those are the
checks that matter most and none of them need a live Redis.

THE SAFETY PROPERTY UNDER TEST, stated once:

    A wrong "yes, quote it" costs one Jupiter call.
    A wrong "no, skip it" costs a trade.

So every uncertainty — feed off, Redis down, key missing, JSON malformed,
value stale — must resolve to YES. The only "no" allowed is positive
evidence that nothing moved. Half these checks exist to prove the asymmetry
holds in the failing direction, because that is the direction that is
silent when it breaks.
"""

from __future__ import annotations

import importlib
import json
import os
import shutil
import socket
import subprocess
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

FAILURES: list[str] = []
CHECKS = 0


def check(label: str, condition: bool, detail: str = "") -> None:
    global CHECKS
    CHECKS += 1
    if condition:
        print(f"  \033[32m✓\033[0m {label}")
    else:
        FAILURES.append(f"{label}{f' — {detail}' if detail else ''}")
        print(f"  \033[31m✗\033[0m {label}" + (f" — {detail}" if detail else ""))


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def test_failure_modes() -> None:
    """Every unknown resolves to 'ask Jupiter'. No server needed."""
    print("\n\033[1m1. every uncertainty defaults to asking Jupiter\033[0m")
    import modules.feed_client as fc

    importlib.reload(fc)
    fc.ENABLED = False
    ok, why = fc.FeedClient().should_quote("r")
    check("FEED_ENABLED=false -> quote", ok, why)

    importlib.reload(fc)
    fc.ENABLED = True
    fc.REDIS_URL = "redis://127.0.0.1:1/0"          # nothing listening
    ok, why = fc.FeedClient().should_quote("r")
    check("Redis unreachable -> quote (and no crash)", ok, why)

    client = fc.FeedClient()
    client._client()
    check("an outage is remembered, not retried per lookup",
          client._unavailable_until > time.monotonic(),
          "a down Redis on every route of every cycle would replace the "
          "latency this module exists to remove")


def test_against_real_redis() -> None:
    print("\n\033[1m2. the real contract, against a real server\033[0m")
    if not shutil.which("redis-server"):
        print("  \033[33m·\033[0m redis-server not installed — section skipped")
        return
    try:
        import redis  # noqa: F401
    except ImportError:
        print("  \033[33m·\033[0m redis package not installed — section skipped")
        return

    port = _free_port()
    proc = subprocess.Popen(
        ["redis-server", "--port", str(port), "--save", "", "--appendonly", "no"],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        import redis

        url = f"redis://127.0.0.1:{port}/0"
        r = None
        for _ in range(50):
            try:
                r = redis.Redis.from_url(url, decode_responses=True)
                r.ping()
                break
            except Exception:  # noqa: BLE001
                time.sleep(0.1)
        if r is None:
            print("  \033[33m·\033[0m server never came up — section skipped")
            return

        import modules.feed_client as fc
        importlib.reload(fc)
        fc.ENABLED = True
        fc.REDIS_URL = url
        fc.MAX_AGE_MS = 250.0
        fc.STALE_ACTION = "skip"
        feed = fc.FeedClient()

        def write(dex: str, ra: int, rb: int, slot: int, age_ms: int = 0) -> None:
            r.set(f"feed:sol-usdc:{dex}:state", json.dumps({
                "reserve_a": ra, "reserve_b": rb,
                "mint_a": "So1", "mint_b": "EPj",
                "slot": slot, "ts_ms": int(time.time() * 1000) - age_ms}))
            r.sadd("feed:sol-usdc:pools", dex)

        r.flushall()
        write("raydium", 100, 200, 1000)
        check("pools are discovered, not hardcoded", feed.pools() == ["raydium"])

        ok, _ = feed.should_quote("USDC->SOL->USDC")
        check("first look quotes (nothing seen yet)", ok)

        ok, why = feed.should_quote("USDC->SOL->USDC")
        check("unchanged pool is SKIPPED — the entire saving", not ok, why)

        write("raydium", 101, 199, 1001)
        ok, _ = feed.should_quote("USDC->SOL->USDC")
        check("a moved pool quotes again", ok)

        # Reserves identical, slot advanced: must still count as moved. A
        # rounding-sized swap can leave both reserves numerically equal.
        write("raydium", 101, 199, 1002)
        ok, _ = feed.should_quote("USDC->SOL->USDC")
        check("slot change alone counts as movement", ok,
              "reserves-only fingerprinting would miss a tiny swap")

        # Per-route isolation: another route has its own memory.
        ok, _ = feed.should_quote("USDT->SOL->USDT")
        check("a different route is judged independently", ok)

        write("raydium", 300, 400, 2000, age_ms=900)
        ok, why = feed.should_quote("USDC->SOL->USDC")
        check("stale feed SKIPS by default", not ok, why)

        fc.STALE_ACTION = "jupiter"
        feed2 = fc.FeedClient()
        ok, why = feed2.should_quote("USDC->SOL->USDC")
        check("stale feed falls back when told to", ok, why)

        # A malformed row is a miss, never an exception.
        r.flushall()
        r.sadd("feed:sol-usdc:pools", "raydium")
        r.set("feed:sol-usdc:raydium:state", "{{{ not json")
        ok, why = fc.FeedClient().should_quote("USDC->SOL->USDC")
        check("malformed JSON -> quote, no crash", ok, why)

        # Listed pool with no state key at all.
        r.delete("feed:sol-usdc:raydium:state")
        ok, why = fc.FeedClient().should_quote("USDC->SOL->USDC")
        check("missing state key -> quote, no crash", ok, why)
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()


def test_never_prices_locally() -> None:
    """The design claim: this module does no AMM maths, on purpose."""
    print("\n\033[1m3. it computes no price — that is the point\033[0m")
    src = (Path(__file__).resolve().parent.parent
           / "modules" / "feed_client.py").read_text(encoding="utf-8")
    code = "\n".join(l for l in src.splitlines()
                     if not l.lstrip().startswith("#"))

    # Reserves alone cannot price a concentrated-liquidity pool, and this
    # deployment's real routes are Whirlpool / Meteora DLMM / Raydium CLMM.
    # A constant-product formula applied to those would be confidently wrong
    # on live money, so it must not appear here at all.
    for forbidden in ("* reserve_b", "reserve_a *", "/ reserve_a", "/ reserve_b",
                      "amount_out", "get_amount_out", "0.9975", "9975"):
        check(f"no AMM arithmetic: `{forbidden}`", forbidden not in code,
              "reserves cannot price a CLMM pool — see the module docstring")

    from modules.feed_client import FeedClient

    check("the public question is should_quote, not price",
          hasattr(FeedClient, "should_quote") and not hasattr(FeedClient, "price"))


def test_wired_into_the_scan_loop() -> None:
    """Built but not called is the same as not built."""
    print("\n\033[1m4. the gate is actually wired into the scan loop\033[0m")
    src = (Path(__file__).resolve().parent.parent
           / "modules" / "solana_arb.py").read_text(encoding="utf-8")
    code = "\n".join(l for l in src.splitlines()
                     if not l.lstrip().startswith("#"))

    check("scan loop imports the feed", "from modules.feed_client import get_feed" in code)
    check("and calls should_quote", "should_quote(route_key)" in code)

    # ORDERING: the gate must sit before the quote it exists to avoid, and
    # after the prune plan — those answer different questions and swapping
    # them would spend budget on routes the planner had already dropped.
    gate = code.find("should_quote(route_key)")
    quote = code.find("await self.scan_route(client,")
    plan = code.find("cycle_plan.get(")
    check("gate runs BEFORE the quote", 0 < gate < quote,
          "a gate after the call it saves is decoration")
    check("gate runs AFTER the prune plan", 0 < plan < gate)

    # A failure inside the gate must not stop the scan.
    check("the gate cannot break a cycle",
          "feed check skipped" in src and "feed_ok, feed_why = True" in code,
          "any error must fall through to quoting, not raise")


def test_inert_by_default() -> None:
    """FEED_ENABLED=false must be byte-identical to today's behaviour."""
    print("\n\033[1m5. off by default, provably\033[0m")
    import modules.feed_client as fc

    saved = os.environ.pop("FEED_ENABLED", None)
    try:
        importlib.reload(fc)
        check("FEED_ENABLED defaults to false", fc.ENABLED is False)
        ok, why = fc.FeedClient().should_quote("USDC->SOL")
        check("so the gate always says quote", ok and "disabled" in why,
              "an upgrade must change nothing until it is asked to")
    finally:
        if saved is not None:
            os.environ["FEED_ENABLED"] = saved
        importlib.reload(fc)


def main() -> int:
    print("\n\033[1m══════ feed_client: trigger, not price ══════\033[0m")
    test_failure_modes()
    test_against_real_redis()
    test_never_prices_locally()
    test_wired_into_the_scan_loop()
    test_inert_by_default()

    print()
    if FAILURES:
        print(f"\033[31m❌ {len(FAILURES)} of {CHECKS} checks FAILED\033[0m")
        for f in FAILURES:
            print(f"   • {f}")
        print()
        return 1
    print(f"\033[32m✅ all {CHECKS} checks passed\033[0m\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())