pg-stand-sync/tests_tui.py

#!/usr/bin/env python3
"""Тесты ядра и TUI. Textual гоняется headless через App.run_test() + Pilot.

Запуск: .venv/Scripts/python.exe -m pytest tests_tui.py -q
Настоящий терминал не открывается, к postgresql.lan никто не ходит: список баз подменяется,
внешние утилиты — короткие python -c.
"""

from __future__ import annotations

import asyncio
import os
import sys
import threading
import time
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent))

import pg_stand_sync as core
import pg_stand_sync_tui as tui
from textual.widgets import Checkbox, Collapsible, Input

PY = sys.executable
DBS = [("zpas", "1284 MB"), ("zpas_arch", "312 MB"), ("zpas_test", "88 MB"),
       ("postgres", "8 MB"), ("registry", "44 MB")]


def script(body: str) -> list:
    return [PY, "-c", body]


def child_env() -> dict:
    # ядро выставляет дочерним PGCLIENTENCODING=UTF8; для python -c аналог — PYTHONIOENCODING
    env = dict(os.environ)
    env["PYTHONIOENCODING"] = "utf-8"
    return env


@pytest.fixture(autouse=True)
def clean_hooks():
    """Хуки — глобальные: тест, забывший их снять, сломает соседние."""
    yield
    core.set_sink(None)
    core.set_progress(None)
    core.set_password_prompt(None)
    core.clear_cancel()
    core._password_cache.clear()


@pytest.fixture
def cfg(tmp_path):
    return {
        "dump_dir": str(tmp_path / "dumps"),
        "keep_dumps": 5,
        "source": {"host": "stand.example", "port": 5432, "user": "postgres",
                   "maintenance_database": "postgres", "password": "s3cret"},
        "target": {"host": "localhost", "port": 5432, "user": "postgres", "database": None,
                   "maintenance_database": "postgres", "recreate": True, "password": "s3cret"},
        "restore": {"jobs": 4},
        "post_restore_sql": [],
    }


# --------------------------------------------------------------------------- ядро


def test_matches_common_filter():
    assert core.matches("zpas_core", "zpa")
    assert core.matches("zpas_core", "ZPA")
    assert core.matches("zpas_core", "zp core")
    assert not core.matches("zpas_test", "zp core")
    assert core.matches("что угодно", "")


def test_pick_uses_matches(monkeypatch):
    """Текстовый режим и TUI обязаны фильтровать одинаково — через одну функцию."""
    answers = iter(["zp core", "1"])
    monkeypatch.setattr(core, "ask", lambda prompt: next(answers))
    monkeypatch.setattr(sys.stdin, "isatty", lambda: True, raising=False)
    got = core.pick(["zpas_core", "zpas_test", "registry"], "базы", lambda s: s)
    assert got == ["zpas_core"]


def test_dry_run_does_not_spawn(monkeypatch, capsys):
    monkeypatch.setattr(core.subprocess, "Popen",
                        lambda *a, **k: pytest.fail("dry-run не должен запускать процесс"))
    core.run(script("print(1)"), child_env(), "проверка", dry_run=True)
    out = capsys.readouterr().out.strip().splitlines()
    assert len(out) == 1 and "проверка:" in out[0]


def test_step_error_text_is_stable():
    with pytest.raises(core.StepError) as err:
        core.run(script("import sys; sys.exit(3)"), child_env(), "шаг", dry_run=False)
    assert str(err.value) == "шаг завершился с кодом 3"


def test_text_mode_output_format(capsys):
    core.run(script("print('первая'); print('вторая')"), child_env(), "шаг", dry_run=False)
    lines = capsys.readouterr().out.splitlines()
    assert lines[1] == "   первая" and lines[2] == "   вторая"
    assert lines[0].startswith("[") and "шаг: " in lines[0]
    assert "готово за" in lines[3]


def test_output_streams_before_exit():
    seen = []
    core.set_sink(lambda kind, text: seen.append((time.monotonic(), kind, text)))
    started = time.monotonic()
    core.run(script("import time,sys\n"
                    "for i in range(5):\n"
                    "    print(i, flush=True); time.sleep(0.15)"),
             child_env(), "поток", dry_run=False)
    finished = time.monotonic()
    first_out = next(t for t, kind, _ in seen if kind == "out")
    assert first_out - started < (finished - started) / 2
    assert [text for kind, text in ((k, t) for _, k, t in seen) if kind == "out"] == list("01234")


