From 1ed5e5069b0c147ebb4adbcee9c6d4866c054643 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Tue, 3 Nov 2026 00:03:58 +0100 Subject: [PATCH] Add the task log: recent tasks with status, Enter shows the log l on the main screen opens "Tasks", the recent cluster tasks newest first. l on the info screen shows only that guest's tasks. Columns: start, duration (or "running"), node, type, VMID ("-" for node tasks), user and status. A failed task gets a "!" in the leading column. Enter opens the task's log in a plain-text widget, so the output is never parsed as markup. `task list -o json` has no start or end times, so the list comes from `/cluster/tasks`. The upid is validated (it must start with "UPID:") before it goes in as a positional argument. Review fixes: - A finished task without a start time no longer crashes the app; its duration shows "-". - "running" now means there is no endtime. A finished task with an empty status shows "unknown" and is flagged. Read-only live check: 25 tasks listed, and the log of the newest (vzdelsnapshot 104) ends with TASK OK. Closes #10 Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/plans/guest-tui-mvp.md | 40 +++++ src/cv4pve_tui/app.py | 127 ++++++++++++++ src/cv4pve_tui/cli.py | 12 ++ tests/test_app.py | 321 ++++++++++++++++++++++++++++++++++++ tests/test_cli.py | 45 +++++ 6 files changed, 546 insertions(+) diff --git a/README.md b/README.md index f7dbee2..1a65006 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ uv run cv4pve-tui | `Enter` then `n` | in info: edit network (bridge, VLAN, firewall, IPv4/IPv6; VM IP via cloud-init at next boot); `u` there = undo the last network change of this guest (this session only) | | `Enter` then `p` / `R` / `D` | in info: take a snapshot (name, description, VM: include RAM) / roll back to the snapshot under the cursor (confirms; starts the guest again if it is running) / delete the snapshot under the cursor (confirms) | | `t` | shell in a running LXC container (`ssh -t root@ pct enter `) | +| `l` | tasks: recent cluster tasks (start, duration, node, type, VMID, user, status; `!` marks failures); `Enter` shows the log of the task under the cursor, `g` reloads. In info: `l` = only this guest's tasks | | `g` | refresh | | `c` | switch cv4pve-cli context (cluster) | | `q` | quit | diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index 8ba747f..f140b7b 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -551,3 +551,43 @@ CLI facts (read-only probes): - otherwise `-`. A failure of `cli.current` keeps today's behaviour (inline "unavailable", status unknown -> not running). + +--- + +# Task log (issue #10) + +## Decision (user-approved) + +`l` on the main screen: all recent cluster tasks. `l` on the info screen: the tasks of +that guest. Enter on a task shows its log. + +CLI facts (read-only probes): + +- `task list -o json` lacks start/end times. `api get /cluster/tasks -o json` has them: + list of dicts `upid`, `type` (e.g. `vzdump`, `qmstart`, `vzsnapshot`), `id` (VMID as a + string, `""` for node tasks), `node`, `user`, `starttime`, `endtime` (epoch; missing + or null while running), `status` (`OK`, an error text, or missing while running), + `saved`, `tokenid`. +- `task log ` prints the log as plain text lines (e.g. `TASK OK` at the end). + +### `cli.py` + +- `tasks() -> list[dict]` - `/cluster/tasks`, newest first by `starttime`. +- `task_log(upid) -> str` - `task log `; ValueError unless `upid` starts with + `UPID:` (positional argument). + +### `app.py` (UI contract) + +- `TasksScreen(ModalScreen)`, opened by `l` on the main screen (all tasks) and on the + info screen (`vmid` filter: only tasks whose `id == str(vmid)`). Frame `#tasks_section` + titled "Tasks" or "Tasks: ()". `DataTable#tasks` (row cursor): Start + (`YYYY-MM-DD HH:MM` local), Duration (`1h 2m`, `3m 12s`, `45s`; `running` without + endtime), Node, Type, VMID (`-` for node tasks), User, Status (`OK`, the error text, or + `running`). A failed task (status present and not `OK`) gets a `!` in a leading + column. `Static#tasks_note` says "No tasks" when empty. Loaded in a thread worker; + failure -> error notify, screen stays. Keys: `enter` open the log, `g` reload, + `escape` close. +- `TaskLogScreen` (DialogScreen style, wide box), title "Log: ", + a read-only plain-text log widget `#tasklog` (no markup parsing), loaded in a worker; + failure -> error notify. `escape` closes. +- Main-screen footer shows `l Tasks`. diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index c04b76a..a801c60 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -18,6 +18,7 @@ from textual.widgets import ( Header, Input, Label, + Log, OptionList, Select, Static, @@ -417,6 +418,124 @@ class NetScreen(DialogScreen): ) +class TaskLogScreen(DialogScreen): + """The log of one task, as plain text (task output is never parsed as markup).""" + + BINDINGS = [("escape", "dismiss", "Close")] # noqa: RUF012 - Textual idiom + DEFAULT_CSS = """ + TaskLogScreen #dialog { width: 120; height: 90%; } + TaskLogScreen Log { height: 1fr; } + """ + + def __init__(self, entry: dict) -> None: + super().__init__() + self.entry = entry + + def compose(self) -> ComposeResult: + t = self.entry + with self.box(f"Log: {t.get('type', '')} {t.get('id') or t.get('node', '')}"): + yield Log(id="tasklog") + + def on_mount(self) -> None: + self._load_worker() + + @work(thread=True) + def _load_worker(self) -> None: + try: + text = cli.task_log(self.entry["upid"]) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.app.call_from_thread(self.app.notify, _message(exc), severity="error") + return + self.app.call_from_thread(self.query_one("#tasklog", Log).write, text) + + +def _duration(task: dict) -> str: + end, start = task.get("endtime"), task.get("starttime") + if not end: + return "running" + if not start: + return "-" # finished, but the start is unknown + secs = max(0, int(end - start)) + if secs >= 3600: + return f"{secs // 3600}h {secs % 3600 // 60}m" + return f"{secs // 60}m {secs % 60}s" if secs >= 60 else f"{secs}s" + + +class TasksScreen(ModalScreen): + """Recent cluster tasks, or those of one guest (vmid); Enter shows a task's log.""" + + BINDINGS = [ # noqa: RUF012 - Textual idiom + ("escape", "dismiss", "Close"), + ("g", "reload", "Refresh"), + ] + DEFAULT_CSS = """ + TasksScreen { align: center middle; } + TasksScreen #tasks_section { + width: 100%; height: 100%; border: round $accent; border-title-style: bold; + padding: 0 1; background: $surface; + } + TasksScreen DataTable { height: 1fr; } + """ + + def __init__(self, guest: dict | None = None) -> None: + super().__init__() + self.guest = guest + + def compose(self) -> ComposeResult: + title = "Tasks" + if self.guest: + title += f": {self.guest['name']} ({self.guest['vmid']})" + with Vertical(id="tasks_section") as section: + section.border_title = Text(title) # guest names: never markup + yield DataTable(id="tasks", cursor_type="row") + yield Static("", id="tasks_note", markup=False) + yield Footer() + + def on_mount(self) -> None: + self.rows: list[dict] = [] + self.query_one("#tasks", DataTable).add_columns( + "", "Start", "Duration", "Node", "Type", "VMID", "User", "Status" + ) + self.action_reload() + + def action_reload(self) -> None: + self._load_worker() + + @work(thread=True) + def _load_worker(self) -> None: + try: + tasks = cli.tasks() + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.app.call_from_thread(self.app.notify, _message(exc), severity="error") + return + if self.guest: + tasks = [t for t in tasks if t.get("id") == str(self.guest["vmid"])] + self.app.call_from_thread(self._fill, tasks) + + def _fill(self, tasks: list[dict]) -> None: + self.rows = tasks + table = self.query_one("#tasks", DataTable) + table.clear() + for t in tasks: + # running = no endtime yet; a finished task without a status text is "unknown" + status = "running" if not t.get("endtime") else t.get("status") or "unknown" + _add_plain( + table, + "!" if status not in ("OK", "running") else "", + _when(t.get("starttime")), + _duration(t), + t.get("node", ""), + t.get("type", ""), + t.get("id") or "-", + t.get("user", ""), + status, + ) + self.query_one("#tasks_note", Static).update("" if tasks else "No tasks") + + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + self.app.push_screen(TaskLogScreen(self.rows[event.cursor_row])) + + class InfoScreen(ModalScreen): """Backups, backup job and network of one guest (read-only), plus backup now.""" @@ -431,6 +550,7 @@ class InfoScreen(ModalScreen): ("p", "snapshot", "Take snapshot"), ("R", "rollback", "Roll back"), ("D", "delete_snapshot", "Delete snapshot"), + ("l", "tasks", "Tasks"), ] DEFAULT_CSS = """ InfoScreen DataTable { height: auto; max-height: 12; } @@ -501,6 +621,9 @@ class InfoScreen(ModalScreen): def action_reload(self) -> None: self._load_worker() + def action_tasks(self) -> None: + self.app.push_screen(TasksScreen(self.guest)) + @work(thread=True) def _load_worker(self) -> None: self.load() @@ -808,6 +931,7 @@ class GuestApp(App): ("t", "shell", "Shell"), ("g", "reload", "Refresh"), ("c", "context", "Context"), + ("l", "tasks", "Tasks"), ("q", "quit", "Quit"), ] @@ -856,6 +980,9 @@ class GuestApp(App): self._show_context() self.action_reload() + def action_tasks(self) -> None: + self.push_screen(TasksScreen()) + def _show_context_frame(self, sub_title: str) -> None: self.context_name = sub_title.removeprefix("context: ") self._update_frame() diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index 920d8e2..75be38a 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -448,3 +448,15 @@ def delete_snapshot(guest: dict, name: str) -> None: if args[2] != "snapshot" or args[4:6] != (str(guest["vmid"]), name): raise RuntimeError("refusing to send a delete that is not a snapshot delete") run(*args) + + +def tasks() -> list[dict]: + """Recent cluster tasks, newest first; running ones lack endtime and status.""" + items = json.loads(run("api", "get", "/cluster/tasks", "-o", "json")) + return sorted(items, key=lambda t: t.get("starttime") or 0, reverse=True) + + +def task_log(upid: str) -> str: + if not isinstance(upid, str) or not upid.startswith("UPID:"): + raise ValueError("not a task id (UPID)") + return run("task", "log", upid) diff --git a/tests/test_app.py b/tests/test_app.py index 68519e6..2d03797 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -74,6 +74,12 @@ class Fake: # issue #12 (live usage); NB the name "current" is taken by the context-name attribute above self.usage = {} # vmid -> overrides merged into the status/current dict self.current_error, self.current_calls = None, [] + # issue #10 (task log) + self.task_list = [ + dict(t) for t in TASKS + ] # newest first, as cli.tasks returns them + self.tasks_error, self.tasks_calls = None, 0 + self.log_text, self.log_error, self.log_calls = "TASK OK\n", None, [] def list_guests(self): self.list_calls += 1 @@ -213,7 +219,31 @@ class Fake: def delete_snapshot(self, guest, name): self._snap_op(self.delete_calls, (guest["vmid"], name)) + def tasks(self): + self.tasks_calls += 1 + if self.tasks_error: + raise self.tasks_error + return [dict(t) for t in self.task_list] + def task_log(self, upid): + self.log_calls.append(upid) + if self.log_error: + raise self.log_error + return self.log_text + + +T_TASK = 1_762_600_000 +TASKS = [ # newest first; one running (no endtime/status), one failed, one node task + {"upid": "UPID:mainframe:0001:qmstop:108:root@pam:", "type": "qmstop", "id": "108", "node": "mainframe", + "user": "root@pam", "starttime": T_TASK + 1000}, + {"upid": "UPID:mainframe:0002:qmstart:108:root@pam:", "type": "qmstart", "id": "108", "node": "mainframe", + "user": "root@pam", "starttime": T_TASK, "endtime": T_TASK + 192, "status": "OK"}, + {"upid": "UPID:mainframe:0003:aptupdate::root@pam:", "type": "aptupdate", "id": "", "node": "mainframe", + "user": "root@pam", "starttime": T_TASK - 5000, "endtime": T_TASK - 5000 + 3720, "status": "OK"}, + {"upid": "UPID:mainframe:0004:vzsnapshot:100:root@pam:", "type": "vzsnapshot", "id": "100", "node": "mainframe", + "user": "root@pam", "starttime": T_TASK - 9000, "endtime": T_TASK - 9000 + 45, + "status": "snapshot 'x' already exists"}, +] # fmt: skip BACKUP_JOB = {"id": "backup-1", "enabled": 1, "storage": "truenas-backup", "all": 1, "vmid": None, "exclude": None, "node": None, "schedule": "mon,wed,fri 04:00", "mode": "snapshot", "compress": "zstd", "prune-backups": {"keep-last": "3"}} # fmt: skip @@ -255,6 +285,8 @@ def fake(monkeypatch): "take_snapshot", "rollback_snapshot", "delete_snapshot", + "tasks", + "task_log", ): monkeypatch.setattr(cli, name, getattr(f, name), raising=False) for name in ("list_guests", "guest_config", "disks", "action", "set_resources", @@ -2850,3 +2882,292 @@ async def test_table_usage_of_acted_guest_comes_from_live_status(fake): row = [str(c) for c in app.query_one("#guests", DataTable).get_row("100")] assert "running" in row assert row[-2:] == ["50.0", "50.0"] + + +# --- issue #10: task log ----------------------------------------------------------- +# spec: docs/plans/guest-tui-mvp.md ("Task log (issue #10)" > app.py) + +TASK_COLUMNS = 8 # leading "!" column + Start, Duration, Node, Type, VMID, User, Status + + +async def open_tasks(app, pilot): + await pilot.press("l") + await settle(app, pilot) + assert isinstance(app.screen, appmod.TasksScreen) + return app.screen.query_one("#tasks", DataTable) + + +def log_text(widget): + """Text of the #tasklog widget, whatever the widget class (the spec only says read-only plain text).""" + if hasattr(widget, "text"): + return widget.text + if hasattr(widget, "lines"): + return "\n".join(str(getattr(ln, "text", ln)) for ln in widget.lines) + return str(widget.render()) + + +def local_task_date(epoch): + return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M") + + +async def test_l_on_main_screen_lists_all_tasks_newest_first(fake): # `l` main screen + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + assert table.row_count == 4 + assert [cells(table, i)[4] for i in range(4)] == [ + "qmstop", + "qmstart", + "aptupdate", + "vzsnapshot", + ] + assert app.screen.query_one("#tasks_section").border_title == "Tasks" + assert table.cursor_type == "row" + + +async def test_tasks_columns_and_formatting( + fake, +): # columns, duration, VMID "-", Status + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + assert len(table.columns) == TASK_COLUMNS + running, ok, node_task, failed = (cells(table, i) for i in range(4)) + assert ok[1:] == [ + local_task_date(T_TASK), + "3m 12s", + "mainframe", + "qmstart", + "108", + "root@pam", + "OK", + ] + assert running[1:] == [local_task_date(T_TASK + 1000), "running", "mainframe", "qmstop", "108", + "root@pam", "running"] # fmt: skip + assert node_task[2] == "1h 2m" + assert node_task[5] == "-" # node task has no VMID + assert failed[2] == "45s" + assert failed[7] == "snapshot 'x' already exists" + + +async def test_failed_task_gets_bang_in_leading_column_only(fake): # "!" for failed + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + marks = [cells(table, i)[0].strip() for i in range(4)] + assert marks == ["", "", "", "!"] # running, OK, OK, failed + + +async def test_l_on_info_screen_shows_only_that_guests_tasks( + fake, +): # `l` info screen, vmid filter + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) # VM 108 docker + table = await open_tasks(app, pilot) + assert [cells(table, i)[4] for i in range(table.row_count)] == [ + "qmstop", + "qmstart", + ] + assert ( + app.screen.query_one("#tasks_section").border_title == "Tasks: docker (108)" + ) + + +async def test_info_tasks_filter_matches_vmid_exactly( + fake, +): # id == str(vmid), not a prefix + fake.task_list = [ + dict(TASKS[1], id="1080"), + dict(TASKS[1], id="10"), + dict(TASKS[1], id="108"), + ] + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + table = await open_tasks(app, pilot) + assert [cells(table, i)[5] for i in range(table.row_count)] == ["108"] + + +async def test_no_tasks_note_when_empty(fake): # Static#tasks_note "No tasks" + fake.task_list = [] + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + assert table.row_count == 0 + assert "No tasks" in str(app.screen.query_one("#tasks_note", Static).render()) + + +async def test_info_tasks_empty_for_guest_without_tasks( + fake, +): # filter leaving nothing -> "No tasks" + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info( + app, pilot, down=1 + ) # CT 100; only a vzsnapshot task of 100 exists + table = await open_tasks(app, pilot) + assert table.row_count == 1 + fake.task_list = [t for t in TASKS if t["id"] != "100"] + await pilot.press("g") + await settle(app, pilot) + assert table.row_count == 0 + assert "No tasks" in str(app.screen.query_one("#tasks_note", Static).render()) + + +async def test_tasks_failure_notifies_error_and_app_keeps_running( + fake, +): # failure -> error notify + fake.tasks_error = cli_error("Error: tasks boom") + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await open_tasks(app, pilot) + assert any("boom" in e for e in errors(app)) + assert isinstance(app.screen, appmod.TasksScreen) and app.is_running + + +async def test_enter_opens_log_of_the_cursor_tasks_upid(fake): # Enter -> TaskLogScreen + fake.log_text = "starting\nTASK OK\n" + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await open_tasks(app, pilot) + await pilot.press("down") # second row: qmstart + await pilot.press("enter") + await settle(app, pilot) + assert isinstance(app.screen, appmod.TaskLogScreen) + assert fake.log_calls == [TASKS[1]["upid"]] + assert "TASK OK" in log_text(app.screen.query_one("#tasklog")) + assert str(app.screen.query_one("#dialog").border_title) == "Log: qmstart 108" + + +async def test_log_title_for_node_task_names_the_node( + fake, +): # title " " + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await open_tasks(app, pilot) + await pilot.press("down", "down", "enter") # aptupdate, no vmid + await settle(app, pilot) + assert ( + str(app.screen.query_one("#dialog").border_title) + == "Log: aptupdate mainframe" + ) + + +async def test_log_with_markup_like_text_is_shown_literally(fake): # no markup parsing + fake.log_text = "error in [/b] and [bold]x[/bold] and [red\nTASK OK\n" + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await open_tasks(app, pilot) + await pilot.press("enter") + await settle(app, pilot) + assert isinstance(app.screen, appmod.TaskLogScreen) and app.is_running + shown = log_text(app.screen.query_one("#tasklog")) + assert "error in [/b] and [bold]x[/bold] and [red" in shown + + +async def test_task_log_failure_notifies_error( + fake, +): # task_log failure -> error notify + fake.log_error = cli_error("Error: log boom") + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await open_tasks(app, pilot) + await pilot.press("enter") + await settle(app, pilot) + assert any("boom" in e for e in errors(app)) + assert app.is_running + + +async def test_g_reloads_tasks(fake): # `g` reload + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + before = fake.tasks_calls + fake.task_list = fake.task_list[:2] + await pilot.press("g") + await settle(app, pilot) + assert fake.tasks_calls == before + 1 + assert table.row_count == 2 + + +async def test_escape_closes_log_then_tasks_then_returns_to_info( + fake, +): # escape closes each screen + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await open_info(app, pilot) + await open_tasks(app, pilot) + await pilot.press("enter") + await settle(app, pilot) + assert isinstance(app.screen, appmod.TaskLogScreen) + await pilot.press("escape") + await settle(app, pilot) + assert isinstance(app.screen, appmod.TasksScreen) + await pilot.press("escape") + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + + +async def test_escape_closes_tasks_screen_on_main(fake): # escape closes + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await open_tasks(app, pilot) + await pilot.press("escape") + await settle(app, pilot) + assert not isinstance(app.screen, appmod.TasksScreen) + assert app.is_running + + +async def test_footer_on_main_screen_shows_l_tasks(fake): # footer `l Tasks` + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + shown = [ab.binding for ab in app.active_bindings.values() if ab.binding.show] + assert any(b.key == "l" and b.description == "Tasks" for b in shown) + + +# --- #10 review findings ---------------------------------------------------------------- + + +async def test_task_with_endtime_but_no_starttime_does_not_crash(fake): + fake.task_list = [ + {"upid": "UPID:mainframe:a:vzdump::root@pam:", "type": "vzdump", "id": "", "node": "mainframe", + "user": "root@pam", "endtime": T_TASK, "status": "OK"}, + {"upid": "UPID:mainframe:b:vzdump::root@pam:", "type": "vzdump", "id": "", "node": "mainframe", + "user": "root@pam", "starttime": None, "endtime": T_TASK, "status": "OK"}, + ] # fmt: skip + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + assert table.row_count == 2 and app.is_running + assert [str(table.get_row_at(i)[2]) for i in range(2)] == [ + "-", + "-", + ] # duration unknown + + +async def test_finished_task_with_empty_status_is_not_shown_as_running(fake): + fake.task_list = [ + {"upid": "UPID:mainframe:c:qmstart:108:root@pam:", "type": "qmstart", "id": "108", "node": "mainframe", + "user": "root@pam", "starttime": T_TASK, "endtime": T_TASK + 5, "status": ""}, + ] # fmt: skip + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + table = await open_tasks(app, pilot) + row = [str(c) for c in table.get_row_at(0)] + assert "running" not in row + assert row[0] == "!" # finished without OK: flagged diff --git a/tests/test_cli.py b/tests/test_cli.py index 834bf0e..6c9a0bd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1295,3 +1295,48 @@ def test_take_snapshot_refuses_description_that_looks_like_an_option(fake): {"vmid": 104, "node": "mainframe", "type": "lxc"}, "snap1", "-y" ) assert fake.calls == [] + + +# --- issue #10: task log (spec: "Task log (issue #10)" > cli.py) ------------------------ + +UPID = "UPID:mainframe:0001:00A:68F:qmstart:108:root@pam:" + + +def test_tasks_argv_is_cluster_tasks_json(fake): # tasks(): api get /cluster/tasks + fake.result = ("[]", "", 0) + assert cli.tasks() == [] + assert fake.argv == ["cv4pve-cli", "api", "get", "/cluster/tasks", "-o", "json"] + + +def test_tasks_newest_first_including_running_without_endtime( + fake, +): # tasks(): newest first + raw = [ + {"upid": "a", "starttime": 100, "endtime": 150, "status": "OK"}, + {"upid": "c", "starttime": 300}, # running: no endtime, no status + {"upid": "b", "starttime": 200, "endtime": None}, + ] + fake.result = (json.dumps(raw), "", 0) + assert [t["upid"] for t in cli.tasks()] == ["c", "b", "a"] + + +def test_tasks_failure_raises_clierror(monkeypatch): # failure surfaces as CliError + FakeRun(monkeypatch, "Error: boom", returncode=1) + with pytest.raises(cli.CliError): + cli.tasks() + + +def test_task_log_argv_and_returns_text(fake): # task_log(): task log + fake.result = ("starting\nTASK OK\n", "", 0) + out = cli.task_log(UPID) + assert fake.argv == ["cv4pve-cli", "task", "log", UPID] + assert "TASK OK" in out + + +@pytest.mark.parametrize("bad", ["-y", "--help", "", "108", "x UPID:a"]) +def test_task_log_rejects_non_upid_without_calling_subprocess( + fake, bad +): # ValueError unless UPID: + with pytest.raises(ValueError): + cli.task_log(bad) + assert fake.calls == []