"""Tests for the close task model: seeding, instantiation, roll-forward,
aggregation, deadline computation, owner resolution, and the HTTP surface.
"""
from __future__ import annotations

import uuid

import pytest
import pytest_asyncio
from httpx import AsyncClient

from app.core.exceptions import ValidationError
from app.domains.finance import close_service
from app.domains.finance import close_task_service as svc
from app.domains.finance.close_models import (
    RUN_CLOSED,
    RUN_SOFT_CLOSED,
    TASK_COMPLETED,
    TASK_IN_PROGRESS,
    TASK_NOT_STARTED,
    FinanceCloseTask,
)

# ── service-level ──

@pytest.mark.asyncio
async def test_seed_is_idempotent(db_session):
    r1 = await svc.ensure_seeded(db_session)
    assert r1["templates"] == 129
    assert r1["entities"] == 9
    templates = await svc.list_templates(db_session)
    assert len(templates) == 129
    entities = await svc.list_entities(db_session)
    assert {e.code for e in entities} >= {"USP", "TEJ", "Nebula", "Cayman"}
    # Second call creates nothing new.
    r2 = await svc.ensure_seeded(db_session)
    assert r2["templates_created"] == 0
    assert len(await svc.list_templates(db_session)) == 129


@pytest.mark.asyncio
async def test_owner_resolution_via_email_hint(db_session):
    from app.identity.person_models import IdentityPerson

    pid = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=pid, hris_person_id="sf-cici", work_email="chenchao@pingcap.com",
    ))
    await db_session.flush()
    await svc.ensure_seeded(db_session)
    templates = await svc.list_templates(db_session)
    cici = [t for t in templates if t.owner_label == "Cici"]
    assert cici, "expected Cici-owned templates"
    assert all(t.owner_person_id == str(pid) for t in cici)
    # An owner with no email hint (Grace) stays label-only — no guessing.
    grace = [t for t in templates if t.owner_label == "Grace"]
    assert grace and all(t.owner_person_id is None for t in grace)


@pytest.mark.asyncio
async def test_instantiate_and_aggregate(db_session):
    res = await svc.instantiate_from_templates(db_session, "2026-06")
    assert res["created"] == 129
    tasks = await svc.list_tasks(db_session, "2026-06")
    assert len(tasks) == 129
    # "On going" tasks start in progress; others not started.
    ongoing = [t for t in tasks if t.frequency == "ongoing"]
    assert ongoing and all(t.status == TASK_IN_PROGRESS for t in ongoing)
    assert all(t.status == TASK_NOT_STARTED for t in tasks if t.frequency != "ongoing")
    agg = await svc.aggregate(db_session, "2026-06")
    assert agg["overall"]["total"] == 129
    assert {b["process"] for b in agg["by_process"]} >= {"AP", "Payroll", "Reporting"}
    # Per-owner rollup matches the captured owner distribution.
    by_owner = {b["owner_label"]: b["total"] for b in agg["by_owner"]}
    assert by_owner.get("Elaine") == 44 and by_owner.get("Huan") == 36
    # Idempotent re-instantiate.
    assert (await svc.instantiate_from_templates(db_session, "2026-06"))["created"] == 0


@pytest.mark.asyncio
async def test_deadlines_match_calendar(db_session):
    await svc.instantiate_from_templates(db_session, "2026-06")
    tasks = {t.instance_key: t for t in await svc.list_tasks(db_session, "2026-06")}
    # ap-02 = Day-2 -> Jun 29; rr-02 = Day+1 -> Jul 1; ap-16 = Day+5 -> Jul 6.
    assert str(tasks["ap-02"].deadline) == "2026-06-29"
    assert str(tasks["rr-02"].deadline) == "2026-07-01"
    assert str(tasks["ap-16"].deadline) == "2026-07-06"
    # "On going" task has no deadline.
    assert tasks["ap-08"].deadline is None


@pytest.mark.asyncio
async def test_roll_forward_resets_status_and_recomputes(db_session):
    await svc.instantiate_from_templates(db_session, "2026-06")
    tasks = await svc.list_tasks(db_session, "2026-06")
    # Complete one monthly task.
    monthly = next(t for t in tasks if t.frequency == "monthly")
    await svc.set_status(db_session, monthly.id, status=TASK_COMPLETED, actor_id=None)

    res = await svc.roll_forward(db_session, from_period="2026-06", to_period="2026-07")
    assert res["created"] == 129
    new_tasks = {t.instance_key: t for t in await svc.list_tasks(db_session, "2026-07")}
    # The completed task is reset.
    assert new_tasks[monthly.instance_key].status == TASK_NOT_STARTED
    # Deadlines recomputed for the new period (July close: Day+1 = Sat Aug 1).
    assert str(new_tasks["rr-02"].deadline) == "2026-08-01"