def test_progress_hook_marks_step():
    events = []
    core.set_progress(lambda event, what, elapsed: events.append((event, what)))
    core.run(script("print('x')"), child_env(), "шаг", dry_run=False)
    assert events == [("start", "шаг"), ("done", "шаг")]


def test_progress_hook_silent_in_text_mode(capsys):
    core.run(script("print('x')"), child_env(), "шаг", dry_run=False)
    assert "start" not in capsys.readouterr().out


def test_cancel_kills_running_process():
    core.set_sink(lambda kind, text: None)
    box = {}

    def worker():
        try:
            core.run(script("import time; time.sleep(30)"), child_env(), "долго", False)
        except BaseException as exc:
            box["exc"] = exc

    thread = threading.Thread(target=worker)
    thread.start()
    time.sleep(1.0)
    core.request_cancel()
    thread.join(timeout=10)
    assert not thread.is_alive()
    assert isinstance(box.get("exc"), (core.Cancelled, core.StepError))
    assert not core._procs


def test_cancel_before_spawn(monkeypatch):
    core.request_cancel()
    monkeypatch.setattr(core.subprocess, "Popen",
                        lambda *a, **k: pytest.fail("после отмены процесс не запускают"))
    with pytest.raises(core.Cancelled):
        core.run(script("print(1)"), child_env(), "шаг", dry_run=False)


def test_spawn_refuses_after_cancel(monkeypatch):
    """Проверка флага и регистрация в _procs — под одним замком, иначе процесс переживёт отмену."""
    core.request_cancel()
    monkeypatch.setattr(core.subprocess, "Popen",
                        lambda *a, **k: pytest.fail("после отмены процесс не запускают"))
    with pytest.raises(core.Cancelled):
        core._spawn(script("print(1)"), child_env(), None)


def test_password_prompt_replaces_getpass(monkeypatch):
    monkeypatch.setattr(core.getpass, "getpass",
                        lambda *a, **k: pytest.fail("getpass под TUI звать нельзя"))
    calls = []
    core.set_password_prompt(lambda label, node: calls.append(label) or "hunter2")
    node = {"host": "stand", "user": "postgres", "port": 5432}
    assert core.resolve_password(node, "стенд") == "hunter2"
    assert core.resolve_password(node, "стенд") == "hunter2"
    assert calls == ["стенд"]


def test_password_needed_and_priming():
    node = {"host": "stand", "user": "postgres", "port": 5432, "password_env": "NOPE_PG"}
    assert core.password_needed(node, "стенд")
    core.prime_password(node, "стенд", "abc")
    assert not core.password_needed(node, "стенд")
    core.forget_password(node, "стенд")
    assert core.password_needed(node, "стенд")
    assert not core.password_needed({"password": "x", "host": "h", "user": "u"}, "стенд")


def test_password_never_in_argv(cfg, monkeypatch):
    seen = []
    monkeypatch.setattr(core, "run", lambda cmd, env, what, dry: seen.append((cmd, env)))
    tools = {"pg_dump": "pg_dump", "pg_restore": "pg_restore", "psql": "psql"}
    core.dump(cfg, tools, "zpas", Path(cfg["dump_dir"]) / "zpas.dump", dry_run=True)
    core.recreate_target(cfg, tools, "zpas", dry_run=True)
    core.restore(cfg, tools, Path(cfg["dump_dir"]) / "zpas.dump", "zpas", True, dry_run=True)
    assert seen
    for cmd, env in seen:
        assert env["PGPASSWORD"] == "s3cret"
        assert not any("s3cret" in str(part) for part in cmd)


def test_restore_error_becomes_warning(cfg, monkeypatch):
    kinds = []
    core.set_sink(lambda kind, text: kinds.append((kind, text)))
    monkeypatch.setattr(core, "run", lambda *a, **k: (_ for _ in ()).throw(
        core.StepError("restore в zpas@localhost завершился с кодом 1")))
    path = Path(cfg["dump_dir"])
    path.mkdir(parents=True, exist_ok=True)
    (path / "zpas.dump").write_bytes(b"x")
    core.restore(cfg, {"pg_restore": "pg_restore"}, path / "zpas.dump", "zpas", True, False)
    assert any(kind == "warn" and "внимание:" in text for kind, text in kinds)

    cfg["restore"]["exit_on_error"] = True
    with pytest.raises(core.StepError):
        core.restore(cfg, {"pg_restore": "pg_restore"}, path / "zpas.dump", "zpas", True, False)


