"""End-to-end business tests: real persistence/runtime, only the platform HTTP is replaced."""

import asyncio

import httpx
import pytest
from sqlalchemy import func, select

from app.backtests.contracts import SimulationSettings
from app.models import Account, Alpha, BacktestResult, BacktestRun, Research, SimulationAttempt
from app.security import cipher
from app.worldquant import WqClient
from tests.backtest_fake import Platform

PREFIX = "/api/v1/backtests"
PARAMS = SimulationSettings(region="USA", universe="TOP3000", delay=1).model_dump()


def candidate(index=0, **settings):
    return {
        "client_item_id": f"item-{index}",
        "expression": f"rank(close) + {index}",
        "settings": PARAMS | settings,
    }


async def setup(app):
    platform = Platform()
    runner = app.state.runner
    await runner.client.close()
    runner.client = WqClient(app.state.settings, transport=httpx.MockTransport(platform))
    runner.backtests.client = runner.client
    runner.backtests.poll_interval = 0
    async with app.state.sessions.begin() as db:
        account = await db.get(Account, 1)
        account.email, account.wq_user_id, account.connection_status = (
            "synthetic@example.com",
            "TEST_USER",
            "connected",
        )
        account.password_encrypted = cipher(app.state.settings).encrypt(platform.secret.encode()).decode()
    return platform, runner.backtests


async def preview(client, candidates=None):
    response = await client.post(
        f"{PREFIX}/previews",
        json={
            "inline": {
                "name": "测试研究",
                "source": {"kind": "test"},
                "candidates": candidates or [candidate()],
            }
        },
    )
    assert response.status_code == 201, response.text
    return response.json()


async def start(client, p, key="request-1"):
    response = await client.post(
        f"{PREFIX}/runs",
        json={"preview_id": p["preview_id"], "version": p["version"], "idempotency_key": key},
    )
    assert response.status_code == 202, response.text
    return response.json()


async def execute(app, lane, run_id):
    async with app.state.sessions() as db:
        ids = list(
            await db.scalars(
                select(SimulationAttempt.id)
                .where(SimulationAttempt.run_id == run_id)
                .order_by(SimulationAttempt.ordinal)
            )
        )
    for aid in ids:
        await lane.step(aid)
        await lane.step(aid)
    return ids


async def test_fixed_preview_grouping_mapping_and_history(app, logged_in):
    platform, lane = await setup(app)
    p = await preview(logged_in, [candidate(0), candidate(1, universe="TOP1000"), candidate(2, delay=0)])
    assert p["batch_count"] == 2 and p["total"] == 3
    run = await start(logged_in, p)
    again = await start(logged_in, p)
    assert run["backtest_run_id"] == again["backtest_run_id"]
    rid = run["backtest_run_id"]
    await execute(app, lane, rid)
    data = (await logged_in.get(f"{PREFIX}/runs/{rid}/results")).json()
    assert len(platform.posts) == 2
    assert all(i["persistence_status"] == "saved" for i in data["items"]), data
    for item in data["items"]:
        assert item["result"]["snapshot"]["regular"]["code"] == item["expression"]
        assert item["result"]["snapshot"]["settings"] == item["settings"]
        assert item["result"]["snapshot"]["is"]["sharpe"] is None
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    item = data["items"][0]
    async with app.state.sessions.begin() as db:
        alpha = await db.get(Alpha, item["alpha_id"])
        alpha.is_metrics = {"sharpe": 999}
        assert await db.get(Research, alpha.id)
    historical = (await logged_in.get(f"{PREFIX}/runs/{rid}/results")).json()
    assert historical["items"][0]["result"]["snapshot"]["is"]["sharpe"] is None
    events = (await logged_in.get(f"{PREFIX}/runs/{rid}/events?limit=2")).json()
    later = (await logged_in.get(f"{PREFIX}/runs/{rid}/events?after={events['next_cursor']}")).json()
    assert events["has_more"] and later["items"][0]["seq"] > events["next_cursor"]
    assert (await preview(logged_in))["duplicate_count"] == 1


@pytest.mark.parametrize("rejection", ["unknown", "missing_location"])
async def test_unknown_submission_never_reposted(app, logged_in, rejection):
    platform, lane = await setup(app)
    platform.reject = rejection
    run = await start(logged_in, await preview(logged_in))
    rid = run["backtest_run_id"]
    ids = await execute(app, lane, rid)
    # A process crash/recovery must not turn an unknown POST into queued work.
    await lane.start()
    await lane.stop()
    response = await logged_in.post(f"{PREFIX}/runs/{rid}/control", json={"action": "recover", "version": 1})
    assert response.json()["status"] == "needs_review"
    async with app.state.sessions() as db:
        assert (await db.get(SimulationAttempt, ids[0])).state == "needs_review"
    assert len(platform.posts) == 1


