cv4pve-cli-tui/tests/test_app.py

"""Textual pilot tests for cv4pve_tui.app; spec: docs/plans/guest-tui-mvp.md ("UI contract").

All cli functions are monkeypatched: cv4pve-cli is never executed.
"""

import contextlib
import subprocess
import threading

import pytest
from textual.screen import ModalScreen
from textual.widgets import Button, DataTable, Input, OptionList, Select, Static

from cv4pve_tui import cli
from cv4pve_tui.app import ConfirmScreen, ContextScreen, EditScreen, GuestApp

VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu", "status": "running",
      "maxcpu": 4, "maxmem": 8192 * 2**20, "maxdisk": 10 * 2**30}  # fmt: skip
CT = {"vmid": 100, "name": "backrest", "node": "mainframe", "type": "lxc", "status": "stopped",
      "maxcpu": 1, "maxmem": 512 * 2**20, "maxdisk": 8 * 2**30}  # fmt: skip


class Fake:
    """Recorder standing in for the cli module functions the app calls."""

    def __init__(self):
        self.guests = [VM, CT]
        self.pending = []
        self.disk_map = {"scsi0": "10G"}
        self.list_calls = 0
        self.actions, self.sets, self.resizes, self.pending_calls = [], [], [], []
        self.action_error = None
        self.action_hook = None
        self.live_status, self.status_calls = {}, []
        self.current = "mainframe"
        self.names = ["mainframe", "mainframe-edge"]
        self.contexts_error = None
        self.contexts_calls = 0
        self.use_calls = []
        self.use_error = None
        self.use_hook = None
        self.guests_by_context = {}
        self.list_hook = None

    def list_guests(self):
        self.list_calls += 1
        snapshot = list(self.guests_by_context.get(self.current, self.guests))
        if (
            self.list_hook
        ):  # runs after reading: simulates a slow reply from that cluster
            self.list_hook()
        return snapshot

    def guest_config(self, guest):
        return {"cores": guest["maxcpu"]}

    def disks(self, config, kind):
        return dict(self.disk_map)

    def action(self, name, guest):
        self.actions.append((name, guest["vmid"]))
        if self.action_hook:
            self.action_hook()
        if self.action_error:
            raise self.action_error

    def set_resources(self, guest, cores=None, memory_mb=None):
        given = {"cores": cores, "memory_mb": memory_mb}
        self.sets.append({k: v for k, v in given.items() if v is not None})

    def resize_disk(self, guest, disk, grow_gb):
        self.resizes.append((guest["vmid"], disk, grow_gb))

    def pending_keys(self, guest):
        self.pending_calls.append(guest["vmid"])
        return list(self.pending)

    def status(self, guest):
        self.status_calls.append(guest["vmid"])
        return self.live_status.get(guest["vmid"], guest["status"])

    def contexts(self):
        self.contexts_calls += 1
        if self.contexts_error:
            raise self.contexts_error
        return [
            {"name": n, "host": f"{n}.home.arpa:8006", "auth": "api-token",
             "current": n == self.current}
            for n in self.names
        ]  # fmt: skip

    def use_context(self, name):
        self.use_calls.append(name)
        if self.use_hook:
            self.use_hook()
        if self.use_error:
            raise self.use_error
        self.current = name


@pytest.fixture
def fake(monkeypatch):
    f = Fake()
    for name in ("list_guests", "guest_config", "disks", "action", "set_resources",
                 "resize_disk", "pending_keys", "status", "contexts", "use_context"):  # fmt: skip
        monkeypatch.setattr(cli, name, getattr(f, name))
    return f


def make_app():
    app = GuestApp()
    app.seen = []  # (message, severity) of every notify
    original = app.notify

    def spy(message, **kwargs):
        app.seen.append((str(message), kwargs.get("severity", "information")))
        return original(message, **kwargs)

    app.notify = spy
    return app


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


def text_of(screen):
    return " ".join(str(w.render()) for w in screen.query(Static))


def errors(app):
    return [m for m, sev in app.seen if sev == "error"]


async def press_button(app, pilot, selector):
    app.screen.query_one(selector, Button).press()
    await settle(app, pilot)


# --- table ------------------------------------------------------------------


