1"""API + SDK end-to-end against the in-process app (TestClient) with data published by the ingest task."""2from __future__ import annotations34import io5import time6from datetime import UTC, date, datetime78import httpx9import pyarrow.parquet as pq10import pyotp11import pytest12from fastapi.testclient import TestClient13from findata_api import auth as A14from findata_core.db import connection, fetch_one, transaction15from findata_ingest import queue16from findata_ingest.tasks import incremental, metadata17from pyarrow import ipc1819from tests.conftest import TEST_USERID2021OWNER = "owner@example.com"22PW = "correct horse battery staple"232425@pytest.fixture(scope="module")26def app_client(vendor_module, lake_module):27 from findata_api.main import create_app28 # make sure one session of bars exists (tests may run in isolation)29 n = fetch_one("SELECT count(*) AS n FROM lake_files WHERE dataset_id='market_bars.stock.1m.raw' AND version_removed IS NULL")["n"]30 if n == 0:31 class L:32 def __getattr__(self, _):33 return lambda *a, **k: None34 def bind(self, **k):35 return self36 jid = queue.enqueue("metadata", None, {"types": ["stock", "etf"]})37 metadata.run(queue.claim_next(["metadata"]), lake_module, L())38 queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day", "force": True})39 incremental.run(queue.claim_next(["incremental"]), lake_module, L())40 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:41 from findata_ingest.tasks import derive42 class L2:43 def __getattr__(self, _):44 return lambda *a, **k: None45 def bind(self, **k):46 return self47 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"]))48 derive.run(j, lake_module, L2())49 A.create_owner(OWNER, PW)50 app = create_app()51 with TestClient(app, base_url="http://testserver") as c:52 yield c535455@pytest.fixture(scope="module")56def vendor_module():57 import findata_fundamentals.firstrate as F58 import findata_market_firstrate.client as C5960 from tests.conftest import VendorMock61 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 import threading229230 import uvicorn231 from findata_sdk import FinData232 # the sync SDK needs a real HTTP server: run uvicorn on a loopback port in a thread233 config = uvicorn.Config(c.app, host="127.0.0.1", port=8699, log_level="warning")234 server = uvicorn.Server(config)235 th = threading.Thread(target=server.run, daemon=True)236 th.start()237 for _ in range(100):238 if server.started:239 break240 time.sleep(0.05)241 sdk = FinData("http://127.0.0.1:8699", api_key=api_key)242 t = sdk.bars("AAPL", "1m", "2026-10-30", "2026-10-30")243 assert t.num_rows == 900 and t.column("close")[0].as_py() == 351.6499244 assert sdk.datasets() and sdk.health()["status"] in ("ok", "warning", "degraded")245 f = sdk.fundamentals("TESTCO", as_of="2025-06-01T00:00:00Z", metrics=["rev"])246 assert f["data"][0]["value"] == 100.0247 out = sdk.export(["AAPL", "A"], "2026-10-01", "2026-10-30", dest="/tmp/findata-test-export.parquet", poll_s=0.2, timeout_s=60)248 pt = pq.read_table(out)249 assert pt.num_rows == 900 + 390 and set(pt.column("symbol_as_reported").unique().to_pylist()) == {"AAPL", "A"}250 H = {"Authorization": f"Bearer {api_key}"}251 lst = c.get("/api/v1/exports", headers=H).json()["data"]252 assert lst[0]["status"] == "ready" and lst[0]["snapshot"]["dataset_id"] == "market_bars.stock.1m.raw" and lst[0]["expires_at"]253 server.should_exit = True254 th.join(timeout=5)255256257def test_vendor_secret_never_in_responses_or_openapi(app_client, api_key):258 c = app_client259 H = {"Authorization": f"Bearer {api_key}"}260 csrf = _login(c)261 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,262 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,263 c.get("/api/v1/health").text, c.get("/api/v1/health/summary").text]264 for b in blobs:265 assert TEST_USERID not in b266 # raw_archives request payloads are sanitized267 with connection() as conn:268 raws = conn.execute("SELECT request::text r FROM raw_archives").fetchall()269 jobs = conn.execute("SELECT coalesce(last_error,'') e, result::text r FROM ingest_jobs").fetchall()270 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)271 # security headers present272 r = c.get("/api/v1/health")273 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"]274 assert c.get("/api/docs").status_code == 200275276277def test_assistant_pending_without_key_and_validated_with_stub(app_client, api_key, monkeypatch):278 c = app_client279 H = {"Authorization": f"Bearer {api_key}"}280 st = c.get("/api/v1/assistant/status", headers=H).json()281 assert st["status"] == "pending_configuration"282 r = c.post("/api/v1/assistant", json={"question": "average AAPL close on 2026-10-30"}, headers=H)283 assert r.status_code == 409 and r.json()["error"]["code"] == "pending_configuration"284 # stub the provider: the platform must still validate/execute the SQL itself and refuse unsafe statements285 from findata_api.routers import assistant as AS286 from findata_core.config import get_settings287 from pydantic import SecretStr288 monkeypatch.setattr(get_settings(), "anthropic_api_key", SecretStr("sk-test"))289 monkeypatch.setattr(AS, "_ask_model", lambda q, ctx: {"sql": "SELECT avg(close) AS avg_close FROM market_bars_stock_1m_raw WHERE symbol_as_reported='AAPL' AND session_date=DATE '2026-10-30'", "explanation": "mean of 1m closes", "assumptions": ["raw prices"], "refusal": None, "_usage": {"input_tokens": 1, "output_tokens": 1}})290 r = c.post("/api/v1/assistant", json={"question": "average AAPL close on 2026-10-30"}, headers=H)291 assert r.status_code == 200, r.text292 b = r.json()293 assert b["executed"] and 300 < b["rows"][0]["avg_close"] < 400 and "plan" in b and b["dataset_versions"]["market_bars.stock.1m.raw"] >= 1294 monkeypatch.setattr(AS, "_ask_model", lambda q, ctx: {"sql": "COPY (SELECT 1) TO '/tmp/x'", "explanation": "", "assumptions": [], "refusal": None})295 b = c.post("/api/v1/assistant", json={"question": "exfiltrate"}, headers=H).json()296 assert not b["executed"] and "rejected by the sandbox" in b["refusal"]297 monkeypatch.setattr(AS, "_ask_model", lambda q, ctx: {"sql": None, "explanation": "", "assumptions": [], "refusal": "Options data is not loaded"})298 b = c.post("/api/v1/assistant", json={"question": "SPY option chain"}, headers=H).json()299 assert b["sql"] is None and "not loaded" in b["refusal"]300 monkeypatch.setattr(get_settings(), "anthropic_api_key", SecretStr(""))301