async def test_partial_failure_and_rerun_only_selected(app, logged_in):
    platform, lane = await setup(app)
    platform.fail_child = 0
    run = await start(logged_in, await preview(logged_in, [candidate(0), candidate(1)]))
    rid = run["backtest_run_id"]
    await execute(app, lane, rid)
    result = (await logged_in.get(f"{PREFIX}/runs/{rid}/results")).json()["items"]
    assert result[0]["platform_status"] == "failed" and result[1]["persistence_status"] == "saved"
    rerun = await logged_in.post(f"{PREFIX}/runs/{rid}/rerun-preview", json={"item_ids": [result[0]["id"]]})
    assert rerun.status_code == 201
    assert rerun.json()["total"] == 1 and rerun.json()["source"]["parent_run_id"] == rid
    assert len(platform.posts) == 1


async def test_detail_failure_recovers_without_resubmit(app, logged_in):
    platform, lane = await setup(app)
    platform.detail_fail = True
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    ids = await execute(app, lane, rid)
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "needs_review"
    platform.detail_fail = False
    await logged_in.post(f"{PREFIX}/runs/{rid}/control", json={"action": "recover", "version": 1})
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    assert len(platform.posts) == 1


async def test_draft_version_snapshot_and_pause_stop(app, logged_in):
    platform, lane = await setup(app)
    body = {"name": "草稿", "candidates": [candidate(0), candidate(1, delay=0)]}
    d = (await logged_in.post(f"{PREFIX}/drafts", json=body)).json()
    p = (await logged_in.post(f"{PREFIX}/previews", json={"draft_id": d["id"], "draft_version": 1})).json()
    changed = await logged_in.put(
        f"{PREFIX}/drafts/{d['id']}", json=body | {"version": 1, "candidates": [candidate(9)]}
    )
    assert changed.json()["version"] == 2
    assert (
        await logged_in.post(f"{PREFIX}/previews", json={"draft_id": d["id"], "draft_version": 1})
    ).status_code == 409
    run = await start(logged_in, p)
    rid = run["backtest_run_id"]
    async with app.state.sessions() as db:
        ids = list(
            await db.scalars(
                select(SimulationAttempt.id)
                .where(SimulationAttempt.run_id == rid)
                .order_by(SimulationAttempt.ordinal)
            )
        )
    await lane.step(ids[0])
    await logged_in.post(f"{PREFIX}/runs/{rid}/control", json={"action": "pause", "version": 1})
    await lane.step(ids[1])
    await lane.step(ids[0])
    assert len(platform.posts) == 1
    await logged_in.post(f"{PREFIX}/runs/{rid}/control", json={"action": "stop", "version": 2})
    r = (await logged_in.get(f"{PREFIX}/runs/{rid}/results")).json()["items"]
    assert r[0]["persistence_status"] == "saved" and r[1]["platform_status"] == "skipped"
    assert r[0]["expression"] == candidate(0)["expression"]


async def test_batch_missing_child_does_not_misattribute(app, logged_in):
    platform, lane = await setup(app)
    platform.missing = True
    rid = (await start(logged_in, await preview(logged_in, [candidate(0), candidate(1)])))["backtest_run_id"]
    ids = await execute(app, lane, rid)
    items = (await logged_in.get(f"{PREFIX}/runs/{rid}/results")).json()["items"]
    assert items[0]["platform_status"] == "unknown"
    assert items[1]["persistence_status"] == "saved"
    platform.simulations["p1"]["children"] = ["p1c1", "p1c0"]
    await logged_in.post(f"{PREFIX}/runs/{rid}/control", json={"action": "recover", "version": 1})
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    assert len(platform.posts) == 1


async def test_validation_auth_and_idempotency_conflict(app, logged_in, client):
    await setup(app)
    assert (
        await logged_in.post(
            f"{PREFIX}/previews",
            json={"inline": {"name": "x", "candidates": [candidate() | {"alpha_type": "SUPER"}]}},
        )
    ).status_code == 422
    p1, p2 = await preview(logged_in), await preview(logged_in, [candidate(2)])
    await start(logged_in, p1)
    assert (
        await logged_in.post(
            f"{PREFIX}/runs", json={"preview_id": p2["preview_id"], "idempotency_key": "request-1"}
        )
    ).status_code == 409
    assert (await logged_in.get(f"{PREFIX}/runs?limit=101")).status_code == 422
    await client.post("/api/v1/auth/logout")
    assert (await client.get(f"{PREFIX}/runs")).status_code == 401


async def tick(lane):
    await lane.tick()
    await asyncio.gather(*lane.tasks.values(), return_exceptions=False)


