"""迁移 617 —— 云 IAM 三件套按账号串链 + 三对具名边。

这条迁移的价值不是「优化」,是**消除一个有据可查的静默故障**:
`cmdb.scopes_sync` / `roles_sync` / `grants_sync` 共用同一把
`cloud_sync:{integration_id}` 锁,却作为三条相隔几分钟的独立 cron 跑 ——
`cmdb.scopes.aws-primary` 因此在 2026-07-07..08-05 **连续 31 天里有 29 天
没跑**(roles 04:09 起跑要 ~15 分钟,scopes 04:18 撞进被持有的锁)。串链之后
它们**在结构上不可能再争同一把锁**。

盯五件事:

1. **形状守卫**。13 个账号 × 2 条边,逐行写死 cron 不现实,所以按「仍是播种时
   那个小时」的形状匹配。管理员把某一行挪到别的小时,就是一个这条迁移不知道的
   判断,必须放过。
2. **上游不存在就不连**。某个账号在这个部署里没有 scopes 行,它本该保证的顺序
   在这里也不存在,把下游留在自己的 cron 上才是对的。
3. **具名边里那条是在修 bug**,不是在记录现状:`usage_signal_refresh` 自称
   「排在 idle_reclaim 之前」,而 02:40 在 02:00 **之后**。
4. **不碰账单路径**。断言迁移一条 cost/billing 边都没连 —— 那类边会让下游比
   今天更早开始,而它们拉的是厂商前一天的账单批次。
5. **降级**。还原成同小时、名字派生的分钟(原分钟不可恢复),而不是让整个车队
   都落在 :00 —— 那正是迁移 418 消掉的惊群。
"""
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

_VERSIONS = pathlib.Path(__file__).resolve().parents[2] / "alembic" / "versions"
JOBS = "scheduled_jobs"


def _load(stem: str):
    path = next(_VERSIONS.glob(f"{stem}_*.py"))
    spec = importlib.util.spec_from_file_location(f"m{stem}", path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


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


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        # 615 provides the column these tests exercise; the shard may or
        # may not have run it yet depending on file order.
        ctx = MigrationContext.configure(c)
        with Operations.context(ctx):
            _load("615").upgrade()
        try:
            yield c
        finally:
            c.execute(
                sa.text(f"UPDATE {JOBS} SET chain_after_job_id = NULL "
                        "WHERE name LIKE 'cmdb.%probe617%'"),
            )
            c.execute(
                sa.text(f"DELETE FROM {JOBS} WHERE name LIKE '%probe617%'"),
            )
    engine.dispose()


def _seed(conn, name: str, cron: str) -> str:
    jid = str(uuid.uuid4())
    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, 'Asia/Shanghai', 1, 0, 0, "
            " NOW(), NOW())",
        ),
        {"id": jid, "n": name, "c": cron},
    )
    return jid


def _row(conn, name: str):
    return conn.execute(
        sa.text(
            f"SELECT trigger_type, cron_expr, chain_after_job_id "
            f"FROM {JOBS} WHERE name = :n",
        ),
        {"n": name},
    ).fetchone()


def test_triad_is_chained_per_account(conn):
    scopes = _seed(conn, "cmdb.scopes.probe617", "40 4 * * *")
    _seed(conn, "cmdb.roles.probe617", "9 4 * * *")
    _seed(conn, "cmdb.grants.probe617", "23 5 * * *")

    _run(conn, "617", "upgrade")

    roles = _row(conn, "cmdb.roles.probe617")
    grants = _row(conn, "cmdb.grants.probe617")
    assert roles.trigger_type == "chained"
    assert roles.cron_expr is None
    assert roles.chain_after_job_id == scopes
    assert grants.trigger_type == "chained"
    # grants chains behind ROLES, not behind scopes — the lock is shared
    # by all three, so only a full serial order removes the contention.
    assert grants.chain_after_job_id == _row_id(conn, "cmdb.roles.probe617")
    # The root keeps its clock; a chain has to start somewhere.
    assert _row(conn, "cmdb.scopes.probe617").trigger_type == "cron"


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


def test_shape_guard_leaves_a_retuned_row_alone(conn):
    """An operator who moved this row to a different hour made a decision
    this migration knows nothing about."""
    _seed(conn, "cmdb.scopes.probe617b", "40 4 * * *")
    _seed(conn, "cmdb.roles.probe617b", "9 6 * * *")  # moved to 06:xx

    _run(conn, "617", "upgrade")

    roles = _row(conn, "cmdb.roles.probe617b")
    assert roles.trigger_type == "cron"
    assert roles.cron_expr == "9 6 * * *"
    assert roles.chain_after_job_id is None


