1"""Integration: lake publication, durable queue, incremental sync idempotency, resume, derive, compaction."""2from __future__ import annotations34import json5from datetime import UTC, date, datetime67import pyarrow as pa8import pytest910from findata_core.db import connection, fetch_one, transaction11from findata_ingest import queue12from findata_ingest.lake import live_files13from findata_ingest.reader import apply_adjustment, read_bars, run_sql14from findata_ingest.tasks import backfill, compaction, derive, incremental, metadata15from tests.conftest import build_update_zip1617DS = "market_bars.stock.1m.raw"181920def _job(kind, dataset_id, params, run=True):21 jid = queue.enqueue(kind, dataset_id, params, created_by="test")22 job = queue.claim_next([kind])23 assert job and str(job["id"]) == jid24 return job252627def _rows(ds=DS):28 with connection() as conn:29 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()303132def test_queue_retry_backoff_and_dead_letter():33 jid = queue.enqueue("quality", None, {"x": 1}, max_attempts=2, dedupe_key="q-test")34 assert queue.enqueue("quality", None, {"x": 1}, dedupe_key="q-test") is None # deduped while queued35 job = queue.claim_next(["quality"])36 assert job["attempts"] == 137 queue.finish(job, False, error=f"boom userid=SECRETVALUE")38 r = fetch_one("SELECT status, last_error, next_run_at > now() later FROM ingest_jobs WHERE id=%s", (jid,))39 assert r["status"] == "queued" and r["later"] and "SECRETVALUE" not in r["last_error"]40 with transaction() as conn:41 conn.execute("UPDATE ingest_jobs SET next_run_at=now() WHERE id=%s", (jid,))42 job = queue.claim_next(["quality"])43 queue.finish(job, False, error="boom again")44 assert fetch_one("SELECT status FROM ingest_jobs WHERE id=%s", (jid,))["status"] == "dead"454647def test_metadata_then_incremental_is_idempotent(vendor, lake, log):48 metadata.run(_job("metadata", None, {"types": ["stock", "etf"]}), lake, log)49 with connection() as conn:50 n = conn.execute("SELECT count(*) FROM instruments").fetchone()["count"]51 delisted = conn.execute("SELECT status FROM instruments WHERE symbol='TWTR-DELISTED'").fetchone()["status"]52 splits = conn.execute("SELECT count(*) FROM corporate_actions WHERE action_type='split'").fetchone()["count"]53 assert n == 5 and delisted == "delisted" and splits == 45455 # first incremental: 3 symbols, one session56 r1 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day"}), lake, log)57 assert r1["rows_published"] == 900 + 390 + 532 or r1["rows_published"] > 150058 assert r1["version"] == 1 and r1["files_added"] == 1 and r1["symbols"] == 359 v1 = _rows()60 cov = fetch_one("SELECT count(*) n FROM coverage WHERE dataset_id=%s", (DS,))["n"]61 assert cov == 362 raw = fetch_one("SELECT status, path FROM raw_archives WHERE dataset_id=%s", (DS,))63 assert raw["status"] == "published" and (lake.root / raw["path"]).exists() # immutable raw kept6465 # re-run: vendor unchanged → skipped, nothing duplicated66 r2 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day"}), lake, log)67 assert r2.get("skipped") and _rows() == v16869 # forced re-fetch of the identical archive → recognised by sha256, no new version70 r3 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day", "force": True}), lake, log)71 assert r3.get("skipped") and r3["reason"] == "archive already ingested" and _rows() == v17273 # vendor republishes the same session with a corrected bar → supersede (no duplicates, version 2)74 fixed = (lake.root.parent / "x").exists() # noqa: F84175 import pathlib76 txt = (pathlib.Path(__file__).parent / "fixtures" / "firstrate" / "AAPL_day_1min_UNADJUSTED.txt").read_text()77 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")78 vendor.archive_override = build_update_zip(mutate={"AAPL": txt})79 vendor.last_update = "2026-10-30"80 r4 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day", "force": True}), lake, log)81 assert not r4.get("skipped") and r4["version"] == 2 and r4["files_retired"] == 182 v2 = _rows()83 assert v2["n"] == v1["n"] and v2["f"] == 184 t, meta = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))85 assert t.num_rows == 900 and meta["version"] == 286 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]87 assert row["close"] == 348.2788 # snapshot reproducibility: version 1 still readable from retired file set89 t1, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30), version=1)90 assert t1.num_rows == 90091 vendor.archive_override = None92 # no duplicate (instrument, ts) anywhere in the live set93 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,94 {"market_bars_stock_1m_raw": [str(lake.abs(f["path"])) for f in _live()]})[0].to_pylist()[0]95 assert list(dup.values())[0] == 0969798def _live():99 with connection() as conn:100 return live_files(conn, DS)101102103def test_derive_5m_from_published_session(vendor, lake, log):104 job = queue.claim_next(["derive"]) # enqueued by the incremental run105 assert job is not None106 r = derive.run(job, lake, log)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 job = _job("compaction", DS, {"min_age_days": 0})140 r = compaction.run(job, lake, log)141 assert r["compacted_files"] == 1 and r["base_files_written"] == 3142 live = _live()143 assert all(f["layout"] == "base" for f in live) and len(live) == 3144 t, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))145 assert t.num_rows == 900146 # the retired delta file was moved out of the live tree147 assert not any(p.name.endswith(".parquet") for p in (lake.root / "curated" / DS / "delta").rglob("*.parquet"))148 assert any((lake.root / "_retired").rglob("*.parquet"))149150151def test_run_sql_sandbox_rejects_mutations():152 views = {"market_bars_stock_1m_raw": []}153 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'"]:154 with pytest.raises(ValueError):155 run_sql(bad, 10, 5, views)156 t, meta = run_sql("SELECT 1 AS one", 10, 5, views)157 assert t.to_pylist() == [{"one": 1}]158