"""API + SDK end-to-end against the in-process app (TestClient) with data published by the ingest task."""
from __future__ import annotations

import io
import json
import time
from datetime import UTC, date, datetime

import httpx
import pyarrow.ipc as ipc
import pyarrow.parquet as pq
import pyotp
import pytest
from fastapi.testclient import TestClient

from findata_api import auth as A
from findata_core.db import connection, fetch_one, transaction
from findata_ingest import queue
from findata_ingest.tasks import incremental, metadata
from tests.conftest import TEST_USERID

OWNER = "owner@example.com"
PW = "correct horse battery staple"


@pytest.fixture(scope="module")
def app_client(vendor_module, lake_module):
    from findata_api.main import create_app
    # make sure one session of bars exists (tests may run in isolation)
    n = fetch_one("SELECT count(*) AS n FROM lake_files WHERE dataset_id='market_bars.stock.1m.raw' AND version_removed IS NULL")["n"]
    if n == 0:
        class L:
            def __getattr__(self, _):
                return lambda *a, **k: None
            def bind(self, **k):
                return self
        jid = queue.enqueue("metadata", None, {"types": ["stock", "etf"]})
        metadata.run(queue.claim_next(["metadata"]), lake_module, L())
        queue.enqueue("incremental", "market_bars.stock.1m.raw", {"type": "stock", "period": "day", "force": True})
        incremental.run(queue.claim_next(["incremental"]), lake_module, L())
    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:
        from findata_ingest.tasks import derive
        class L2:
            def __getattr__(self, _):
                return lambda *a, **k: None
            def bind(self, **k):
                return self
        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"]))
        derive.run(j, lake_module, L2())
    A.create_owner(OWNER, PW)
    app = create_app()
    with TestClient(app, base_url="http://testserver") as c:
        yield c


@pytest.fixture(scope="module")
def vendor_module():
    from tests.conftest import VendorMock
    import findata_market_firstrate.client as C
    import findata_fundamentals.firstrate as F
    vm = VendorMock()
    old = (C.DEFAULT_TRANSPORT, F.DEFAULT_TRANSPORT)
    C.DEFAULT_TRANSPORT = httpx.MockTransport(vm)
    F.DEFAULT_TRANSPORT = httpx.MockTransport(vm)
    yield vm
    C.DEFAULT_TRANSPORT, F.DEFAULT_TRANSPORT = old


@pytest.fixture(scope="module")
def lake_module():
    from findata_ingest.lake import Lake
    return Lake()


def _login(c: TestClient) -> str:
    r = c.post("/api/v1/auth/login", json={"email": OWNER, "password": PW})
    assert r.status_code == 200, r.text
    return r.json()["csrf_token"]


def test_unauthenticated_access_fails(app_client):
    c = app_client
    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"]:
        r = c.get(path)
        assert r.status_code == 401, path
        assert r.json()["error"]["code"] == "unauthenticated" and r.headers["x-request-id"]
    assert c.post("/api/v1/query", json={"sql": "select 1"}).status_code == 401
    assert c.get("/api/v1/health").json() == {"status": c.get("/api/v1/health").json()["status"], "service": "findata-os"}
    r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2026-10-30", "end": "2026-10-30"}, headers={"Authorization": "Bearer fd_bad.token"})
    assert r.status_code == 401 and r.json()["error"]["code"] == "invalid_api_key"


