From d04462b1e5f43c2377835f88850f45ee0197e9fc Mon Sep 17 00:00:00 2001 From: David Brochart Date: Mon, 19 Oct 2026 15:20:28 +0200 Subject: [PATCH 1/3] Support writing to a document --- docs/usage/CLI.md | 35 ++++ examples/plotext.ipynb | 149 ++++++++--------- plugins/cell/txl_cell/main.py | 8 +- .../txl_notebook_editor/main.py | 19 ++- .../txl_remote_contents/main.py | 1 + src/jpterm/app.py | 21 +++ src/jpterm/cli.py | 4 + src/jpterm/show.py | 40 ++--- src/jpterm/write.py | 74 +++++++++ tests/test_server_cli.py | 76 ++++++++- tests/test_show.py | 30 ++-- tests/test_write.py | 152 ++++++++++++++++++ 12 files changed, 493 insertions(+), 116 deletions(-) create mode 100644 src/jpterm/app.py create mode 100644 src/jpterm/write.py create mode 100644 tests/test_write.py diff --git a/docs/usage/CLI.md b/docs/usage/CLI.md index ec436ba..5151b96 100644 --- a/docs/usage/CLI.md +++ b/docs/usage/CLI.md @@ -17,3 +17,38 @@ If you pass `--json` then the content will not be represented as a rich output b ```bash jpterm show path/to/notebook --json | jq -r '.cells[0].source' ``` + +Use `--cell CELL_ID` to show a single notebook cell, including its saved outputs: + +```bash +jpterm show notebook.ipynb --cell CELL_ID +jpterm show notebook.ipynb --cell CELL_ID --json +``` + +To display IDs use `--show-cell-ids`: + +```bash +jpterm show notebook.ipynb --show-cell-ids +``` + +This also works with `--cell CELL_ID`. JSON output already includes cell IDs. + +Use `write` to create or replace a text file: + +```bash +jpterm write script.py --source 'print("hello")' +cat script.py | jpterm write another.py --stdin +``` + +For notebooks, specify `--cell CELL_ID` to replace a cell's source: + +```bash +jpterm write notebook.ipynb --cell CELL_ID --source 'print("hello")' +cat script.py | jpterm write notebook.ipynb --cell CELL_ID --stdin +jpterm --server http://127.0.0.1:8000 write notebook.ipynb --cell CELL_ID --source 'print(42)' +``` + +Provide exactly one of `--source` or `--stdin`; `--source ''` clears the content. +When a code cell's source changes, its saved outputs and execution count are cleared. +Other cells are preserved, and no cells are executed. Writing a whole notebook without +`--cell` is not supported. diff --git a/examples/plotext.ipynb b/examples/plotext.ipynb index e91beff..e09cf4b 100644 --- a/examples/plotext.ipynb +++ b/examples/plotext.ipynb @@ -1,82 +1,73 @@ { - "cells": [ - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from ypywidgets_textual.plotext import Plotext\n", - "plt = Plotext()" - ] + "cells": [ + { + "source": "from ypywidgets_textual.plotext import Plotext\nplt = Plotext()", + "cell_type": "code", + "metadata": {}, + "execution_count": null, + "id": "eff14ab7-9e42-40a3-847f-8fab1c8320da", + "outputs": [] + }, + { + "cell_type": "code", + "source": "plt", + "metadata": {}, + "outputs": [], + "id": "fff14ab7-9e42-40a3-847f-8fab1c8320db", + "execution_count": null + }, + { + "source": "plt.title('Scatter Plot')", + "metadata": {}, + "cell_type": "code", + "execution_count": null, + "id": "fff14ab7-9e42-40a3-847f-8fab1c8321da", + "outputs": [] + }, + { + "id": "fff14ab7-9e42-40a3-847f-8fab1c8420da", + "source": "x = [0, 1, 2, 3, 4, 5]\ny = [5, 6, 7, 1, 2, 3]\nplt.scatter(x, y)", + "execution_count": null, + "metadata": {}, + "outputs": [], + "cell_type": "code" + }, + { + "execution_count": null, + "id": "fff14ab7-9e42-40a3-847f-8fab1c9320da", + "metadata": {}, + "outputs": [], + "source": "plt.title('Hello')", + "cell_type": "code" + }, + { + "id": "fff14ab7-9e42-40a3-847f-8fab1d8320da", + "metadata": {}, + "cell_type": "code", + "outputs": [], + "execution_count": null, + "source": "plt.clear_data()" + } + ], + "metadata": { + "language_info": { + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "file_extension": ".py", + "name": "python", + "version": "3.11.5", + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "mimetype": "text/x-python" + }, + "kernelspec": { + "language": "python", + "display_name": "Python 3 (ipykernel)", + "name": "python3" + } }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "plt" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "plt.title('Scatter Plot')" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "x = [0, 1, 2, 3, 4, 5]\n", - "y = [5, 6, 7, 1, 2, 3]\n", - "plt.scatter(x, y)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "plt.title('Hello')" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "plt.clear_data()" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.11.5" - } - }, - "nbformat": 4, - "nbformat_minor": 4 + "nbformat": 4, + "nbformat_minor": 5 } diff --git a/plugins/cell/txl_cell/main.py b/plugins/cell/txl_cell/main.py index 6b47997..c9b5d2a 100644 --- a/plugins/cell/txl_cell/main.py +++ b/plugins/cell/txl_cell/main.py @@ -7,6 +7,7 @@ from anyio import create_memory_object_stream, create_task_group, sleep from fps import Module from pycrdt import Doc, Map, MapEvent, Text +from rich.markup import escape from rich.text import Text as RichText from textual.app import ComposeResult from textual.containers import Container @@ -63,6 +64,7 @@ def __init__( task_group, show_execution_count: bool = True, show_border: bool = True, + show_cell_id: bool = False, ) -> None: super().__init__() self.ycell = ycell @@ -73,6 +75,7 @@ def __init__( self.task_group = task_group self.show_execution_count = show_execution_count self.show_border = show_border + self.show_cell_id = show_cell_id self.outputs = [] self.update(mount=False) self.ycell.observe_deep(self.on_change) @@ -180,7 +183,10 @@ async def observe_cell_changes(self): def get_execution_count(self, value): execution_count = " " if value is None else str(value).removesuffix(".0") - return r"[green]In \[[#66ff00]" + execution_count + r"[/#66ff00]]:[/green]" + prompt = r"[green]In \[[#66ff00]" + execution_count + r"[/#66ff00]]:[/green]" + if self.show_cell_id: + prompt += " " + escape(self.ycell.get("id") or "(missing)") + return prompt def compose(self) -> ComposeResult: if self.show_execution_count: diff --git a/plugins/notebook_editor/txl_notebook_editor/main.py b/plugins/notebook_editor/txl_notebook_editor/main.py index 41552bb..c14c589 100644 --- a/plugins/notebook_editor/txl_notebook_editor/main.py +++ b/plugins/notebook_editor/txl_notebook_editor/main.py @@ -140,13 +140,24 @@ async def create_empty_notebook(self, kernel_name: str) -> str: await self.contents.save(path, ynb) return path - async def open(self, path: str | None = None) -> None: + async def open( + self, path: str | None = None, *, cell_id: str | None = None, show_cell_ids: bool = False + ) -> None: if path is None: kernel_name = await self.select_kernel() path = await self.create_empty_notebook(kernel_name) self.main_area.set_label(path) self.path = path self.ynb = await self.contents.get(self.path, type="notebook", format="json") + self.cell_index = None + self.show_cell_ids = show_cell_ids + if cell_id is not None: + self.cell_index = next( + (i for i, cell in enumerate(self.ynb.source["cells"]) + if cell.get("id") == cell_id), None + ) + if self.cell_index is None: + raise ValueError(f"Cell ID not found: {cell_id}") self.update() if not self.app.cli: self.nb_change_target = StapledObjectStream( @@ -166,9 +177,11 @@ def update(self): if kernel_name: self.kernel = self.kernels(kernel_name) self.kernel.kernel.busy.connect(self.watch_busy) - for i_cell in range(self.ynb.cell_number): + indices = range(self.ynb.cell_number) if self.cell_index is None else [self.cell_index] + for i_cell in indices: cell = self.cell_factory( - self.ynb.ycells[i_cell], self.language, self.kernel + self.ynb.ycells[i_cell], self.language, self.kernel, + **({"show_cell_id": True} if self.show_cell_ids else {}), ) if self.app.cli: cell.source.read_only = True diff --git a/plugins/remote_contents/txl_remote_contents/main.py b/plugins/remote_contents/txl_remote_contents/main.py index 93065fa..8ec7930 100644 --- a/plugins/remote_contents/txl_remote_contents/main.py +++ b/plugins/remote_contents/txl_remote_contents/main.py @@ -161,6 +161,7 @@ async def save( cookies=self.cookies, ) self.cookies.update(r.cookies) + r.raise_for_status() async def websocket_provider(self, room_id, ydoc, session_id=None, synced=None): ws_url = f"{self.ws_url}/api/collaboration/room/{room_id}" diff --git a/src/jpterm/app.py b/src/jpterm/app.py new file mode 100644 index 0000000..f51a9f1 --- /dev/null +++ b/src/jpterm/app.py @@ -0,0 +1,21 @@ +"""Shared application setup for CLI commands.""" + +from fps import get_root_module + +from txl.app import AppModule + + +def create_app(settings, timeout, inline=False): + config = {"type": AppModule, "config": {"cli": True, "inline": inline}} + for setting in settings: + if "=" not in setting: + raise ValueError(f"No '=' while setting a module parameter: {setting}") + key, value = setting.split("=", 1) + node = config + parts = key.split(".") + for name in parts[:-1]: + node = node.setdefault("modules", {}).setdefault(name, {}) + node.setdefault("config", {})[parts[-1]] = value + module = get_root_module({"app": config}) + module._global_start_timeout = timeout + return module diff --git a/src/jpterm/cli.py b/src/jpterm/cli.py index ecf46af..6b9a758 100644 --- a/src/jpterm/cli.py +++ b/src/jpterm/cli.py @@ -9,6 +9,7 @@ from txl.cli import set_main, txl_main from .show import show +from .write import write def jpterm_main(kwargs): @@ -78,11 +79,14 @@ def command(ctx, **kwargs): else: if kwargs["collaborative"] and not kwargs["server"]: raise click.UsageError("--collaborative requires --server for CLI commands.") + if kwargs["collaborative"] and ctx.invoked_subcommand == "write": + raise click.UsageError("write does not support --collaborative yet.") ctx.obj = jpterm_main(kwargs) # Keep the existing TUI options on the root command. command.params.extend(click.command()(_main).params) command.add_command(show) + command.add_command(write) command.add_command(logo) command() diff --git a/src/jpterm/show.py b/src/jpterm/show.py index f3aaf6c..1a270a5 100644 --- a/src/jpterm/show.py +++ b/src/jpterm/show.py @@ -5,27 +5,20 @@ import anyio import rich_click as click -from fps import get_root_module from textual import events from textual.geometry import Size -from txl.app import AppModule from txl.base import Contents, Editors +from .app import create_app -async def load_document(path, settings, timeout, render=False): - config = {"type": AppModule, "config": {"cli": True, "inline": render}} - for setting in settings: - if "=" not in setting: - raise ValueError(f"No '=' while setting a module parameter: {setting}") - key, value = setting.split("=", 1) - node = config - parts = key.split(".") - for name in parts[:-1]: - node = node.setdefault("modules", {}).setdefault(name, {}) - node.setdefault("config", {})[parts[-1]] = value - module = get_root_module({"app": config}) - module._global_start_timeout = timeout + +async def load_document(path, settings, timeout, render=False, cell_id=None, show_cell_ids=False): + if cell_id is not None and PurePosixPath(path).suffix != ".ipynb": + raise ValueError("--cell requires a notebook (.ipynb).") + if show_cell_ids and PurePosixPath(path).suffix != ".ipynb": + raise ValueError("--show-cell-ids requires a notebook (.ipynb).") + module = create_app(settings, timeout, inline=render) with anyio.fail_after(timeout): async with module: if render: @@ -41,7 +34,10 @@ async def load_document(path, settings, timeout, render=False): app.screen.styles.padding = 0 app.screen.styles.scrollbar_size_vertical = 0 await app.screen.mount(editor) - await editor.open(path) + if cell_id is None and not show_cell_ids: + await editor.open(path) + else: + await editor.open(path, cell_id=cell_id, show_cell_ids=show_cell_ids) def show_full_height(): height = max(1, editor.virtual_size.height + editor.styles.gutter.height) @@ -62,16 +58,24 @@ def show_full_height(): kind = "notebook" if PurePosixPath(path).suffix == ".ipynb" else "file" document = await contents.get(path, type=kind) source = document.source + if cell_id is not None: + source = next( + (cell for cell in source["cells"] if cell.get("id") == cell_id), None + ) + if source is None: + raise ValueError(f"Cell ID not found: {cell_id}") return source @click.command() @click.argument("path") @click.option("--json", "as_json", is_flag=True, help="Print the document source as JSON.") +@click.option("--cell", "cell_id", help="Show only the notebook cell with this ID.") +@click.option("--show-cell-ids", is_flag=True, help="Display cell IDs after their prompts.") @click.option("--timeout", type=click.FloatRange(min=0, min_open=True), default=30.0, show_default=True, help="Maximum time to load and synchronize, in seconds.") @click.pass_context -def show(ctx, path, as_json, timeout): +def show(ctx, path, as_json, cell_id, show_cell_ids, timeout): """Show document PATH inline using its registered editor. Display the full document and return immediately. This command does not execute cells. @@ -80,7 +84,7 @@ def show(ctx, path, as_json, timeout): options = ctx.obj try: source = anyio.run( - load_document, path, options["set_"], timeout, not as_json, + load_document, path, options["set_"], timeout, not as_json, cell_id, show_cell_ids, backend=options["backend"], ) if as_json: diff --git a/src/jpterm/write.py b/src/jpterm/write.py new file mode 100644 index 0000000..f031dc0 --- /dev/null +++ b/src/jpterm/write.py @@ -0,0 +1,74 @@ +"""Write text files and notebook cell sources through the contents service.""" + +from importlib.metadata import entry_points +from pathlib import PurePosixPath + +import anyio +import rich_click as click + +from txl.base import Contents + +from .app import create_app + + +async def write_document(path, source, cell_id, settings, timeout): + notebook = PurePosixPath(path).suffix == ".ipynb" + if notebook and cell_id is None: + raise ValueError("Writing a notebook requires --cell CELL_ID.") + if not notebook and cell_id is not None: + raise ValueError("--cell requires a notebook (.ipynb).") + module = create_app(settings, timeout) + with anyio.fail_after(timeout): + async with module: + contents = await module.get(Contents) + if notebook: + document = await contents.get(path, type="notebook") + notebook_source = document.source + cell = next( + (cell for cell in notebook_source["cells"] if cell.get("id") == cell_id), None + ) + if cell is None: + raise ValueError(f"Cell ID not found: {cell_id}") + if cell["source"] == source: + return + cell["source"] = source + if cell["cell_type"] == "code": + cell["outputs"] = [] + cell["execution_count"] = None + document.source = notebook_source + else: + factory = next(iter(entry_points(group="jupyter_ydoc", name="file"))).load() + document = factory() + document.source = source + await contents.save(path, document) + + +@click.command() +@click.argument("path") +@click.option("--cell", "cell_id", help="Replace the source of the notebook cell with this ID.") +@click.option("--source", help="Replacement text (use an empty string to clear it).") +@click.option("--stdin", "from_stdin", is_flag=True, help="Read replacement text from stdin.") +@click.option("--timeout", type=click.FloatRange(min=0, min_open=True), default=30.0, + show_default=True, help="Maximum time to load and save, in seconds.") +@click.pass_context +def write(ctx, path, cell_id, source, from_stdin, timeout): + """Write a text file or replace a notebook cell's source. + + Provide exactly one of --source or --stdin. Changed code cells have their saved + outputs and execution count cleared. This command does not execute cells. + """ + if (source is not None) == from_stdin: + raise click.UsageError("Provide exactly one of --source or --stdin.") + if from_stdin: + source = click.get_text_stream("stdin").read() + options = ctx.obj + try: + anyio.run( + write_document, path, source, cell_id, options["set_"], timeout, + backend=options["backend"], + ) + except Exception as exc: + while getattr(exc, "exceptions", None): + exc = exc.exceptions[0] + message = "Timed out writing document" if isinstance(exc, TimeoutError) else str(exc) + raise click.ClickException(message) from exc diff --git a/tests/test_server_cli.py b/tests/test_server_cli.py index 9fe45b8..86124d3 100644 --- a/tests/test_server_cli.py +++ b/tests/test_server_cli.py @@ -37,6 +37,8 @@ def jupyverse_server(tmp_path_factory): } }, "cells": [ + {"id": "other-cell", "cell_type": "markdown", "metadata": {}, + "source": "unselected_cell_content"}, { "id": "server-cell", "cell_type": "code", @@ -50,6 +52,7 @@ 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.text = "SERVER LICENSE\n\nOnly available from the server.\n" # Write exact bytes so Windows does not translate LF into CRLF. (server_root / "LICENSE").write_bytes(server.text.encode("utf-8")) @@ -104,7 +107,7 @@ def jupyverse_server(tmp_path_factory): @pytest.fixture def run_cli(jupyverse_server): - def run(*arguments, expected_code=0): + def run(*arguments, expected_code=0, command="show", input=None): environment = dict(os.environ, PYTHONIOENCODING="utf-8", NO_COLOR="1", COLUMNS="80") result = subprocess.run( [ @@ -113,7 +116,7 @@ def run(*arguments, expected_code=0): "from jpterm.cli import main; main()", "--server", jupyverse_server.url, - "show", + command, *arguments, ], cwd=jupyverse_server.client_root, @@ -121,6 +124,7 @@ def run(*arguments, expected_code=0): capture_output=True, text=True, encoding="utf-8", + input=input, timeout=20, ) assert result.returncode == expected_code, result.stdout + result.stderr @@ -131,9 +135,15 @@ def run(*arguments, expected_code=0): return run -def test_show_notebook(run_cli, jupyverse_server): - result = run_cli("remote.ipynb") +@pytest.mark.parametrize("show_cell_ids", [False, True]) +def test_show_notebook(run_cli, jupyverse_server, show_cell_ids): + arguments = ["remote.ipynb"] + if show_cell_ids: + arguments.append("--show-cell-ids") + result = run_cli(*arguments) output = result.stdout + result.stderr + assert ("In [7]: server-cell" in output) == show_cell_ids + assert ("In []: other-cell" in output) == show_cell_ids for line in jupyverse_server.source_lines: assert output.count(line) == 1 assert output.count("saved_server_output") == 1 @@ -152,12 +162,42 @@ def test_show_json(run_cli, jupyverse_server, path): elif path == "LICENSE_CRLF": assert source == jupyverse_server.crlf_text else: - cell = source["cells"][0] + cell = source["cells"][1] assert cell["id"] == "server-cell" - assert cell["source"] == jupyverse_server.notebook["cells"][0]["source"] + assert cell["source"] == jupyverse_server.notebook["cells"][1]["source"] assert cell["execution_count"] == 7 +@pytest.mark.parametrize("as_json", [False, True]) +def test_show_cell(run_cli, jupyverse_server, as_json): + arguments = ["remote.ipynb", "--cell", "server-cell"] + if as_json: + arguments.append("--json") + result = run_cli(*arguments) + if as_json: + notebook = json.loads(run_cli("remote.ipynb", "--json").stdout) + assert json.loads(result.stdout) == notebook["cells"][1] + else: + output = result.stdout + result.stderr + assert "unselected_cell_content" not in output + assert output.count("saved_server_output") == 1 + for line in jupyverse_server.source_lines: + assert output.count(line) == 1 + + +@pytest.mark.parametrize("as_json", [False, True]) +@pytest.mark.parametrize("path, message", [ + ("remote.ipynb", "Cell ID not found: missing"), + ("LICENSE", "--cell requires a notebook"), +]) +def test_invalid_cell_selection(run_cli, as_json, path, message): + arguments = [path, "--cell", "missing"] + if as_json: + arguments.append("--json") + result = run_cli(*arguments, expected_code=1) + assert message in result.stderr + + def test_show_plain_text(run_cli, jupyverse_server): result = run_cli("LICENSE") output = result.stdout + result.stderr @@ -175,3 +215,27 @@ def test_show_markdown(run_cli, jupyverse_server): def test_missing_file(run_cli, jupyverse_server): result = run_cli("missing.ipynb", expected_code=1) assert "404" in result.stderr + + +@pytest.mark.parametrize("from_stdin", [False, True]) +def test_write_server_text(run_cli, jupyverse_server, from_stdin): + path = f"written-{from_stdin}.txt" + source = "server write π\nsecond line\n" + arguments = [path, "--stdin"] if from_stdin else [path, "--source", source] + run_cli(*arguments, command="write", input=source if from_stdin else None) + assert json.loads(run_cli(path, "--json").stdout) == source + assert not (jupyverse_server.client_root / path).exists() + run_cli(path, "--source", "", command="write") + assert json.loads(run_cli(path, "--json").stdout) == "" + + +def test_write_server_cell(run_cli): + path = "writable.ipynb" + before = json.loads(run_cli(path, "--json").stdout) + run_cli(path, "--cell", "server-cell", "--source", "print(42)", command="write") + after = json.loads(run_cli(path, "--json").stdout) + before["cells"][1].update(source="print(42)", outputs=[], execution_count=None) + assert after == before + result = run_cli(path, "--cell", "missing", "--source", "bad", command="write", expected_code=1) + assert "Cell ID not found" in result.stderr + assert json.loads(run_cli(path, "--json").stdout) == after diff --git a/tests/test_show.py b/tests/test_show.py index 9b43fbd..7134fc6 100644 --- a/tests/test_show.py +++ b/tests/test_show.py @@ -5,6 +5,7 @@ import anyio import httpx2 +import pytest from click.testing import CliRunner from fps import Module, get_root_module from jpterm.cli import jpterm_main, main @@ -183,10 +184,15 @@ async def check(): anyio.run(check) -def test_inline_uses_notebook_editor_without_starting_kernel(): +@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): 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 = {"cells": [], "metadata": {"kernelspec": {"name": "python3"}}} + notebook = { + "cells": [{"id": f"cell-{i}"} for i in range(len(sources))], + "metadata": {"kernelspec": {"name": "python3"}}, + } document = SimpleNamespace( source=notebook, ycells=sources, cell_number=len(sources), observe=lambda callback: None ) @@ -211,8 +217,9 @@ def exit(self, *args, **kwargs): app = TestApp() - def factory(cell, language, kernel): + 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.source = widget @@ -232,12 +239,17 @@ async def start(self): self.put(_Editors(Static(), Static(), main_area), Editors) with patch("txl.app.modules", {"test": TestModule, "notebook_editor": NotebookEditorModule}): - anyio.run(load_document, "example.ipynb", [], 1, True) + anyio.run(load_document, "example.ipynb", [], 1, True, cell_id, show_cell_ids) assert app.inline_requested - assert used_cells == sources + expected = sources if cell_id is None else [sources[12]] + assert used_cells == expected assert app.rendered - assert app.final_height > 40 - for source in sources: + if cell_id is None: + assert app.final_height > 40 + else: + assert "source_11" not in app.output + assert "source_13" not in app.output + for source in expected: for line in source.splitlines(): assert app.output.count(line) == 1 kernels.assert_not_called() @@ -326,10 +338,10 @@ def test_json_and_inline(): code, output = invoke(["show", "example.ipynb", "--json"]) assert code == 0 assert json.loads(output) == notebook - assert not load.call_args.args[-1] + assert not load.call_args.args[-3] code, output = invoke(["show", "example.ipynb"]) assert code == 0 - assert load.call_args.args[-1] + assert load.call_args.args[-3] def test_logo_command(): diff --git a/tests/test_write.py b/tests/test_write.py new file mode 100644 index 0000000..c3c5061 --- /dev/null +++ b/tests/test_write.py @@ -0,0 +1,152 @@ +import json +import os +import subprocess +import sys +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import anyio +import httpx2 +import pytest +from fps import Module +from jpterm.write import write_document +from jupyter_ydoc import YNotebook +from textual.app import App +from txl_remote_contents.main import RemoteContents + +from txl.base import Contents + + +@pytest.fixture +def notebook(): + document = YNotebook() + document.source = { + "nbformat": 4, "nbformat_minor": 5, + "metadata": {"custom": "preserved"}, + "cells": [ + {"id": "code", "cell_type": "code", "source": "old code", + "metadata": {"custom": "preserved"}, "execution_count": 7, + "outputs": [{"output_type": "stream", "name": "stdout", "text": "old output"}]}, + {"id": "markdown", "cell_type": "markdown", "source": "old markdown", + "metadata": {}}, + ], + } + return document + + +@pytest.fixture +def write_with_contents(): + def run(contents, path, source, cell_id=None): + class TestModule(Module): + async def start(self): + self.put(contents, Contents) + self.put(App(), App) + + with patch("txl.app.modules", {"test": TestModule}): + anyio.run(write_document, path, source, cell_id, [], 5) + + return run + + +@pytest.mark.parametrize("cell_id", ["code", "markdown"]) +def test_write_cell_preserves_other_content(notebook, write_with_contents, cell_id): + before = notebook.source + contents = SimpleNamespace(get=AsyncMock(return_value=notebook), save=AsyncMock()) + write_with_contents(contents, "test.ipynb", "new source\n", cell_id) + expected = before + cell = next(cell for cell in expected["cells"] if cell["id"] == cell_id) + cell["source"] = "new source\n" + if cell_id == "code": + cell["outputs"] = [] + cell["execution_count"] = None + assert notebook.source == expected + contents.save.assert_awaited_once_with("test.ipynb", notebook) + + +def test_unchanged_cell_keeps_outputs(notebook, write_with_contents): + before = notebook.source + contents = SimpleNamespace(get=AsyncMock(return_value=notebook), save=AsyncMock()) + write_with_contents(contents, "test.ipynb", "old code", "code") + assert notebook.source == before + contents.save.assert_not_awaited() + + +def test_unknown_cell_does_not_save(notebook, write_with_contents): + before = notebook.source + contents = SimpleNamespace(get=AsyncMock(return_value=notebook), save=AsyncMock()) + with pytest.raises(ValueError, match="Cell ID not found: missing"): + write_with_contents(contents, "test.ipynb", "new source", "missing") + assert notebook.source == before + contents.save.assert_not_awaited() + + +@pytest.fixture +def run_write(tmp_path): + def run(*arguments, input=None): + return subprocess.run( + [sys.executable, "-c", "from jpterm.cli import main; main()", *arguments], + input=input, capture_output=True, text=True, encoding="utf-8", cwd=tmp_path, + env=dict(os.environ, PYTHONIOENCODING="utf-8", NO_COLOR="1"), timeout=20, + ) + return run + + +@pytest.mark.parametrize("exists", [False, True]) +@pytest.mark.parametrize("source", ["", "hello π\nsecond line\n"]) +def test_write_local_text(run_write, tmp_path, exists, source): + path = tmp_path / "LICENSE" + if exists: + path.write_text("old content", encoding="utf-8") + result = run_write("write", "LICENSE", "--source", source) + assert result.returncode == 0, result.stderr + assert result.stdout == "" + assert path.read_text(encoding="utf-8") == source + + +def test_write_stdin(run_write, tmp_path): + source = "first line\nsecond line π\n" + result = run_write("write", "script.py", "--stdin", input=source) + assert result.returncode == 0, result.stderr + assert (tmp_path / "script.py").read_text(encoding="utf-8") == source + + +def test_write_local_notebook(run_write, tmp_path, notebook): + path = tmp_path / "test.ipynb" + path.write_text(json.dumps(notebook.source), encoding="utf-8") + result = run_write("write", "test.ipynb", "--cell", "code", "--source", "print(42)") + assert result.returncode == 0, result.stderr + cell = json.loads(path.read_text())["cells"][0] + assert cell["source"] == "print(42)" + assert cell["outputs"] == [] + assert cell["execution_count"] is None + + +@pytest.mark.parametrize("arguments, message", [ + (["write", "file.txt"], "Provide exactly one"), + (["write", "file.txt", "--source", "text", "--stdin"], "Provide exactly one"), + (["write", "file.txt", "--cell", "code", "--source", "text"], "requires a notebook"), + (["write", "test.ipynb", "--source", "text"], "requires --cell"), + (["--server", "http://server", "--collaborative", "write", "file.txt", + "--source", "text"], "does not support --collaborative"), +]) +def test_invalid_write(run_write, tmp_path, arguments, message): + result = run_write(*arguments) + assert result.returncode != 0 + assert message in result.stderr + assert not list(tmp_path.iterdir()) + + +def test_remote_save_reports_http_error(): + async def check(): + response = httpx2.Response( + 403, request=httpx2.Request("PUT", "http://server/api/contents/file.txt") + ) + client = AsyncMock() + client.__aenter__.return_value = client + client.put.return_value = response + contents = RemoteContents("http://server", {}, httpx2.Cookies(), False, None) + with patch("txl_remote_contents.main.httpx2.AsyncClient", return_value=client): + with pytest.raises(httpx2.HTTPStatusError): + await contents.save("file.txt", SimpleNamespace(source="text")) + + anyio.run(check) From 69e7968f9ba7c1985f39dfe9c5067989486ff815 Mon Sep 17 00:00:00 2001 From: David Brochart Date: Mon, 19 Oct 2026 15:30:48 +0200 Subject: [PATCH 2/3] Fix tests --- .../local_contents/txl_local_contents/main.py | 8 +++---- tests/test_server_cli.py | 7 +++++- tests/test_write.py | 24 +++++++++++++++---- 3 files changed, 30 insertions(+), 9 deletions(-) diff --git a/plugins/local_contents/txl_local_contents/main.py b/plugins/local_contents/txl_local_contents/main.py index 269b34a..fa4f419 100644 --- a/plugins/local_contents/txl_local_contents/main.py +++ b/plugins/local_contents/txl_local_contents/main.py @@ -27,11 +27,11 @@ async def get( if await p.is_file(): jupyter_ydoc = ydocs[type]() if type == "file": - jupyter_ydoc.source = await p.read_text() + jupyter_ydoc.source = await p.read_text(encoding="utf-8") elif type == "blob": jupyter_ydoc.source = await p.read_bytes() elif type == "notebook": - jupyter_ydoc.source = json.loads(await p.read_text()) + jupyter_ydoc.source = json.loads(await p.read_text(encoding="utf-8")) return jupyter_ydoc async def save( @@ -42,11 +42,11 @@ async def save( p = Path(path) source = jupyter_ydoc.source if isinstance(source, dict): - await p.write_text(json.dumps(source, indent=2)) + await p.write_text(json.dumps(source, indent=2), encoding="utf-8") elif isinstance(source, bytes): await p.write_bytes(source) else: - await p.write_text(source) + await p.write_text(source, encoding="utf-8") class LocalContentsModule(Module): diff --git a/tests/test_server_cli.py b/tests/test_server_cli.py index 86124d3..904024a 100644 --- a/tests/test_server_cli.py +++ b/tests/test_server_cli.py @@ -11,6 +11,7 @@ import httpx2 import pytest +from rich.text import Text @pytest.fixture(scope="module") @@ -75,6 +76,7 @@ def jupyverse_server(tmp_path_factory): cwd=server_root, stdout=log, stderr=subprocess.STDOUT, + env=dict(os.environ, PYTHONUTF8="1"), ) try: deadline = time.monotonic() + 20 @@ -92,7 +94,8 @@ def jupyverse_server(tmp_path_factory): pass time.sleep(0.1) if not ready: - pytest.fail(f"Jupyverse did not become ready:\n{server.log_path.read_text()}") + server_log = server.log_path.read_text(encoding="utf-8") + pytest.fail(f"Jupyverse did not become ready:\n{server_log}") yield server finally: if server.process.poll() is None: @@ -127,6 +130,8 @@ def run(*arguments, expected_code=0, command="show", input=None): input=input, timeout=20, ) + result.stdout = Text.from_ansi(result.stdout).plain + 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 diff --git a/tests/test_write.py b/tests/test_write.py index c3c5061..1e14431 100644 --- a/tests/test_write.py +++ b/tests/test_write.py @@ -11,6 +11,7 @@ from fps import Module from jpterm.write import write_document from jupyter_ydoc import YNotebook +from rich.text import Text from textual.app import App from txl_remote_contents.main import RemoteContents @@ -82,12 +83,16 @@ def test_unknown_cell_does_not_save(notebook, write_with_contents): @pytest.fixture def run_write(tmp_path): - def run(*arguments, input=None): - return subprocess.run( + def run(*arguments, input=None, env=None): + result = subprocess.run( [sys.executable, "-c", "from jpterm.cli import main; main()", *arguments], input=input, capture_output=True, text=True, encoding="utf-8", cwd=tmp_path, - env=dict(os.environ, PYTHONIOENCODING="utf-8", NO_COLOR="1"), timeout=20, + env={**os.environ, "PYTHONIOENCODING": "utf-8", "NO_COLOR": "1", **(env or {})}, + timeout=20, ) + result.stdout = Text.from_ansi(result.stdout).plain + result.stderr = Text.from_ansi(result.stderr).plain + return result return run @@ -110,12 +115,23 @@ def test_write_stdin(run_write, tmp_path): assert (tmp_path / "script.py").read_text(encoding="utf-8") == source +def test_unicode_file_with_non_utf8_locale(run_write, tmp_path): + source = "Unicode text: π\n" + env = {"PYTHONUTF8": "0", "PYTHONCOERCECLOCALE": "0", "LC_ALL": "C"} + result = run_write("write", "file.txt", "--stdin", input=source, env=env) + assert result.returncode == 0, result.stderr + assert (tmp_path / "file.txt").read_text(encoding="utf-8") == source + result = run_write("show", "file.txt", "--json", env=env) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout) == source + + def test_write_local_notebook(run_write, tmp_path, notebook): path = tmp_path / "test.ipynb" path.write_text(json.dumps(notebook.source), encoding="utf-8") result = run_write("write", "test.ipynb", "--cell", "code", "--source", "print(42)") assert result.returncode == 0, result.stderr - cell = json.loads(path.read_text())["cells"][0] + cell = json.loads(path.read_text(encoding="utf-8"))["cells"][0] assert cell["source"] == "print(42)" assert cell["outputs"] == [] assert cell["execution_count"] is None From e008392208356d8e64dc7347267a0fb0a97e4058 Mon Sep 17 00:00:00 2001 From: David Brochart Date: Mon, 19 Oct 2026 15:34:25 +0200 Subject: [PATCH 3/3] again --- tests/test_server_cli.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_server_cli.py b/tests/test_server_cli.py index 904024a..1d6f6e7 100644 --- a/tests/test_server_cli.py +++ b/tests/test_server_cli.py @@ -228,7 +228,8 @@ def test_write_server_text(run_cli, jupyverse_server, from_stdin): source = "server write π\nsecond line\n" arguments = [path, "--stdin"] if from_stdin else [path, "--source", source] run_cli(*arguments, command="write", input=source if from_stdin else None) - assert json.loads(run_cli(path, "--json").stdout) == source + # Jupyverse writes text using the server platform's native line endings. + assert json.loads(run_cli(path, "--json").stdout) == source.replace("\n", os.linesep) assert not (jupyverse_server.client_root / path).exists() run_cli(path, "--source", "", command="write") assert json.loads(run_cli(path, "--json").stdout) == ""