def parse(argv):
    return core.build_parser().parse_args(argv)


def test_tui_enabled_matrix(monkeypatch):
    monkeypatch.setattr(core, "tui_available", lambda: True)
    monkeypatch.setattr(sys.stdout, "isatty", lambda: True, raising=False)
    monkeypatch.setattr(sys.stdin, "isatty", lambda: True, raising=False)
    monkeypatch.delenv("PG_STAND_SYNC_NO_TUI", raising=False)
    assert core.tui_enabled(parse([]))
    assert core.tui_enabled(parse(["menu"]))
    assert not core.tui_enabled(parse(["--no-tui"]))
    assert not core.tui_enabled(parse(["list"]))
    monkeypatch.setenv("PG_STAND_SYNC_NO_TUI", "1")
    assert not core.tui_enabled(parse([]))
    monkeypatch.delenv("PG_STAND_SYNC_NO_TUI")
    monkeypatch.setattr(sys.stdout, "isatty", lambda: False, raising=False)
    assert not core.tui_enabled(parse([]))
    monkeypatch.setattr(sys.stdout, "isatty", lambda: True, raising=False)
    monkeypatch.setattr(core, "tui_available", lambda: False)
    assert not core.tui_enabled(parse([]))


# --------------------------------------------------------------------------- TUI


def make_app(cfg, monkeypatch, dbs=DBS, failure=None):
    def fake_list(tools, node, label):
        if failure:
            raise core.StepError(failure)
        return list(dbs)

    monkeypatch.setattr(core, "list_databases", fake_list)
    tools = {"pg_dump": "pg_dump", "pg_restore": "pg_restore", "psql": "psql"}
    return tui.PgSyncApp(cfg, tools, dry_run=True)


async def settle(app, pilot):
    await pilot.pause()
    await app.workers.wait_for_complete()
    await pilot.pause()


