chuanqi/pipeline_bot/tests/test_db_live.py

"""live 门禁测试 — 触真实远程 MySQL, 无网络/库不可达时整组 skip。

run: pytest tests/test_db_live.py -v
"""
from __future__ import annotations

import pytest

pymysql = pytest.importorskip("pymysql")

from pbot.db.stores import (  # noqa: E402
    AccountStore,
    RoleStore,
    RunLogStore,
    ServerStore,
    ZoneStore,
    init_all,
)


@pytest.fixture(scope="module")
def conn() -> object:
    try:
        from pbot.db.engine import get_connection
        real = get_connection()
    except Exception as e:  # noqa: BLE001
        pytest.skip(f"远程 MySQL 不可达: {e}")
    yield real
    real.close()


def test_init_tables(conn: object) -> None:
    n = init_all(conn)  # type: ignore[arg-type]
    assert n == 11
    with conn.cursor() as cur:  # type: ignore[attr-defined]
        cur.execute("SHOW TABLES LIKE 'pb_%'")
        tables = {r["Tables_in_996cq (pb_%)"] for r in cur.fetchall()}
    assert len(tables) == 11


def test_seed_data(conn: object) -> None:
    servers = ServerStore(conn)  # type: ignore[arg-type]
    server = servers.get_by_name("玖扬复古176")
    assert server is not None
    accounts = AccountStore(conn).list_for_server(int(server["id"]))  # type: ignore[arg-type]
    assert len(accounts) == 3
    tasks = servers.tasks_with_scripts(int(server["id"]))
    assert len(tasks) == 10
    seqs = [int(t["seq"]) for t in tasks]
    assert seqs == sorted(seqs)


def test_zone_latest(conn: object) -> None:
    zone = ZoneStore(conn).latest_enabled()  # type: ignore[arg-type]
    assert zone is None or int(zone["state"]) == 1


def test_role_crud_roundtrip(conn: object) -> None:  # type: ignore[arg-type]
    """角色 upsert + 清理(临时数据)。"""
    servers = ServerStore(conn)
    server = servers.get_by_name("玖扬复古176")
    assert server is not None
    with conn.cursor() as cur:
        cur.execute(
            "SELECT id FROM pb_zone WHERE server_id=%s "
            "ORDER BY open_time DESC LIMIT 1", (server["id"],))
        zone = cur.fetchone()
    if zone is None:
        pytest.skip("pb_zone 无数据(未扫描)")
    roles = RoleStore(conn)
    from datetime import datetime
    acct = AccountStore(conn).list_for_server(int(server["id"]))[0]
    row = roles.ensure_for_zone(
        account_id=int(acct["id"]), server_id=int(server["id"]),
        zone_id=int(zone["id"]), server_name="t", zone_name="t",
        spawn_at=datetime.now())  # noqa: DTZ005 - 库侧 naive 定案
    rid = int(row["id"])
    roles.set_role_name(rid, "测试角色")
    roles.set_progress(rid, 3, 10)
    got = roles.get(rid)
    assert got is not None
    assert got["role_name"] == "测试角色"
    assert int(got["progress_done"]) == 3
    with conn.cursor() as cur:
        cur.execute("DELETE FROM pb_role WHERE id=%s", (rid,))
        cur.execute(
            "DELETE FROM pb_role_task_progress WHERE role_id=%s", (rid,))


def test_runlog_roundtrip(conn: object) -> None:  # type: ignore[arg-type]
    servers = ServerStore(conn)
    server = servers.get_by_name("玖扬复古176")
    assert server is not None
    roles = RoleStore(conn)
    existing = roles.list_with_names()
    if not existing:
        pytest.skip("无角色行(未分配)")
    rid = int(existing[0]["id"])
    runs = RunLogStore(conn)
    run_id = runs.start(rid, "manual")
    runs.set_pid(run_id, 999999)
    runs.finish(run_id, 0, "test")
    with conn.cursor() as cur:
        cur.execute("DELETE FROM pb_run_log WHERE id=%s", (run_id,))