from datetime import UTC, datetime, timedelta

from sqlalchemy import event


def boot(client):
    response = client.post(
        "/api/v1/setup/initialize",
        json={"username": "owner", "password": "correct horse battery staple"},
    )
    assert response.status_code == 201
    return client.get("/api/v1/lists").json()[0]


def test_recurring_calendar_exceptions_and_scopes(client):
    inbox = boot(client)
    task = client.post(
        "/api/v1/tasks",
        json={"title": "站会", "list_id": inbox["id"], "due_at": "2026-09-01T09:00:00Z"},
    ).json()
    normal_task = client.post(
        "/api/v1/tasks",
        json={"title": "月内普通任务", "list_id": inbox["id"], "due_at": "2026-09-12T09:00:00Z"},
    ).json()
    client.post(
        "/api/v1/tasks",
        json={"title": "月外普通任务", "list_id": inbox["id"], "due_at": "2026-10-02T09:00:00Z"},
    )
    recurrence = client.post(
        "/api/v1/recurrences", json={"task_id": task["id"], "rrule": "FREQ=WEEKLY;BYDAY=TU,TH;COUNT=5"}
    )
    assert recurrence.status_code == 201
    recurrence_id = recurrence.json()["id"]
    yearly = client.post(
        "/api/v1/tasks",
        json={"title": "年度任务", "list_id": inbox["id"], "due_at": "2026-09-06T09:00:00Z"},
    ).json()
    assert client.post(
        "/api/v1/recurrences", json={"task_id": yearly["id"], "rrule": "FREQ=YEARLY;INTERVAL=1;BYMONTH=9;BYMONTHDAY=6"}
    ).status_code == 201
    calendar = client.get(
        "/api/v1/calendar", params={"start": "2026-09-01", "end": "2026-09-30"}
    ).json()
    assert any(row.get("id") == normal_task["id"] for row in calendar)
    assert any(row["title"] == "年度任务" for row in calendar)
    assert not any(row["title"] == "月外普通任务" for row in calendar)
    occurrences = [row for row in calendar if row["recurrence_id"] == recurrence_id]
    assert len(occurrences) == 5
    assert occurrences[0]["title"] == "站会"

    at = occurrences[1]["occurrence_at"]
    edited = client.patch(
        f"/api/v1/recurrences/{recurrence_id}",
        params={"scope": "this", "occurrence_at": at},
        json={"title": "特殊站会"},
    )
    assert edited.status_code == 200
    client.post(f"/api/v1/recurrences/{recurrence_id}/complete", json={"occurrence_at": at})
    changed = client.get(
        "/api/v1/calendar", params={"start": "2026-09-01", "end": "2026-09-30"}
    ).json()
    exception = next(row for row in changed if row.get("occurrence_at") == at)
    assert exception["title"] == "特殊站会" and exception["completed"] is True
    assert client.delete(
        f"/api/v1/recurrences/{recurrence_id}",
        params={"scope": "this", "occurrence_at": occurrences[2]["occurrence_at"]},
    ).status_code == 204
    remaining = client.get(
        "/api/v1/calendar", params={"start": "2026-09-01", "end": "2026-09-30"}
    ).json()
    assert len([row for row in remaining if row["recurrence_id"] == recurrence_id]) == 4
    assert any(row.get("id") == normal_task["id"] for row in remaining)