# ── #1656: concurrent get-then-insert must not 500 ──

@pytest.mark.asyncio
async def test_ensure_run_idempotent_on_duplicate_period(db_session, monkeypatch):
    """A concurrent ``ensure_run`` sees ``get_run() is None`` but its INSERT
    trips ``uq_finance_close_run_period`` (the winner already committed). The
    SAVEPOINT must confine the IntegrityError and re-select the surviving row
    — not poison the session. Before the fix this 500'd."""
    run1 = await close_service.ensure_run(db_session, "2026-09")
    await db_session.flush()

    # Force the race: make the FIRST get_run return None (as a concurrent
    # caller would see before the winner's row is visible), so the INSERT
    # path runs against the already-present unique row. The retry re-select
    # (unpatched) then finds the winner.
    real_get_run = close_service.get_run
    calls = {"n": 0}

    async def _get_run_race(db, period):
        calls["n"] += 1
        if calls["n"] == 1 and period == "2026-09":
            return None
        return await real_get_run(db, period)

    monkeypatch.setattr(close_service, "get_run", _get_run_race)
    run2 = await close_service.ensure_run(db_session, "2026-09")
    assert run2.id == run1.id
    # Session still usable — a follow-up commit succeeds.
    await db_session.commit()


@pytest.mark.asyncio
async def test_ensure_run_recovers_cross_transaction_winner(db_session, db_engine):
    """Regression for the review finding: the IntegrityError recovery must use
    a LOCKING current-read, not a snapshot read. Under REPEATABLE READ this
    transaction's snapshot is fixed at the first ``get_run`` (before the winner
    commits), so a plain re-select can't see the concurrently-committed winner
    → returns None → re-raises → the very 500 the guard exists to prevent. A
    ``FOR UPDATE`` current read sees the committed winner. This reproduces the
    genuine cross-transaction race (the sibling test above uses the same
    session, where the winner row is trivially self-visible)."""
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    period = "2026-11"
    # 1. Establish db_session's REPEATABLE READ snapshot NOW (before winner
    #    commits) — this consistent read fixes the snapshot with no row.
    assert await close_service.get_run(db_session, period) is None

    # 2. A DIFFERENT transaction creates + commits the winning run.
    other = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with other() as s2:
        await close_service.ensure_run(s2, period)
        await s2.commit()

    # 3. ensure_run here: get_run (snapshot) still sees None → INSERT hits the
    #    committed unique row → IntegrityError → locking re-select finds the
    #    winner. Pre-fix this raised (snapshot re-select → None) and 500'd.
    run = await close_service.ensure_run(db_session, period)
    assert run is not None
    assert run.period == period
    # Session is not poisoned — a follow-up commit succeeds.
    await db_session.commit()


@pytest.mark.asyncio
async def test_instantiate_idempotent_task_insert_helper(db_session):
    """The per-task insert helper tolerates a concurrent duplicate: a row that
    already exists under ``uq_finance_close_task_period_key`` yields a skip
    (``False``), not an IntegrityError that poisons the outer session."""
    await svc.instantiate_from_templates(db_session, "2026-06")
    await db_session.flush()
    existing = (await svc.list_tasks(db_session, "2026-06"))[0]
    # A brand-new task object colliding on (period, instance_key).
    dup = FinanceCloseTask(
        period="2026-06",
        instance_key=existing.instance_key,
        template_source_key=existing.template_source_key,
        status=TASK_NOT_STARTED,
        process=existing.process,
        description=existing.description,
    )
    created = await svc._insert_task_idempotent(db_session, dup)
    assert created is False
    # Session survives the swallowed duplicate.
    await db_session.commit()


# ── #1657: task-list mutations frozen after close ──

@pytest.mark.asyncio
async def test_task_mutations_blocked_after_close(db_session):
    await svc.instantiate_from_templates(db_session, "2026-06")
    run = await close_service.ensure_run(db_session, "2026-06")
    await close_service.set_status(db_session, run, new_status=RUN_CLOSED, actor_id=None)
    await db_session.flush()

    task = (await svc.list_tasks(db_session, "2026-06"))[0]
    with pytest.raises(ValidationError):
        await svc.set_status(db_session, task.id, status=TASK_COMPLETED, actor_id=None)
    with pytest.raises(ValidationError):
        await svc.set_comment(
            db_session, task.id, comment="x",
            commenter_person_id=None, commenter_label=None,
        )
    with pytest.raises(ValidationError):
        await svc.update_task(db_session, task.id, fields={"owner_label": "New Owner"})
    with pytest.raises(ValidationError):
        await svc.delete_task(db_session, task.id)
    with pytest.raises(ValidationError):
        await svc.create_task(
            db_session, "2026-06",
            fields={"process": "AP", "description": "adhoc after close"},
            actor_id=None,
        )


