"""迁移 613 —— 节奏纠偏 + 台账体检 cron.

这是一批**数据**迁移,风险点和 DDL 迁移不同,只盯三件事:

1. **不覆盖管理员的决定。** 每一条节奏改写都带 ``AND cron_expr = :expected``
   / ``AND interval_seconds = :expected`` 守卫。生产上这些行是可以在管理台
   直接改的,一个已经被调过节奏的行代表一个这次迁移不知道的判断 —— 静默盖掉
   它比留着旧默认值更糟。这条守卫写错的形状是「本地看起来全对、生产上把别人
   的调参抹了」,所以必须有测试盯住。
2. **幂等。** 迁移会在多个环境上跑,新 cron 行没有自然唯一键之外的保护;
   重复插入的代价是同一个任务被挂两次、每天跑两遍。
3. **可逆。** downgrade 必须还原成迁移前的值,而且要用同样的守卫 —— 升级后
   又被人改过的行,降级时同样不能盖。

顺带钉住一条语义不变量:两个半年度开盘任务改成 ``1-7`` of ``6,12``,必须和
handler 自己的开窗谓词 ``[boundary, boundary+7d)`` 完全同集合。这两处一旦
错开,故障形状是「今年这一轮根本没开盘」,而且要等半年才会被发现。
"""
from __future__ import annotations

import importlib.util
import pathlib
import uuid

import pytest
import sqlalchemy as sa
from alembic.migration import MigrationContext
from alembic.operations import Operations

from ..conftest import _TEST_SYNC_URL

_MIGRATION = (
    pathlib.Path(__file__).resolve().parents[2]
    / "alembic" / "versions" / "613_scheduled_job_capacity_hygiene.py"
)

JOBS = "scheduled_jobs"


def _load():
    spec = importlib.util.spec_from_file_location("m613", _MIGRATION)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _run(conn, direction: str = "upgrade") -> None:
    ctx = MigrationContext.configure(conn)
    with Operations.context(ctx):
        getattr(_load(), direction)()


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        try:
            yield c
        finally:
            c.execute(sa.text(f"DELETE FROM {JOBS} WHERE name LIKE 'probe613.%'"))
            c.execute(
                sa.text(f"DELETE FROM {JOBS} WHERE name = 'scheduler.ledger_reconcile'"),
            )
            for name in (
                "cmdb.resource_directory_refresh",
                "cmdb.resource_inventory_refresh",
                "approval_registry.owner_confirmation_semiannual",
                "auditor.access_self_attestation_auto_open_semiannual",
                "cmdb.auto_owner_grant_backfill",
            ):
                c.execute(
                    sa.text(f"DELETE FROM {JOBS} WHERE name = :n"), {"n": name},
                )
    engine.dispose()


def _seed_cron(conn, name: str, cron: str) -> None:
    conn.execute(
        sa.text(
            f"INSERT INTO {JOBS} (id, name, display_name, handler_key, "
            " trigger_type, cron_expr, timezone, enabled, run_count, "
            " failure_count, created_at, updated_at) "
            "VALUES (:id, :n, :n, :n, 'cron', :c, 'UTC', 1, 0, 0, NOW(), NOW())",
        ),
        {"id": str(uuid.uuid4()), "n": name, "c": cron},
    )


def _seed_interval(conn, name: str, seconds: int) -> None:
    conn.execute(
        sa.text(
            f"INSERT INTO {JOBS} (id, name, display_name, handler_key, "
            " trigger_type, interval_seconds, timezone, enabled, run_count, "
            " failure_count, created_at, updated_at) "
            "VALUES (:id, :n, :n, :n, 'interval', :s, 'UTC', 1, 0, 0, "
            " NOW(), NOW())",
        ),
        {"id": str(uuid.uuid4()), "n": name, "s": seconds},
    )


def _cron_of(conn, name: str) -> str | None:
    return conn.execute(
        sa.text(f"SELECT cron_expr FROM {JOBS} WHERE name = :n"), {"n": name},
    ).scalar()


def _interval_of(conn, name: str) -> int | None:
    return conn.execute(
        sa.text(f"SELECT interval_seconds FROM {JOBS} WHERE name = :n"),
        {"n": name},
    ).scalar()


