TUI-пульт на Textual вместо текстового меню
Списки баз стенда и локальных дампов на одном экране, фильтр сужает их на каждом набранном символе. Enter по базе открывает окно параметров дампа с галочкой автоподнятия: снята — только dump, поднята — dump плюс restore. Правки полей уезжают в копию конфига, config.json не трогается. Ядро получило четыре хука для UI (лог, границы шагов, пароль, отмена), run() читает вывод утилит построчно через Popen, поэтому журнал наполняется во время работы pg_dump. Текстовый режим, CLI и --dry-run ведут себя как раньше; без textual или вне tty пульт не запускается. Журнал разворачивается на весь экран по F4 и сам на старте операции. Раскладка адаптивная: narrow ниже 100 колонок, short ниже 32 строк. Ctrl-хоткеи продублированы кириллицей — русская раскладка их не глушит. Запуск через run.ps1 (venv, UTF-8-консоль, -Setup и -Tests). Проверка — tests_tui.py: 46 тестов, TUI гоняется headless через Pilot.
This commit is contained in:
+972
@@ -0,0 +1,972 @@
|
||||
#!/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 Input, Switch
|
||||
|
||||
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", Switch).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", Switch)
|
||||
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}"
|
||||
Reference in New Issue
Block a user