def test_calendar_respects_timezone_boundaries_and_moved_occurrences(client):
    inbox = boot(client)
    inside = client.post(
        "/api/v1/tasks",
        json={"title": "上海九月第一刻", "list_id": inbox["id"], "due_at": "2026-08-31T16:30:00Z"},
    ).json()
    client.post(
        "/api/v1/tasks",
        json={"title": "上海十月第一刻", "list_id": inbox["id"], "due_at": "2026-09-30T16:30:00Z"},
    )
    recurring = client.post(
        "/api/v1/tasks",
        json={"title": "跨月重复任务", "list_id": inbox["id"], "due_at": "2026-10-01T09:00:00Z"},
    ).json()
    recurrence = client.post(
        "/api/v1/recurrences", json={"task_id": recurring["id"], "rrule": "FREQ=WEEKLY;COUNT=2"}
    ).json()
    client.patch(
        f"/api/v1/recurrences/{recurrence['id']}",
        params={"scope": "this", "occurrence_at": "2026-10-01T09:00:00Z"},
        json={"due_at": "2026-09-20T09:00:00Z"},
    )

    calendar = client.get(
        "/api/v1/calendar",
        params={"start": "2026-09-01", "end": "2026-09-30", "timezone_offset": 480},
    ).json()

    assert any(row.get("id") == inside["id"] for row in calendar)
    assert not any(row["title"] == "上海十月第一刻" for row in calendar)
    assert any(row.get("recurrence_id") == recurrence["id"] and row["due_at"].startswith("2026-09-20") for row in calendar)


def test_habit_logs_support_date_range_filter(client):
    boot(client)
    habit = client.post("/api/v1/habits", json={"name": "跑步", "kind": "boolean", "schedule_type": "daily"}).json()
    hid = habit["id"]
    for day in ("2026-08-01", "2026-08-15", "2026-09-01"):
        client.post(f"/api/v1/habits/{hid}/logs", json={"day": day, "value": 1})

    all_logs = client.get(f"/api/v1/habits/{hid}/logs").json()
    ranged = client.get(f"/api/v1/habits/{hid}/logs", params={"from": "2026-08-10", "to": "2026-08-31"}).json()

    assert len(all_logs) == 3
    assert [row["day"] for row in ranged] == ["2026-08-15"]


def test_habit_grid_uses_a_bounded_number_of_queries(client, monkeypatch):
    boot(client)
    for index in range(6):
        client.post("/api/v1/habits", json={"name": f"习惯 {index}", "kind": "boolean", "schedule_type": "daily"})

    from backend.db import get_engine

    statement_count = 0
    engine = get_engine().sync_engine

    def count_queries(*_):
        nonlocal statement_count
        statement_count += 1

    event.listen(engine, "before_cursor_execute", count_queries)
    try:
        response = client.get("/api/v1/habits/grid", params={"week": "2026-09-01"})
    finally:
        event.remove(engine, "before_cursor_execute", count_queries)

    assert response.status_code == 200
    assert len(response.json()["habits"]) == 6
    assert statement_count <= 5


def test_habits_numeric_accumulation_pause_archive_grid_and_stats(client):
    boot(client)
    habit = client.post(
        "/api/v1/habits",
        json={"name": "喝水", "kind": "numeric", "target": 8, "schedule_type": "daily", "max_value": 10},
    )
    assert habit.status_code == 201
    habit_id = habit.json()["id"]
    today = datetime.now(UTC).date().isoformat()
    for value in (6, 7):
        assert client.post(f"/api/v1/habits/{habit_id}/logs", json={"day": today, "value": value}).status_code == 200
    assert client.get(f"/api/v1/habits/{habit_id}/logs").json()[0]["value"] == 10
    assert client.put(f"/api/v1/habits/{habit_id}/logs/{today}", json={"value": 8}).json()["value"] == 8
    yesterday = (datetime.now(UTC).date() - timedelta(days=1)).isoformat()
    assert client.post(f"/api/v1/habits/{habit_id}/pauses", json={"start_date": yesterday, "end_date": today}).status_code == 201
    grid = client.get("/api/v1/habits/grid", params={"week": yesterday}).json()
    assert len(grid["days"]) == 7 and grid["habits"][0]["cells"]
    assert grid["habits"][0]["kind"] == "numeric"
    assert grid["habits"][0]["target"] == 8
    assert grid["habits"][0]["max_value"] == 10
    assert grid["habits"][0]["stats"]["total"] == 8
    assert grid["habits"][0]["stats"]["completed_days"] == 1
    stats = client.get(f"/api/v1/habits/{habit_id}/stats").json()
    assert stats["total"] == 8 and stats["completed_days"] == 1
    assert client.delete(f"/api/v1/habits/{habit_id}").status_code == 204
    archived = client.get("/api/v1/habits", params={"archived": True}).json()
    assert archived[0]["id"] == habit_id
    assert client.get(f"/api/v1/habits/{habit_id}/stats").json()["total"] == 8