def test_login_bad_password_then_good_then_csrf(app_client):
    c = app_client
    r = c.post("/api/v1/auth/login", json={"email": OWNER, "password": "nope"})
    assert r.status_code == 401
    csrf = _login(c)
    me = c.get("/api/v1/auth/me").json()
    assert me["email"] == OWNER and me["role"] == "owner" and "admin" in me["scopes"] and me["totp_enabled"] is False
    # state-changing call without CSRF header is refused
    r = c.post("/api/v1/admin/jobs", json={"kind": "quality"})
    assert r.status_code == 403 and r.json()["error"]["code"] == "csrf"
    r = c.post("/api/v1/admin/jobs", json={"kind": "quality"}, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 201
    with transaction() as conn:
        conn.execute("UPDATE ingest_jobs SET status='cancelled' WHERE kind='quality' AND status='queued'")


def test_totp_enrolment_and_mfa_gate(app_client):
    c = app_client
    csrf = _login(c)
    r = c.post("/api/v1/auth/totp/setup", headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200 and r.json()["otpauth_uri"].startswith("otpauth://totp/")
    secret = r.json()["secret"]
    r = c.post("/api/v1/auth/totp/verify", json={"code": "000000"}, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    r = c.post("/api/v1/auth/totp/verify", json={"code": pyotp.TOTP(secret).now()}, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200 and r.json()["totp_enabled"]
    # new login now requires MFA
    r = c.post("/api/v1/auth/login", json={"email": OWNER, "password": PW})
    assert r.json()["mfa_required"] is True
    csrf = r.json()["csrf_token"]
    assert c.get("/api/v1/catalog/datasets").status_code == 401  # mfa_required
    r = c.post("/api/v1/auth/mfa", json={"code": "123456"}, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 401
    r = c.post("/api/v1/auth/mfa", json={"code": pyotp.TOTP(secret).now()}, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200
    assert c.get("/api/v1/catalog/datasets").status_code == 200
    # disable TOTP again for the remaining tests (keep the flow simple)
    with transaction() as conn:
        conn.execute("UPDATE users SET totp_enabled=false WHERE email=%s", (OWNER,))


@pytest.fixture(scope="module")
def api_key(app_client):
    csrf = _login(app_client)
    r = app_client.post("/api/v1/admin/keys", json={"name": "sdk-test", "scopes": ["read:market", "read:fundamentals", "query", "export"]},
                        headers={"X-CSRF-Token": csrf})
    assert r.status_code == 201, r.text
    tok = r.json()["token"]
    app_client.cookies.clear()
    return tok


def test_api_key_scopes_and_bars(app_client, api_key):
    c = app_client
    H = {"Authorization": f"Bearer {api_key}"}
    assert c.get("/api/v1/admin/jobs", headers=H).status_code == 403  # no admin scope
    r = c.get("/api/v1/catalog/datasets", headers=H)
    assert r.status_code == 200
    ds = {d["id"]: d for d in r.json()["data"]}
    assert ds["market_bars.stock.1m.raw"]["status"] == "active" and ds["market_ticks"]["status"] == "unsupported"
    assert ds["fundamentals.sec_companyfacts"]["status"] == "pending_configuration"
    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"]
    r = c.get("/api/v1/market/bars", params={"symbol": "aapl", "interval": "1m", "start": "2026-10-30", "end": "2026-10-30", "limit": 400}, headers=H)
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["meta"]["rows"] == 400 and body["meta"]["rows_matching"] == 900 and body["next_cursor"]
    assert body["data"][0]["timestamp_utc"] == "2026-10-30T08:00:00Z" and body["meta"]["timestamp_convention"] == "bar_open"
    assert r.headers["X-Dataset-Version"].isdigit() and r.headers["X-Dataset-Id"] == "market_bars.stock.1m.raw"
    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)
    assert r2.json()["data"][0]["timestamp_utc"] > body["data"][-1]["timestamp_utc"]
    # 5-minute derived
    r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "interval": "5m", "start": "2026-10-30", "end": "2026-10-30"}, headers=H)
    assert r.status_code == 200 and 150 <= r.json()["meta"]["rows_matching"] <= 192
    # unbounded range refused
    r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2020-01-01", "end": "2026-10-30"}, headers=H)
    assert r.status_code == 413 and r.json()["error"]["code"] == "range_too_large"
    # arrow transport
    r = c.get("/api/v1/market/bars", params={"symbol": "AAPL", "start": "2026-10-30", "end": "2026-10-30", "format": "arrow", "limit": 10000}, headers=H)
    with ipc.open_stream(io.BytesIO(r.content)) as reader:
        t = reader.read_all()
    assert t.num_rows == 900
    # sessions / coverage / quality
    assert c.get("/api/v1/market/sessions", params={"symbol": "AAPL"}, headers=H).json()["data"][0]["bar_count"] == 900
    assert c.get("/api/v1/coverage/heatmap", params={"start": "2026-10-01", "end": "2026-10-31"}, headers=H).json()["sessions"][0]["symbols"] == 3
    assert "codes" in c.get("/api/v1/quality", headers=H).json()
    assert c.get("/api/v1/instruments", params={"query": "aap"}, headers=H).json()["data"][0]["symbol"] == "AAPL"


def test_fundamentals_as_of_hides_later_filings_and_restatements(app_client, api_key):
    c = app_client
    H = {"Authorization": f"Bearer {api_key}"}
    with transaction() as conn:
        conn.execute("DELETE FROM fundamentals_facts WHERE entity_id='frd:TESTCO'")
        # Q1 filed 2025-05-01 (available 2025-05-02 04:00Z), Q2 filed 2025-08-01 ; Q1 'rev' restated later (observed 2026-01-15)
        rows = [
            ("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)),
            ("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),
            ("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),
        ]
        for m, v, pe, fd, av, ob, rev, rs, sup in rows:
            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) "
                         "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))
    # as of 2025-06-01: only Q1, original value 100 (restatement observed 2026-01 must not leak)
    r = c.get("/api/v1/fundamentals/TESTCO", params={"as_of": "2025-06-01T00:00:00Z", "metrics": "rev"}, headers=H)
    assert r.status_code == 200, r.text
    d = r.json()["data"]
    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"] == 1
    # as of 2025-08-01 (filing day itself): Q2 not yet available (policy: next day)
    d = c.get("/api/v1/fundamentals/TESTCO", params={"as_of": "2025-08-01T20:00:00Z", "metrics": "rev"}, headers=H).json()["data"]
    assert [x["fiscal_period_end"] for x in d] == ["2025-03-31"]
    # as of 2025-08-02 05:00Z: Q2 visible
    d = c.get("/api/v1/fundamentals/TESTCO", params={"as_of": "2025-08-02T05:00:00Z", "metrics": "rev"}, headers=H).json()["data"]
    assert sorted(x["value"] for x in d) == [100.0, 120.0]
    # as of now: restated Q1 value 105 (revision 2) is the current knowledge
    d = c.get("/api/v1/fundamentals/TESTCO", params={"metrics": "rev"}, headers=H).json()["data"]
    assert sorted(x["value"] for x in d) == [105.0, 120.0] and any(x["restatement_flag"] for x in d)


def test_query_sandbox_and_quotas(app_client, api_key):
    c = app_client
    H = {"Authorization": f"Bearer {api_key}"}
    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)
    assert r.status_code == 200, r.text
    rows = {x["s"]: x["n"] for x in r.json()["rows"]}
    assert rows["AAPL"] == 900 and r.json()["meta"]["engine"].startswith("duckdb")
    for bad in ["DROP TABLE x", "SELECT 1; SELECT 2", "SELECT * FROM read_csv('/etc/passwd')", "COPY (SELECT 1) TO '/tmp/x'"]:
        r = c.post("/api/v1/query", json={"sql": bad}, headers=H)
        assert r.status_code == 400 and r.json()["error"]["code"] in ("query_rejected", "query_error"), bad
    with connection() as conn:
        u = conn.execute("SELECT * FROM usage_daily WHERE principal LIKE 'api_key:%'").fetchall()
    assert u and u[0]["requests"] > 0 and u[0]["rows_returned"] > 0


def test_export_roundtrip_and_sdk(app_client, api_key):
    c = app_client
    from findata_sdk import FinData
    import threading
    import uvicorn
    # the sync SDK needs a real HTTP server: run uvicorn on a loopback port in a thread
    config = uvicorn.Config(c.app, host="127.0.0.1", port=8699, log_level="warning")
    server = uvicorn.Server(config)
    th = threading.Thread(target=server.run, daemon=True)
    th.start()
    for _ in range(100):
        if server.started:
            break
        time.sleep(0.05)
    sdk = FinData("http://127.0.0.1:8699", api_key=api_key)
    t = sdk.bars("AAPL", "1m", "2026-10-30", "2026-10-30")
    assert t.num_rows == 900 and t.column("close")[0].as_py() == 351.6499
    assert sdk.datasets() and sdk.health()["status"] in ("ok", "warning", "degraded")
    f = sdk.fundamentals("TESTCO", as_of="2025-06-01T00:00:00Z", metrics=["rev"])
    assert f["data"][0]["value"] == 100.0
    out = sdk.export(["AAPL", "A"], "2026-10-01", "2026-10-30", dest="/tmp/findata-test-export.parquet", poll_s=0.2, timeout_s=60)
    pt = pq.read_table(out)
    assert pt.num_rows == 900 + 390 and set(pt.column("symbol_as_reported").unique().to_pylist()) == {"AAPL", "A"}
    H = {"Authorization": f"Bearer {api_key}"}
    lst = c.get("/api/v1/exports", headers=H).json()["data"]
    assert lst[0]["status"] == "ready" and lst[0]["snapshot"]["dataset_id"] == "market_bars.stock.1m.raw" and lst[0]["expires_at"]
    server.should_exit = True
    th.join(timeout=5)


def test_vendor_secret_never_in_responses_or_openapi(app_client, api_key):
    c = app_client
    H = {"Authorization": f"Bearer {api_key}"}
    csrf = _login(c)
    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,
             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,
             c.get("/api/v1/health").text, c.get("/api/v1/health/summary").text]
    for b in blobs:
        assert TEST_USERID not in b
    # raw_archives request payloads are sanitized
    with connection() as conn:
        raws = conn.execute("SELECT request::text r FROM raw_archives").fetchall()
        jobs = conn.execute("SELECT coalesce(last_error,'') e, result::text r FROM ingest_jobs").fetchall()
    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)
    # security headers present
    r = c.get("/api/v1/health")
    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"]
    assert c.get("/api/docs").status_code == 200