def test_rewrites_only_rows_still_on_the_seeded_cadence(conn):
    module = _load()
    _seed_cron(conn, "cmdb.resource_directory_refresh", "7,22,37,52 * * * *")
    # An operator has already halved this one by hand; the migration
    # knows nothing about why, so it must leave it exactly as found.
    _seed_cron(conn, "cmdb.resource_inventory_refresh", "17 */2 * * *")
    _seed_interval(conn, "cmdb.auto_owner_grant_backfill", 600)

    _run(conn, "upgrade")

    assert _cron_of(conn, "cmdb.resource_directory_refresh") == "7 * * * *"
    assert _cron_of(conn, "cmdb.resource_inventory_refresh") == "17 */2 * * *"
    assert _interval_of(conn, "cmdb.auto_owner_grant_backfill") == 3600
    assert module.LEDGER_RECONCILE_JOB["name"] == "scheduler.ledger_reconcile"


def test_seeding_the_new_cron_is_idempotent(conn):
    _run(conn, "upgrade")
    _run(conn, "upgrade")
    count = conn.execute(
        sa.text(f"SELECT COUNT(*) FROM {JOBS} WHERE name = :n"),
        {"n": "scheduler.ledger_reconcile"},
    ).scalar()
    assert count == 1


def test_new_cron_row_points_at_a_registered_handler(conn):
    """The worker's boot guard hard-crashes on an enabled row whose
    handler is unknown, so a seed/handler mismatch takes the whole worker
    down on the next deploy rather than failing quietly."""
    from app.scheduled_jobs.registry import get_handler

    _run(conn, "upgrade")
    handler_key = conn.execute(
        sa.text(f"SELECT handler_key FROM {JOBS} WHERE name = :n"),
        {"n": "scheduler.ledger_reconcile"},
    ).scalar()
    assert get_handler(handler_key) is not None


def test_downgrade_restores_and_respects_later_edits(conn):
    _seed_cron(conn, "cmdb.resource_directory_refresh", "7,22,37,52 * * * *")
    _seed_cron(
        conn, "approval_registry.owner_confirmation_semiannual", "30 8 * * *",
    )
    _run(conn, "upgrade")
    # Someone tunes one row AFTER the upgrade. Downgrade must not stomp
    # that either — the guard is symmetric on purpose.
    conn.execute(
        sa.text(
            f"UPDATE {JOBS} SET cron_expr = '3 * * * *' "
            "WHERE name = 'cmdb.resource_directory_refresh'",
        ),
    )

    _run(conn, "downgrade")

    assert _cron_of(conn, "cmdb.resource_directory_refresh") == "3 * * * *"
    assert (
        _cron_of(conn, "approval_registry.owner_confirmation_semiannual")
        == "30 8 * * *"
    )
    assert conn.execute(
        sa.text(f"SELECT COUNT(*) FROM {JOBS} WHERE name = :n"),
        {"n": "scheduler.ledger_reconcile"},
    ).scalar() == 0


@pytest.mark.parametrize(
    ("month", "day", "expected"),
    [
        (6, 1, True), (6, 7, True), (6, 8, False),
        (12, 1, True), (12, 7, True), (12, 8, False),
        (7, 1, False), (1, 1, False),
    ],
)
def test_semiannual_cron_matches_the_handler_open_window(month, day, expected):
    """The new cron must fire on exactly the days the handler accepts.

    Both handlers open only inside ``[boundary, boundary + 7 days)`` for
    Jun 1 / Dec 1. If the cron were narrower the round would silently
    never open; if it were wider we would be back to 351 pointless runs
    a year. Checked against the handler's own predicate, not against a
    second copy of the date arithmetic.
    """
    from datetime import UTC, datetime

    from app.domains.approval_registry.confirmation import due_boundary

    moment = datetime(2026, month, day, 8, 30, tzinfo=UTC)
    cron_matches = month in (6, 12) and 1 <= day <= 7
    assert cron_matches == expected
    assert (due_boundary(moment, window_days=7) is not None) == expected