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)

        # ── THE SILENT MISMATCH ───────────────────────────────────────
        # The two briefs specified `feed:sol-usdc:<dex>:state` and
        # `feed:solusdc:<dex>`. Read the wrong one and this module works
        # perfectly: it finds nothing, falls back to Jupiter, reports no
        # error, and buys exactly nothing forever. describe() is the only
        # thing standing between that and a week of "why is the budget
        # still at 99%".
        print("\n\033[1m2b. a key mismatch cannot hide\033[0m")
        r.flushall()
        r.set("feed:solusdc:raydium", json.dumps({          # the WRONG prefix
            "reserve_a": 1, "reserve_b": 2, "slot": 1,
            "ts_ms": int(time.time() * 1000)}))
        r.sadd("feed:solusdc:pools", "raydium")
        importlib.reload(fc)
        fc.ENABLED = True
        fc.REDIS_URL = url
        fc.KEY_PREFIX = "feed:sol-usdc"                     # what we read
        d = fc.FeedClient().describe()
        check("reads nothing under its own prefix", d["pools"] == [])
        check("but SEES the writer's actual prefix",
              "feed:solusdc" in d["found_prefixes"], str(d["found_prefixes"]))
        check("and calls it a mismatch, not a quiet market",
              any("MISMATCH" in p for p in d["problems"]), str(d["problems"]))
        check("and prints the exact command that fixes it",
              "FEED_KEY_PREFIX=feed:solusdc" in d["hint"], d["hint"])

        # Reading the right prefix must produce NO problem — a diagnostic
        # that cries wolf on a healthy feed gets ignored on a broken one.
        fc.KEY_PREFIX = "feed:solusdc"
        d = fc.FeedClient().describe()
        check("a correctly-configured feed reports clean",
              d["pools"] == ["raydium"] and not d["problems"], str(d))

        # Redis up, writer never started: a different problem, said
        # differently. Blaming the prefix here would send the operator to
        # change a setting that is already right.
        r.flushall()
        d = fc.FeedClient().describe()
        check("no data at all is blamed on the WRITER, not the prefix",
              any("not running" in w for w in d["warnings"]) and not d["hint"],
              str(d["warnings"]))

        # ── BLOCKER vs NOISE ──────────────────────────────────────────
        # /doctor's ⛔ list is for things that stop the bot trading, and
        # it is read top-down. On the live box this module put
        #
        #   • pool feed: Redis unreachable ... trading is unaffected ...
        #
        # in that list, directly above the IAM denial that genuinely was
        # blocking every trade. A false blocker is worse than a warning:
        # it dilutes the one list the operator is meant to act on, and
        # this one said "trading is unaffected" in its own text.
        check("an absent writer is NOT a blocker",
              not d["problems"],
              "expected until garden-feed exists; the Jupiter fallback is "
              "where the bot has always been")

        importlib.reload(fc)
        fc.ENABLED = True
        fc.REDIS_URL = "redis://127.0.0.1:1/0"          # nothing listening
        d = fc.FeedClient().describe()
        check("an unreachable Redis is NOT a blocker either",
              not d["problems"] and d["warnings"], str(d))

        # But a MISMATCH still is — that one is silently costing something.
        importlib.reload(fc)
        fc.ENABLED = True
        fc.REDIS_URL = url
        fc.KEY_PREFIX = "feed:sol-usdc"
        r.set("feed:solusdc:raydium", json.dumps({
            "reserve_a": 1, "reserve_b": 2, "slot": 1,
            "ts_ms": int(time.time() * 1000)}))
        r.sadd("feed:solusdc:pools", "raydium")
        d = fc.FeedClient().describe()
        check("a real KEY MISMATCH is still a blocker",
              any("MISMATCH" in p for p in d["problems"]) and d["hint"],
              "this is the one case that is silently costing budget")

        # MISMATCH is the ONLY blocker. A writer that has started and not
        # yet registered a pool is a writer that is STARTING — waiting is
        # the correct response, so it must not enter the ⛔ list. Qodo
        # caught this contradicting the contract stated in describe()'s own
        # docstring, two screens above the code that broke it.
        r.flushall()
        r.set("feed:sol-usdc:raydium:state", json.dumps({
            "reserve_a": 1, "reserve_b": 2, "slot": 1,
            "ts_ms": int(time.time() * 1000)}))          # no :pools member
        d = fc.FeedClient().describe()
        check("a writer with no pools registered yet is NOT a blocker",
              not d["problems"] and d["warnings"], str(d))

        # Every warning must name its own subsystem. /doctor prefixes
        # problems with "pool feed:" but forwards warnings verbatim, so a
        # bare "could not scan Redis" loses its owner among the warnings of
        # five other subsystems.
        for w in d["warnings"]:
            check("warning names its subsystem", w.startswith("pool feed"),
                  w[:80])

        # And the chat path must give the same reason /doctor does.
        from modules.bot_chat import BotChat

        importlib.reload(fc)
        fc.ENABLED = True
        fc.REDIS_URL = "redis://127.0.0.1:1/0"
        said = BotChat()._say_feed()
        check("chat reports the WARNING, not a bare 'no data'",
              "no data" not in said and "unreachable" in said,
              "reading only problems left the real reason — and the command "
              "that silences it — unread in warnings")
    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())