"""迁移 635 —— 两列,以及它们各自防的那件事。

* `assistant_escalation_routes.backup_owner_person_id`:主负责人休假两周,那两周里
  每一次「转人工」都会把员工和一个不会回话的人关进一个房间 —— 而员工会**以为**
  有人在看。
* `assistant_ticket_chats.escalated_at`:升级只该发生一次。没有这一列,每一次催办
  都会再拉一次人 —— 一个房间会慢慢变成整个 IT 团队。

盯的是**幂等**与**类型**:这条迁移会在已经有这两列的库上再跑(中断续跑、本地
`create_all` 建过的测试库),而 `backup_owner_person_id` 的类型必须与
`identity_persons.id` 完全一致(含排序规则)—— TiDB 上不一致就是 3780,而本地
MySQL 复现不出来。
"""
from __future__ import annotations

import importlib.util
import pathlib

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" / "635_assistant_room_escalation.py"
)

ROUTES = "assistant_escalation_routes"
CHATS = "assistant_ticket_chats"


def _load():
    spec = importlib.util.spec_from_file_location("m635", _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:
        yield c
    engine.dispose()


def _columns(conn, table: str) -> dict[str, dict]:
    return {
        c["name"]: c for c in sa.inspect(conn).get_columns(table)
    }


def test_both_columns_exist_after_the_migration(conn):
    """测试库由 ``create_all`` 建过,所以这两列已经在 —— 迁移必须**空操作**
    而不是报错。生产上中断续跑走的是同一条路。
    """
    _run(conn)
    assert "backup_owner_person_id" in _columns(conn, ROUTES)
    assert "escalated_at" in _columns(conn, CHATS)


def test_running_it_twice_is_a_no_op(conn):
    _run(conn)
    _run(conn)
    assert "backup_owner_person_id" in _columns(conn, ROUTES)


def test_the_backup_column_matches_the_person_id_column(conn):
    """类型与排序规则必须和父表一致 —— TiDB 上不一致就是 3780(「Referencing
    column are incompatible」),而本地 MySQL 单一排序规则复现不出来。
    """
    rows = conn.execute(sa.text(
        "SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH "
        "FROM information_schema.COLUMNS "
        "WHERE TABLE_SCHEMA = DATABASE() "
        "AND ((TABLE_NAME = :routes AND COLUMN_NAME = 'backup_owner_person_id') "
        "  OR (TABLE_NAME = 'identity_persons' AND COLUMN_NAME = 'id'))",
    ), {"routes": ROUTES}).mappings().all()
    by_table = {r["TABLE_NAME"]: r for r in rows}
    assert ROUTES in by_table, "备份负责人这一列没建出来"
    assert "identity_persons" in by_table
    assert (
        by_table[ROUTES]["DATA_TYPE"] == by_table["identity_persons"]["DATA_TYPE"]
    )
    assert (
        by_table[ROUTES]["CHARACTER_MAXIMUM_LENGTH"]
        == by_table["identity_persons"]["CHARACTER_MAXIMUM_LENGTH"]
    )