File size: 5,303 Bytes
2e10c7a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """Ingest/db review fixes (2026-09-14): DSN parsing, corpus/replay DB resilience, token stats for the selector's universe."""
from __future__ import annotations
import threading
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
import httpx
import psycopg
import pytest
from fly_trader import config
from fly_trader.db import schema
from fly_trader.db.connection import database_name
from fly_trader.ingest import corpus_pull, discovery
def _tx_on(conn):
@contextmanager
def tx():
yield conn
return tx
# ---- 4: DSN parsing ----
@pytest.mark.parametrize("url,name", [("postgresql://fly_trader:pw@localhost/fly_trader", "fly_trader"),
("postgresql:///fly_trader_test", "fly_trader_test"),
("dbname=fly_trader_test user=fly_trader host=/tmp", "fly_trader_test"),
("host=localhost user=bob", "fly_trader")])
def test_database_name_reads_urls_and_key_value_dsns(url, name):
assert database_name(url) == name
def test_ensure_database_admin_dsn_keeps_user_and_host(monkeypatch):
seen = []
class Conn:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def execute(self, sql, params=None):
class R:
def fetchone(self):
return (1,)
return R()
monkeypatch.setattr(schema.psycopg, "connect", lambda conninfo, **kw: seen.append(conninfo) or Conn())
assert schema.ensure_database("postgresql://fly_trader:pw@localhost/fly_trader") is False
assert psycopg.conninfo.conninfo_to_dict(seen[0]) == {"user": "fly_trader", "password": "pw", "host": "localhost", "dbname": "postgres"}
schema.ensure_database("dbname=fly_trader user=fly_trader host=/tmp")
assert psycopg.conninfo.conninfo_to_dict(seen[1]) == {"user": "fly_trader", "host": "/tmp", "dbname": "postgres"}
# ---- 2 / 3: corpus + replay resilience ----
def test_db_retry_rides_out_postgres_errors(monkeypatch):
monkeypatch.setattr(corpus_pull, "_sleep", lambda s, stop: None)
calls = []
def flaky(x):
calls.append(x)
if len(calls) < 3:
raise psycopg.OperationalError("server closed the connection unexpectedly")
return x * 2
def bug():
raise ValueError("not a database error")
assert corpus_pull._db_retry(None, "t", flaky, 21) == 42 and len(calls) == 3
stopped = threading.Event()
stopped.set()
assert corpus_pull._db_retry(stopped, "t", flaky, 1) is None
with pytest.raises(ValueError):
corpus_pull._db_retry(None, "t", bug)
def test_swap_api_4xx_is_permanent():
api = corpus_pull.SwapApi(corpus_pull.Pacer(0.0, 0.0), corpus_pull.Status(), None)
api.c = httpx.Client(base_url="http://swap.test", transport=httpx.MockTransport(lambda req: httpx.Response(404, text="not found")))
with pytest.raises(corpus_pull.Permanent):
api.get("/v2/coins/M/candles", {})
def test_corpus_requeues_cooled_transient_errors_only(db_conn, monkeypatch):
monkeypatch.setattr(corpus_pull, "transaction", _tx_on(db_conn))
grad = datetime.fromisoformat(config.CORPUS_PULL_BEFORE).replace(tzinfo=timezone.utc) - timedelta(days=1)
rows = {"TestFix3Pending": ("pending", None, 0.0),
"TestFix3TransientOld": ("error", "RuntimeError: swap-api /v2/coins/M/candles: gave up after retries", corpus_pull.RETRY_ERROR_H + 1),
"TestFix3TransientNew": ("error", "RuntimeError: swap-api /v2/coins/M/candles: gave up after retries", 0.5),
"TestFix3Permanent": ("error", "permanent: Permanent: swap-api /v2/coins/M/candles HTTP 404: nf", corpus_pull.RETRY_ERROR_H + 1)}
for m, (st, err, age_h) in rows.items():
db_conn.execute("INSERT INTO corpus_tokens (mint, graduated_at, status, last_error, updated_at) VALUES (%s, %s, %s, %s, now() - %s * interval '1 hour')",
(m, grad, st, err, age_h))
picked = {r["mint"] for r in corpus_pull._next_pending(limit=1_000_000)}
assert {"TestFix3Pending", "TestFix3TransientOld"} <= picked
assert not {"TestFix3TransientNew", "TestFix3Permanent"} & picked
# ---- stats for the selector's universe: token_stats only, no tokens/watch_pools side effects ----
def test_refresh_stats_covers_selector_universe_without_watching(db_conn):
mint = "TestUniverseMint111111111111111111111111111"
db_conn.execute("INSERT INTO pump_minutes (mint, ts, close, resq_sol) VALUES (%s, date_trunc('minute', now()) - interval '5 minutes', 1e-7, 30.0)", (mint,))
asked = []
class Client:
def search_many(self, mints):
asked.append(list(mints))
return [{"id": mint, "organicScore": 42.0}] if mint in mints else []
c = discovery.refresh_stats(db_conn, Client())
assert c["universe"] >= 1 and c["refreshed"] == 1 and any(mint in a for a in asked)
assert db_conn.execute("SELECT organic_score FROM token_stats WHERE mint = %s", (mint,)).fetchone()["organic_score"] == 42.0
assert db_conn.execute("SELECT 1 FROM tokens WHERE mint = %s", (mint,)).fetchone() is None
assert db_conn.execute("SELECT 1 FROM watch_pools WHERE mint = %s", (mint,)).fetchone() is None
|