1"""Integration: lake publication, durable queue, incremental sync idempotency, resume, derive, compaction."""2from __future__ import annotations34from datetime import UTC, date, datetime56import pyarrow as pa7import pytest8from findata_core.db import connection, fetch_one, transaction9from findata_ingest import queue10from findata_ingest.lake import live_files11from findata_ingest.reader import apply_adjustment, read_bars, run_sql12from findata_ingest.tasks import backfill, compaction, derive, incremental, metadata1314from tests.conftest import build_update_zip1516DS = "market_bars.stock.1m.raw"171819def _job(kind, dataset_id, params, run=True):20 jid = queue.enqueue(kind, dataset_id, params, created_by="test")21 job = queue.claim_next([kind])22 assert job and str(job["id"]) == jid23 return job242526def _rows(ds=DS):27 with connection() as conn:28 return conn.execute("SELECT coalesce(sum(row_count),0) n, count(*) f FROM lake_files WHERE dataset_id=%s AND version_removed IS NULL", (ds,)).fetchone()293031def test_queue_retry_backoff_and_dead_letter():32 jid = queue.enqueue("quality", None, {"x": 1}, max_attempts=2, dedupe_key="q-test")33 assert queue.enqueue("quality", None, {"x": 1}, dedupe_key="q-test") is None # deduped while queued34 job = queue.claim_next(["quality"])35 assert job["attempts"] == 136 queue.finish(job, False, error="boom userid=SECRETVALUE")37 r = fetch_one("SELECT status, last_error, next_run_at > now() later FROM ingest_jobs WHERE id=%s", (jid,))38 assert r["status"] == "queued" and r["later"] and "SECRETVALUE" not in r["last_error"]39 with transaction() as conn:40 conn.execute("UPDATE ingest_jobs SET next_run_at=now() WHERE id=%s", (jid,))41 job = queue.claim_next(["quality"])42 queue.finish(job, False, error="boom again")43 assert fetch_one("SELECT status FROM ingest_jobs WHERE id=%s", (jid,))["status"] == "dead"444546def test_metadata_then_incremental_is_idempotent(vendor, lake, log):47 metadata.run(_job("metadata", None, {"types": ["stock", "etf"]}), lake, log)48 with connection() as conn:49 n = conn.execute("SELECT count(*) FROM instruments").fetchone()["count"]50 delisted = conn.execute("SELECT status FROM instruments WHERE symbol='TWTR-DELISTED'").fetchone()["status"]51 splits = conn.execute("SELECT count(*) FROM corporate_actions WHERE action_type='split'").fetchone()["count"]52 assert n == 5 and delisted == "delisted" and splits == 45354 # first incremental: 3 symbols, one session55 r1 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day"}), lake, log)56 assert r1["rows_published"] == 900 + 390 + 532 or r1["rows_published"] > 150057 assert r1["version"] == 1 and r1["files_added"] == 1 and r1["symbols"] == 358 v1 = _rows()59 cov = fetch_one("SELECT count(*) n FROM coverage WHERE dataset_id=%s", (DS,))["n"]60 assert cov == 361 raw = fetch_one("SELECT status, path FROM raw_archives WHERE dataset_id=%s", (DS,))62 assert raw["status"] == "published" and (lake.root / raw["path"]).exists() # immutable raw kept6364 # re-run: vendor unchanged → skipped, nothing duplicated65 r2 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day"}), lake, log)66 assert r2.get("skipped") and _rows() == v16768 # forced re-fetch of the identical archive → recognised by sha256, no new version69 r3 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day", "force": True}), lake, log)70 assert r3.get("skipped") and r3["reason"] == "archive already ingested" and _rows() == v17172 # vendor republishes the same session with a corrected bar → supersede (no duplicates, version 2)73 fixed = (lake.root.parent / "x").exists()74 import pathlib75 txt = (pathlib.Path(__file__).parent / "fixtures" / "firstrate" / "AAPL_day_1min_UNADJUSTED.txt").read_text()76 txt = txt.replace("2026-10-30 09:30:00,346.63,348.51,346.62,348.26,1119437", "2026-10-30 09:30:00,346.63,348.51,346.62,348.27,1119437")77 vendor.archive_override = build_update_zip(mutate={"AAPL": txt})78 vendor.last_update = "2026-10-30"79 r4 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day", "force": True}), lake, log)80 assert not r4.get("skipped") and r4["version"] == 2 and r4["files_retired"] == 181 v2 = _rows()82 assert v2["n"] == v1["n"] and v2["f"] == 183 t, meta = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))84 assert t.num_rows == 900 and meta["version"] == 285 row = t.filter(pa.compute.equal(t.column("timestamp_utc"), pa.scalar(datetime(2026, 10, 30, 13, 30, tzinfo=UTC), pa.timestamp("us", tz="UTC")))).to_pylist()[0]86 assert row["close"] == 348.2787 # snapshot reproducibility: version 1 still readable from retired file set88 t1, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30), version=1)89 assert t1.num_rows == 90090 vendor.archive_override = None91 # no duplicate (instrument, ts) anywhere in the live set92 dup = run_sql("SELECT count(*) FROM (SELECT instrument_id, timestamp_utc, count(*) c FROM market_bars_stock_1m_raw GROUP BY 1,2 HAVING c > 1)", 10, 10,93 {"market_bars_stock_1m_raw": [str(lake.abs(f["path"])) for f in _live()]})[0].to_pylist()[0]94 assert list(dup.values())[0] == 0959697def _live():98 with connection() as conn:99 return live_files(conn, DS)100101102def test_derive_5m_from_published_session(vendor, lake, log):103 job = queue.claim_next(["derive"]) # enqueued by the incremental run104 assert job is not None105 r = derive.run(job, lake, log)106 queue.finish(job, True, r)107 assert r["rows"] > 0108 t, _ = read_bars("market_bars.stock.5m.raw", ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))109 assert 150 <= t.num_rows <= 192 and t.column("interval")[0].as_py() == "5m"110 one, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))111 assert abs(sum(t.column("volume").to_pylist()) - sum(one.column("volume").to_pylist())) < 1e-6112113114def test_adjustment_applied_at_query_time(vendor, lake, log):115 with transaction() as conn:116 conn.execute("INSERT INTO corporate_actions(instrument_id, action_type, ex_date, value, source_id) VALUES ('firstrate:stock:AAPL','split','2026-10-31',2,'firstrate') ON CONFLICT DO NOTHING")117 raw, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))118 adj, notes = apply_adjustment(raw, "split")119 assert abs(adj.column("close")[0].as_py() * 2 - raw.column("close")[0].as_py()) < 1e-9120 assert abs(adj.column("volume")[0].as_py() - raw.column("volume")[0].as_py() * 2) < 1e-6121 assert adj.column("adjustment_type")[0].as_py() == "split" and raw.column("adjustment_type")[0].as_py() == "raw"122 with transaction() as conn:123 conn.execute("DELETE FROM corporate_actions WHERE ex_date='2026-10-31'")124125126def test_backfill_resumes_from_checkpoint_and_respects_delta_dates(vendor, lake, log):127 job = _job("backfill", DS, {"type": "stock", "ticker_range": "A", "delete_raw_after_publish": False})128 # simulate a crash after AAL was published: pre-seed the checkpoint129 queue.set_checkpoint(f"job:{job['id']}", "state", {"done_symbols": ["AAL"]})130 r = backfill.run(job, lake, log)131 # all fixture rows fall on 2026-10-30 which already lives in a delta file → excluded, nothing duplicated132 assert r["symbols"] == 0 and r["skipped_rows_in_delta"] > 0133 ck = queue.get_checkpoint(f"job:{job['id']}", "state")134 assert set(ck["done_symbols"]) == {"A", "AAL", "AAPL"}135 assert _rows()["f"] == 1136137138def test_compaction_merges_delta_into_base_and_checks_invariant(vendor, lake, log):139 with transaction() as conn: # compaction defers while backfill/derive jobs are pending on the dataset140 conn.execute("UPDATE ingest_jobs SET status='cancelled' WHERE status IN ('queued','running') AND kind IN ('derive','backfill')")141 job = _job("compaction", DS, {"min_age_days": 0})142 r = compaction.run(job, lake, log)143 assert r["compacted_files"] == 1 and r["base_files_written"] == 3144 live = _live()145 assert all(f["layout"] == "base" for f in live) and len(live) == 3146 t, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))147 assert t.num_rows == 900148 # the retired delta file was moved out of the live tree149 assert not any(p.name.endswith(".parquet") for p in (lake.root / "curated" / DS / "delta").rglob("*.parquet"))150 assert any((lake.root / "_retired").rglob("*.parquet"))151152153def test_run_sql_sandbox_rejects_mutations():154 views = {"market_bars_stock_1m_raw": []}155 for bad in ["DROP TABLE x", "SELECT 1; SELECT 2", "COPY (SELECT 1) TO '/tmp/x'", "INSTALL httpfs", "SELECT * FROM read_csv('/etc/passwd')", "ATTACH 'x.db'"]:156 with pytest.raises(ValueError):157 run_sql(bad, 10, 5, views)158 t, meta = run_sql("SELECT 1 AS one", 10, 5, views)159 assert t.to_pylist() == [{"one": 1}]160