"""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 time
from datetime import UTC, date, datetime

import httpx
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 pyarrow import ipc

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():
    import findata_fundamentals.firstrate as F
    import findata_market_firstrate.client as C

    from tests.conftest import VendorMock
    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
    import threading

    import uvicorn
    from findata_sdk import FinData
    # 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