"""迁移 632 —— 群聊的三个开关必须**看得见**,而且不能覆盖管理员的决定。

盯两件事,它们各自防一种「上线之后才发现」的伤害:

1. **种子行必须真的落下。** `platform_settings` 是「有行才看得见」的表 —— 设置页
   列的就是这张表。不种行,管理员在界面上找不到这几个开关,只能靠读文档知道它们
   存在;而一个找不到的开关,和没有开关的区别只在于文档里那一句话。
2. **已经存在的行绝不覆盖。** 管理员可能在这条迁移之前就手工建过其中一行(比如
   提前关掉任意群)。一次发布把他的决定改回默认值,是这类种子最典型也最难发现的
   伤害:界面上看起来一切正常,行为却回到了出厂设置。

顺带钉住取值本身:工单群**开**、任意群**关**。这两个默认值是 2026-09-08 与用户
确认的产品决定,写反任何一个都是行为上的重大改变 —— 前者让 Helix 自己建的房间
继续沉默,后者让全公司任何群都能 @ 它。
"""
from __future__ import annotations

import importlib.util
import json
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" / "632_assistant_group_channel.py"
)

KEYS = (
    "assistant.group_chat_enabled",
    "assistant.group_chat_any_enabled",
    "assistant.group_turns_per_hour",
)


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


def _seed_only(conn) -> None:
    """跑迁移里的种子那一半。

    DDL 那一半在测试库上早已由 `Base.metadata.create_all` 建好(列已经在),
    迁移自己的存在性守卫会跳过它 —— 所以直接跑 upgrade 就只剩种子在动。
    """
    ctx = MigrationContext.configure(conn)
    with Operations.context(ctx):
        _load().upgrade()


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        c.execute(
            sa.text("DELETE FROM platform_settings WHERE `key` IN :keys").bindparams(
                sa.bindparam("keys", expanding=True),
            ),
            {"keys": list(KEYS)},
        )
        try:
            yield c
        finally:
            c.execute(
                sa.text(
                    "DELETE FROM platform_settings WHERE `key` IN :keys",
                ).bindparams(sa.bindparam("keys", expanding=True)),
                {"keys": list(KEYS)},
            )
    engine.dispose()


def _values(conn) -> dict[str, object]:
    rows = conn.execute(
        sa.text(
            "SELECT `key`, `value`, `description` FROM platform_settings "
            "WHERE `key` IN :keys",
        ).bindparams(sa.bindparam("keys", expanding=True)),
        {"keys": list(KEYS)},
    ).mappings().all()
    out: dict[str, object] = {}
    for row in rows:
        raw = row["value"]
        out[row["key"]] = json.loads(raw) if isinstance(raw, str) else raw
        assert row["description"], f"{row['key']} 没有说明 —— 设置页上就是一行裸键名"
    return out


def test_the_three_switches_become_visible_rows(conn):
    """种下来才在设置页上存在。"""
    _seed_only(conn)
    values = _values(conn)
    assert set(values) == set(KEYS)
    # 工单群开、任意群关 —— 写反任何一个都是行为上的重大改变。
    assert values["assistant.group_chat_enabled"] is True
    assert values["assistant.group_chat_any_enabled"] is False
    assert values["assistant.group_turns_per_hour"] == 120


def test_an_administrators_existing_choice_survives_the_migration(conn):
    """管理员提前关掉过任意群 —— 一次发布不能把它悄悄改回来。"""
    conn.execute(
        sa.text(
            "INSERT INTO platform_settings (`key`, `value`, `description`) "
            "VALUES (:k, :v, :d)",
        ),
        {
            "k": "assistant.group_chat_enabled",
            "v": json.dumps(False),
            "d": "管理员自己关的",
        },
    )
    _seed_only(conn)
    values = _values(conn)
    assert values["assistant.group_chat_enabled"] is False, (
        "已经存在的行被覆盖了 —— 管理员的决定在一次发布里被静默改回默认值"
    )
    # 其余两行照常种下,不因为一行已存在就整批跳过。
    assert values["assistant.group_chat_any_enabled"] is False
    assert values["assistant.group_turns_per_hour"] == 120


def test_running_it_twice_changes_nothing(conn):
    """迁移在生产上可能被重跑(中断续跑),第二次必须是空操作。"""
    _seed_only(conn)
    first = _values(conn)
    _seed_only(conn)
    assert _values(conn) == first