@pytest.mark.asyncio
async def test_task_mutations_blocked_after_soft_close_and_allowed_on_reopen(db_session):
    await svc.instantiate_from_templates(db_session, "2026-06")
    run = await close_service.ensure_run(db_session, "2026-06")
    await close_service.set_status(
        db_session, run, new_status=RUN_SOFT_CLOSED, actor_id=None,
    )
    await db_session.flush()
    task = (await svc.list_tasks(db_session, "2026-06"))[0]
    with pytest.raises(ValidationError):
        await svc.set_status(db_session, task.id, status=TASK_COMPLETED, actor_id=None)

    # Reopen → mutations allowed again.
    from app.domains.finance.close_models import RUN_IN_PROGRESS
    await close_service.set_status(
        db_session, run, new_status=RUN_IN_PROGRESS, actor_id=None,
    )
    await db_session.flush()
    updated = await svc.set_status(
        db_session, task.id, status=TASK_COMPLETED, actor_id=None,
    )
    assert updated.status == TASK_COMPLETED


# ── HTTP ──

@pytest_asyncio.fixture
async def seeded_period(client: AsyncClient, finance_headers):
    await client.post("/api/v1/finance/close/restore-defaults", headers=finance_headers)
    r = await client.post("/api/v1/finance/close/runs/2026-06/instantiate", headers=finance_headers)
    assert r.status_code == 200, r.text
    return "2026-06"


@pytest.mark.asyncio
async def test_http_list_tasks_and_aggregate(client, finance_headers, seeded_period):
    r = await client.get(f"/api/v1/finance/close/runs/{seeded_period}/tasks", headers=finance_headers)
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert len(data["tasks"]) == 129
    assert data["aggregate"]["overall"]["total"] == 129
    # Filter by process.
    r = await client.get(
        f"/api/v1/finance/close/runs/{seeded_period}/tasks?process=AP", headers=finance_headers,
    )
    assert len(r.json()["data"]["tasks"]) == 19


@pytest.mark.asyncio
async def test_http_task_lifecycle(client, finance_headers, seeded_period):
    r = await client.get(f"/api/v1/finance/close/runs/{seeded_period}/tasks?process=Tax", headers=finance_headers)
    task = r.json()["data"]["tasks"][0]
    tid = task["id"]
    # status -> completed
    r = await client.post(f"/api/v1/finance/close/tasks/{tid}/status", headers=finance_headers,
                          json={"status": "completed"})
    assert r.status_code == 200 and r.json()["data"]["status"] == "completed"
    assert r.json()["data"]["completed_at"] is not None
    # comment
    r = await client.post(f"/api/v1/finance/close/tasks/{tid}/comment", headers=finance_headers,
                          json={"comment": "done by EOD", "commenter_label": "Grace"})
    assert r.json()["data"]["comment"] == "done by EOD"
    # update fields (reassign owner label + change day -> recompute deadline)
    r = await client.patch(f"/api/v1/finance/close/tasks/{tid}", headers=finance_headers,
                          json={"calendar_day": "Day+1", "owner_label": "Huan"})
    assert r.json()["data"]["deadline"] == "2026-07-01"
    assert r.json()["data"]["owner_label"] == "Huan"


@pytest.mark.asyncio
async def test_http_create_and_delete_adhoc(client, finance_headers, seeded_period):
    r = await client.post(f"/api/v1/finance/close/runs/{seeded_period}/tasks", headers=finance_headers,
                          json={"process": "AP", "description": "ad-hoc check", "calendar_day": "Day+2"})
    assert r.status_code == 200, r.text
    tid = r.json()["data"]["id"]
    assert r.json()["data"]["template_source_key"] is None
    r = await client.delete(f"/api/v1/finance/close/tasks/{tid}", headers=finance_headers)
    assert r.status_code == 200


@pytest.mark.asyncio
async def test_http_bulk_ops(client, finance_headers, seeded_period):
    r = await client.get(f"/api/v1/finance/close/runs/{seeded_period}/tasks?process=Payroll", headers=finance_headers)
    ids = [t["id"] for t in r.json()["data"]["tasks"][:5]]
    r = await client.post("/api/v1/finance/close/tasks/bulk-status", headers=finance_headers,
                          json={"task_ids": ids, "status": "na"})
    assert r.json()["data"]["updated"] == 5
    r = await client.post("/api/v1/finance/close/tasks/bulk-reassign", headers=finance_headers,
                          json={"task_ids": ids, "owner_label": "Cici"})
    assert r.json()["data"]["updated"] == 5