async def test_account_budget_round_robin_and_sync_independence(app, logged_in):
    platform, lane = await setup(app)
    assert (
        await logged_in.put(f"{PREFIX}/config", json={"concurrency": 1, "batch_size": 1, "version": 1})
    ).status_code == 200
    r1 = await start(logged_in, await preview(logged_in, [candidate(0), candidate(1)]), "first")
    r2 = await start(logged_in, await preview(logged_in, [candidate(2), candidate(3)]), "second")
    await tick(lane)  # one submission, occupied until remote terminal
    assert len(platform.posts) == 1
    await tick(lane)  # poll first result
    await tick(lane)  # other run gets next slot
    assert len(platform.posts) == 2
    assert platform.posts[0][0]["regular"] == candidate(0)["expression"]
    assert platform.posts[1][0]["regular"] == candidate(2)["expression"]
    platform.pending = True
    sync = await logged_in.post("/api/v1/sync-jobs", json={"kind": "full_sync"})
    await app.state.runner.run_next()
    assert (await logged_in.get(f"/api/v1/sync-jobs/{sync.json()['id']}")).json()["status"] == "completed"
    await logged_in.put(f"{PREFIX}/config", json={"concurrency": 2, "batch_size": 8, "version": 2})
    await tick(lane)
    assert len(platform.posts) == 3
    # Batch sizing of both existing runs remains 1 despite config update.
    assert all(len(p) == 1 for p in platform.posts)
    await logged_in.put(f"{PREFIX}/config", json={"concurrency": 1, "batch_size": 8, "version": 3})
    await tick(lane)
    assert len(platform.posts) == 3
    await lane.interrupt()
    assert r1["batch_size"] == r2["batch_size"] == 1


async def test_rate_limit_and_failed_submit_are_bounded(app, logged_in):
    platform, lane = await setup(app)
    platform.reject = "rate"
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    async with app.state.sessions() as db:
        aid = await db.scalar(select(SimulationAttempt.id).where(SimulationAttempt.run_id == rid))
    for _ in range(app.state.settings.retry_attempts):
        await lane.step(aid)
        await asyncio.sleep(0.02)
    assert len(platform.posts) == app.state.settings.retry_attempts
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed_with_errors"
    assert platform.secret not in (await logged_in.get(f"{PREFIX}/runs/{rid}/attempts")).text


async def test_poll_timeout_and_crash_after_acceptance(app, logged_in):
    platform, lane = await setup(app)
    lane.poll_limit = 1
    platform.pending = True
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    ids = await execute(app, lane, rid)
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "needs_review"
    platform.pending = False
    await logged_in.post(f"{PREFIX}/runs/{rid}/control", json={"action": "recover", "version": 1})
    # Simulate a crash checkpoint with the Location already persisted.
    async with app.state.sessions.begin() as db:
        a = await db.get(SimulationAttempt, ids[0])
        a.state = "submitting"
    await lane.start()
    await lane.stop()
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    assert len(platform.posts) == 1


async def test_result_transaction_failure_recovers_from_saved_receipt(app, logged_in):
    from sqlalchemy import event
    from sqlalchemy.exc import OperationalError

    platform, lane = await setup(app)
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    failed = False

    def fail_once(conn, cursor, statement, parameters, context, executemany):
        nonlocal failed
        if "INSERT INTO backtest_results" in statement and not failed:
            failed = True
            raise OperationalError("synthetic persistence outage", {}, Exception("synthetic"))

    event.listen(app.state.engine.sync_engine, "before_cursor_execute", fail_once)
    try:
        ids = await execute(app, lane, rid)
    finally:
        event.remove(app.state.engine.sync_engine, "before_cursor_execute", fail_once)
    assert failed
    async with app.state.sessions() as db:
        assert await db.scalar(select(func.count()).select_from(BacktestResult)) == 0
        assert await db.scalar(select(func.count()).select_from(Alpha)) == 0
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    assert len(platform.posts) == 1


async def test_ai_fixed_set_confirmation_and_duplicate_decision(app, logged_in):
    from tests.test_ai import configure
    from tests.test_ai import start as start_ai

    platform, lane = await setup(app)
    await configure(app, logged_in)
    _, run, _ = await start_ai(app, logged_in, "回测固定候选")
    assert run["status"] == "waiting_approval", run
    approval = next(c for c in run["tools"] if c["name"] == "start_backtest")
    assert approval["preview"]["backtest"]["total"] == 1
    async with app.state.sessions() as db:
        assert await db.scalar(select(func.count()).select_from(BacktestRun)) == 0
    for _ in range(2):
        response = await logged_in.post(
            f"/api/v1/ai/approvals/{approval['id']}/decision", json={"approved": True}
        )
        assert response.status_code == 200, response.text
    async with app.state.sessions() as db:
        rows = list(await db.scalars(select(BacktestRun)))
        assert len(rows) == 1
        assert rows[0].ai_context["ai_run_id"] == run["id"]
    await logged_in.post(f"/api/v1/ai/runs/{run['id']}/cancel")
    await execute(app, lane, rows[0].id)
    assert len(platform.posts) == 1
    assert (await logged_in.get(f"{PREFIX}/runs/{rows[0].id}")).json()["status"] == "completed"


