"""Test harness: isolated lake in a temp dir, a fresh `findata_test` PostgreSQL database (localhost), and a mocked FirstRate transport built from the licensed fixture excerpts (one session, three tickers). No real vendor calls.""" from __future__ import annotations import io import os import subprocess import tempfile import zipfile from pathlib import Path TEST_USERID = "test-userid-SECRET-9f8e7d6c" ROOT = Path(__file__).resolve().parents[1] LAKE_DIR = Path(os.environ.get("FINDATA_LAKE") or tempfile.mkdtemp(prefix="findata-lake-")) os.environ.setdefault("FINDATA_LAKE", str(LAKE_DIR)) os.environ.update({ "FINDATA_ENV": "test", "FINDATA_ROOT": str(ROOT), "FINDATA_DATABASE_URL": os.environ.get("FINDATA_TEST_DATABASE_URL", "postgresql://127.0.0.1:5432/findata_test"), "FINDATA_SECRET_KEY": "test-secret-key-test-secret-key-0123456789", "FINDATA_PUBLIC_URL": "http://testserver", "FRD_USERID": TEST_USERID, "SEC_EDGAR_USER_AGENT": "", "FINDATA_LAKE_HARD_MIN_FREE_GB": "1", "FINDATA_BACKUP_TARGET": "", }) import httpx import pytest FIX = ROOT / "tests" / "fixtures" def _reset_db() -> None: url = os.environ["FINDATA_DATABASE_URL"] subprocess.run(["psql", url, "-q", "-c", "DROP SCHEMA public CASCADE; CREATE SCHEMA public;"], check=True, capture_output=True) from findata_core.db import close_pool, migrate close_pool() migrate(verbose=False) @pytest.fixture(scope="session", autouse=True) def database(): _reset_db() yield from findata_core.db import close_pool close_pool() def build_update_zip(tickers: list[str] | None = None, mutate: dict[str, str] | None = None, period: str = "day") -> bytes: """Zip shaped like the vendor's `stock_update_day_1min_UNADJUSTED.zip`.""" tickers = tickers or ["AAPL", "A", "AAL"] buf = io.BytesIO() with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: for t in tickers: data = (FIX / "firstrate" / f"{t}_day_1min_UNADJUSTED.txt").read_text() if mutate and t in mutate: data = mutate[t] zf.writestr(zipfile.ZipInfo(f"{t}_{period}_1min_UNADJUSTED.txt", date_time=(2026, 10, 30, 18, 6, 0)), data, compress_type=zipfile.ZIP_DEFLATED) return buf.getvalue() def build_full_zip(tickers: list[str]) -> bytes: buf = io.BytesIO() with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: for t in tickers: zf.writestr(zipfile.ZipInfo(f"{t}_1min.txt", date_time=(2026, 10, 30, 18, 6, 0)), (FIX / "firstrate" / f"{t}_day_1min_UNADJUSTED.txt").read_text(), compress_type=zipfile.ZIP_DEFLATED) return buf.getvalue() class VendorMock: """httpx.MockTransport handler emulating FirstRate endpoints; records sanitized calls; asserts auth.""" def __init__(self): self.calls: list[str] = [] self.last_update = "2026-10-30" self.archive_override: bytes | None = None self.fail_next_download = 0 def __call__(self, request: httpx.Request) -> httpx.Response: url = request.url assert url.params.get("userid") == TEST_USERID, "vendor calls must carry the userid" self.calls.append(f"{url.path}?{'&'.join(f'{k}={v}' for k, v in sorted(url.params.items()) if k != 'userid')}") if url.path == "/api/last_update": return httpx.Response(200, text=self.last_update if url.params.get("type") != "fx" else self.last_update.replace("-", "")) if url.path == "/api/ticker_listing": t = url.params.get("type") body = "Ticker,Name,First Date,Last Date\n" if t == "stock": body += "A,Agilent Technologies Inc,2000-01-04,2026-10-30\nAAL,American Airlines,2005-09-27,2026-10-30\nAAPL,Apple Inc,2000-01-03,2026-10-30\nTWTR-DELISTED,Twitter Inc,2013-11-07,2022-10-27\n" else: body += "SPY,SPDR S&P 500,2000-01-03,2026-10-30\n" return httpx.Response(200, text=body) if url.path == "/api/update_log": return httpx.Response(200, text="Ticker,Last Date\nTWTR,2022-10-27\n") if url.path == "/api/meta_file": kind = url.params.get("metafile_type") buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: if kind == "splits": zf.writestr("AAPL.txt", "2020-08-31,4\n2014-06-09,7\n2005-02-28,2\n2000-06-21,2\n") zf.writestr("_splits_readme.txt", "readme") else: zf.writestr("AAPL_divs.txt", "2026-08-10,0.27\n2026-05-11,0.27\n") return httpx.Response(302, headers={"location": "https://cdn.example/meta.zip"}) if False else httpx.Response(200, content=buf.getvalue(), headers={"content-type": "application/zip"}) if url.path == "/api/data_file": if self.fail_next_download: self.fail_next_download -= 1 return httpx.Response(503, text="busy") if url.params.get("period") == "full": letter = url.params.get("ticker_range") tick = [t for t in ["A", "AAL", "AAPL"] if t.startswith(letter)] return httpx.Response(200, content=build_full_zip(tick), headers={"content-type": "application/zip"}) body = self.archive_override or build_update_zip(period=url.params.get("period", "day")) # emulate the CDN redirect if url.host == "firstratedata.com": return httpx.Response(302, headers={"location": f"https://x01.frd-data.com/stock_update_{url.params.get('period')}_1min_UNADJUSTED_tok.zip?userid={TEST_USERID}"}) return httpx.Response(200, content=body, headers={"content-type": "application/zip"}) if url.host == "x01.frd-data.com": body = self.archive_override or build_update_zip() return httpx.Response(200, content=body, headers={"content-type": "application/zip"}) if url.path.startswith("/api-f/v1/financials/all/"): p = (FIX / "fundamentals" / "financials_all_AAPL_quarterly.json").read_text() return httpx.Response(200, text=p, headers={"content-type": "application/json"}) if url.path.startswith("/api-f/v1/earnings_release/t/"): return httpx.Response(200, text=(FIX / "fundamentals" / "earnings_AAPL.json").read_text(), headers={"content-type": "application/json"}) return httpx.Response(404, text=f"no route {url.path} userid={url.params.get('userid')}") @pytest.fixture def vendor(monkeypatch): vm = VendorMock() import findata_fundamentals.firstrate as F import findata_market_firstrate.client as C monkeypatch.setattr(C, "DEFAULT_TRANSPORT", httpx.MockTransport(vm)) monkeypatch.setattr(F, "DEFAULT_TRANSPORT", httpx.MockTransport(vm)) return vm @pytest.fixture def lake(): from findata_ingest.lake import Lake return Lake(LAKE_DIR) class Log: def __getattr__(self, name): def f(*a, **k): return None return f def bind(self, **k): return self @pytest.fixture def log(): return Log()