def test_boolean_interval_habit_schedule(client):
    boot(client)
    habit = client.post(
        "/api/v1/habits",
        json={"name": "拉伸", "kind": "boolean", "schedule_type": "interval", "interval_days": 2},
    )
    assert habit.status_code == 201
    assert client.post(
        f"/api/v1/habits/{habit.json()['id']}/logs", json={"day": datetime.now(UTC).date().isoformat(), "value": 1}
    ).json()["value"] == 1


def test_attachment_security_ownership_and_size(client, tmp_path, monkeypatch):
    monkeypatch.setenv("DODO_ATTACHMENT_DIR", str(tmp_path / "uploads"))
    inbox = boot(client)
    task = client.post("/api/v1/tasks", json={"title": "文件", "list_id": inbox["id"]}).json()
    uploaded = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("notes.txt", b"safe text", "text/plain")},
    )
    assert uploaded.status_code == 201
    attachment = uploaded.json()
    assert attachment["filename"] == "notes.txt"
    assert client.get(f"/api/v1/attachments/{attachment['id']}").content == b"safe text"
    bad = client.post(
        f"/api/v1/tasks/{task['id']}/attachments",
        files={"file": ("../evil.exe", b"x", "application/x-msdownload")},
    )
    assert bad.status_code == 400
    assert client.delete(f"/api/v1/attachments/{attachment['id']}").status_code == 204


def test_tasks_can_be_loaded_by_list_and_completion_filter(client):
    inbox = boot(client)
    other = client.post("/api/v1/lists", json={"name": "其他清单"}).json()
    client.post("/api/v1/tasks", json={"title": "收集箱未完成", "list_id": inbox["id"]})
    done = client.post("/api/v1/tasks", json={"title": "收集箱已完成", "list_id": inbox["id"]}).json()
    client.patch(f"/api/v1/tasks/{done['id']}", json={"completed": True, "version": done["version"]})
    client.post("/api/v1/tasks", json={"title": "其他清单未完成", "list_id": other["id"]})

    response = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "completed": False})

    assert response.status_code == 200
    assert [item["title"] for item in response.json()["items"]] == ["收集箱未完成"]


def test_tasks_support_numbered_pagination_with_total(client):
    inbox = boot(client)
    for index in range(5):
        client.post("/api/v1/tasks", json={"title": f"任务 {index}", "list_id": inbox["id"]})

    first = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 1, "page_size": 2})
    second = client.get("/api/v1/tasks", params={"list_id": inbox["id"], "page": 2, "page_size": 2})

    assert first.status_code == 200
    assert first.json()["total"] == 5
    assert first.json()["page"] == 1
    assert first.json()["page_size"] == 2
    assert len(first.json()["items"]) == 2
    assert len(second.json()["items"]) == 2
    assert {item["id"] for item in first.json()["items"]}.isdisjoint(
        {item["id"] for item in second.json()["items"]}
    )


def test_tasks_support_due_range_pagination(client):
    inbox = boot(client)
    client.post("/api/v1/tasks", json={"title": "今天", "list_id": inbox["id"], "due_at": "2026-09-05T08:00:00Z"})
    client.post("/api/v1/tasks", json={"title": "以后", "list_id": inbox["id"], "due_at": "2026-09-08T08:00:00Z"})
    client.post("/api/v1/tasks", json={"title": "无日期", "list_id": inbox["id"]})

    response = client.get(
        "/api/v1/tasks",
        params={"due_from": "2026-09-05T00:00:00Z", "due_to": "2026-09-06T00:00:00Z", "page": 1},
    )

    assert response.status_code == 200
    assert response.json()["total"] == 1
    assert [item["title"] for item in response.json()["items"]] == ["今天"]