@pytest.mark.asyncio
async def test_http_roll_forward_endpoint(client, finance_headers, seeded_period):
    r = await client.post("/api/v1/finance/close/roll-forward", headers=finance_headers,
                          json={"from_period": "2026-06", "to_period": "2026-07"})
    assert r.status_code == 200, r.text
    assert r.json()["data"]["created"] == 129
    r = await client.get("/api/v1/finance/close/runs/2026-07/tasks", headers=finance_headers)
    assert len(r.json()["data"]["tasks"]) == 129


@pytest.mark.asyncio
async def test_http_templates_and_entities(client, finance_headers, seeded_period):
    r = await client.get("/api/v1/finance/close/templates", headers=finance_headers)
    assert len(r.json()["data"]) == 129
    r = await client.get("/api/v1/finance/close/entities", headers=finance_headers)
    assert len(r.json()["data"]) == 9


@pytest.mark.asyncio
async def test_http_export_xlsx(client, finance_headers, auditor_headers, seeded_period):
    # Full period → a real .xlsx workbook (zip magic bytes), audited.
    r = await client.get(
        f"/api/v1/finance/close/runs/{seeded_period}/tasks/export.xlsx",
        headers=finance_headers,
    )
    assert r.status_code == 200, r.text
    assert "spreadsheetml" in r.headers["content-type"]
    assert r.content[:2] == b"PK"  # xlsx is a zip
    assert r.headers["X-Export-Returned"] == "129"

    # ids= narrows to exactly those rows (matches the on-screen filter).
    r2 = await client.get(
        f"/api/v1/finance/close/runs/{seeded_period}/tasks?process=AP", headers=finance_headers,
    )
    ap_ids = [t["id"] for t in r2.json()["data"]["tasks"][:3]]
    r3 = await client.get(
        f"/api/v1/finance/close/runs/{seeded_period}/tasks/export.xlsx",
        params={"ids": ",".join(str(i) for i in ap_ids)}, headers=finance_headers,
    )
    assert r3.status_code == 200
    assert r3.headers["X-Export-Returned"] == "3"

    # Auditor has finance view scope → may export too.
    r4 = await client.get(
        f"/api/v1/finance/close/runs/{seeded_period}/tasks/export.xlsx", headers=auditor_headers,
    )
    assert r4.status_code == 200


@pytest.mark.asyncio
async def test_http_anchor_change_recomputes_task_deadlines(client, finance_headers, seeded_period):
    # Move Day+1 anchor to Jul 2 -> rr-02 (Day+1) deadline should follow.
    r = await client.patch("/api/v1/finance/close/runs/2026-06/anchors", headers=finance_headers,
                          json={"anchor_plus_1": "2026-07-02"})
    assert r.status_code == 200, r.text
    r = await client.get("/api/v1/finance/close/runs/2026-06/tasks?q=Recognize OP", headers=finance_headers)
    rr02 = next(t for t in r.json()["data"]["tasks"] if t["template_source_key"] == "rr-02")
    assert rr02["deadline"] == "2026-07-02"


@pytest.mark.asyncio
async def test_create_task_rejects_day_zero(client, finance_headers, seeded_period):
    # Day+0 / Day-0 match the token regex but resolve to no deadline — reject.
    r = await client.post(f"/api/v1/finance/close/runs/{seeded_period}/tasks", headers=finance_headers,
                          json={"process": "AP", "description": "bad day", "calendar_day": "Day+0"})
    assert r.status_code >= 400


@pytest.mark.asyncio
async def test_search_escapes_like_wildcards(client, finance_headers, seeded_period):
    # A literal '%' must match literally, not act as a match-all wildcard.
    # Exactly one seeded task has a '%' in its text ("...20% for Alibaba...").
    r = await client.get(f"/api/v1/finance/close/runs/{seeded_period}/tasks?q=%25", headers=finance_headers)
    assert r.status_code == 200
    tasks = r.json()["data"]["tasks"]
    assert 0 < len(tasks) < 129  # not a match-all
    assert all("%" in tk["description"] for tk in tasks)


@pytest.mark.asyncio
async def test_http_rbac(client, employee_headers, auditor_headers, finance_headers, seeded_period):
    r = await client.get("/api/v1/finance/close/runs/2026-06/tasks", headers=employee_headers)
    assert r.status_code == 403
    r = await client.get("/api/v1/finance/close/runs/2026-06/tasks", headers=auditor_headers)
    assert r.status_code == 200
    r = await client.post("/api/v1/finance/close/runs/2026-06/instantiate", headers=auditor_headers)
    assert r.status_code == 403