def test_missing_predecessor_leaves_the_successor_on_its_cron(conn):
    """No scopes row for this account means the ordering it was meant to
    guarantee does not exist here either."""
    _seed(conn, "cmdb.roles.probe617c", "9 4 * * *")
    _run(conn, "617", "upgrade")
    roles = _row(conn, "cmdb.roles.probe617c")
    assert roles.trigger_type == "cron"
    assert roles.chain_after_job_id is None


def test_upgrade_is_idempotent(conn):
    _seed(conn, "cmdb.scopes.probe617d", "40 4 * * *")
    _seed(conn, "cmdb.roles.probe617d", "9 4 * * *")
    _run(conn, "617", "upgrade")
    first = _row(conn, "cmdb.roles.probe617d").chain_after_job_id
    _run(conn, "617", "upgrade")
    assert _row(conn, "cmdb.roles.probe617d").chain_after_job_id == first


def test_downgrade_restores_a_spread_cron_in_the_same_hour(conn):
    """Not the original minute — the upgrade nulls it and it is not
    recoverable. What must hold is the hour, and that the fleet does not
    all land on :00 (the thundering herd migration 418 removed)."""
    _seed(conn, "cmdb.scopes.probe617e", "40 4 * * *")
    _seed(conn, "cmdb.roles.probe617e", "9 4 * * *")
    _seed(conn, "cmdb.grants.probe617e", "23 5 * * *")
    _run(conn, "617", "upgrade")

    _run(conn, "617", "downgrade")

    roles = _row(conn, "cmdb.roles.probe617e")
    grants = _row(conn, "cmdb.grants.probe617e")
    assert roles.trigger_type == "cron"
    assert roles.chain_after_job_id is None
    assert roles.cron_expr.endswith(" 4 * * *"), roles.cron_expr
    assert grants.cron_expr.endswith(" 5 * * *"), grants.cron_expr
    minute = int(roles.cron_expr.split()[0])
    assert 0 <= minute <= 59


def test_no_billing_job_is_ever_chained_by_this_migration():
    """The load-bearing constraint of the whole migration.

    Every chain edge makes its successor start EARLIER than today's gap
    allows. For a job that pulls yesterday's vendor line items, earlier
    means possibly reading a batch the vendor has not finished writing —
    a data-correctness change wearing a scheduling change's costume.
    Checked against the declared edge lists, so adding a billing edge
    later fails here rather than in a month-end reconciliation.
    """
    module = _load("617")
    billing_markers = ("cost_facts", "billing", "cost_recon", "monthclose")
    touched = [
        name
        for edge in module.TRIAD_EDGES
        for name in (edge[0], edge[1])
    ] + [
        name
        for edge in module.NAMED_EDGES
        for name in (edge[0], edge[1])
    ]
    offenders = [
        n for n in touched if any(m in n for m in billing_markers)
    ]
    assert not offenders, (
        f"617 must not chain anything on the billing path: {offenders}"
    )


def test_named_edges_point_at_registered_handlers():
    from app.scheduled_jobs.registry import get_handler

    module = _load("617")
    for successor, predecessor, _cron in module.NAMED_EDGES:
        assert get_handler(successor) is not None, successor
        assert get_handler(predecessor) is not None, predecessor


def test_the_inverted_pair_is_actually_inverted_today():
    """Pins the bug this migration fixes, so the fix cannot be quietly
    reverted into a no-op.

    ``software_seat.usage_signal_refresh`` (migration 612) says it runs
    «排在 software_seat.idle_reclaim 之前» and is seeded at 02:40 UTC,
    while ``idle_reclaim`` is seeded at 02:00 UTC — forty minutes the
    wrong way round, with ``idle_reclaim`` genuinely reading the table
    the collector fills.
    """
    seed_612 = (_VERSIONS / "612_seat_usage_signals.py").read_text()
    seed_204 = next(_VERSIONS.glob("204_*.py")).read_text()
    assert '"40 2 * * *"' in seed_612 or "'40 2 * * *'" in seed_612
    assert "0 2 * * *" in seed_204
    assert "idle_reclaim 之前" in seed_612

    module = _load("617")
    pair = [
        e for e in module.NAMED_EDGES
        if e[0] == "software_seat.idle_reclaim"
    ]
    assert pair, "617 must chain idle_reclaim behind the collector"
    assert pair[0][1] == "software_seat.usage_signal_refresh"