1"""API + SDK end-to-end against the in-process app (TestClient) with data published by the ingest task."""2from __future__ import annotations34import io5import json6import time7from datetime import UTC, date, datetime89import httpx10import pyarrow.ipc as ipc11import pyarrow.parquet as pq12import pyotp13import pytest14from fastapi.testclient import TestClient1516from findata_api import auth as A17from findata_core.db import connection, fetch_one, transaction18from findata_ingest import queue19from findata_ingest.tasks import incremental, metadata20from tests.conftest import TEST_USERID2122OWNER = "owner@example.com"23PW = "correct horse battery staple"242526@pytest.fixture(scope="module")27def app_client(vendor_module, lake_module):28 from findata_api.main import create_app29 # make sure one session of bars exists (tests may run in isolation)30 n = fetch_one("SELECT count(*) AS n FROM lake_files WHERE dataset_id='market_bars.stock.1m.raw' AND version_removed IS NULL")["n"]31 if n == 0:32 class L:33 def __getattr__(self, _):34 return lambda *a, **k: None35 def bind(self, **k):36 return self37 jid = queue.enqueue("metadata", None, {"types": ["stock", "etf"]})38 metadata.run(queue.claim_next(["metadata"]), lake_module, L())39 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day", "force": True})40 incremental.run(queue.claim_next(["incremental"]), lake_module, L())41 if fetch_one("SELECT count(*) AS n FROM lake_files WHERE dataset_id='market_bars.stock.5m.raw' AND version_removed IS NULL")["n"] == 0:42 from findata_ingest.tasks import derive43 class L2:44 def __getattr__(self, _):45 return lambda *a, **k: None46 def bind(self, **k):47 return self48 j = queue.claim_next(["derive"]) or (queue.enqueue("derive", "market_bars.stock.5m.raw", {"source_dataset": "market_bars.stock.1m.raw", "target_interval": "5m", "dates": ["2026-10-30"]}) and queue.claim_next(["derive"]))49 derive.run(j, lake_module, L2())50 A.create_owner(OWNER, PW)51 app = create_app()52 with TestClient(app, base_url="http://testserver") as c:53 yield c545556@pytest.fixture(scope="module")57def vendor_module():58 from tests.conftest import VendorMock59 import findata_market_firstrate.client as C60 import findata_fundamentals.firstrate as F61 vm = VendorMock()62 old = (C.DEFAULT_TRANSPORT, F.DEFAULT_TRANSPORT)63 C.DEFAULT_TRANSPORT = httpx.MockTransport(vm)64 F.DEFAULT_TRANSPORT = httpx.MockTransport(vm)65 yield vm66 C.DEFAULT_TRANSPORT, F.DEFAULT_TRANSPORT = old676869@pytest.fixture(scope="module")70def lake_module():71 from findata_ingest.lake import Lake72 return Lake()737475def _login(c: TestClient) -> str:76 r = c.post("/api/v1/auth/login", json={"email": OWNER, "password": PW})77 assert r.status_code == 200, r.text78 return r.json()["csrf_token"]798081def test_unauthenticated_access_fails(app_client):82 c = app_client83 for path in ["/api/v1/catalog/datasets", "/api/v1/market/bars?symbol=AAPL&start=2026-10-30&end=2026-10-30", "/api/v1/admin/jobs", "/api/v1/query/views"]:84 r = c.get(path)85 assert r.status_code == 401, path86 assert r.json()["error"]["code"] == "unauthenticated" and r.headers["x-request-id"]87 assert c.post("/api/v1/query", json={"sql": "select 1"}).status_code == 40188 assert c.get("/api/v1/health").json() == {"status": c.get("/api/v1/health").json()["status"], "service": "findata-os"}89 r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2026-10-30", "end": "2026-10-30"}, headers={"Authorization": "Bearer fd_bad.token"})90 assert r.status_code == 401 and r.json()["error"]["code"] == "invalid_api_key"919293def test_login_bad_password_then_good_then_csrf(app_client):94 c = app_client95 r = c.post("/api/v1/auth/login", json={"email": OWNER, "password": "nope"})96 assert r.status_code == 40197 csrf = _login(c)98 me = c.get("/api/v1/auth/me").json()99 assert me["email"] == OWNER and me["role"] == "owner" and "admin" in me["scopes"] and me["totp_enabled"] is False100 # state-changing call without CSRF header is refused101 r = c.post("/api/v1/admin/jobs", json={"kind": "quality"})102 assert r.status_code == 403 and r.json()["error"]["code"] == "csrf"103 r = c.post("/api/v1/admin/jobs", json={"kind": "quality"}, headers={"X-CSRF-Token": csrf})104 assert r.status_code == 201105 with transaction() as conn:106 conn.execute("UPDATE ingest_jobs SET status='cancelled' WHERE kind='quality' AND status='queued'")107108109def test_totp_enrolment_and_mfa_gate(app_client):110 c = app_client111 csrf = _login(c)112 r = c.post("/api/v1/auth/totp/setup", headers={"X-CSRF-Token": csrf})113 assert r.status_code == 200 and r.json()["otpauth_uri"].startswith("otpauth://totp/")114 secret = r.json()["secret"]115 r = c.post("/api/v1/auth/totp/verify", json={"code": "000000"}, headers={"X-CSRF-Token": csrf})116 assert r.status_code == 400117 r = c.post("/api/v1/auth/totp/verify", json={"code": pyotp.TOTP(secret).now()}, headers={"X-CSRF-Token": csrf})118 assert r.status_code == 200 and r.json()["totp_enabled"]119 # new login now requires MFA120 r = c.post("/api/v1/auth/login", json={"email": OWNER, "password": PW})121 assert r.json()["mfa_required"] is True122 csrf = r.json()["csrf_token"]123 assert c.get("/api/v1/catalog/datasets").status_code == 401 # mfa_required124 r = c.post("/api/v1/auth/mfa", json={"code": "123456"}, headers={"X-CSRF-Token": csrf})125 assert r.status_code == 401126 r = c.post("/api/v1/auth/mfa", json={"code": pyotp.TOTP(secret).now()}, headers={"X-CSRF-Token": csrf})127 assert r.status_code == 200128 assert c.get("/api/v1/catalog/datasets").status_code == 200129 # disable TOTP again for the remaining tests (keep the flow simple)130 with transaction() as conn:131 conn.execute("UPDATE users SET totp_enabled=false WHERE email=%s", (OWNER,))132133134@pytest.fixture(scope="module")135def api_key(app_client):136 csrf = _login(app_client)137 r = app_client.post("/api/v1/admin/keys", json={"name": "sdk-test", "scopes": ["read:market", "read:fundamentals", "query", "export"]},138 headers={"X-CSRF-Token": csrf})139 assert r.status_code == 201, r.text140 tok = r.json()["token"]141 app_client.cookies.clear()142 return tok143144145def test_api_key_scopes_and_bars(app_client, api_key):146 c = app_client147 H = {"Authorization": f"Bearer {api_key}"}148 assert c.get("/api/v1/admin/jobs", headers=H).status_code == 403 # no admin scope149 r = c.get("/api/v1/catalog/datasets", headers=H)150 assert r.status_code == 200151 ds = {d["id"]: d for d in r.json()["data"]}152 assert ds["market_bars.stock.1m.raw"]["status"] == "active" and ds["market_ticks"]["status"] == "unsupported"153 assert ds["fundamentals.sec_companyfacts"]["status"] == "pending_configuration"154 assert "no redistribution" in ds["market_bars.stock.1m.raw"]["license_text"].lower() or "may not be redistributed" in ds["market_bars.stock.1m.raw"]["license_text"]155 r = c.get("/api/v1/market/bars", params={"symbol": "aapl", "interval": "1m", "start": "2026-10-30", "end": "2026-10-30", "limit": 400}, headers=H)156 assert r.status_code == 200, r.text157 body = r.json()158 assert body["meta"]["rows"] == 400 and body["meta"]["rows_matching"] == 900 and body["next_cursor"]159 assert body["data"][0]["timestamp_utc"] == "2026-10-30T08:00:00Z" and body["meta"]["timestamp_convention"] == "bar_open"160 assert r.headers["X-Dataset-Version"].isdigit() and r.headers["X-Dataset-Id"] == "market_bars.stock.1m.raw"161 r2 = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2026-10-30", "end": "2026-10-30", "limit": 400, "cursor": body["next_cursor"]}, headers=H)162 assert r2.json()["data"][0]["timestamp_utc"] > body["data"][-1]["timestamp_utc"]163 # 5-minute derived164 r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "interval": "5m", "start": "2026-10-30", "end": "2026-10-30"}, headers=H)165 assert r.status_code == 200 and 150 <= r.json()["meta"]["rows_matching"] <= 192166 # unbounded range refused167 r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2020-01-01", "end": "2026-10-30"}, headers=H)168 assert r.status_code == 413 and r.json()["error"]["code"] == "range_too_large"169 # arrow transport170 r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2026-10-30", "end": "2026-10-30", "format": "arrow", "limit": 10000}, headers=H)171 with ipc.open_stream(io.BytesIO(r.content)) as reader:172 t = reader.read_all()173 assert t.num_rows == 900174 # sessions / coverage / quality175 assert c.get("/api/v1/market/sessions", params={"symbol": "AAPL"}, headers=H).json()["data"][0]["bar_count"] == 900176 assert c.get("/api/v1/coverage/heatmap", params={"start": "2026-10-01", "end": "2026-10-31"}, headers=H).json()["sessions"][0]["symbols"] == 3177 assert "codes" in c.get("/api/v1/quality", headers=H).json()178 assert c.get("/api/v1/instruments", params={"query": "aap"}, headers=H).json()["data"][0]["symbol"] == "AAPL"179180181def test_fundamentals_as_of_hides_later_filings_and_restatements(app_client, api_key):182 c = app_client183 H = {"Authorization": f"Bearer {api_key}"}184 with transaction() as conn:185 conn.execute("DELETE FROM fundamentals_facts WHERE entity_id='frd:TESTCO'")186 # Q1 filed 2025-05-01 (available 2025-05-02 04:00Z), Q2 filed 2025-08-01 ; Q1 'rev' restated later (observed 2026-01-15)187 rows = [188 ("rev", 100.0, date(2025, 3, 31), date(2025, 5, 1), datetime(2025, 5, 2, 4, tzinfo=UTC), datetime(2025, 5, 2, 12, tzinfo=UTC), 1, False, datetime(2026, 1, 15, tzinfo=UTC)),189 ("rev", 105.0, date(2025, 3, 31), date(2025, 5, 1), datetime(2025, 5, 2, 4, tzinfo=UTC), datetime(2026, 1, 15, tzinfo=UTC), 2, True, None),190 ("rev", 120.0, date(2025, 6, 30), date(2025, 8, 1), datetime(2025, 8, 2, 4, tzinfo=UTC), datetime(2025, 8, 2, 12, tzinfo=UTC), 1, False, None),191 ]192 for m, v, pe, fd, av, ob, rev, rs, sup in rows:193 conn.execute("INSERT INTO fundamentals_facts(entity_id, metric, value, unit, fiscal_period_end, fiscal_period_type, filing_date, available_at, observed_at, source_id, revision_id, restatement_flag, superseded_at) "194 "VALUES ('frd:TESTCO',%s,%s,'USD',%s,'q',%s,%s,%s,'firstrate_fundamentals',%s,%s,%s)", (m, v, pe, fd, av, ob, rev, rs, sup))195 # as of 2025-06-01: only Q1, original value 100 (restatement observed 2026-01 must not leak)196 r = c.get("/api/v1/fundamentals/TESTCO", params={"as_of": "2025-06-01T00:00:00Z", "metrics": "rev"}, headers=H)197 assert r.status_code == 200, r.text198 d = r.json()["data"]199 assert len(d) == 1 and d[0]["value"] == 100.0 and d[0]["revision_id"] == 1 and r.json()["meta"]["facts_hidden_by_as_of"] == 1200 # as of 2025-08-01 (filing day itself): Q2 not yet available (policy: next day)201 d = c.get("/api/v1/fundamentals/TESTCO", params={"as_of": "2025-08-01T20:00:00Z", "metrics": "rev"}, headers=H).json()["data"]202 assert [x["fiscal_period_end"] for x in d] == ["2025-03-31"]203 # as of 2025-08-02 05:00Z: Q2 visible204 d = c.get("/api/v1/fundamentals/TESTCO", params={"as_of": "2025-08-02T05:00:00Z", "metrics": "rev"}, headers=H).json()["data"]205 assert sorted(x["value"] for x in d) == [100.0, 120.0]206 # as of now: restated Q1 value 105 (revision 2) is the current knowledge207 d = c.get("/api/v1/fundamentals/TESTCO", params={"metrics": "rev"}, headers=H).json()["data"]208 assert sorted(x["value"] for x in d) == [105.0, 120.0] and any(x["restatement_flag"] for x in d)209210211def test_query_sandbox_and_quotas(app_client, api_key):212 c = app_client213 H = {"Authorization": f"Bearer {api_key}"}214 r = c.post("/api/v1/query", json={"sql": "SELECT symbol_as_reported s, count(*) n FROM market_bars_stock_1m_raw GROUP BY 1 ORDER BY 1", "max_rows": 10}, headers=H)215 assert r.status_code == 200, r.text216 rows = {x["s"]: x["n"] for x in r.json()["rows"]}217 assert rows["AAPL"] == 900 and r.json()["meta"]["engine"].startswith("duckdb")218 for bad in ["DROP TABLE x", "SELECT 1; SELECT 2", "SELECT * FROM read_csv('/etc/passwd')", "COPY (SELECT 1) TO '/tmp/x'"]:219 r = c.post("/api/v1/query", json={"sql": bad}, headers=H)220 assert r.status_code == 400 and r.json()["error"]["code"] in ("query_rejected", "query_error"), bad221 with connection() as conn:222 u = conn.execute("SELECT * FROM usage_daily WHERE principal LIKE 'api_key:%'").fetchall()223 assert u and u[0]["requests"] > 0 and u[0]["rows_returned"] > 0224225226def test_export_roundtrip_and_sdk(app_client, api_key):227 c = app_client228 from findata_sdk import FinData229 import threading230 import uvicorn231 # the sync SDK needs a real HTTP server: run uvicorn on a loopback port in a thread232 config = uvicorn.Config(c.app, host="127.0.0.1", port=8699, log_level="warning")233 server = uvicorn.Server(config)234 th = threading.Thread(target=server.run, daemon=True)235 th.start()236 for _ in range(100):237 if server.started:238 break239 time.sleep(0.05)240 sdk = FinData("http://127.0.0.1:8699", api_key=api_key)241 t = sdk.bars("AAPL", "1m", "2026-10-30", "2026-10-30")242 assert t.num_rows == 900 and t.column("close")[0].as_py() == 351.6499243 assert sdk.datasets() and sdk.health()["status"] in ("ok", "warning", "degraded")244 f = sdk.fundamentals("TESTCO", as_of="2025-06-01T00:00:00Z", metrics=["rev"])245 assert f["data"][0]["value"] == 100.0246 out = sdk.export(["AAPL", "A"], "2026-10-01", "2026-10-30", dest="/tmp/findata-test-export.parquet", poll_s=0.2, timeout_s=60)247 pt = pq.read_table(out)248 assert pt.num_rows == 900 + 390 and set(pt.column("symbol_as_reported").unique().to_pylist()) == {"AAPL", "A"}249 H = {"Authorization": f"Bearer {api_key}"}250 lst = c.get("/api/v1/exports", headers=H).json()["data"]251 assert lst[0]["status"] == "ready" and lst[0]["snapshot"]["dataset_id"] == "market_bars.stock.1m.raw" and lst[0]["expires_at"]252 server.should_exit = True253 th.join(timeout=5)254255256def test_vendor_secret_never_in_responses_or_openapi(app_client, api_key):257 c = app_client258 H = {"Authorization": f"Bearer {api_key}"}259 csrf = _login(c)260 blobs = [c.get("/api/v1/openapi.json").text, c.get("/api/v1/catalog/datasets", headers=H).text, c.get("/api/v1/catalog/sources", headers=H).text,261 c.get("/api/v1/admin/settings").text, c.get("/api/v1/admin/storage").text, c.get("/api/v1/admin/audit").text, c.get("/api/v1/admin/jobs").text,262 c.get("/api/v1/health").text, c.get("/api/v1/health/summary").text]263 for b in blobs:264 assert TEST_USERID not in b265 # raw_archives request payloads are sanitized266 with connection() as conn:267 raws = conn.execute("SELECT request::text r FROM raw_archives").fetchall()268 jobs = conn.execute("SELECT coalesce(last_error,'') e, result::text r FROM ingest_jobs").fetchall()269 assert all(TEST_USERID not in x["r"] for x in raws) and all(TEST_USERID not in x["e"] + x["r"] for x in jobs)270 # security headers present271 r = c.get("/api/v1/health")272 assert r.headers["X-Content-Type-Options"] == "nosniff" and r.headers["X-Frame-Options"] == "DENY" and "default-src 'self'" in r.headers["Content-Security-Policy"]273 assert c.get("/api/docs").status_code == 200274