def test_filter_narrows_on_every_keystroke(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            lst = app.dashboard.query_one("#tbl-db", tui.DbList)
            assert lst.option_count == len(DBS)
            counts = []
            for char in "zpa":
                await pilot.press(char)
                await pilot.pause()
                counts.append(lst.option_count)
            # Enter не нажимался ни разу
            assert counts == [3, 3, 3]
            assert app.dashboard.query_one("#flt-db").value == "zpa"
            for char in "s_a":
                await pilot.press(char)
                await pilot.pause()
            assert [row.name for row in lst.rows] == ["zpas_arch"]
            # счётчик живёт на рамке панели: у самого OptionList border: none, подпись не рисуется
            assert app.dashboard.query_one("#pane-db").border_subtitle == "показано 1/5"
    asyncio.run(scenario())


def test_escape_clears_filter_then_marks(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            lst = app.dashboard.query_one("#tbl-db", tui.DbList)
            await pilot.press("z", "p", "a")
            await pilot.pause()
            assert lst.option_count == 3
            await pilot.press("escape")
            await pilot.pause()
            assert app.dashboard.query_one("#flt-db").value == ""
            assert lst.option_count == len(DBS)
            lst.marked.add("zpas")
            await pilot.press("escape")
            await pilot.pause()
            assert not lst.marked
    asyncio.run(scenario())


def test_space_marks_only_in_list_and_survives_filter(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            lst = app.dashboard.query_one("#tbl-db", tui.DbList)
            flt = app.dashboard.query_one("#flt-db")

            await pilot.press("space")          # фокус в фильтре — это пробел, не отметка
            await pilot.pause()
            assert flt.value == " " and not lst.marked
            flt.value = ""
            await pilot.pause()

            await pilot.press("tab")
            await pilot.pause()
            assert app.focused is lst
            await pilot.press("space")
            await pilot.press("down")
            await pilot.press("space")
            await pilot.pause()
            assert lst.marked == {"zpas", "zpas_arch"}

            flt.focus()
            await pilot.press("r", "e", "g")     # фильтр прячет отмеченные
            await pilot.pause()
            assert [row.name for row in lst.rows] == ["registry"]
            assert lst.marked == {"zpas", "zpas_arch"}
            assert "отмечено 2" in app.dashboard.query_one("#pane-db").border_subtitle
            # состав операции не зависит от фильтра: отмечено 2 — значит и берутся обе
            assert lst.selection() == ["zpas", "zpas_arch"]
    asyncio.run(scenario())


def painted(app) -> str:
    """Что реально нарисовано на экране: border_subtitle без рамки в SVG не попадает вовсе."""
    return "\n".join(strip.text for strip in app.screen._compositor.render_strips())


def test_counter_is_actually_rendered(cfg, monkeypatch):
    """border_subtitle без рамки не рисуется — требование «показано/всего» проверяем по экрану."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            assert "показано 5/5" in painted(app)
            await pilot.press("q", "q", "q")
            await pilot.pause()
            screen = painted(app)
            assert "показано 0/5" in screen
            assert "ничего не найдено по фильтру" in screen
    asyncio.run(scenario())


def test_rows_fit_the_pane_width(cfg, monkeypatch):
    """Строки списков собираются под реальную ширину: перенос ломает колонки и съедает место."""
    dump_dir = Path(cfg["dump_dir"])
    dump_dir.mkdir(parents=True, exist_ok=True)
    (dump_dir / "zpas_analytics_reporting_daily-20260820-031332.dump").write_bytes(b"x")
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(100, 40)) as pilot:
            await settle(app, pilot)
            for widget_id, kind in (("#tbl-db", tui.DbList), ("#tbl-dump", tui.DumpList)):
                lst = app.dashboard.query_one(widget_id, kind)
                width = lst.content_size.width
                assert all(len(lst.line(row)) <= width for row in lst.rows)
    asyncio.run(scenario())


def test_multi_sync_refused_when_target_database_fixed(cfg, monkeypatch):
    """target.database слил бы отмеченные базы в одну — do_sync берёт имя из конфига."""
    cfg["target"]["database"] = "zpas_local"
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            dash.query_one("#tbl-db", tui.DbList).marked.update({"zpas", "zpas_arch"})
            dash.sync_selected()
            for _ in range(20):
                await pilot.pause()
            assert not isinstance(app.screen, tui.DbOpModal)
    asyncio.run(scenario())


def test_focus_returns_after_operation(cfg, monkeypatch):
    """Панели на время операции disabled: Textual снимает фокус, и фильтр молчит до Tab."""
    app = make_app(cfg, monkeypatch)
    monkeypatch.setattr(core, "do_dump", lambda *a: [])

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            assert app.focused.id == "flt-db"
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            await settle(app, pilot)
            assert app.focused is not None and app.focused.id == "flt-db"
            await pilot.press("z")
            await pilot.pause()
            assert dash.query_one("#flt-db").value == "z"
    asyncio.run(scenario())


def test_failed_operation_forgets_password(cfg, monkeypatch):
    """Опечатка в модалке иначе живёт в кэше до перезапуска: модалка больше не покажется."""
    cfg["target"].pop("password")
    app = make_app(cfg, monkeypatch)
    monkeypatch.setattr(core, "do_restore", lambda *a: (_ for _ in ()).throw(
        core.StepError("DROP DATABASE завершился с кодом 1")))
    core.prime_password(cfg["target"], "локальная", "typo")

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            dash.begin(tui.OpSpec("restore", dump_file=Path("x.dump"), target_db="zpas"))
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            assert not dash.op_active
            assert core.password_needed(cfg["target"], "локальная")
    asyncio.run(scenario())


def test_priority_hotkeys_survive_input_focus(cfg, monkeypatch):
    """Input биндит ctrl+a/ctrl+d/ctrl+u: без priority=True действия молча не сработают."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            lst = app.dashboard.query_one("#tbl-db", tui.DbList)
            flt = app.dashboard.query_one("#flt-db")
            await pilot.press("z", "p")
            await pilot.pause()
            await pilot.press("ctrl+a")
            await pilot.pause()
            assert lst.marked == {"zpas", "zpas_arch", "zpas_test"}
            assert flt.value == "zp"
            await pilot.press("ctrl+u")
            await pilot.pause()
            assert not lst.marked and flt.value == "zp"

            await pilot.press("left")
            await pilot.press("ctrl+d")          # у Input это delete_right
            await pilot.pause()
            await pilot.pause()
            assert flt.value == "zp"
            assert isinstance(app.screen, tui.DbOpModal)
            await pilot.press("escape")
            await pilot.pause()
    asyncio.run(scenario())


def test_arrows_from_filter_move_list_cursor(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            lst = app.dashboard.query_one("#tbl-db", tui.DbList)
            flt = app.dashboard.query_one("#flt-db")
            assert lst.highlighted == 0
            await pilot.press("down", "down")
            await pilot.pause()
            assert lst.highlighted == 2
            assert app.focused is flt
    asyncio.run(scenario())


def test_connection_error_does_not_kill_app(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch, failure="список баз stand.example: неверный пароль")

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            assert app.is_running
            error = app.dashboard.query_one("#db-error")
            assert error.has_class("shown")
            assert "неверный пароль" in str(error.content)
            assert app.dashboard.query_one("#pane-db").has_class("pane-error")
            await pilot.press("f5")              # повтор попытки не роняет приложение
            await settle(app, pilot)
            assert app.is_running
    asyncio.run(scenario())


def test_live_log_and_finish(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    def fake_dump(cfg_, tools_, dbs_, dry_run_):
        core.log(f"дамп {dbs_[0]}@stand.example: pg_dump …")
        for i in range(200):
            core._emit("out", f"строка {i}")
        return []

    monkeypatch.setattr(core, "do_dump", fake_dump)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            assert dash.op_active
            assert app.dashboard.query_one("#tbl-db").disabled
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            assert not dash.op_active
            assert not app.dashboard.query_one("#tbl-db").disabled
            body = [text for kind, _, text in dash.journal if kind == "out"]
            assert body == [f"строка {i}" for i in range(200)]
            assert any(kind == "ok" for kind, _, _ in dash.journal)
    asyncio.run(scenario())


def test_operation_failure_is_reported_not_raised(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)
    monkeypatch.setattr(core, "do_dump", lambda *a: (_ for _ in ()).throw(
        core.StepError("дамп zpas@stand.example завершился с кодом 1")))

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            assert app.is_running and not dash.op_active
            assert any(kind == "err" and "кодом 1" in text for kind, _, text in dash.journal)
    asyncio.run(scenario())


def test_password_modal_answers_worker_thread(cfg, monkeypatch):
    """resolve_password из середины do_* обязан спрашивать модалкой, а не вешать приложение."""
    cfg["source"].pop("password")
    app = make_app(cfg, monkeypatch)
    got = {}

    def fake_dump(cfg_, tools_, dbs_, dry_run_):
        got["password"] = core.resolve_password(cfg_["source"], "стенд")
        return []

    monkeypatch.setattr(core, "do_dump", fake_dump)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            for _ in range(60):
                await pilot.pause()
                if isinstance(app.screen, tui.PasswordModal):
                    break
            assert isinstance(app.screen, tui.PasswordModal)
            await pilot.press("h", "u", "n", "t")
            await pilot.press("enter")
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            assert got.get("password") == "hunt"
    asyncio.run(scenario())


def test_ctrl_c_exits_with_zero(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            await pilot.press("ctrl+c")
            for _ in range(40):
                await pilot.pause()
                if not app.is_running:
                    break
        assert app.return_value == 0
    asyncio.run(scenario())


def test_ctrl_c_during_operation_asks_and_returns_130(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)
    release = threading.Event()
    cancels = []
    monkeypatch.setattr(core, "do_dump", lambda *a: release.wait(10))
    monkeypatch.setattr(core, "request_cancel", lambda: cancels.append(1) or True)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            app.dashboard.begin(tui.OpSpec("dump", dbs=["zpas"]))
            await pilot.pause()
            await pilot.press("ctrl+c")
            for _ in range(40):
                await pilot.pause()
                if isinstance(app.screen, tui.QuitConfirm):
                    break
            assert isinstance(app.screen, tui.QuitConfirm)
            await pilot.press("tab")
            await pilot.press("enter")
            for _ in range(40):
                await pilot.pause()
                if not app.is_running:
                    break
            release.set()
        # 130 достижим только через ветку «операция идёт» — она же зовёт request_cancel
        assert app.return_value == 130
        assert cancels
    asyncio.run(scenario())


def test_exit_releases_thread_waiting_for_password(cfg, monkeypatch):
    """Колбэк модалки при разборе экранов не приходит: без отпускания поток ждёт до таймаута."""
    cfg["source"].pop("password")
    app = make_app(cfg, monkeypatch)
    box = {}

    def fake_dump(cfg_, tools_, dbs_, dry_run_):
        started = time.monotonic()
        try:
            core.resolve_password(cfg_["source"], "стенд")
        except core.Cancelled:
            box["cancelled"] = True
        box["waited"] = time.monotonic() - started
        return []

    monkeypatch.setattr(core, "do_dump", fake_dump)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            app.dashboard.begin(tui.OpSpec("dump", dbs=["zpas"]))
            for _ in range(60):
                await pilot.pause()
                if isinstance(app.screen, tui.PasswordModal):
                    break
            assert isinstance(app.screen, tui.PasswordModal)
            app.exit(0)
            for _ in range(40):
                await pilot.pause()
                if not app.is_running:
                    break
        for _ in range(50):
            if "waited" in box:
                break
            await asyncio.sleep(0.1)
        assert box.get("cancelled") and box["waited"] < 30
    asyncio.run(scenario())


def test_exit_leaves_cancel_flag_set(cfg, monkeypatch):
    """Снятый на выходе флаг дал бы ещё живому потоку доработать DROP DATABASE уже без UI."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            await pilot.press("ctrl+c")
            for _ in range(40):
                await pilot.pause()
                if not app.is_running:
                    break
        assert core.cancel_requested()
    asyncio.run(scenario())


def test_reload_works_after_cancelled_operation(cfg, monkeypatch):
    """Флаг отмены, доживший до reload(), краснит панель баз и не лечится F5."""
    app = make_app(cfg, monkeypatch)
    monkeypatch.setattr(core, "do_dump", lambda *a: (_ for _ in ()).throw(core.Cancelled()))

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            core.request_cancel()
            dash.op_active = True
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            await settle(app, pilot)
            assert not core.cancel_requested()
            assert not dash.query_one("#db-error").has_class("shown")
    asyncio.run(scenario())


def test_restore_pane_enter_opens_confirm(cfg, monkeypatch):
    dump_dir = Path(cfg["dump_dir"])
    dump_dir.mkdir(parents=True, exist_ok=True)
    (dump_dir / "zpas-20260820-031332.dump").write_bytes(b"x" * 2048)
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dumps = app.dashboard.query_one("#tbl-dump", tui.DumpList)
            assert dumps.option_count == 1
            app.dashboard.query_one("#flt-dump").focus()
            await pilot.pause()
            await pilot.press("enter")
            for _ in range(40):
                await pilot.pause()
                if isinstance(app.screen, tui.ConfirmModal):
                    break
            assert isinstance(app.screen, tui.ConfirmModal)
            assert app.screen.spec.kind == "restore"
            assert app.screen.spec.target_db == "zpas"
            assert app.focused.id == "cancel"     # recreate=true -> согласие не по умолчанию
            await pilot.press("escape")
            await pilot.pause()
    asyncio.run(scenario())


def _open_db_dialog(app, pilot, autorestore=True):
    """Открыть диалог операции над базой под курсором и дождаться его появления."""
    app.dashboard.open_db_op(autorestore=autorestore)

    async def wait():
        for _ in range(40):
            await pilot.pause()
            if isinstance(app.screen, tui.DbOpModal):
                return app.screen
        return app.screen
    return wait()


def test_db_dialog_carries_dump_options(cfg, monkeypatch):
    """Клик по базе открывает параметры дампа, а не голое подтверждение."""
    app = make_app(cfg, monkeypatch)
    captured = {}
    monkeypatch.setattr(core, "do_sync",
                        lambda cfg_, tools, dbs, target, no_recreate, dry:
                        captured.update(cfg=cfg_, dbs=dbs, target=target))

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            modal = await _open_db_dialog(app, pilot)
            assert isinstance(modal, tui.DbOpModal)
            modal.query_one("#schema-only", Checkbox).value = True
            modal.query_one("#compress", Input).value = "0"
            modal.query_one("#ex-schemas", Input).value = "audit, tmp"
            modal.action_start()
            for _ in range(60):
                await pilot.pause()
                if captured:
                    break
            assert captured["cfg"]["dump"]["schema_only"] is True
            assert captured["cfg"]["dump"]["compress"] == 0
            assert captured["cfg"]["dump"]["exclude_schemas"] == ["audit", "tmp"]
            # исходный конфиг диалог не трогает: правки живут в копии
            assert "schema_only" not in (cfg.get("dump") or {})
    asyncio.run(scenario())


def test_autorestore_switch_picks_operation(cfg, monkeypatch):
    """Снятая галочка — это dump, поднятая — dump + restore; выбор запоминается на сессию."""
    app = make_app(cfg, monkeypatch)
    calls = []
    monkeypatch.setattr(core, "do_dump", lambda *a, **k: calls.append("dump"))
    monkeypatch.setattr(core, "do_sync", lambda *a, **k: calls.append("sync"))

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            modal = await _open_db_dialog(app, pilot)
            switch = modal.query_one("#autorestore", Checkbox)
            assert switch.value is True                     # главный сценарий по умолчанию
            switch.value = False
            await pilot.pause()
            assert modal.query_one("#target-block").has_class("hidden")
            modal.action_start()
            for _ in range(60):
                await pilot.pause()
                if calls:
                    break
            assert calls == ["dump"]
            assert app.dashboard.op_defaults["autorestore"] is False
    asyncio.run(scenario())


def test_db_dialog_rejects_bad_compress(cfg, monkeypatch):
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            modal = await _open_db_dialog(app, pilot)
            modal.query_one("#compress", Input).value = "42"
            modal.action_start()
            await pilot.pause()
            assert isinstance(app.screen, tui.DbOpModal)    # диалог не закрылся
            await pilot.press("escape")
            await pilot.pause()
    asyncio.run(scenario())


def test_version_mismatch_noise_gets_explained():
    """SET из дампа новее сервера — шум: строку-объяснение выдаём один раз на вызов."""
    seen = []
    core.set_sink(lambda kind, text: seen.append((kind, text)))
    core.run(script("print('pg_restore: error: unrecognized configuration parameter \"x\"')\n"
                    "print('pg_restore: error: unrecognized configuration parameter \"y\"')"),
             child_env(), "restore", dry_run=False)
    hints = [text for kind, text in seen if kind == "warn"]
    assert hints == [core.NEWER_CLIENT_HINT]


def test_zoom_toggles_and_scrolls_log(cfg, monkeypatch):
    """F4 разворачивает журнал, стрелки листают его, Esc возвращает списки."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(100, 30)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            for i in range(120):
                dash._write(("out", None, f"строка {i}"))
            await pilot.pause()
            await pilot.press("f4")
            await pilot.pause()
            assert dash.zoomed and dash.has_class("zoom-log")
            assert not dash.query_one("#board").display
            log = dash.query_one("#log")
            log.scroll_end(animate=False)
            await pilot.pause()
            bottom = log.scroll_offset.y
            await pilot.press("pageup")
            await pilot.pause()
            assert log.scroll_offset.y < bottom          # журнал листается клавишами
            await pilot.press("escape")
            await pilot.pause()
            assert not dash.zoomed
            assert dash.query_one("#board").display
    asyncio.run(scenario())


def test_operation_switches_to_log(cfg, monkeypatch):
    """Во время операции смотреть нечего, кроме журнала: пульт переключается сам."""
    app = make_app(cfg, monkeypatch)
    # операция должна быть заметно длиннее одного тика, иначе состояние «во время» не поймать
    monkeypatch.setattr(core, "do_dump", lambda *a, **k: time.sleep(0.4))

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            assert not dash.zoomed
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            await pilot.pause()
            assert dash.zoomed                            # сразу на старте, а не по завершении
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            # авто-разворот сам и сворачивается: иначе фокус застрял бы в журнале
            assert not dash.zoomed
            assert app.focused is dash.query_one("#flt-db")

            dash.toggle_zoom()                            # развернули руками
            await pilot.pause()
            dash.begin(tui.OpSpec("dump", dbs=["zpas"]))
            for _ in range(80):
                await pilot.pause()
                if not dash.op_active:
                    break
            assert dash.zoomed                            # ручной выбор пульт не отменяет
    asyncio.run(scenario())


def test_layout_adapts_to_small_console(cfg, monkeypatch):
    """Адаптив: на низком и узком окне панели не вылезают за экран."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(80, 24)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            assert dash.has_class("narrow") and dash.has_class("short")
            board = dash.query_one("#board").size.height
            bottom = dash.query_one("#bottom").size.height
            assert board + bottom <= 24 - 4               # шапка 2 + статус 1 + футер 1
            assert not dash.query_one("#target-card").display
            await pilot.resize_terminal(160, 50)
            await pilot.pause()
            assert not dash.has_class("narrow") and not dash.has_class("short")
            assert dash.query_one("#target-card").display
    asyncio.run(scenario())


def test_rows_do_not_wrap_when_scrollbar_takes_a_column(cfg, monkeypatch):
    """Скроллбар отъедает колонку, и строка ровно по ширине рвёт хвост «1277 MB» пополам."""
    many = [("homeassistant", "1277 MB")] + [(f"filler_{i:02d}", f"{i * 13} MB")
                                             for i in range(40)]
    app = make_app(cfg, monkeypatch, dbs=many)
    # scrollbar-gutter: stable уводит полосу в поток — так же ведёт себя настоящий терминал
    app.CSS = "#tbl-db { overflow-y: scroll; scrollbar-gutter: stable; }"

    async def scenario():
        async with app.run_test(size=(200, 40)) as pilot:
            await settle(app, pilot)
            lst = app.dashboard.query_one("#tbl-db", tui.DbList)
            assert lst.option_count == len(many)
            assert max(len(lst.line(row)) for row in lst.rows) <= lst.content_size.width - 1
            assert lst.virtual_size.height == lst.option_count      # ни одна строка не перенеслась
    asyncio.run(scenario())


def test_hotkeys_work_in_russian_layout(cfg, monkeypatch):
    """При русской раскладке терминал присылает Ctrl+«в» — хоткей обязан сработать так же."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(120, 40)) as pilot:
            await settle(app, pilot)
            dash = app.dashboard
            lst = dash.query_one("#tbl-db", tui.DbList)
            dash.query_one("#flt-db", Input).focus()
            await pilot.pause()

            await pilot.press("ctrl+ф")                  # ctrl+a — отметить показанные
            await pilot.pause()
            assert lst.marked, "ctrl+ф не отметил базы"
            await pilot.press("ctrl+г")                   # ctrl+u — снять отметки
            await pilot.pause()
            assert not lst.marked

            await pilot.press("ctrl+в")                   # ctrl+d — окно операции
            for _ in range(40):
                await pilot.pause()
                if isinstance(app.screen, tui.DbOpModal):
                    break
            assert isinstance(app.screen, tui.DbOpModal)
            await pilot.press("escape")
            await pilot.pause()
            assert dash.query_one("#flt-db", Input).value == ""   # символы не утекли в фильтр
    asyncio.run(scenario())


def test_ru_layout_covers_every_ctrl_hotkey():
    """Каждый Ctrl-хоткей пульта имеет кириллический двойник с тем же действием."""
    pairs = {b.key: b.action for b in tui.PgSyncApp.BINDINGS}
    for key, action in list(pairs.items()):
        if key.startswith("ctrl+") and len(key) == 6 and key[5].isascii():
            twin = f"ctrl+{tui.RU_LAYOUT[key[5]]}"
            assert pairs.get(twin) == action, f"нет двойника для {key}"


def test_op_dialog_fits_and_keeps_buttons_visible(cfg, monkeypatch):
    """Форма не должна выдавливать ряд кнопок за нижнюю границу диалога."""
    app = make_app(cfg, monkeypatch)

    async def scenario():
        async with app.run_test(size=(80, 24)) as pilot:
            await settle(app, pilot)
            app.dashboard.open_db_op(autorestore=True)
            for _ in range(40):
                await pilot.pause()
                if isinstance(app.screen, tui.DbOpModal):
                    break
            modal = app.screen
            modal.query_one(Collapsible).collapsed = False      # самый длинный вариант формы
            await pilot.pause()
            dialog = modal.query_one(".dialog")
            start = modal.query_one("#start")
            assert dialog.size.height <= 24
            assert start.region.height > 0
            assert dialog.region.y <= start.region.y
            assert start.region.bottom <= dialog.region.bottom
            form = modal.query_one(".form")
            assert form.virtual_size.height > form.size.height   # длинное уезжает в скролл
            await pilot.press("escape")
            await pilot.pause()
    asyncio.run(scenario())