Add LXC shell via ssh and pct enter

Press t on a running container to suspend the TUI and run
`ssh -t root@<node> pct enter <vmid>`. cv4pve-cli has no shell or exec
for LXC, so this goes around it and needs key-based root SSH to the
PVE node.

- VMs, stopped containers and busy guests get a warning instead.
- Only exit code 255 (ssh failed) is reported as an error. Other codes
  are the last status of the container's shell.
- OSError is caught inside suspend(). Textual's suspend() has no
  try/finally, so an exception escaping it would leave the app
  suspended.
- The guest table reloads when the shell returns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DuckbootandClaude Opus 5.5 committed 2026-11-02 20:35:22 +01:00
1 parent 62b79ab29e
commit 40feb125b4
4 files changed
+272 -1

No files matched your search

+180
View File
@@ -3,6 +3,7 @@
All cli functions are monkeypatched: cv4pve-cli is never executed.
"""
import contextlib
import subprocess
import threading
@@ -727,3 +728,182 @@ 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