Mark the LXC shell interactive test as done in the plan

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

README.md

@@ -20,10 +20,13 @@ uv run cv4pve-tui
 | `x` | stop - hard power-off (confirm) |
 | `r` | reboot (confirm) |
 | `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) |
+| `t` | shell in a running LXC container (`ssh -t root@<node> pct enter <vmid>`) |
 | `g` | refresh |
 | `c` | switch cv4pve-cli context (cluster) |
 | `q` | quit |
 
+The shell key needs key-based root SSH to the PVE node (LXC only).
+
 Switching context runs `cv4pve-cli config use`, which changes cv4pve-cli's global current
 context: other shells and scripts follow, and it stays after the app exits.
 

docs/plans/guest-tui-mvp.md

@@ -179,3 +179,60 @@ Adding/editing/deleting contexts, `config verify`, restoring the original contex
       (only `config use` and read-only listing; no guest actions); final context
       is `mainframe`.
 - [ ] Push `feat/guest-tui-mvp`, merge to `main` (user-approved).
+
+---
+
+# LXC shell
+
+## Problem
+
+Open an interactive shell in a running LXC container from the TUI. cv4pve-cli 2.4.0 has
+no console/shell/exec for LXC (verified: no command; the Proxmox API has no LXC exec),
+so the app goes around it: SSH to the PVE node as root and run `pct enter <vmid>`.
+
+## Decision
+
+Option 1 from the discussion: `ssh -t root@<node> pct enter <vmid>`. The node name is
+the SSH host (resolves via DNS here: `mainframe`, `mainframe-edge`); the user's key is in
+root's `authorized_keys` (verified: `ssh -o BatchMode=yes root@mainframe pct status 104`
+-> `status: running`). This bypasses cv4pve-cli and its API token.
+
+## `app.py` (UI contract)
+
+- Key `t` ("Shell") on the selected guest.
+- Only for `type == "lxc"` and `status == "running"`; otherwise a warning notify that
+  says why (not a container / not running) and nothing runs. Empty table: nothing.
+- Busy guest (action running) -> warning notify, nothing runs.
+- Runs on the main thread inside `with self.suspend():` -
+  `subprocess.run(["ssh", "-t", f"root@{node}", "pct", "enter", str(vmid)])`
+  (no capture, no timeout: the user's terminal is handed over; `import subprocess` in
+  app.py so tests patch `cv4pve_tui.app.subprocess.run`).
+- Exit 255 (ssh itself failed) -> error notify naming the guest and the code. Other
+  exit codes are the container shell's last status (`exit` after a failed command
+  gives 1) and are not errors.
+- Inside `suspend()`, right before ssh, the terminal gets a full reset (`ESC c`, what
+  `reset` sends) so the container shell starts on a clean screen without leftover TUI
+  terminal modes (user request). Done locally, not in the container: no dependency on
+  `reset` existing there, and `pct enter` stays unchanged.
+- `OSError` from `subprocess.run` is caught *inside* the `suspend()` block: Textual
+  8.2.8's `suspend()` has no try/finally, so an exception escaping it leaves the app
+  suspended.
+- The guest table reloads after the shell returns (the user may have stopped or
+  rebooted the container from inside).
+- `textual.app.SuspendNotSupported` (headless / unsupported driver) and `OSError`
+  (no `ssh` binary) -> error notify, app keeps running.
+- Tests: `App.suspend` raises `SuspendNotSupported` under `run_test`, so tests replace
+  `app.suspend` with a no-op context manager (e.g. `contextlib.nullcontext`) unless
+  testing that error path.
+
+## Out of scope
+
+VM consoles (serial/VNC), configurable SSH user/host/options, termproxy/websocket.
+
+## Steps
+
+- [x] Tests for the UI contract above - red first.
+- [x] Implementation - full suite green, ruff clean. README key table gets `t`.
+- [x] Interactive test by the user: `t` on CT 104, `exit`, back in the table; screen
+      clean after the reset.
+- [ ] Push `feat/lxc-shell`, merge to `main` (user-approved).

src/cv4pve_tui/app.py

@@ -1,7 +1,10 @@
 """Textual app for managing Proxmox guests through cv4pve-cli."""
 
+import subprocess
+import sys
+
 from textual import work
-from textual.app import App, ComposeResult
+from textual.app import App, ComposeResult, SuspendNotSupported
 from textual.containers import Horizontal, Vertical
 from textual.screen import ModalScreen
 from textual.widgets import (
@@ -132,6 +135,7 @@ class GuestApp(App):
         ("x", "act('stop')", "Stop"),
         ("r", "act('reboot')", "Reboot"),
         ("e", "edit", "Edit"),
+        ("t", "shell", "Shell"),
         ("g", "reload", "Refresh"),
         ("c", "context", "Context"),
         ("q", "quit", "Quit"),
@@ -336,6 +340,40 @@ class GuestApp(App):
             return
         self._open_edit(guest)
 
+    def action_shell(self) -> None:
+        guest = self._selected()
+        if guest is None:
+            return
+        who = f"{guest['name']} ({guest['vmid']})"
+        if guest["type"] != "lxc":
+            self.notify(f"{who} is not a container", severity="warning")
+            return
+        if guest["status"] != "running":
+            self.notify(f"{who} is not running", severity="warning")
+            return
+        if guest["vmid"] in self.busy:
+            self.notify(f"{who} is busy", severity="warning")
+            return
+        cmd = ["ssh", "-t", f"root@{guest['node']}", "pct", "enter", str(guest["vmid"])]
+        error = None
+        try:
+            with self.suspend():
+                try:  # inside: Textual's suspend() skips the resume if its body raises
+                    _reset_terminal()
+                    code = subprocess.run(cmd, check=False).returncode
+                except OSError as exc:
+                    error = _message(exc)
+        except SuspendNotSupported as exc:
+            error = _message(exc)
+        else:
+            if (
+                error is None and code == 255
+            ):  # other codes are the shell's own last status
+                error = "ssh failed (exit 255)"
+        if error:
+            self.notify(f"Shell in {who} failed: {error}", severity="error")
+        self.action_reload()
+
     @work(thread=True)
     def _open_edit(self, guest: dict) -> None:
         try:
@@ -390,6 +428,12 @@ class GuestApp(App):
         self._start_job(guest, "edit", job)
 
 
+def _reset_terminal() -> None:
+    """Full terminal reset (ESC c, what `reset` sends): clean screen, no leftover TUI modes."""
+    sys.stdout.write("\033c")
+    sys.stdout.flush()
+
+
 def _scaled(value, unit: int):
     return "-" if value is None else round(value / unit)
 

tests/test_app.py

@@ -3,6 +3,7 @@
 All cli functions are monkeypatched: cv4pve-cli is never executed.
 """
 
+import contextlib
 import subprocess
 import threading
 
@@ -727,3 +728,196 @@ async def test_context_refused_while_post_action_reload_runs(fake):
         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