def test_trash_supports_numbered_pagination_with_total(client):
    inbox = boot(client)
    for index in range(3):
        task = client.post("/api/v1/tasks", json={"title": f"删除 {index}", "list_id": inbox["id"]}).json()
        client.delete(f"/api/v1/tasks/{task['id']}")

    response = client.get("/api/v1/trash", params={"page": 2, "page_size": 2})

    assert response.status_code == 200
    assert response.json()["total"] == 3
    assert response.json()["page"] == 2
    assert len(response.json()["items"]) == 1


def test_ticktick_preview_import_dedupe_and_json_restore(client):
    boot(client)
    csv_data = "Title,List Name,Due Date,Status,ID\nImported,Inbox,2026-10-01,0,ext-1\n"
    preview = client.post("/api/v1/import/ticktick/preview", files={"file": ("tasks.csv", csv_data, "text/csv")})
    assert preview.status_code == 200 and preview.json()["valid"] == 1
    for _ in range(2):
        response = client.post("/api/v1/import/ticktick", files={"file": ("tasks.csv", csv_data, "text/csv")})
        assert response.status_code == 200
    assert response.json()["skipped"] == 1
    assert len(client.get("/api/v1/tasks").json()["items"]) == 1

    export = client.get("/api/v1/export").json()
    assert export["version"] == 1 and export["tasks"][0]["external_id"] == "ext-1"
    client.delete(f"/api/v1/tasks/{export['tasks'][0]['id']}")
    restored = client.post("/api/v1/restore", params={"mode": "replace"}, json=export)
    assert restored.status_code == 200
    assert len(client.get("/api/v1/tasks").json()["items"]) == 1
    assert client.post("/api/v1/restore", json={"version": 999}).status_code == 422


def test_audit_logs_cover_task_and_collection_operations(client):
    boot(client)
    folder = client.post("/api/v1/folders", json={"name": "F"}).json()
    task_list = client.post("/api/v1/lists", json={"name": "L", "folder_id": folder["id"]}).json()
    task = client.post("/api/v1/tasks", json={"title": "T", "list_id": task_list["id"]}).json()
    client.patch(f"/api/v1/tasks/{task['id']}", json={"completed": True, "version": task["version"]})
    client.delete(f"/api/v1/tasks/{task['id']}")
    client.post(f"/api/v1/tasks/{task['id']}/restore")
    client.patch(f"/api/v1/lists/{task_list['id']}", json={"name": "L2"})
    client.delete(f"/api/v1/folders/{folder['id']}")
    logs = client.get("/api/v1/audit-logs").json()
    pairs = {(row["entity_type"], row["action"]) for row in logs}
    assert {("task", "create"), ("task", "complete"), ("task", "delete"), ("task", "restore"), ("list", "update"), ("folder", "delete")} <= pairs


def test_sessions_csrf_headers_revocation_and_docs(client):
    boot(client)
    assert client.get("/api/docs").status_code == 200
    sessions = client.get("/api/v1/sessions").json()
    assert len(sessions) == 1 and sessions[0]["current"] is True
    assert client.delete(f"/api/v1/sessions/{sessions[0]['id']}").status_code == 204
    assert client.get("/api/v1/me").status_code == 401

    anonymous = client.__class__(client.app)
    with anonymous:
        assert anonymous.get("/api/docs").status_code == 401


def test_login_rate_limit_is_progressive(client):
    boot(client)
    client.post("/api/v1/auth/logout")
    statuses = [
        client.post("/api/v1/auth/login", json={"username": "owner", "password": "wrong password"}).status_code
        for _ in range(8)
    ]
    assert 429 in statuses


def test_security_headers(client):
    response = client.get("/health/live")
    assert response.headers["x-content-type-options"] == "nosniff"
    assert response.headers["content-security-policy"]