diff --git a/docs/usage/CLI.md b/docs/usage/CLI.md index de42daf..008289a 100644 --- a/docs/usage/CLI.md +++ b/docs/usage/CLI.md @@ -53,6 +53,59 @@ When a code cell's source changes, its saved outputs and execution count are cle Other cells are preserved, and no cells are executed. Writing a whole notebook without `--cell` is not supported. +Use `execute` to run a notebook locally or on a Jupyter server. +Add `--cell CELL_ID` to run only one code cell: + +```bash +jpterm execute notebook.ipynb +jpterm execute notebook.ipynb --cell CELL_ID +jpterm --server http://127.0.0.1:8000 execute notebook.ipynb +jpterm --server http://127.0.0.1:8000 execute notebook.ipynb --cell CELL_ID +jpterm --server http://127.0.0.1:8000 execute notebook.ipynb --cell CELL_ID --json +``` + +Local execution starts a detached kernel in the notebook's directory and reuses +it on later calls. Each notebook's resolved absolute path identifies its kernel. +The local TUI also reuses this session. It leaves an existing kernel running on +exit, and shuts down kernels it started itself. +Avoid simultaneous jpterm calls for the same notebook; execution and notebook +saves are not coordinated between CLI processes. + +Manage detached local kernels with: + +```bash +jpterm kernel start notebook.ipynb +jpterm kernel list +jpterm kernel list --json +jpterm kernel stop notebook.ipynb +``` + +Add `--server` to manage notebook kernels on a Jupyter server: + +```bash +jpterm --server http://127.0.0.1:8000 kernel start notebook.ipynb +jpterm --server http://127.0.0.1:8000 kernel list +jpterm --server http://127.0.0.1:8000 kernel list --json +jpterm --server http://127.0.0.1:8000 kernel stop notebook.ipynb +``` + +`kernel start` uses the notebook's kernelspec (or `--kernel NAME` for a new +kernel) and waits until the kernel is ready. It reuses an existing live kernel +without executing cells or saving the notebook. `execute` still starts a kernel +automatically when needed. + +Remote listing shows notebook sessions, including those started by other clients. +Paths are relative to the server's root. Remote `stop` deletes the matching +notebook session and lets the server manage kernel shutdown. Both backends +support `--json` for listing; remote entries include kernel and session IDs, +while local entries include the kernel or launcher PID. + +With `--server`, execution reuses the notebook's session, creating +one with the notebook's kernelspec (or `--kernel NAME`) when needed. The kernel +stays running so variables and imports persist between calls. Without `--cell`, +code cells execute in notebook order. +Execution stops at the first cell error and returns a nonzero exit status. + For a server with real-time collaboration enabled, add `--collaborative` before the command to read and write the live shared document: diff --git a/docs/usage/TUI.md b/docs/usage/TUI.md index b28cd94..39ab60a 100644 --- a/docs/usage/TUI.md +++ b/docs/usage/TUI.md @@ -21,6 +21,22 @@ $ jpterm --help The easiest way to launch jpterm is simply by entering `jpterm` and hitting *Enter*. Jpterm will run locally on your machine, without needing a Jupyter server. This is the TUI mode (Text User Interface), where jpterm takes control of your terminal and runs as a full-screen application. There is also a [CLI](CLI.md) mode (Command Line Interface). +In local mode, opening a notebook reuses its running jpterm kernel when one +exists. For example, you can prepare a persistent kernel with +`jpterm kernel start notebook.ipynb`, then open that notebook in the TUI. + +Press `Ctrl+W` to close the active tab. If a document has unsaved changes, +choose **Cancel** to keep editing or **Discard changes** to close it without +saving. You can cancel, press `Ctrl+S` to save, then close the tab. + +Closing a notebook tab leaves its kernel running until it is stopped explicitly +or cleaned up when the TUI exits. + +On exit, the TUI shuts down local notebook kernels it started itself. Kernels +that were already running remain available after the TUI exits. Use +`jpterm kernel stop notebook.ipynb` to stop a persistent kernel explicitly. Avoid simultaneous execution or saves +from the TUI and CLI for the same notebook; they are not coordinated. + Jpterm can also run as a client to a Jupyter server, just like when you use JupyterLab. While JupyterLab is a Web client that runs in the browser, jpterm runs in the terminal. The two main Jupyter servers are [jupyter-server](https://github.com/jupyter-server/jupyter_server) and [jupyverse](https://github.com/jupyter-server/jupyverse). You can install and launch them like so: ```bash diff --git a/docs/usage/key_bindings.md b/docs/usage/key_bindings.md index 4f72d16..3a4b681 100644 --- a/docs/usage/key_bindings.md +++ b/docs/usage/key_bindings.md @@ -2,3 +2,6 @@ Depending on the context, some commands can be triggered using key bindings. For - [Notebook editor](../plugins/notebook_editor.md#key-bindings) - [Console](../plugins/console.md#key-bindings) + +In the TUI, `Ctrl+W` closes the active tab, with confirmation for unsaved changes. +Use `Ctrl+S` to save a text file or notebook before closing it. diff --git a/plugins/cell/txl_cell/main.py b/plugins/cell/txl_cell/main.py index c9b5d2a..3841c34 100644 --- a/plugins/cell/txl_cell/main.py +++ b/plugins/cell/txl_cell/main.py @@ -77,8 +77,10 @@ def __init__( self.show_border = show_border self.show_cell_id = show_cell_id self.outputs = [] + self._view_tasks = [] + self._model_subscriptions = [] self.update(mount=False) - self.ycell.observe_deep(self.on_change) + self._cell_subscription = self.ycell.observe_deep(self.on_change) self.styles.height = "auto" self.cell_change_events = StapledObjectStream( *create_memory_object_stream[Any](max_buffer_size=math.inf) @@ -86,8 +88,16 @@ def __init__( self.widget_change_events = StapledObjectStream( *create_memory_object_stream[Any](max_buffer_size=math.inf) ) - task_group.create_task(self.observe_cell_changes()) - task_group.create_task(self.observe_widget_changes()) + self._view_tasks.append(task_group.create_task(self.observe_cell_changes())) + self._view_tasks.append(task_group.create_task(self.observe_widget_changes())) + + async def on_unmount(self): + self.ycell.unobserve(self._cell_subscription) + for text, subscription in self._model_subscriptions: + text.unobserve(subscription) + await self.source.stop() + for task in self._view_tasks: + task.cancel() def on_click(self): self.clicked = True @@ -159,13 +169,16 @@ async def observe_cell_changes(self): # this is a widget is_widget = True room_id = f"ywidget:{inserted.guid}" - self.task_group.create_task( + self._view_tasks.append(self.task_group.create_task( self.contents.websocket_provider(room_id, inserted), - ) + )) inserted["_model_name"] = model_name = Text() - model_name.observe(partial( + subscription = model_name.observe(partial( self.on_widget_change, inserted )) + self._model_subscriptions.append( + (model_name, subscription) + ) else: #output = json.loads(str(inserted)) output = inserted @@ -222,7 +235,7 @@ def update(self, mount: bool = True): language=language, show_border=self.show_border, ) - self.task_group.create_task(self.source.start()) + self._view_tasks.append(self.task_group.create_task(self.source.start())) if mount: self.mount(self.source) @@ -238,9 +251,12 @@ def get_output_widget(self, output): guid = output["guid"] ywidget_doc = Doc() room_id = f"ywidget:{guid}" - self.task_group.create_task(self.contents.websocket_provider(room_id, ywidget_doc)) + self._view_tasks.append(self.task_group.create_task( + self.contents.websocket_provider(room_id, ywidget_doc) + )) ywidget_doc["_model_name"] = model_name = Text() - model_name.observe(partial(self.on_widget_change, ywidget_doc)) + subscription = model_name.observe(partial(self.on_widget_change, ywidget_doc)) + self._model_subscriptions.append((model_name, subscription)) return output_type = output["output_type"] diff --git a/plugins/jpterm/txl_jpterm/close_tab.py b/plugins/jpterm/txl_jpterm/close_tab.py new file mode 100644 index 0000000..e1e51bb --- /dev/null +++ b/plugins/jpterm/txl_jpterm/close_tab.py @@ -0,0 +1,52 @@ +from textual.app import ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, Static + + +class ConfirmCloseTab(ModalScreen[bool]): + DEFAULT_CSS = """ + ConfirmCloseTab { + align: center middle; + background: $background 60%; + } + #close-tab-dialog { + width: 60; + max-width: 90%; + height: auto; + padding: 1 2; + border: thick $accent; + background: $surface; + } + #close-tab-buttons { + height: 3; + margin-top: 1; + align: right middle; + } + #close-tab-buttons Button { + margin-left: 1; + min-width: 0; + padding: 0 1; + } + """ + BINDINGS = [("escape", "cancel", "Cancel")] + + def __init__(self, title): + super().__init__() + self.tab_title = title + + def compose(self) -> ComposeResult: + with Vertical(id="close-tab-dialog"): + yield Static(f"Close {self.tab_title}?\nUnsaved changes will be lost.", markup=False) + with Horizontal(id="close-tab-buttons"): + yield Button("Cancel", id="keep-editing") + yield Button("Discard changes", variant="error", id="discard-changes") + + def on_mount(self): + self.query_one("#keep-editing", Button).focus() + + def action_cancel(self): + self.dismiss(False) + + def on_button_pressed(self, event: Button.Pressed): + self.dismiss(event.button.id == "discard-changes") diff --git a/plugins/jpterm/txl_jpterm/main.py b/plugins/jpterm/txl_jpterm/main.py index ce22d1d..7119d6e 100644 --- a/plugins/jpterm/txl_jpterm/main.py +++ b/plugins/jpterm/txl_jpterm/main.py @@ -2,11 +2,13 @@ from fps import Module from textual._context import active_app from textual.app import App +from textual.binding import Binding from textual.containers import Container from textual.reactive import var from txl.base import Editors, FileBrowser, Footer, Header, Launcher, MainArea +from .close_tab import ConfirmCloseTab from .footer import Footer as _Footer from .header import Header as _Header from .main_area import MainArea as _MainArea @@ -18,6 +20,7 @@ class Jpterm(App): ("f", "toggle_files", "Toggle Files"), ("l", "open_launcher", "Open Launcher"), ("q", "quit", "Quit"), + Binding("ctrl+w", "close_tab", "Close Tab", priority=True), ] show_browser = var(True) cli: bool = False @@ -54,6 +57,22 @@ def action_toggle_files(self) -> None: def action_open_launcher(self) -> None: self.main_area.show(self.launcher, "Launcher") + async def action_close_tab(self) -> None: + if self.cli or isinstance(self.screen, ConfirmCloseTab): + return + widget = self.main_area.shown + if widget not in self.main_area.widget_to_tab: + return + tab, dirty = self.main_area.widget_to_tab[widget] + if dirty: + async def confirm(discard): + if discard: + await self.main_area.close(widget) + + self.push_screen(ConfirmCloseTab(tab.label_text.removeprefix("+ ")), confirm) + else: + await self.main_area.close(widget) + class JptermModule(Module): def __init__(self, name: str, driver_class=None): diff --git a/plugins/jpterm/txl_jpterm/main_area.py b/plugins/jpterm/txl_jpterm/main_area.py index bb7c2ca..e3b4cca 100644 --- a/plugins/jpterm/txl_jpterm/main_area.py +++ b/plugins/jpterm/txl_jpterm/main_area.py @@ -28,19 +28,41 @@ def show(self, widget: Widget, title: Optional[str] = None, mount: bool = True): if self.tabs is None: self.tabs = Tabs(tab) self.mount(self.tabs) - self.mount(widget) - mount = False else: self.tabs.add_tab(tab) self.tabs.active = tab.id self.widget_to_tab[widget] = (tab, False) self.mounted.append(widget) - if mount: + if widget.parent is None: self.mount(widget) else: tab, dirty = self.widget_to_tab[widget] self.tabs.active = tab.id + async def close(self, widget=None): + widget = self.shown if widget is None else widget + if widget not in self.widget_to_tab: + return + launcher = getattr(self.app, "launcher", None) + if widget is launcher and len(self.mounted) == 1: + return + index = self.mounted.index(widget) + tab, _ = self.widget_to_tab.pop(widget) + self.mounted.remove(widget) + was_shown = self.shown is widget + if was_shown: + self.shown = None + await self.tabs.remove_tab(tab) + if widget is launcher: + widget.display = False + else: + await widget.remove() + if was_shown: + if self.mounted: + self.show(self.mounted[max(0, index - 1)]) + elif launcher is not None: + self.show(launcher, "Launcher") + def get_label(self) -> str: tab = self.tabs.active_tab return tab.label_text @@ -70,6 +92,8 @@ def clear_dirty(self, widget: Widget) -> None: def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None: widgets = list(self.widget_to_tab.keys()) tabs = [tab for tab, dirty in self.widget_to_tab.values()] + if event.tab not in tabs: + return widget = widgets[tabs.index(event.tab)] if self.shown is not None: self.shown.display = False diff --git a/plugins/local_kernels/pyproject.toml b/plugins/local_kernels/pyproject.toml index 529261d..7ba7516 100644 --- a/plugins/local_kernels/pyproject.toml +++ b/plugins/local_kernels/pyproject.toml @@ -11,7 +11,7 @@ requires-python = ">=3.11" license = "MIT" keywords = [] classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",] -dependencies = [ "txl ==0.3.4", "txl_kernel >=0.3.11,<0.4.0", "pyzmq >=27.1.0", "ipykernel",] +dependencies = [ "txl ==0.3.4", "txl_kernel >=0.3.11,<0.4.0", "pyzmq >=27.1.0", "ipykernel", "psutil >=5.8",] [[project.authors]] name = "David Brochart" email = "david.brochart@gmail.com" diff --git a/plugins/local_kernels/txl_local_kernels/connect.py b/plugins/local_kernels/txl_local_kernels/connect.py index 3fe0273..fc1eabd 100644 --- a/plugins/local_kernels/txl_local_kernels/connect.py +++ b/plugins/local_kernels/txl_local_kernels/connect.py @@ -78,6 +78,18 @@ async def launch_kernel( kernel_cwd: str | None, capture_output: bool, ) -> Process: + cmd = kernel_command(kernelspec_path, connection_file_path) + if capture_output: + stdout = subprocess.DEVNULL + stderr = subprocess.STDOUT + else: + stdout = None + stderr = None + process = await open_process(cmd, stdout=stdout, stderr=stderr, cwd=kernel_cwd or None) + return process + + +def kernel_command(kernelspec_path: str, connection_file_path: str) -> list[str]: with open(kernelspec_path) as f: kernelspec = json.load(f) cmd = [s.format(connection_file=connection_file_path) for s in kernelspec["argv"]] @@ -87,16 +99,7 @@ async def launch_kernel( "python%i.%i" % sys.version_info[:2], }: cmd[0] = sys.executable - if capture_output: - stdout = subprocess.DEVNULL - stderr = subprocess.STDOUT - else: - stdout = None - stderr = None - if not kernel_cwd: - kernel_cwd = None - process = await open_process(cmd, stdout=stdout, stderr=stderr, cwd=kernel_cwd) - return process + return cmd def create_socket(channel: str, cfg: cfg_t, identity: Optional[bytes] = None) -> Socket: ip = cfg["ip"] diff --git a/plugins/local_kernels/txl_local_kernels/driver.py b/plugins/local_kernels/txl_local_kernels/driver.py index a04ea14..25140b4 100644 --- a/plugins/local_kernels/txl_local_kernels/driver.py +++ b/plugins/local_kernels/txl_local_kernels/driver.py @@ -24,13 +24,18 @@ def __init__( write_connection_file: bool = True, capture_kernel_output: bool = True, comm_handlers=[], + launch: bool = True, + owned_session=None, ) -> None: super().__init__(task_group) self.task_group = task_group self.capture_kernel_output = capture_kernel_output - self.kernelspec_path = kernelspec_path or find_kernelspec(kernel_name) + self.launch = launch + self.owned_session = owned_session + self.owned_session_record = owned_session.read() if owned_session is not None else None + self.kernelspec_path = kernelspec_path or (find_kernelspec(kernel_name) if launch else "") self.kernel_cwd = kernel_cwd - if not self.kernelspec_path: + if launch and not self.kernelspec_path: raise RuntimeError("Could not find a kernel, maybe you forgot to install one?") if write_connection_file: self.connection_file_path, self.connection_cfg = _write_connection_file(connection_file) @@ -61,12 +66,13 @@ async def restart(self, startup_timeout: float = float("inf")) -> None: self.listen_channels() async def start(self, startup_timeout: float = float("inf"), connect: bool = True) -> None: - self.kernel_process = await launch_kernel( - self.kernelspec_path, - self.connection_file_path, - self.kernel_cwd, - self.capture_kernel_output, - ) + if self.launch: + self.kernel_process = await launch_kernel( + self.kernelspec_path, + self.connection_file_path, + self.kernel_cwd, + self.capture_kernel_output, + ) if connect: await self.connect(startup_timeout) self.started.set() @@ -87,11 +93,26 @@ def listen_channels(self): self.channel_tasks.append(self.task_group.create_task(self._recv_shell())) async def stop(self) -> None: - self.kernel_process.kill() - await self.kernel_process.wait() - os.remove(self.connection_file_path) + process = getattr(self, "kernel_process", None) + if process is not None: + if process.returncode is None: + process.kill() + await process.wait() for task in self.channel_tasks: task.cancel() + for task in self.channel_tasks: + await task.wait() + for channel in ("shell_channel", "control_channel", "iopub_channel"): + socket = getattr(self, channel, None) + if socket is not None: + socket.close(linger=0) + if self.launch and os.path.exists(self.connection_file_path): + os.remove(self.connection_file_path) + if self.owned_session is not None: + await self.owned_session.stop(expected_record=self.owned_session_record) + self.owned_session = None + if self in kernel_drivers: + kernel_drivers.remove(self) async def _recv_iopub(self): while True: diff --git a/plugins/local_kernels/txl_local_kernels/main.py b/plugins/local_kernels/txl_local_kernels/main.py index 5258e7a..e7fcc8d 100644 --- a/plugins/local_kernels/txl_local_kernels/main.py +++ b/plugins/local_kernels/txl_local_kernels/main.py @@ -2,7 +2,7 @@ from pathlib import Path from typing import Any -from anyio import create_task_group, sleep +from anyio import CancelScope, create_task_group, sleep from fps import Module from pycrdt import Map @@ -10,17 +10,36 @@ from .driver import KernelDriver, kernel_drivers from .kernelspec import kernelspec_dirs +from .sessions import LocalSession class LocalKernels(Kernels): comm_handlers = [] - def __init__(self, kernel_name: str | None = None): - self.kernel = KernelDriver(self.task_group, kernel_name, comm_handlers=self.comm_handlers) + def __init__( + self, kernel_name: str | None = None, kernel_cwd: str = "", connection_file: str = "", + path: str | None = None, + ): + owned_session = None + if path is not None and not connection_file: + session = LocalSession(path) + connection_file = session.connection_file(kernel_name) + if session.launched: + owned_session = session + self.kernel = KernelDriver( + self.task_group, kernel_name, kernel_cwd=kernel_cwd, + comm_handlers=self.comm_handlers, + connection_file=connection_file, write_connection_file=not connection_file, + launch=not connection_file, + owned_session=owned_session, + ) async def execute(self, ycell: Map): await self.kernel.execute(ycell) + async def aclose(self): + await self.kernel.stop() + class LocalKernelspecs(Kernelspecs): async def get(self) -> dict[str, Any]: @@ -46,8 +65,10 @@ class _LocalKernels(LocalKernels): await sleep(float("inf")) async def stop(self) -> None: - for kernel_driver in kernel_drivers: - self.tg.start_soon(kernel_driver.stop) + with CancelScope(shield=True): + for kernel_driver in list(kernel_drivers): + if kernel_driver.task_group is self.tg: + await kernel_driver.stop() self.tg.cancel_scope.cancel() diff --git a/plugins/local_kernels/txl_local_kernels/paths.py b/plugins/local_kernels/txl_local_kernels/paths.py index deae183..f47679f 100644 --- a/plugins/local_kernels/txl_local_kernels/paths.py +++ b/plugins/local_kernels/txl_local_kernels/paths.py @@ -81,7 +81,7 @@ def jupyter_data_dir(): def jupyter_runtime_dir(): if "JUPYTER_RUNTIME_DIR" in os.environ: - return os.environ("JUPYTER_RUNTIME_DIR") + return os.environ["JUPYTER_RUNTIME_DIR"] return os.path.join(jupyter_data_dir(), "runtime") diff --git a/plugins/local_kernels/txl_local_kernels/sessions.py b/plugins/local_kernels/txl_local_kernels/sessions.py new file mode 100644 index 0000000..690faf0 --- /dev/null +++ b/plugins/local_kernels/txl_local_kernels/sessions.py @@ -0,0 +1,150 @@ +"""Notebook kernel sessions shared by the CLI and TUI.""" + +import hashlib +import json +import os +import subprocess +import uuid +from pathlib import Path + +import anyio +import psutil + +from .connect import kernel_command, write_connection_file +from .kernelspec import find_kernelspec +from .paths import jupyter_runtime_dir + +# Keep child handles available for reaping while this process is alive. +_children = [] + + +def runtime_directory(): + directory = Path(jupyter_runtime_dir()) / "jpterm" + directory.mkdir(parents=True, exist_ok=True, mode=0o700) + return directory + + +def session_process(record): + """Check process identity as well as PID, which may have been reused.""" + try: + process = psutil.Process(record["pid"]) + if (process.create_time() == record["created"] + and process.status() != psutil.STATUS_ZOMBIE): + return process + except psutil.NoSuchProcess: + pass + return None + + +def process_alive(process): + try: + return process.is_running() and process.status() != psutil.STATUS_ZOMBIE + except psutil.NoSuchProcess: + return False + + +def list_sessions(): + records = [] + for path in sorted(runtime_directory().glob("*.session.json")): + record = json.loads(path.read_text(encoding="utf-8")) + records.append({ + "path": record["path"], "kernel_name": record["kernel_name"], + "pid": record["pid"], "alive": session_process(record) is not None, + }) + return records + + +class LocalSession: + def __init__(self, path, timeout=30): + self.path = os.path.normcase(str(Path(path).resolve())) + self.timeout = timeout + digest = hashlib.sha256(self.path.encode("utf-8")).hexdigest() + directory = runtime_directory() + self.record_path = directory / f"{digest}.session.json" + self.connection_path = directory / f"{digest}.connection.json" + self.launched = False + + def read(self): + try: + record = json.loads(self.record_path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None + if record["path"] != self.path: + raise ValueError("Local kernel session path does not match notebook") + return record + + def connection_file(self, kernel_name): + self.launched = False + record = self.read() + if record is not None and session_process(record) is not None: + if not self.connection_path.exists(): + raise ValueError("Kernel connection file is missing; stop the kernel first") + return str(self.connection_path) + spec = find_kernelspec(kernel_name) + if not spec: + raise ValueError("Could not find a kernel; provide --kernel NAME or install one") + # Create private connection files even when replacing a stale session. + fd = os.open(self.connection_path, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600) + os.close(fd) + write_connection_file(str(self.connection_path)) + command = kernel_command(spec, str(self.connection_path)) + environment = dict(os.environ) + # Detached kernels must not monitor the lifetime of this CLI process. + for key in ("JPY_PARENT_PID", "JPY_INTERRUPT_EVENT"): + environment.pop(key, None) + options = ( + {"creationflags": subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP} + if os.name == "nt" else {"start_new_session": True} + ) + process = None + temporary = self.record_path.with_suffix(f".{uuid.uuid4().hex}.tmp") + try: + process = subprocess.Popen( + command, cwd=Path(self.path).parent, env=environment, + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + close_fds=True, **options, + ) + record = { + "path": self.path, "kernel_name": kernel_name, "pid": process.pid, + "created": psutil.Process(process.pid).create_time(), + } + fd = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(record, stream) + os.replace(temporary, self.record_path) + except BaseException: + if process is not None: + process.kill() + process.wait() + self.connection_path.unlink(missing_ok=True) + temporary.unlink(missing_ok=True) + raise + _children[:] = [child for child in _children if child.poll() is None] + _children.append(process) + self.launched = True + return str(self.connection_path) + + async def stop(self, expected_record=None): + record = self.read() + if record is None or (expected_record is not None and record != expected_record): + return False + process = session_process(record) + if process is not None: + # Windows venv launchers can run the interpreter in a child process. + # psutil verifies process identity before signaling each captured PID. + try: + children = process.children(recursive=True) + except psutil.NoSuchProcess: + children = [] + for target in [process, *reversed(children)]: + try: + target.kill() + except psutil.NoSuchProcess: + pass + with anyio.fail_after(self.timeout): + while any(process_alive(target) for target in [process, *children]): + await anyio.sleep(0.05) + _children[:] = [child for child in _children if child.poll() is None] + self.record_path.unlink(missing_ok=True) + self.connection_path.unlink(missing_ok=True) + return True diff --git a/plugins/notebook_editor/txl_notebook_editor/main.py b/plugins/notebook_editor/txl_notebook_editor/main.py index c14c589..1914b83 100644 --- a/plugins/notebook_editor/txl_notebook_editor/main.py +++ b/plugins/notebook_editor/txl_notebook_editor/main.py @@ -81,6 +81,7 @@ def __init__( self.main_area = main_area self.experimental = experimental self.kernel = None + self._observe_task = None self.cells = [] self.cell_i = 0 self.cell_copy = None @@ -158,7 +159,9 @@ async def open( ) if self.cell_index is None: raise ValueError(f"Cell ID not found: {cell_id}") - self.update() + mounts = self.update() + for mount in mounts: + await mount if not self.app.cli: self.nb_change_target = StapledObjectStream( *anyio.create_memory_object_stream[Any](max_buffer_size=math.inf) @@ -167,15 +170,23 @@ async def open( *anyio.create_memory_object_stream[Any](max_buffer_size=math.inf) ) self.ynb.observe(self.on_change) - self.task_group.create_task(self.observe_nb_changes()) + self._observe_task = self.task_group.create_task(self.observe_nb_changes()) + + def on_unmount(self): + if self._observe_task is not None: + self.ynb.unobserve() + self._observe_task.cancel() + if self.kernel is not None: + self.kernel.kernel.busy.disconnect(self.watch_busy) def update(self): + mounts = [] ipynb = self.ynb.source self.language = ipynb.get("metadata", {}).get("kernelspec", {}).get("language", None) if self.kernel is None and not self.app.cli: kernel_name = ipynb.get("metadata", {}).get("kernelspec", {}).get("name") if kernel_name: - self.kernel = self.kernels(kernel_name) + self.kernel = self.kernels(kernel_name, path=self.path) self.kernel.kernel.busy.connect(self.watch_busy) indices = range(self.ynb.cell_number) if self.cell_index is None else [self.cell_index] for i_cell in indices: @@ -189,10 +200,11 @@ def update(self): cell.source.show_cursor = False cell.source.styles.max_height = None cell.source.styles.scrollbar_size_vertical = 0 - self.mount(cell) + mounts.append(self.mount(cell)) self.cells.append(cell) if self.cells and not self.app.cli: self.cells[self.cell_i].select() + return mounts def on_change(self, target, events): self.nb_change_target.send_nowait(target) @@ -213,7 +225,7 @@ async def observe_nb_changes(self): if self.kernel is None: kernel_name = kernelspec.get("name") if kernel_name: - self.kernel = self.kernels(kernel_name) + self.kernel = self.kernels(kernel_name, path=self.path) self.kernel.kernel.busy.connect(self.watch_busy) elif target == "state": if "dirty" in events.keys: diff --git a/plugins/remote_kernels/txl_remote_kernels/driver.py b/plugins/remote_kernels/txl_remote_kernels/driver.py index ee35e93..2879f2d 100644 --- a/plugins/remote_kernels/txl_remote_kernels/driver.py +++ b/plugins/remote_kernels/txl_remote_kernels/driver.py @@ -28,9 +28,11 @@ def __init__( url: str, kernel_name: str | None = "", comm_handlers=[], + path: str | None = None, ) -> None: super().__init__(task_group) self.task_group = task_group + self.path = path self.kernel_name = kernel_name parsed_url = parse.urlparse(url) self.base_url = parse.urljoin(url, parsed_url.path).rstrip("/") @@ -49,31 +51,45 @@ def __init__( async def start(self): i = str(uuid.uuid4()) async with httpx2.AsyncClient() as client: - body = { - "kernel": {"name": self.kernel_name}, - "name": i, - "path": i, - "type": "notebook", - } - r = await client.post( - f"{self.base_url}/api/sessions", - json=body, - params={**self.query_params}, - cookies=self.cookies, - ) - d = r.json() - self.cookies.update(r.cookies) - self.session_id = d["id"] - self.kernel_id = d["kernel"]["id"] + session = None + if self.path is not None: + r = await client.get( + f"{self.base_url}/api/sessions", + params=self.query_params, cookies=self.cookies, + ) + r.raise_for_status() + self.cookies.update(r.cookies) + session = next( + (item for item in r.json() if item["path"] == self.path + and item["type"] == "notebook"), None + ) + if session is None: + body = { + "kernel": {"name": self.kernel_name}, + "name": self.path.rsplit("/", 1)[-1] if self.path else i, + "path": self.path if self.path is not None else i, + "type": "notebook", + } + r = await client.post( + f"{self.base_url}/api/sessions", json=body, + params=self.query_params, cookies=self.cookies, + ) + r.raise_for_status() + self.cookies.update(r.cookies) + session = r.json() + # The channels client identity must be unique even when reusing a session. + self.session_id = i + self.kernel_id = session["kernel"]["id"] r = await client.get( f"{self.base_url}/api/kernels/{self.kernel_id}", + params=self.query_params, cookies=self.cookies, ) if r.status_code != 200 or self.kernel_id != r.json()["id"]: return async with httpx2.AsyncClient(cookies=self.cookies) as client, client.websocket( f"{self.ws_url}/api/kernels/{self.kernel_id}/channels", - params={"session_id": self.session_id}, + params={**self.query_params, "session_id": self.session_id}, subprotocols=["v1.kernel.websocket.jupyter.org"], ) as self.websocket: recv_task = self.task_group.create_task(self._recv()) diff --git a/plugins/remote_kernels/txl_remote_kernels/main.py b/plugins/remote_kernels/txl_remote_kernels/main.py index 7db79be..3a3f7d8 100644 --- a/plugins/remote_kernels/txl_remote_kernels/main.py +++ b/plugins/remote_kernels/txl_remote_kernels/main.py @@ -20,9 +20,10 @@ def __init__( self, url: str, kernel_name: str | None, + path: str | None = None, ): self.kernel = KernelDriver( - self.task_group, url, kernel_name, comm_handlers=self.comm_handlers + self.task_group, url, kernel_name, comm_handlers=self.comm_handlers, path=path ) async def execute(self, ycell: Map): @@ -77,7 +78,6 @@ def __init__(self, *args, **kwargs): async def stop(self) -> None: self.tg.cancel_scope.cancel() - class RemoteKernelspecsModule(Module): def __init__(self, name: str, url: str = "http://127.0.0.1:8000"): super().__init__(name) diff --git a/plugins/remote_kernels/txl_remote_kernels/sessions.py b/plugins/remote_kernels/txl_remote_kernels/sessions.py new file mode 100644 index 0000000..be33ff3 --- /dev/null +++ b/plugins/remote_kernels/txl_remote_kernels/sessions.py @@ -0,0 +1,49 @@ +"""Manage notebook sessions without opening kernel channels.""" + +from pathlib import PurePosixPath +from urllib.parse import parse_qs, quote, urlsplit, urlunsplit + +import anyio +import httpx2 + + +class RemoteSessions: + def __init__(self, url, timeout=30): + parsed = urlsplit(url) + self.base_url = urlunsplit(parsed._replace(query="", fragment="")).rstrip("/") + self.params = parse_qs(parsed.query) + self.timeout = timeout + + async def _list(self, client): + response = await client.get(f"{self.base_url}/api/sessions", params=self.params) + response.raise_for_status() + return [session for session in response.json() + if session.get("type") == "notebook" and session.get("kernel") is not None] + + async def list(self): + with anyio.fail_after(self.timeout): + async with httpx2.AsyncClient(timeout=self.timeout) as client: + sessions = await self._list(client) + return [ + {"path": session["path"], "kernel_name": session["kernel"]["name"], + "kernel_id": session["kernel"]["id"], "session_id": session["id"], + "alive": session["kernel"].get("execution_state") != "dead"} + for session in sessions + ] + + async def stop(self, path): + path = str(PurePosixPath(path)).lstrip("/") + with anyio.fail_after(self.timeout): + async with httpx2.AsyncClient(timeout=self.timeout) as client: + sessions = [session for session in await self._list(client) + if session["path"] == path] + if not sessions: + return False + if len(sessions) > 1: + raise ValueError(f"Multiple notebook sessions found for: {path}") + session_id = quote(sessions[0]["id"], safe="") + response = await client.delete( + f"{self.base_url}/api/sessions/{session_id}", params=self.params, + ) + response.raise_for_status() + return True diff --git a/plugins/text_editor/txl_text_editor/main.py b/plugins/text_editor/txl_text_editor/main.py index 4594370..061f2f8 100644 --- a/plugins/text_editor/txl_text_editor/main.py +++ b/plugins/text_editor/txl_text_editor/main.py @@ -27,6 +27,9 @@ def __init__(self, contents: Contents, main_area: MainArea, task_group) -> None: self.contents = contents self.main_area = main_area self.task_group = task_group + self.editor = None + self._observe_task = None + self._closed = False async def open(self, path: str) -> None: self.path = path @@ -56,10 +59,20 @@ async def open(self, path: str) -> None: *create_memory_object_stream[Any](max_buffer_size=math.inf) ) self.ytext.observe(self.on_change) - self.task_group.start_soon(self.observe_changes) + self._observe_task = self.task_group.create_task(self.observe_changes()) async def close(self) -> None: - await self.editor.stop() + if self._closed: + return + self._closed = True + if self._observe_task is not None: + self.ytext.unobserve() + self._observe_task.cancel() + if self.editor is not None: + await self.editor.stop() + + async def on_unmount(self): + await self.close() def on_change(self, target, events): self.change_target.send_nowait(target) diff --git a/src/jpterm/cli.py b/src/jpterm/cli.py index b477911..fcd7950 100644 --- a/src/jpterm/cli.py +++ b/src/jpterm/cli.py @@ -8,6 +8,8 @@ from txl.app import disabled, modules from txl.cli import set_main, txl_main +from .execute import execute +from .kernel import kernel from .show import show from .write import write @@ -85,6 +87,8 @@ def command(ctx, **kwargs): command.params.extend(click.command()(_main).params) command.add_command(show) command.add_command(write) + command.add_command(execute) + command.add_command(kernel) command.add_command(logo) command() diff --git a/src/jpterm/execute.py b/src/jpterm/execute.py new file mode 100644 index 0000000..98ed492 --- /dev/null +++ b/src/jpterm/execute.py @@ -0,0 +1,92 @@ +"""Execute notebook cells through the configured kernel and contents services.""" + +from pathlib import PurePosixPath + +import anyio +import rich_click as click + +from txl.base import Contents, Kernels + +from .app import create_app +from .kernel import create_notebook_kernel +from .show import show + + +async def execute_notebook(path, cell_id, settings, timeout, kernel_name=None, remote=False): + if PurePosixPath(path).suffix != ".ipynb": + raise ValueError("Execution requires a notebook (.ipynb).") + module = create_app(settings, timeout) + deadline = anyio.current_time() + timeout + async with module: + with anyio.fail_after(max(0, deadline - anyio.current_time())): + contents = await module.get(Contents) + document = await contents.get(path, type="notebook") + if cell_id is None: + cells = [cell for cell in document.ycells if cell["cell_type"] == "code"] + else: + cell = next( + (cell for cell in document.ycells if cell.get("id") == cell_id), None + ) + if cell is None: + raise ValueError(f"Cell ID not found: {cell_id}") + if cell["cell_type"] != "code": + raise ValueError(f"Cell is not a code cell: {cell_id}") + cells = [cell] + if not cells: + return [] + kernels = await module.get(Kernels) + kernel = create_notebook_kernel(path, document, kernels, kernel_name, remote) + try: + for cell in cells: + await kernel.execute(cell) + errors = [ + output for output in cell["outputs"].to_py() + if output["output_type"] == "error" + ] + if errors: + break + document.dirty = True + await contents.save(path, document) + return errors + finally: + if not remote: + with anyio.CancelScope(shield=True): + await kernel.aclose() + + +@click.command() +@click.argument("path") +@click.option("--cell", "cell_id", help="Execute only the code cell with this ID.") +@click.option("--kernel", "kernel_name", help="Kernel name when starting a new kernel.") +@click.option( + "--json", "as_json", is_flag=True, help="Print the notebook or selected cell as JSON." +) +@click.option("--timeout", type=click.FloatRange(min=0, min_open=True), default=30.0, + show_default=True, help="Maximum time to load, execute and save, in seconds.") +@click.pass_context +def execute(ctx, path, cell_id, kernel_name, as_json, timeout): + """Execute a notebook locally or using its session on a Jupyter server. + + Local execution reuses a detached kernel for this notebook. + With --server, execution reuses the notebook's server session. + Without --cell, run code cells in order, stopping at the first cell error. + Cell errors are saved and produce a nonzero exit status. + """ + remote = bool(ctx.parent.params["server"]) + options = ctx.obj + try: + errors = anyio.run( + execute_notebook, path, cell_id, options["set_"], timeout, kernel_name, remote, + backend=options["backend"], + ) + except Exception as exc: + while getattr(exc, "exceptions", None): + exc = exc.exceptions[0] + message = "Timed out executing notebook" if isinstance(exc, TimeoutError) else str(exc) + raise click.ClickException(message) from exc + ctx.invoke( + show, path=path, cell_id=cell_id, as_json=as_json, + show_cell_ids=False, timeout=timeout, + ) + if errors: + raise click.ClickException(f"{errors[0]['ename']}: {errors[0]['evalue']}") diff --git a/src/jpterm/kernel.py b/src/jpterm/kernel.py new file mode 100644 index 0000000..604fbe7 --- /dev/null +++ b/src/jpterm/kernel.py @@ -0,0 +1,126 @@ +"""Start, inspect and stop notebook kernels locally or on a Jupyter server.""" + +import json +from pathlib import PurePosixPath + +import anyio +import httpx2 +import rich_click as click +from txl_local_kernels.sessions import LocalSession, list_sessions +from txl_remote_kernels.sessions import RemoteSessions + +from txl.base import Contents, Kernels + +from .app import create_app + + +def create_notebook_kernel(path, document, kernels, kernel_name=None, remote=False): + kernel_name = kernel_name or document.source.get("metadata", {}).get( + "kernelspec", {} + ).get("name") + if remote: + return kernels(kernel_name, path=str(PurePosixPath(path)).lstrip("/")) + connection_file = LocalSession(path).connection_file(kernel_name) + return kernels(kernel_name, connection_file=connection_file) + + +async def start_notebook_kernel(path, settings, timeout, kernel_name=None, remote=False): + if PurePosixPath(path).suffix != ".ipynb": + raise ValueError("Starting a kernel requires a notebook (.ipynb).") + module = create_app(settings, timeout) + deadline = anyio.current_time() + timeout + async with module: + with anyio.fail_after(max(0, deadline - anyio.current_time())): + contents = await module.get(Contents) + document = await contents.get(path, type="notebook") + kernels = await module.get(Kernels) + client = create_notebook_kernel(path, document, kernels, kernel_name, remote) + try: + await client.kernel.started.wait() + finally: + if not remote: + with anyio.CancelScope(shield=True): + await client.aclose() + + +def request_error(exc, action): + if isinstance(exc, (TimeoutError, httpx2.TimeoutException)): + return f"Timed out {action} kernels" + if isinstance(exc, httpx2.HTTPStatusError): + return f"Remote kernel request failed (HTTP {exc.response.status_code})" + if isinstance(exc, httpx2.ConnectError): + return "Could not connect to the Jupyter server" + return str(exc) + + +@click.group() +def kernel(): + """Manage notebook kernels locally or with --server.""" + + +@kernel.command("start") +@click.argument("path") +@click.option("--kernel", "kernel_name", help="Kernel name when starting a new kernel.") +@click.option("--timeout", type=click.FloatRange(min=0, min_open=True), default=30.0, + show_default=True, help="Maximum time to load the notebook and wait for readiness.") +@click.pass_context +def start_kernel(ctx, path, kernel_name, timeout): + """Start or reuse notebook PATH's kernel without executing or saving cells.""" + remote = bool(ctx.find_root().params["server"]) + try: + anyio.run( + start_notebook_kernel, path, ctx.obj["set_"], timeout, kernel_name, remote, + backend=ctx.obj["backend"], + ) + except Exception as exc: + while getattr(exc, "exceptions", None): + exc = exc.exceptions[0] + raise click.ClickException(request_error(exc, "starting")) from exc + + +@kernel.command("list") +@click.option("--json", "as_json", is_flag=True, help="Print session information as JSON.") +@click.option("--timeout", type=click.FloatRange(min=0, min_open=True), default=30.0, + show_default=True, help="Maximum time to list server sessions, in seconds.") +@click.pass_context +def list_kernels(ctx, as_json, timeout): + """List notebook kernels (including stale local sessions).""" + server = ctx.find_root().params["server"] + try: + sessions = ( + anyio.run(RemoteSessions(server, timeout).list, backend=ctx.obj["backend"]) + if server else list_sessions() + ) + except Exception as exc: + raise click.ClickException(request_error(exc, "listing")) from exc + if as_json: + click.echo(json.dumps(sessions, indent=2, ensure_ascii=False)) + else: + for session in sessions: + state = "running" if session["alive"] else "stale" + identifier = session["kernel_id"] if server else session["pid"] + click.echo(f"{session['path']}\t{session['kernel_name']}\t{identifier}\t{state}") + + +@kernel.command("stop") +@click.argument("path") +@click.option("--timeout", type=click.FloatRange(min=0, min_open=True), default=30.0, + show_default=True, help="Maximum time to stop the kernel, in seconds.") +@click.pass_context +def stop_kernel(ctx, path, timeout): + """Stop notebook PATH's kernel by removing its local or server session.""" + server = ctx.find_root().params["server"] + + async def stop(): + if server: + return await RemoteSessions(server, timeout).stop(path) + with anyio.fail_after(timeout): + return await LocalSession(path, timeout).stop() + + try: + stopped = anyio.run(stop, backend=ctx.obj["backend"]) + except Exception as exc: + raise click.ClickException(request_error(exc, "stopping")) from exc + if not stopped: + location = "remote" if server else "local" + raise click.ClickException(f"No {location} kernel session for: {path}") diff --git a/tests/test_execute.py b/tests/test_execute.py new file mode 100644 index 0000000..c505c73 --- /dev/null +++ b/tests/test_execute.py @@ -0,0 +1,396 @@ +import json +import os +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import anyio +import psutil +import pytest +from fps import Module +from jpterm.execute import execute_notebook +from jupyter_ydoc import YNotebook +from rich.text import Text +from textual.app import App +from txl_cell.main import _Cell +from txl_local_kernels.main import LocalKernels, LocalKernelsModule +from txl_notebook_editor.main import NotebookEditor + +from txl.base import Contents, Kernels + + +@pytest.fixture(autouse=True) +def isolated_sessions(tmp_path, monkeypatch): + from txl_local_kernels.sessions import LocalSession, list_sessions + + monkeypatch.setenv("JUPYTER_RUNTIME_DIR", str(tmp_path / "runtime")) + yield + + async def cleanup(): + for record in list_sessions(): + await LocalSession(record["path"]).stop() + + anyio.run(cleanup) + + +@pytest.mark.parametrize("fail", [False, True]) +def test_save_once_after_execution(fail): + document = YNotebook() + document.source = { + "nbformat": 4, "nbformat_minor": 5, "metadata": {}, + "cells": [ + {"id": str(i), "cell_type": "code", "source": str(i), + "metadata": {}, "outputs": [], "execution_count": None} + for i in range(3) + ], + } + contents = SimpleNamespace(get=AsyncMock(return_value=document), save=AsyncMock()) + executed = [] + + async def execute(cell): + contents.save.assert_not_awaited() + executed.append(str(cell["id"])) + if fail and cell["id"] == "1": + cell["outputs"].append({ + "output_type": "error", "ename": "ValueError", "evalue": "failed", + "traceback": [], + }) + + class TestModule(Module): + async def start(self): + self.put(contents, Contents) + self.put( + lambda *args, **kwargs: SimpleNamespace(execute=execute, aclose=AsyncMock()), + Kernels, + ) + self.put(App(), App) + + with patch("txl.app.modules", {"test": TestModule}): + errors = anyio.run(execute_notebook, "test.ipynb", None, [], 5, None, True) + assert executed == (["0", "1"] if fail else ["0", "1", "2"]) + assert bool(errors) == fail + contents.save.assert_awaited_once_with("test.ipynb", document) + + +@pytest.fixture +def local_notebook(tmp_path): + directory = tmp_path / "notebooks" + directory.mkdir() + (directory / "input.txt").write_text("relative input", encoding="utf-8") + source = { + "nbformat": 4, "nbformat_minor": 5, + "metadata": {"kernelspec": { + "name": "python3", "display_name": "Python 3", "language": "python", + }}, + "cells": [ + {"id": "markdown", "cell_type": "markdown", "metadata": {}, "source": "# Title"}, + {"id": "define", "cell_type": "code", "metadata": {}, "outputs": [], + "execution_count": None, "source": ( + "import os, json, ipykernel\nfrom pathlib import Path\n" + "counter = globals().get('counter', 0) + 1\n" + "info = {'pid': os.getpid(), 'connection': ipykernel.get_connection_file(), " + "'counter': counter}\n" + "Path('kernel-info.json').write_text(json.dumps(info))\n" + "print(json.dumps(info))" + )}, + {"id": "read", "cell_type": "code", "metadata": {}, "outputs": [], + "execution_count": None, "source": ( + "assert counter >= 1\nprint(Path('input.txt').read_text())" + )}, + ], + } + path = directory / "test.ipynb" + path.write_text(json.dumps(source), encoding="utf-8") + return path, source + + +def run_local(tmp_path, *arguments): + return subprocess.run( + [sys.executable, "-c", "from jpterm.cli import main; main()", "execute", *arguments], + cwd=tmp_path, capture_output=True, text=True, encoding="utf-8", timeout=25, + env={**os.environ, "PYTHONIOENCODING": "utf-8", "NO_COLOR": "1"}, + ) + + +def assert_session_process(session, info): + # On Windows, a virtualenv launcher may be the recorded process while the + # notebook executes inside its child interpreter. + process = psutil.Process(session["pid"]) + assert info["pid"] in {process.pid, *(child.pid for child in process.children(recursive=True))} + + +def assert_kernel_stopped(info): + assert not Path(info["connection"]).exists() + try: + assert psutil.Process(info["pid"]).status() == psutil.STATUS_ZOMBIE + except psutil.NoSuchProcess: + pass + + +def test_local_execution_reuses_kernel_and_notebook_directory(tmp_path, local_notebook): + path, source = local_notebook + previous_pid = None + for iteration in range(1, 3): + result = run_local(tmp_path, "notebooks/test.ipynb", "--json") + assert result.returncode == 0, result.stdout + result.stderr + notebook = json.loads(result.stdout) + saved = YNotebook() + saved.source = json.loads(path.read_text(encoding="utf-8")) + assert notebook == saved.source + assert notebook["cells"][0] == source["cells"][0] + info = json.loads("".join(notebook["cells"][1]["outputs"][0]["text"])) + assert info["counter"] == iteration + if previous_pid is not None: + assert info["pid"] == previous_pid + previous_pid = info["pid"] + assert notebook["cells"][2]["execution_count"] == iteration * 2 + assert "".join(notebook["cells"][2]["outputs"][0]["text"]) == "relative input\n" + assert Path(info["connection"]).exists() + assert not list((tmp_path / "runtime").rglob("*.lock")) + + +@pytest.mark.parametrize("timeout", [False, True]) +def test_local_execution_preserves_kernel_on_failure(tmp_path, local_notebook, timeout): + path, source = local_notebook + source["cells"][2]["source"] = ( + "import time; time.sleep(60)" if timeout else "raise ValueError('cell failed')" + ) + path.write_text(json.dumps(source), encoding="utf-8") + arguments = [str(path), "--json"] + if timeout: + arguments.extend(["--timeout", "5"]) + result = run_local(tmp_path, *arguments) + assert result.returncode == 1, result.stdout + result.stderr + info = json.loads((path.parent / "kernel-info.json").read_text(encoding="utf-8")) + assert Path(info["connection"]).exists() + assert psutil.Process(info["pid"]).is_running() + if timeout: + assert "Timed out executing notebook" in result.stderr + assert json.loads(path.read_text(encoding="utf-8")) == source + # A busy kernel must not be mistaken for a dead one and replaced. + result = run_local(tmp_path, str(path), "--cell", "define", "--timeout", "0.5") + assert result.returncode == 1, result.stdout + result.stderr + sessions = json.loads(kernel_command(tmp_path, "list", "--json").stdout) + assert len(sessions) == 1 + assert_session_process(sessions[0], info) + else: + assert "ValueError: cell failed" in result.stderr + notebook = json.loads(result.stdout) + assert notebook["cells"][2]["outputs"][0]["output_type"] == "error" + saved = YNotebook() + saved.source = json.loads(path.read_text(encoding="utf-8")) + assert notebook == saved.source + + +@pytest.mark.parametrize("cell_id, message", [ + ("missing", "Cell ID not found"), ("markdown", "Cell is not a code cell"), +]) +@pytest.mark.parametrize("force_color", ["0", "1"]) +def test_invalid_local_cell(tmp_path, local_notebook, monkeypatch, cell_id, message, force_color): + path, source = local_notebook + monkeypatch.setenv("FORCE_COLOR", force_color) + result = run_local(tmp_path, str(path), "--cell", cell_id) + assert result.returncode == 1, result.stdout + result.stderr + assert message in Text.from_ansi(result.stderr).plain + assert json.loads(path.read_text(encoding="utf-8")) == source + assert not list((tmp_path / "runtime").rglob("*.session.json")) + + +def kernel_command(tmp_path, *arguments): + return subprocess.run( + [sys.executable, "-c", "from jpterm.cli import main; main()", "kernel", *arguments], + cwd=tmp_path, capture_output=True, text=True, encoding="utf-8", timeout=15, + env={**os.environ, "PYTHONIOENCODING": "utf-8", "NO_COLOR": "1"}, + ) + + +def test_local_cells_share_state_and_can_be_stopped(tmp_path, local_notebook): + path, source = local_notebook + # This fails in a fresh kernel: earlier cells must not be executed implicitly. + result = run_local(tmp_path, "notebooks/../notebooks/test.ipynb", "--cell", "read", "--json") + assert result.returncode == 1 + assert "NameError" in Text.from_ansi(result.stderr).plain + result = run_local(tmp_path, str(path), "--cell", "define", "--json") + assert result.returncode == 0, result.stdout + result.stderr + cell = json.loads(result.stdout) + info = json.loads("".join(cell["outputs"][0]["text"])) + assert cell["execution_count"] == 2 + result = run_local(tmp_path, str(path), "--cell", "read", "--json") + assert result.returncode == 0, result.stdout + result.stderr + assert json.loads(result.stdout)["execution_count"] == 3 + saved = json.loads(path.read_text(encoding="utf-8")) + assert saved["cells"][0] == source["cells"][0] + assert saved["cells"][1]["execution_count"] == 2 + result = kernel_command(tmp_path, "list", "--json") + assert result.returncode == 0, result.stderr + sessions = json.loads(result.stdout) + assert len(sessions) == 1 + assert_session_process(sessions[0], info) + assert sessions[0]["alive"] + assert "key" not in sessions[0] + if os.name == "posix": + assert Path(info["connection"]).stat().st_mode & 0o077 == 0 + result = kernel_command(tmp_path, "stop", str(path)) + assert result.returncode == 0, result.stderr + assert_kernel_stopped(info) + assert json.loads(kernel_command(tmp_path, "list", "--json").stdout) == [] + result = run_local(tmp_path, str(path), "--cell", "define", "--json") + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["execution_count"] == 1 + + +def test_local_kernel_start_is_ready_and_preserves_notebook(tmp_path, local_notebook): + path, source = local_notebook + # An explicit kernel must override the notebook's kernelspec on first launch. + source["metadata"]["kernelspec"]["name"] = "not-installed" + path.write_text(json.dumps(source), encoding="utf-8") + before = path.read_bytes() + result = kernel_command(tmp_path, "start", str(path), "--kernel", "python3") + assert result.returncode == 0, result.stdout + result.stderr + assert result.stdout == "" + assert path.read_bytes() == before + assert not (path.parent / "kernel-info.json").exists() + sessions = json.loads(kernel_command(tmp_path, "list", "--json").stdout) + assert len(sessions) == 1 + assert sessions[0]["alive"] + result = run_local(tmp_path, str(path), "--cell", "define", "--json") + assert result.returncode == 0, result.stdout + result.stderr + assert json.loads(result.stdout)["execution_count"] == 1 + before = path.read_bytes() + result = kernel_command(tmp_path, "start", str(path)) + assert result.returncode == 0, result.stdout + result.stderr + assert path.read_bytes() == before + assert json.loads(kernel_command(tmp_path, "list", "--json").stdout) == sessions + result = run_local(tmp_path, str(path), "--cell", "read", "--json") + assert result.returncode == 0, result.stdout + result.stderr + assert json.loads(result.stdout)["execution_count"] == 2 + + +@pytest.mark.parametrize("path, message", [ + ("file.txt", "requires a notebook"), ("missing.ipynb", "File not found"), +]) +def test_invalid_local_kernel_start(tmp_path, path, message): + result = kernel_command(tmp_path, "start", path) + assert result.returncode == 1, result.stdout + result.stderr + assert message in Text.from_ansi(result.stderr).plain + assert not list((tmp_path / "runtime").rglob("*.session.json")) + + +def test_dead_local_kernel_is_replaced(tmp_path, local_notebook): + path, _ = local_notebook + result = run_local(tmp_path, str(path), "--cell", "define", "--json") + assert result.returncode == 0, result.stderr + info = json.loads("".join(json.loads(result.stdout)["outputs"][0]["text"])) + psutil.Process(info["pid"]).kill() + # Wait for the killed process to exit before requesting a new session. + import time + + for _ in range(100): + try: + if psutil.Process(info["pid"]).status() == psutil.STATUS_ZOMBIE: + break + except psutil.NoSuchProcess: + break + time.sleep(0.02) + result = run_local(tmp_path, str(path), "--cell", "define", "--json") + assert result.returncode == 0, result.stdout + result.stderr + cell = json.loads(result.stdout) + assert cell["execution_count"] == 1 + assert json.loads("".join(cell["outputs"][0]["text"]))["counter"] == 1 + + +def test_stop_does_not_signal_a_reused_pid(tmp_path): + from txl_local_kernels.sessions import LocalSession + + async def check(): + session = LocalSession(tmp_path / "test.ipynb") + session.record_path.write_text(json.dumps({ + "path": session.path, "pid": 12345, "created": 1.0, + }), encoding="utf-8") + with patch("txl_local_kernels.sessions.psutil.Process") as factory: + process = factory.return_value + process.create_time.return_value = 2.0 + assert await session.stop() + process.kill.assert_not_called() + assert not session.record_path.exists() + anyio.run(check) + + +@pytest.mark.parametrize("ownership", ["owned", "attached", "replaced"]) +def test_tui_attaches_by_notebook_path_and_respects_ownership(tmp_path, local_notebook, ownership): + path, source = local_notebook + if ownership == "attached": + result = kernel_command(tmp_path, "start", str(path)) + assert result.returncode == 0, result.stdout + result.stderr + + info = {} + + async def check(): + with anyio.fail_after(40): + async with anyio.create_task_group() as tasks: + class TuiKernels(LocalKernels): + task_group = tasks + + module = LocalKernelsModule("local_kernels") + module.tg = tasks + document = YNotebook() + document.source = source + contents = SimpleNamespace(get=AsyncMock(return_value=document)) + + def cell_factory(cell, language, kernel): + return _Cell(cell, language, kernel, contents, None, tasks) + + app = App() + app.cli = False + try: + async with app.run_test(): + editor = NotebookEditor( + tasks, contents, TuiKernels, {}, cell_factory, SimpleNamespace( + set_dirty=lambda editor: None, clear_dirty=lambda editor: None, + ), + ) + await app.screen.mount(editor) + await editor.open(str(path)) + await editor.kernel.execute(document.ycells[1]) + info.update(json.loads( + (path.parent / "kernel-info.json").read_text(encoding="utf-8") + )) + result = await anyio.to_thread.run_sync( + lambda: run_local(tmp_path, str(path), "--cell", "read", "--json") + ) + assert result.returncode == 0, result.stdout + result.stderr + assert json.loads(result.stdout)["execution_count"] == 2 + await editor.remove() + result = await anyio.to_thread.run_sync( + lambda: run_local(tmp_path, str(path), "--cell", "read", "--json") + ) + assert result.returncode == 0, result.stdout + result.stderr + assert json.loads(result.stdout)["execution_count"] == 3 + if ownership == "replaced": + result = await anyio.to_thread.run_sync( + lambda: kernel_command(tmp_path, "stop", str(path)) + ) + assert result.returncode == 0, result.stderr + result = await anyio.to_thread.run_sync( + lambda: run_local(tmp_path, str(path), "--cell", "define", "--json") + ) + assert result.returncode == 0, result.stdout + result.stderr + info.update(json.loads( + (path.parent / "kernel-info.json").read_text(encoding="utf-8") + )) + finally: + await module.stop() + + anyio.run(check) + sessions = json.loads(kernel_command(tmp_path, "list", "--json").stdout) + if ownership == "owned": + assert sessions == [] + assert_kernel_stopped(info) + else: + assert len(sessions) == 1 + assert_session_process(sessions[0], info) + assert sessions[0]["alive"] + result = run_local(tmp_path, str(path), "--cell", "read", "--json") + assert result.returncode == 0, result.stdout + result.stderr diff --git a/tests/test_kernel.py b/tests/test_kernel.py new file mode 100644 index 0000000..7afade6 --- /dev/null +++ b/tests/test_kernel.py @@ -0,0 +1,72 @@ +from unittest.mock import AsyncMock, patch + +import anyio +import httpx2 +import pytest +from jpterm.kernel import request_error +from txl_remote_kernels.sessions import RemoteSessions + + +def test_remote_sessions_preserve_base_path_and_authentication(): + async def check(): + sessions = [ + {"id": "session/id", "path": "folder/test.ipynb", "type": "notebook", + "kernel": {"id": "kernel-id", "name": "python3", "execution_state": "idle"}}, + {"id": "console", "path": "console", "type": "console", + "kernel": {"id": "console-kernel", "name": "python3"}}, + ] + response = httpx2.Response( + 200, json=sessions, request=httpx2.Request("GET", "http://server/base/api/sessions") + ) + client = AsyncMock() + client.__aenter__.return_value = client + client.get.return_value = response + client.delete.return_value = httpx2.Response( + 204, request=httpx2.Request("DELETE", "http://server/base/api/sessions/session%2Fid") + ) + manager = RemoteSessions("http://server/base/?token=secret") + with patch("txl_remote_kernels.sessions.httpx2.AsyncClient", return_value=client): + assert await manager.list() == [{ + "path": "folder/test.ipynb", "kernel_name": "python3", + "kernel_id": "kernel-id", "session_id": "session/id", "alive": True, + }] + client.get.assert_awaited_once_with( + "http://server/base/api/sessions", params={"token": ["secret"]} + ) + assert await manager.stop("./folder/test.ipynb") + client.delete.assert_awaited_once_with( + "http://server/base/api/sessions/session%2Fid", params={"token": ["secret"]} + ) + client.delete.reset_mock() + assert not await manager.stop("missing.ipynb") + client.delete.assert_not_awaited() + client.get.return_value = httpx2.Response( + 200, json=[sessions[0], dict(sessions[0], id="duplicate")], + request=response.request, + ) + with pytest.raises(ValueError, match="Multiple notebook sessions"): + await manager.stop("folder/test.ipynb") + client.delete.assert_not_awaited() + + anyio.run(check) + + +@pytest.mark.parametrize("action", ["list", "stop"]) +def test_remote_sessions_report_http_errors(action): + async def check(): + client = AsyncMock() + client.__aenter__.return_value = client + client.get.return_value = httpx2.Response( + 403, request=httpx2.Request("GET", "http://server/api/sessions?token=secret") + ) + with patch("txl_remote_kernels.sessions.httpx2.AsyncClient", return_value=client): + manager = RemoteSessions("http://server/?token=secret") + with pytest.raises(httpx2.HTTPStatusError) as caught: + if action == "list": + await manager.list() + else: + await manager.stop("test.ipynb") + assert request_error(caught.value, action) == "Remote kernel request failed (HTTP 403)" + client.delete.assert_not_awaited() + + anyio.run(check) diff --git a/tests/test_local_kernel_lifecycle.py b/tests/test_local_kernel_lifecycle.py new file mode 100644 index 0000000..ad93e81 --- /dev/null +++ b/tests/test_local_kernel_lifecycle.py @@ -0,0 +1,84 @@ +import json +import sys +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import anyio +import psutil +from txl_local_kernels.driver import KernelDriver +from txl_local_kernels.sessions import LocalSession, process_alive + + +def test_stop_waits_for_channel_readers_before_closing_sockets(): + async def check(): + stopped = set() + + async def reader(name, ready): + ready.set() + try: + await anyio.sleep_forever() + finally: + with anyio.CancelScope(shield=True): + await anyio.sleep(0.01) + stopped.add(name) + + async with anyio.create_task_group() as tasks: + driver = KernelDriver.__new__(KernelDriver) + driver.launch = False + driver.owned_session = None + driver.channel_tasks = [] + for name in ("shell", "iopub"): + ready = anyio.Event() + driver.channel_tasks.append(tasks.create_task(reader(name, ready))) + await ready.wait() + + def close(**kwargs): + assert stopped == {"shell", "iopub"} + + for channel in ("shell_channel", "control_channel", "iopub_channel"): + setattr(driver, channel, SimpleNamespace(close=Mock(side_effect=close))) + await driver.stop() + for channel in ("shell_channel", "control_channel", "iopub_channel"): + getattr(driver, channel).close.assert_called_once_with(linger=0) + + anyio.run(check) + + +def test_stop_terminates_launcher_and_child_interpreter(tmp_path, monkeypatch): + monkeypatch.setenv("JUPYTER_RUNTIME_DIR", str(tmp_path / "runtime")) + child_pid_path = tmp_path / "child.pid" + script = ( + "import pathlib, subprocess, sys\n" + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" + f"pathlib.Path({str(child_pid_path)!r}).write_text(str(child.pid))\n" + "child.wait()\n" + ) + + async def check(): + session = LocalSession(tmp_path / "test.ipynb") + with patch("txl_local_kernels.sessions.kernel_command", return_value=[ + sys.executable, "-c", script, + ]): + session.connection_file("python3") + try: + with anyio.fail_after(10): + while not child_pid_path.exists(): + await anyio.sleep(0.02) + # Creation may be visible before the PID has been written. + while not child_pid_path.read_text(): + await anyio.sleep(0.02) + child = psutil.Process(int(child_pid_path.read_text())) + record = json.loads(session.record_path.read_text()) + launcher = psutil.Process(record["pid"]) + assert child.pid != launcher.pid + descendants = launcher.children(recursive=True) + assert await session.stop() + assert not process_alive(launcher) + assert not process_alive(child) + assert all(not process_alive(process) for process in descendants) + assert not session.record_path.exists() + assert not session.connection_path.exists() + finally: + await session.stop() + + anyio.run(check) diff --git a/tests/test_server_cli.py b/tests/test_server_cli.py index 05dc43e..80fda55 100644 --- a/tests/test_server_cli.py +++ b/tests/test_server_cli.py @@ -57,6 +57,39 @@ def jupyverse_server(tmp_path_factory): (server_root / "remote.ipynb").write_text(json.dumps(server.notebook), encoding="utf-8") (server_root / "writable.ipynb").write_text(json.dumps(server.notebook), encoding="utf-8") (server_root / "rtc.ipynb").write_text(json.dumps(server.notebook), encoding="utf-8") + executable_notebook = { + "nbformat": 4, "nbformat_minor": 5, + "metadata": {"kernelspec": { + "name": "python3", "display_name": "Python 3", "language": "python", + }}, + "cells": [ + {"id": cell_id, "cell_type": "code", "metadata": {}, "source": source, + "outputs": [], "execution_count": None} + for cell_id, source in [ + ("define", "cli_value = 41"), + ("read", "print(cli_value + 1)"), + ("error", "raise ValueError('cell failed')"), + ] + ], + } + (server_root / "execute.ipynb").write_text( + json.dumps(executable_notebook), encoding="utf-8" + ) + for name in ("managed.ipynb", "other-managed.ipynb"): + (server_root / name).write_text(json.dumps(executable_notebook), encoding="utf-8") + for name, fail in [("execute-all.ipynb", False), ("execute-stop.ipynb", True)]: + whole_notebook = json.loads(json.dumps(executable_notebook)) + whole_notebook["cells"].insert(1, { + "id": "markdown", "cell_type": "markdown", "metadata": {}, + "source": "# Skipped during execution", + }) + if not fail: + whole_notebook["cells"].pop() + whole_notebook["cells"].append({ + "id": "last", "cell_type": "code", "metadata": {}, + "source": "print(cli_value + 2)", "outputs": [], "execution_count": None, + }) + (server_root / name).write_text(json.dumps(whole_notebook), encoding="utf-8") (server_root / "rtc.txt").write_text("Original shared text", encoding="utf-8") server.text = "SERVER LICENSE\n\nOnly available from the server.\n" # Write exact bytes so Windows does not translate LF into CRLF. @@ -114,7 +147,8 @@ def jupyverse_server(tmp_path_factory): @pytest.fixture def run_cli(jupyverse_server): - def run(*arguments, expected_code=0, command="show", input=None, collaborative=False): + def run(*arguments, expected_code=0, command="show", input=None, collaborative=False, + cell_error=False): environment = dict(os.environ, PYTHONIOENCODING="utf-8", NO_COLOR="1", COLUMNS="80") result = subprocess.run( [ @@ -139,7 +173,8 @@ def run(*arguments, expected_code=0, command="show", input=None, collaborative=F result.stderr = Text.from_ansi(result.stderr).plain assert result.returncode == expected_code, result.stdout + result.stderr assert "WRONG_LOCAL_CONTENT" not in result.stdout + result.stderr - assert "Traceback" not in result.stderr + if not cell_error: + assert "Traceback" not in result.stderr return result return run @@ -252,6 +287,119 @@ def test_write_server_cell(run_cli): assert json.loads(run_cli(path, "--json").stdout) == after +def test_execute_reuses_notebook_session(run_cli, jupyverse_server): + run_cli("execute.ipynb", "--cell", "define", command="execute") + result = run_cli("execute.ipynb", "--cell", "read", "--json", command="execute") + cell = json.loads(result.stdout) + assert cell == json.loads(run_cli("execute.ipynb", "--cell", "read", "--json").stdout) + assert cell["execution_count"] == 2 + assert "".join(cell["outputs"][0]["text"]) == "42\n" + rendered = run_cli("execute.ipynb", "--cell", "read", command="execute") + shown = run_cli("execute.ipynb", "--cell", "read") + assert rendered.stdout + rendered.stderr == shown.stdout + shown.stderr + saved = json.loads(run_cli("execute.ipynb", "--cell", "read", "--json").stdout) + cell["execution_count"] = 3 + assert saved == cell + result = run_cli( + "execute.ipynb", "--cell", "error", command="execute", expected_code=1, + cell_error=True + ) + assert "ValueError: cell failed" in result.stderr + saved = json.loads(run_cli("execute.ipynb", "--cell", "error", "--json").stdout) + assert saved["outputs"][0]["output_type"] == "error" + with httpx2.Client(trust_env=False) as client: + response = client.get(f"{jupyverse_server.url}/api/sessions") + response.raise_for_status() + sessions = [s for s in response.json() if s["path"] == "execute.ipynb"] + assert len(sessions) == 1 + # Clean up only this test's session after checking that it remains alive. + client.delete(f"{jupyverse_server.url}/api/sessions/{sessions[0]['id']}").raise_for_status() + + +def test_manage_remote_notebook_kernels(run_cli, jupyverse_server): + for path in ("managed.ipynb", "other-managed.ipynb"): + run_cli(path, "--cell", "define", "--json", command="execute") + sessions = json.loads(run_cli("list", "--json", command="kernel").stdout) + assert {session["path"] for session in sessions} == {"managed.ipynb", "other-managed.ipynb"} + target = next(session for session in sessions if session["path"] == "managed.ipynb") + assert target["kernel_name"] == "python3" + assert target["alive"] + text = run_cli("list", command="kernel").stdout + assert "managed.ipynb" in text + assert target["kernel_id"] in text + result = run_cli("stop", "missing.ipynb", command="kernel", expected_code=1) + assert "No remote kernel session" in result.stderr + assert json.loads(run_cli("list", "--json", command="kernel").stdout) == sessions + run_cli("stop", "./managed.ipynb", command="kernel") + remaining = json.loads(run_cli("list", "--json", command="kernel").stdout) + assert [session["path"] for session in remaining] == ["other-managed.ipynb"] + with httpx2.Client(trust_env=False) as client: + response = client.get(f"{jupyverse_server.url}/api/kernels") + response.raise_for_status() + assert target["kernel_id"] not in {kernel["id"] for kernel in response.json()} + result = run_cli("managed.ipynb", "--cell", "define", "--json", command="execute") + assert json.loads(result.stdout)["execution_count"] == 1 + for path in ("managed.ipynb", "other-managed.ipynb"): + run_cli("stop", path, command="kernel") + assert json.loads(run_cli("list", "--json", command="kernel").stdout) == [] + + +def test_remote_kernel_start_is_ready_and_preserves_notebook(run_cli): + path = "managed.ipynb" + before = run_cli(path, "--json").stdout + result = run_cli("start", path, "--kernel", "python3", command="kernel") + assert result.stdout == "" + assert json.loads(run_cli(path, "--json").stdout) == json.loads(before) + sessions = json.loads(run_cli("list", "--json", command="kernel").stdout) + assert len(sessions) == 1 + assert sessions[0]["path"] == path + assert sessions[0]["alive"] + result = run_cli(path, "--cell", "define", "--json", command="execute") + assert json.loads(result.stdout)["execution_count"] == 1 + before = run_cli(path, "--json").stdout + run_cli("start", path, command="kernel") + assert json.loads(run_cli(path, "--json").stdout) == json.loads(before) + assert json.loads(run_cli("list", "--json", command="kernel").stdout) == sessions + result = run_cli(path, "--cell", "read", "--json", command="execute") + assert json.loads(result.stdout)["execution_count"] == 2 + run_cli("stop", path, command="kernel") + + +@pytest.mark.parametrize("fail", [False, True]) +def test_execute_whole_notebook(run_cli, jupyverse_server, fail): + path = "execute-stop.ipynb" if fail else "execute-all.ipynb" + result = run_cli(path, "--json", command="execute", expected_code=int(fail)) + notebook = json.loads(result.stdout) + assert notebook == json.loads(run_cli(path, "--json").stdout) + cells = notebook["cells"] + assert cells[0]["execution_count"] == 1 + assert cells[1]["cell_type"] == "markdown" + assert cells[2]["execution_count"] == 2 + assert "".join(cells[2]["outputs"][0]["text"]) == "42\n" + if fail: + assert "ValueError: cell failed" in result.stderr + assert cells[3]["outputs"][0]["output_type"] == "error" + assert cells[-1]["execution_count"] is None + assert cells[-1]["outputs"] == [] + else: + assert cells[-1]["execution_count"] == 3 + assert "".join(cells[-1]["outputs"][0]["text"]) == "43\n" + with httpx2.Client(trust_env=False) as client: + sessions = client.get(f"{jupyverse_server.url}/api/sessions").json() + session = next(session for session in sessions if session["path"] == path) + client.delete(f"{jupyverse_server.url}/api/sessions/{session['id']}").raise_for_status() + + +@pytest.mark.parametrize("path, cell_id, message", [ + ("remote.ipynb", "missing", "Cell ID not found"), + ("remote.ipynb", "other-cell", "Cell is not a code cell"), + ("LICENSE", "code", "requires a notebook"), +]) +def test_invalid_execute(run_cli, path, cell_id, message): + result = run_cli(path, "--cell", cell_id, command="execute", expected_code=1) + assert message in result.stderr + + @pytest.mark.parametrize("path", ["remote.ipynb", "LICENSE"]) @pytest.mark.parametrize("as_json", [False, True]) def test_collaborative_show(run_cli, path, as_json): diff --git a/tests/test_show.py b/tests/test_show.py index c58f673..2f13ba8 100644 --- a/tests/test_show.py +++ b/tests/test_show.py @@ -185,7 +185,8 @@ async def check(): @pytest.mark.parametrize("cell_id", [None, "cell-12"]) @pytest.mark.parametrize("show_cell_ids", [False, True]) -def test_inline_uses_notebook_editor_without_starting_kernel(cell_id, show_cell_ids): +@pytest.mark.parametrize("slow_mount", [False, True]) +def test_inline_uses_notebook_editor_without_starting_kernel(cell_id, show_cell_ids, slow_mount): sources = [f"source_{index:02}" for index in range(30)] sources[0] = "\n".join((f"long_line_{index:02}" for index in range(40))) notebook = { @@ -216,11 +217,16 @@ def exit(self, *args, **kwargs): app = TestApp() + class CellWidget(Static): + async def on_mount(self): + if slow_mount: + await anyio.sleep(0.05) + def factory(cell, language, kernel, **kwargs): assert kernel is None assert kwargs == ({"show_cell_id": True} if show_cell_ids else {}) used_cells.append(cell) - widget = Static(cell) + widget = CellWidget(cell) widget.source = widget widget.select = lambda: None return widget diff --git a/tests/test_tabs.py b/tests/test_tabs.py new file mode 100644 index 0000000..f513cc3 --- /dev/null +++ b/tests/test_tabs.py @@ -0,0 +1,111 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import anyio +import pytest +from jupyter_ydoc import YFile +from textual.widgets import Button, Static, Tab +from txl_jpterm.close_tab import ConfirmCloseTab +from txl_jpterm.main import Jpterm +from txl_jpterm.main_area import MainArea +from txl_text_editor.main import TextEditor + + +def make_app(): + area = MainArea() + launcher = Button("Launcher", id="test-launcher") + app = Jpterm(Static(), Static(), area) + app.start(launcher, Static()) + return app, area, launcher + + +def test_close_tabs_selects_neighbour_and_restores_launcher(): + async def check(): + app, area, launcher = make_app() + first, second = Button("First"), Button("Second") + async with app.run_test() as pilot: + await pilot.pause() + area.show(first, "First") + await pilot.pause() + area.show(second, "Second") + await pilot.pause() + assert area.shown is second + await pilot.press("ctrl+w") + await pilot.pause() + assert area.shown is first + assert first.has_focus + assert second not in area.mounted + assert second.parent is None + area.show(launcher) + await pilot.pause() + await pilot.press("ctrl+w") + await pilot.pause() + assert area.shown is first + assert launcher not in area.widget_to_tab + await pilot.press("ctrl+w") + await pilot.pause() + assert area.shown is launcher + assert launcher.has_focus + assert area.mounted == [launcher] + assert len(area.tabs.query(Tab)) == 1 + await pilot.press("ctrl+w", "l") + await pilot.pause() + assert area.mounted == [launcher] + assert len(area.tabs.query(Tab)) == 1 + # New documents can still be opened after all document tabs close. + third = Button("Third") + area.show(third, "Third") + await pilot.pause() + assert area.shown is third + + anyio.run(check) + + +@pytest.mark.parametrize("close_action", ["discard", "save"]) +@pytest.mark.parametrize("size", [(80, 24), (40, 18)]) +def test_close_modified_editor_requires_confirmation_and_cleans_up(close_action, size): + async def check(): + async with anyio.create_task_group() as tasks: + app, area, launcher = make_app() + document = YFile() + document.source = "original" + contents = SimpleNamespace(get=AsyncMock(return_value=document), save=AsyncMock()) + async with app.run_test(size=size) as pilot: + await pilot.pause() + editor = TextEditor(contents, area, tasks) + area.show(editor, "example.txt") + await editor.open("example.txt") + await pilot.pause() + editor.editor.focus() + document._ysource += " changed" + await pilot.pause() + assert area.widget_to_tab[editor][1] + await pilot.press("ctrl+w") + assert isinstance(app.screen, ConfirmCloseTab) + await pilot.press("escape") + await pilot.pause() + assert area.shown is editor + assert document.source == "original changed" + contents.save.assert_not_awaited() + if close_action == "discard": + await pilot.press("ctrl+w") + await pilot.click("#discard-changes") + else: + await pilot.press("ctrl+s") + await pilot.pause() + contents.save.assert_awaited_once_with("example.txt", document) + assert not area.widget_to_tab[editor][1] + await pilot.press("ctrl+w") + await pilot.pause() + assert not isinstance(app.screen, ConfirmCloseTab) + assert area.shown is launcher + assert editor not in area.widget_to_tab + assert editor.parent is None + assert editor._closed + # Later shared-document changes must not reach a closed TextInput. + document._ysource += " after close" + await pilot.pause() + await editor.close() # Application shutdown can close it again safely. + tasks.cancel_scope.cancel() + + anyio.run(check) diff --git a/txl/txl/text_input.py b/txl/txl/text_input.py index e0b7fac..5e23d3f 100644 --- a/txl/txl/text_input.py +++ b/txl/txl/text_input.py @@ -9,6 +9,9 @@ class TextInput(TextArea): def __init__(self, ytext: Text, path=None, language=None): self.ytext = ytext + self._subscription = None + self._stopped = False + self.send_change_events = None text = str(ytext) if language is None: if path is not None: @@ -22,12 +25,26 @@ def __init__(self, ytext: Text, path=None, language=None): self.language = language async def start(self): + if self._stopped: + return self.send_change_events, self.receive_change_events = create_memory_object_stream() - self.ytext.observe(self.on_change) - await self.observe_changes() + self._subscription = self.ytext.observe(self.on_change) + try: + await self.observe_changes() + finally: + await self.stop() async def stop(self): - await self.send_change_events.aclose() + self._stopped = True + if self._subscription is not None: + self.ytext.unobserve(self._subscription) + self._subscription = None + if self.send_change_events is not None: + stream, self.send_change_events = self.send_change_events, None + await stream.aclose() + + async def on_unmount(self): + await self.stop() def delete( self,