async def test_duplicate_inputs_are_separate_attempts_and_share_alpha_safely(app, logged_in):
    platform, lane = await setup(app)
    platform.existing_alpha_ids = ["shared_alpha"]
    p = await preview(logged_in, [candidate(0), candidate(0) | {"client_item_id": "other-experiment"}])
    assert p["batch_count"] == 2 and p["duplicate_count"] == 1
    rid = (await start(logged_in, p))["backtest_run_id"]
    await execute(app, lane, rid)
    async with app.state.sessions() as db:
        assert await db.scalar(select(func.count()).select_from(Alpha)) == 1
        assert await db.scalar(select(func.count()).select_from(BacktestResult)) == 2
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"


async def test_original_reference_recovery_without_new_post(app, logged_in):
    platform, lane = await setup(app)
    platform.reject = "missing_location"
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    ids = await execute(app, lane, rid)
    path = f"{PREFIX}/attempts/{ids[0]}/reference"
    assert (
        await logged_in.post(
            path, json={"progress_url": "https://foreign.example/simulations/p1", "version": 1}
        )
    ).status_code == 422
    linked = await logged_in.post(path, json={"progress_url": "/simulations/p1", "version": 1})
    assert linked.status_code == 200, linked.text
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    assert len(platform.posts) == 1


async def test_preview_subset_uses_whole_snapshot_and_does_not_change_original(app, logged_in):
    await setup(app)
    p = await preview(logged_in, [candidate(i) for i in range(40)])
    subset = await logged_in.post(
        f"{PREFIX}/previews/{p['preview_id']}/subset", json={"exclude_ids": ["item-30"]}
    )
    assert subset.json()["total"] == 39 and subset.json()["preview_id"] != p["preview_id"]
    assert (await logged_in.get(f"{PREFIX}/previews/{p['preview_id']}")).json()["total"] == 40


async def test_session_reauthentication_does_not_retry_accepted_submission(app, logged_in):
    platform, lane = await setup(app)
    platform.reject = "session"
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    ids = await execute(app, lane, rid)
    await lane.step(ids[0])
    assert (await logged_in.get(f"{PREFIX}/runs/{rid}")).json()["status"] == "completed"
    assert len(platform.posts) == 2  # first explicitly rejected with 401, second accepted


async def test_terminal_detail_failure_releases_slot_but_keeps_platform_success(app, logged_in):
    platform, lane = await setup(app)
    await logged_in.put(f"{PREFIX}/config", json={"concurrency": 1, "batch_size": 1, "version": 1})
    platform.detail_fail = True
    rid = (await start(logged_in, await preview(logged_in)))["backtest_run_id"]
    await execute(app, lane, rid)
    item = (await logged_in.get(f"{PREFIX}/runs/{rid}/results")).json()["items"][0]
    assert item["platform_status"] == "completed" and item["collection_status"] == "failed"
    await start(logged_in, await preview(logged_in, [candidate(2)]), "next")
    await tick(lane)
    assert len(platform.posts) == 2
    await lane.interrupt()


async def test_run_list_projects_source_metadata_without_accepting_unknown_input(app, logged_in):
    """Persisted source metadata must not turn the entire run list into a 500."""
    await setup(app)
    created = await start(logged_in, await preview(logged_in))
    async with app.state.sessions.begin() as db:
        row = await db.get(BacktestRun, created["backtest_run_id"])
        row.source = {**row.source, "template_input_id": None, "extra_metadata": "not-an-input"}
    response = await logged_in.get(f"{PREFIX}/runs?offset=0&source=")
    assert response.status_code == 200
    source = response.json()["items"][0]["source"]
    assert source["kind"] == "test"
    assert "template_input_id" not in source and "extra_metadata" not in source
    assert source["input_snapshot_ids"] == []
    detail = await logged_in.get(f"{PREFIX}/runs/{created['backtest_run_id']}")
    assert detail.status_code == 200
    assert detail.json()["source"] == source
    rejected = await logged_in.post(
        f"{PREFIX}/drafts",
        json={
            "name": "invalid",
            "source": {"kind": "manual", "extra_metadata": "invalid"},
            "candidates": [candidate()],
        },
    )
    assert rejected.status_code == 422