| """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 |
|
|
|
|
| |
| @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"} |
|
|
|
|
| |
| 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 |
|
|
|
|
| |
| 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 |
|
|