async def test_table_loads_one_row_per_guest_in_list_order(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        table = app.query_one("#guests", DataTable)
        assert table.row_count == 2
        assert [k.value for k in table.rows] == ["108", "100"]
        cells = [str(c) for c in table.get_row("108")]
        assert "docker" in cells and "mainframe" in cells and "running" in cells
        assert "8192" in cells  # memory in MiB


async def test_refresh_key_reloads_guests(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        before = fake.list_calls
        fake.guests = [CT]
        await pilot.press("g")
        await settle(app, pilot)
        assert fake.list_calls == before + 1
        assert app.query_one("#guests", DataTable).row_count == 1


# --- lifecycle actions ------------------------------------------------------


async def test_start_runs_without_confirm_and_refreshes_table(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        before = fake.list_calls
        await pilot.press("s")
        await settle(app, pilot)
        assert fake.actions == [("start", 108)]
        assert not isinstance(app.screen, ModalScreen)
        assert fake.list_calls > before


async def test_action_targets_the_cursor_row(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press("down", "s")
        await settle(app, pilot)
        assert fake.actions == [("start", 100)]


@pytest.mark.parametrize(
    "key,name", [("S", "shutdown"), ("x", "stop"), ("r", "reboot")]
)
async def test_confirmed_action_calls_cli_only_after_yes(fake, key, name):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press(key)
        await pilot.pause()
        assert isinstance(app.screen, ConfirmScreen)
        assert "docker" in text_of(app.screen) and "108" in text_of(app.screen)
        assert fake.actions == []
        await press_button(app, pilot, "#yes")
        assert fake.actions == [(name, 108)]


@pytest.mark.parametrize("key", ["S", "x", "r"])
async def test_declined_action_does_not_call_cli(fake, key):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press(key)
        await pilot.pause()
        await press_button(app, pilot, "#no")
        assert fake.actions == []
        assert not isinstance(app.screen, ModalScreen)


async def test_stop_confirm_says_hard_power_off(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press("x")
        await pilot.pause()
        assert "hard power-off" in text_of(app.screen).lower()


@pytest.mark.parametrize("key", ["s", "S", "x", "r", "e"])
async def test_keys_on_empty_table_do_nothing(fake, key):
    fake.guests = []
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press(key)
        await settle(app, pilot)
        assert fake.actions == [] and fake.sets == [] and fake.resizes == []
        assert not isinstance(app.screen, ModalScreen)


async def test_cli_error_is_notified_with_summary(fake, monkeypatch):
    with monkeypatch.context() as m:  # build a real CliError through cli.run
        m.setattr(cli.subprocess, "run", lambda argv, **kw: subprocess.CompletedProcess(
            argv, 5, "UPID:mainframe:0:0:0:qmstart:108:root@pam:\nstart failed: no quorum\n", ""))  # fmt: skip
        with pytest.raises(cli.CliError) as exc:
            cli.run("do", "start")
    fake.action_error = exc.value
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press("s")
        await settle(app, pilot)
        assert any(exc.value.summary in m for m in errors(app))
        assert not any("UPID" in m for m in errors(app))


async def test_busy_guest_refuses_second_action(fake):
    started, release = threading.Event(), threading.Event()
    fake.action_hook = lambda: (started.set(), release.wait(10))
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        try:
            await pilot.press("s")
            for _ in range(200):
                if started.is_set():
                    break
                await pilot.pause(0.01)
            assert started.is_set()
            assert app.seen, "a working notice is expected while the action runs"
            await pilot.press("s")
            await pilot.pause()
            assert fake.actions == [("start", 108)]
        finally:
            release.set()
        await settle(app, pilot)
        assert fake.actions == [("start", 108)]


# --- edit flow --------------------------------------------------------------


async def open_edit(app, pilot):
    await settle(app, pilot)
    await pilot.press("e")
    await settle(app, pilot)
    assert isinstance(app.screen, EditScreen)


def fill(app, **values):
    for key, value in values.items():
        w = app.screen.query_one(f"#{key}", Select if key == "disk" else Input)
        w.value = value


async def test_edit_applies_only_given_fields_as_ints(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, cores="8")
        await press_button(app, pilot, "#apply")
        assert fake.sets == [{"cores": 8}]
        await pilot.press("e")
        await settle(app, pilot)
        fill(app, cores="2", memory="4096")
        await press_button(app, pilot, "#apply")
        assert fake.sets[-1] == {"cores": 2, "memory_mb": 4096}
        assert all(isinstance(v, int) for v in fake.sets[-1].values())


@pytest.mark.parametrize("field", ["cores", "memory"])
@pytest.mark.parametrize("bad", ["abc", "0", "-4", "2.5"])
async def test_edit_invalid_resource_input_errors_without_cli_call(fake, field, bad):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, **{field: bad})
        await press_button(app, pilot, "#apply")
        assert errors(app)
        assert fake.sets == [] and fake.resizes == []


@pytest.mark.parametrize("bad", ["abc", "0", "-1"])
async def test_edit_invalid_grow_errors_without_cli_call(fake, bad):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, disk="scsi0", grow=bad)
        await press_button(app, pilot, "#apply")
        assert errors(app)
        assert fake.resizes == []


async def test_edit_with_empty_fields_changes_nothing(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        await press_button(app, pilot, "#apply")
        assert fake.sets == [] and fake.resizes == []


async def test_edit_cancel_changes_nothing(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, cores="8")
        await press_button(app, pilot, "#cancel")
        assert fake.sets == [] and fake.resizes == []
        assert not isinstance(app.screen, ModalScreen)


async def test_grow_needs_confirm_then_resizes(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, disk="scsi0", grow="5")
        await press_button(app, pilot, "#apply")
        assert isinstance(app.screen, ConfirmScreen)
        assert fake.resizes == []
        await press_button(app, pilot, "#yes")
        assert fake.resizes == [(108, "scsi0", 5)]


async def test_grow_declined_does_not_resize(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, disk="scsi0", grow="5")
        await press_button(app, pilot, "#apply")
        await press_button(app, pilot, "#no")
        assert fake.resizes == []


@pytest.mark.parametrize(
    "pending,expect_notice", [(["memory"], True), (["cores"], True), ([], False)]
)
async def test_pending_cores_or_memory_notifies_reboot(fake, pending, expect_notice):
    fake.pending = pending
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, memory="4096")
        await press_button(app, pilot, "#apply")
        assert fake.pending_calls == [108]
        assert any("reboot" in m.lower() for m, _ in app.seen) is expect_notice


# --- review findings: robustness --------------------------------------------


async def test_guest_without_size_fields_still_listed(fake):
    # e.g. a guest on an unreachable node: the API omits maxcpu/maxmem/maxdisk
    fake.guests = [
        {
            "vmid": 999,
            "name": "ghost",
            "node": "down",
            "type": "qemu",
            "status": "unknown",
        }
    ]
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        assert app.query_one("#guests", DataTable).row_count == 1


async def test_unexpected_error_in_action_is_notified_not_fatal(fake):
    fake.action_error = KeyError("key")
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press("s")
        await settle(app, pilot)
        assert errors(app)
        assert app.is_running and app.busy == set()


async def test_unexpected_error_in_listing_is_notified_not_fatal(fake, monkeypatch):
    def broken():
        raise ValueError("bad json")

    monkeypatch.setattr(cli, "list_guests", broken)
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        assert errors(app) and app.is_running


async def test_pending_check_failure_does_not_skip_confirmed_grow(fake, monkeypatch):
    def broken(guest):
        raise cli.CliError("Error: boom")

    monkeypatch.setattr(cli, "pending_keys", broken)
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, memory="4096", disk="scsi0", grow="5")
        await press_button(app, pilot, "#apply")
        await press_button(app, pilot, "#yes")
        assert fake.sets == [{"memory_mb": 4096}]
        assert fake.resizes == [(108, "scsi0", 5)]


async def test_declining_grow_cancels_whole_edit(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_edit(app, pilot)
        fill(app, cores="8", disk="scsi0", grow="5")
        await press_button(app, pilot, "#apply")
        await press_button(app, pilot, "#no")
        assert fake.sets == [] and fake.resizes == []


@pytest.mark.parametrize("key", ["S", "e"])
async def test_escape_closes_modal_without_cli_call(fake, key):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        await pilot.press(key)
        await settle(app, pilot)
        assert isinstance(app.screen, ModalScreen)
        await pilot.press("escape")
        await settle(app, pilot)
        assert not isinstance(app.screen, ModalScreen)
        assert fake.actions == [] and fake.sets == []


async def test_table_shows_live_status_after_action_not_lagging_list(fake):
    # /cluster/resources lags a few seconds behind; the acted-on guest must show live status
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        fake.live_status = {108: "stopped"}  # list_guests still says "running"
        await pilot.press("S")
        await settle(app, pilot)
        await press_button(app, pilot, "#yes")
        assert fake.status_calls == [108]
        assert "stopped" in [
            str(c) for c in app.query_one("#guests", DataTable).get_row("108")
        ]


# --- context switching ------------------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("Context switching" > "app.py (UI contract)")


async def open_contexts(app, pilot):
    await settle(app, pilot)
    await pilot.press("c")
    await settle(app, pilot)
    assert isinstance(app.screen, ContextScreen)
    return app.screen.query_one("#contexts", OptionList)


async def choose(app, pilot, index):
    ol = app.screen.query_one("#contexts", OptionList)
    ol.focus()
    ol.highlighted = index
    await pilot.pause()
    await pilot.press("enter")
    await settle(app, pilot)


async def test_sub_title_shows_active_context_after_mount(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        assert app.sub_title == "context: mainframe"


async def test_contexts_failure_notifies_and_sub_title_is_unknown(fake):
    fake.contexts_error = cli.CliError("Error: no config")
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        assert errors(app)
        assert app.sub_title == "context: ?"


async def test_c_opens_context_screen_listing_contexts_with_current_marked(fake):
    app = make_app()
    async with app.run_test() as pilot:
        ol = await open_contexts(app, pilot)
        assert ol.option_count == 2
        options = [ol.get_option_at_index(i) for i in range(2)]
        assert [o.id for o in options] == ["mainframe", "mainframe-edge"]
        assert "*" in str(options[0].prompt)
        assert "*" not in str(options[1].prompt)
        assert "mainframe-edge" in str(options[1].prompt)


async def test_selecting_other_context_switches_and_reloads(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_contexts(app, pilot)
        before = fake.list_calls
        await choose(app, pilot, 1)
        assert fake.use_calls == ["mainframe-edge"]
        assert app.sub_title == "context: mainframe-edge"
        assert fake.list_calls > before
        assert not isinstance(app.screen, ModalScreen)
        assert app.query_one("#guests", DataTable).row_count == 2


async def test_table_is_cleared_while_switch_is_in_progress(fake):
    started, release = threading.Event(), threading.Event()
    fake.use_hook = lambda: (started.set(), release.wait(10))
    app = make_app()
    async with app.run_test() as pilot:
        await open_contexts(app, pilot)
        assert app.query_one("#guests", DataTable).row_count == 2
        try:
            ol = app.screen.query_one("#contexts", OptionList)
            ol.focus()
            ol.highlighted = 1
            await pilot.pause()
            await pilot.press("enter")
            for _ in range(200):
                if started.is_set():
                    break
                await pilot.pause(0.01)
            assert started.is_set()
            await pilot.pause()
            assert app.query_one("#guests", DataTable).row_count == 0
        finally:
            release.set()
        await settle(app, pilot)
        assert app.query_one("#guests", DataTable).row_count == 2


async def test_selecting_current_context_makes_no_cli_call(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_contexts(app, pilot)
        before = fake.list_calls
        await choose(app, pilot, 0)
        assert fake.use_calls == []
        assert not isinstance(app.screen, ModalScreen)
        assert app.sub_title == "context: mainframe"
        assert fake.list_calls == before


async def test_escape_closes_context_screen_without_cli_call(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await open_contexts(app, pilot)
        await pilot.press("escape")
        await settle(app, pilot)
        assert not isinstance(app.screen, ModalScreen)
        assert fake.use_calls == []
        assert app.sub_title == "context: mainframe"


async def test_c_while_guest_busy_warns_and_opens_nothing(fake):
    started, release = threading.Event(), threading.Event()
    fake.action_hook = lambda: (started.set(), release.wait(10))
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        try:
            await pilot.press("s")
            for _ in range(200):
                if started.is_set():
                    break
                await pilot.pause(0.01)
            assert started.is_set()
            await pilot.press("c")
            await pilot.pause()
            assert not isinstance(app.screen, ContextScreen)
            assert any(sev == "warning" for _, sev in app.seen)
        finally:
            release.set()
        await settle(app, pilot)
        assert fake.use_calls == []


async def test_use_context_failure_notifies_rereads_context_and_reloads(fake):
    err = cli.CliError("Error: Context 'mainframe-edge' not found.")

    def hook():  # the CLI state differs from what the app assumed
        fake.current = "mainframe-edge"

    fake.use_hook, fake.use_error = hook, err
    app = make_app()
    async with app.run_test() as pilot:
        await open_contexts(app, pilot)
        contexts_before, list_before = fake.contexts_calls, fake.list_calls
        await choose(app, pilot, 1)
        assert any(err.summary in m for m in errors(app))
        assert fake.contexts_calls > contexts_before
        assert app.sub_title == "context: mainframe-edge"  # whatever contexts() says
        assert fake.list_calls > list_before
        assert app.query_one("#guests", DataTable).row_count == 2


# --- review finding: no old-cluster rows after a switch ----------------------

EDGE = {"vmid": 300, "name": "edge-vm", "node": "edge", "type": "qemu", "status": "running",
        "maxcpu": 2, "maxmem": 2**30, "maxdisk": 2**31}  # fmt: skip


async def until(pilot, cond):
    for _ in range(300):
        if cond():
            return
        await pilot.pause(0.01)
    raise AssertionError("condition not reached")


def blocking_once():
    started, release = threading.Event(), threading.Event()

    def hook():
        if not started.is_set():
            started.set()
            release.wait(10)

    return hook, started, release


def rows(app):
    return [k.value for k in app.query_one("#guests", DataTable).rows]


async def switch_to_edge(app, pilot):
    await pilot.press("c")
    await until(pilot, lambda: isinstance(app.screen, ContextScreen))
    ol = app.screen.query_one("#contexts", OptionList)
    ol.focus()
    ol.highlighted = 1
    await pilot.pause()
    await pilot.press("enter")


async def test_reload_in_flight_before_switch_cannot_restore_old_rows(fake):
    fake.guests_by_context = {"mainframe-edge": [EDGE]}
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        hook, started, release = blocking_once()
        fake.list_hook = hook
        try:
            await pilot.press("g")  # reads mainframe, then stalls
            await until(pilot, started.is_set)
            await switch_to_edge(app, pilot)
            await until(pilot, lambda: rows(app) == ["300"])
        finally:
            release.set()
        await settle(app, pilot)
        assert rows(app) == ["300"]
        assert app.sub_title == "context: mainframe-edge"


async def test_refresh_refused_while_switching(fake):
    started, release = threading.Event(), threading.Event()
    fake.use_hook = lambda: (started.set(), release.wait(10))
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        try:
            await switch_to_edge(app, pilot)
            await until(pilot, started.is_set)
            before = fake.list_calls
            await pilot.press("g")
            await pilot.pause()
            assert fake.list_calls == before
            assert any(sev == "warning" for _, sev in app.seen)
        finally:
            release.set()
        await settle(app, pilot)


async def test_context_refused_while_post_action_reload_runs(fake):
    app = make_app()
    async with app.run_test() as pilot:
        await settle(app, pilot)
        hook, started, release = blocking_once()
        fake.list_hook = hook
        try:
            await pilot.press("s")  # start is instant; the reload after it stalls
            await until(pilot, started.is_set)
            await pilot.press("c")
            await pilot.pause(0.1)
            assert not isinstance(app.screen, ContextScreen)
            assert any(sev == "warning" for _, sev in app.seen)
        finally:
            release.set()
        await settle(app, pilot)


# --- LXC shell --------------------------------------------------------------
# spec: docs/plans/guest-tui-mvp.md ("LXC shell" > "app.py (UI contract)")

RUNNING_CT = {"vmid": 104, "name": "debian", "node": "mainframe", "type": "lxc", "status": "running",
              "maxcpu": 1, "maxmem": 512 * 2**20, "maxdisk": 8 * 2**30}  # fmt: skip
SSH_ARGV = ["ssh", "-t", "root@mainframe", "pct", "enter", "104"]


def patch_shell(monkeypatch, app, rc=0, error=None):
    """Replace subprocess.run and app.suspend with recorders; returns the event log."""
    events = []

    def run(argv, **kwargs):
        events.append(("run", list(argv)))
        if error:
            raise error
        return subprocess.CompletedProcess(argv, rc)

    @contextlib.contextmanager
    def suspend():
        events.append(("enter", None))
        yield
        events.append(("exit", None))

    monkeypatch.setattr("cv4pve_tui.app.subprocess.run", run)
    app.suspend = suspend
    return events


def ssh_runs(events):
    return [argv for kind, argv in events if kind == "run"]


async def press_t(app, pilot):
    await settle(app, pilot)
    await pilot.press("t")
    await settle(app, pilot)


async def test_t_on_running_lxc_runs_ssh_pct_enter_inside_suspend(fake, monkeypatch):
    # spec: runs the ssh argv once, on the main thread, inside `with self.suspend()`
    fake.guests = [RUNNING_CT]
    app = make_app()
    events = patch_shell(monkeypatch, app)
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert events == [("enter", None), ("run", SSH_ARGV), ("exit", None)]
        assert not errors(app)
        assert app.is_running


async def test_shell_exit_nonzero_notifies_error_with_guest_and_code(fake, monkeypatch):
    # spec: non-zero exit -> error notify naming the guest and the exit code
    fake.guests = [RUNNING_CT]
    app = make_app()
    events = patch_shell(monkeypatch, app, rc=255)
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert ssh_runs(events) == [SSH_ARGV]
        assert any(("debian" in m or "104" in m) and "255" in m for m in errors(app))
        assert app.is_running


@pytest.mark.parametrize("guest,why", [(VM, "not a container"), (CT, "not running")])
async def test_t_on_ineligible_guest_warns_and_runs_nothing(
    fake, monkeypatch, guest, why
):
    # spec: only type lxc and status running; otherwise warning (says why), nothing runs
    fake.guests = [guest]
    app = make_app()
    events = patch_shell(monkeypatch, app)
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert events == []
        assert any(sev == "warning" for _, sev in app.seen)
        assert not errors(app)


async def test_t_on_empty_table_does_nothing(fake, monkeypatch):
    # spec: empty table -> nothing
    fake.guests = []
    app = make_app()
    events = patch_shell(monkeypatch, app)
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert events == []
        assert app.seen == []
        assert app.is_running


async def test_t_on_busy_guest_warns_and_runs_nothing(fake, monkeypatch):
    # spec: busy guest (action running) -> warning notify, nothing runs
    started, release = threading.Event(), threading.Event()
    fake.guests = [RUNNING_CT]
    fake.action_hook = lambda: (started.set(), release.wait(10))
    app = make_app()
    events = patch_shell(monkeypatch, app)
    async with app.run_test() as pilot:
        await settle(app, pilot)
        try:
            await pilot.press("s")
            await until(pilot, started.is_set)
            warnings_before = sum(sev == "warning" for _, sev in app.seen)
            await pilot.press("t")
            await pilot.pause()
            assert events == []
            assert sum(sev == "warning" for _, sev in app.seen) > warnings_before
        finally:
            release.set()
        await settle(app, pilot)
        assert events == []


async def test_suspend_not_supported_notifies_error_and_app_keeps_running(
    fake, monkeypatch
):
    # spec: SuspendNotSupported (headless driver, suspend left unpatched) -> error notify
    fake.guests = [RUNNING_CT]
    app = make_app()
    calls = []
    monkeypatch.setattr(
        "cv4pve_tui.app.subprocess.run", lambda argv, **kw: calls.append(argv)
    )
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert calls == []
        assert errors(app)
        assert app.is_running


async def test_missing_ssh_binary_notifies_error_and_app_keeps_running(
    fake, monkeypatch
):
    # spec: OSError (no ssh binary) -> error notify, app keeps running
    fake.guests = [RUNNING_CT]
    app = make_app()
    events = patch_shell(monkeypatch, app, error=FileNotFoundError("ssh"))
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert ssh_runs(events) == [SSH_ARGV]
        assert errors(app)
        assert app.is_running


# --- review findings: LXC shell ----------------------------------------------


async def test_missing_ssh_binary_still_resumes_app(fake, monkeypatch):
    # Textual's suspend() has no try/finally: an exception inside it skips the resume
    fake.guests = [RUNNING_CT]
    app = make_app()
    events = patch_shell(monkeypatch, app, error=FileNotFoundError("ssh"))
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert events[-1] == ("exit", None)


async def test_shell_exit_code_from_inside_container_is_not_an_error(fake, monkeypatch):
    # pct enter returns the shell's last status; only 255 means ssh itself failed
    fake.guests = [RUNNING_CT]
    app = make_app()
    patch_shell(monkeypatch, app, rc=1)
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert errors(app) == []


async def test_table_reloads_after_shell(fake, monkeypatch):
    fake.guests = [RUNNING_CT]
    app = make_app()
    patch_shell(monkeypatch, app)
    async with app.run_test() as pilot:
        await settle(app, pilot)
        before = fake.list_calls
        await press_t(app, pilot)
        await settle(app, pilot)
        assert fake.list_calls > before


async def test_terminal_is_reset_inside_suspend_before_ssh(fake, monkeypatch):
    # user request: a clean screen (and no leftover TUI terminal modes) in the container shell
    fake.guests = [RUNNING_CT]
    app = make_app()
    events = patch_shell(monkeypatch, app)
    resets = []
    monkeypatch.setattr(
        "cv4pve_tui.app._reset_terminal", lambda: resets.append(list(events))
    )
    async with app.run_test() as pilot:
        await press_t(app, pilot)
        assert resets == [[("enter", None)]]  # inside suspend, before ssh ran