fix(convert): refresh pair baseline after canonical pair sync

Author: TTTPOB

jupyter_jcli/commands/convert_cmd.py

@@ -5,23 +5,43 @@
 import click
 import nbformat
 
+from jupyter_jcli import pair_baseline
+from jupyter_jcli.canonicalize import canonicalize_py_text
 from jupyter_jcli.pair_io import create_ipynb_from_parsed, emit_py_percent, update_ipynb_sources
-from jupyter_jcli.parser import parse_ipynb, parse_py_percent
+from jupyter_jcli.parser import ipynb_path_for_py, parse_ipynb, parse_py_percent
 
 
 @click.group()
 def convert():
     """Convert between .ipynb and py:percent (.py) formats."""
 
 
+def _is_canonical_pair(py_path: Path, ipynb_path: Path) -> bool:
+    """Return True when *py_path* and *ipynb_path* are the managed pair."""
+    return ipynb_path_for_py(py_path).resolve(strict=False) == ipynb_path.resolve(strict=False)
+
+
+def _refresh_pair_baseline(py_path: Path) -> None:
+    """Best-effort baseline refresh after a successful canonical pair sync."""
+    try:
+        canonical_text = canonicalize_py_text(py_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError):
+        return
+    pair_baseline.write_baseline(py_path, canonical_text)
+
+
 @convert.command("ipynb-to-py")
 @click.argument("in_ipynb", metavar="<in.ipynb>", type=click.Path(exists=True, dir_okay=False))
 @click.argument("out_py", metavar="<out.py>", type=click.Path(dir_okay=False))
 def ipynb_to_py(in_ipynb: str, out_py: str) -> None:
     """Convert a .ipynb file to py:percent format."""
     parsed = parse_ipynb(in_ipynb)
     text = emit_py_percent(parsed)
-    Path(out_py).write_text(text, encoding="utf-8")
+    in_ipynb_path = Path(in_ipynb)
+    out_py_path = Path(out_py)
+    out_py_path.write_text(text, encoding="utf-8")
+    if _is_canonical_pair(out_py_path, in_ipynb_path):
+        _refresh_pair_baseline(out_py_path)
     click.echo(f"Wrote {out_py}")
 
 
@@ -36,23 +56,27 @@ def py_to_ipynb(in_py: str, out_ipynb: str | None) -> None:
     (outputs and metadata are preserved). Otherwise a new notebook is created.
     """
     parsed = parse_py_percent(in_py)
+    in_py_path = Path(in_py)
 
     # Determine output path
     if out_ipynb is None:
-        py_path = Path(in_py)
-        stem = py_path.stem
+        stem = in_py_path.stem
         if stem.endswith(".dummy"):
             stem = stem[: -len(".dummy")]
-        out_ipynb = str(py_path.parent / f"{stem}.ipynb")
+        out_ipynb = str(in_py_path.parent / f"{stem}.ipynb")
 
     out_path = Path(out_ipynb)
 
     if out_path.exists():
         # Update existing notebook sources only
         update_ipynb_sources(out_path, parsed.cells)
+        if _is_canonical_pair(in_py_path, out_path):
+            _refresh_pair_baseline(in_py_path)
         click.echo(f"Updated {out_ipynb}")
     else:
         # Create a new notebook
         nb = create_ipynb_from_parsed(parsed)
         nbformat.write(nb, str(out_path))
+        if _is_canonical_pair(in_py_path, out_path):
+            _refresh_pair_baseline(in_py_path)
         click.echo(f"Wrote {out_ipynb}")

tests/test_convert_cmd.py

@@ -1,11 +1,14 @@
 """Tests for `j-cli convert`."""
 
+import subprocess
 from pathlib import Path
 
 import nbformat
 import pytest
 from click.testing import CliRunner
 
+from jupyter_jcli import pair_baseline
+from jupyter_jcli.canonicalize import canonicalize_py_text
 from jupyter_jcli.cli import main
 from jupyter_jcli.parser import parse_py_percent, parse_py_percent_text
 
@@ -19,6 +22,46 @@ def _invoke(*args: str):
     return runner.invoke(main, list(args), catch_exceptions=False)
 
 
+@pytest.fixture
+def git_repo(tmp_path: Path) -> Path:
+    subprocess.run(["git", "init"], cwd=str(tmp_path), check=True, capture_output=True)
+    subprocess.run(
+        ["git", "config", "user.email", "test@test.com"],
+        cwd=str(tmp_path),
+        check=True,
+        capture_output=True,
+    )
+    subprocess.run(
+        ["git", "config", "user.name", "Test User"],
+        cwd=str(tmp_path),
+        check=True,
+        capture_output=True,
+    )
+    return tmp_path
+
+
+def _git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["git", *args],
+        cwd=str(repo),
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+
+def _has_ref(repo: Path, rel_py_path: str) -> bool:
+    ref_name = pair_baseline._ref_name(Path(rel_py_path).as_posix())
+    result = subprocess.run(
+        ["git", "for-each-ref", ref_name, "--format=%(refname)"],
+        cwd=str(repo),
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return result.stdout.strip() == ref_name
+
+
 def _make_ipynb(cells: list[tuple[str, str, list]], kernel: str = "python3") -> nbformat.NotebookNode:
     nb = nbformat.v4.new_notebook()
     nb.metadata["kernelspec"] = {"name": kernel, "display_name": kernel, "language": "python"}
@@ -87,6 +130,51 @@ def test_roundtrip_content(self, tmp_path):
         assert parsed.cells[0].source == "a = 1"
         assert parsed.cells[1].source == "b = 2"
 
+    def test_canonical_ipynb_to_py_refreshes_baseline(self, git_repo):
+        nb = _make_ipynb([("code", "x = 1", [])])
+        ipynb = git_repo / "nb.ipynb"
+        nbformat.write(nb, str(ipynb))
+        py = git_repo / "nb.py"
+
+        result = _invoke("convert", "ipynb-to-py", str(ipynb), str(py))
+
+        assert result.exit_code == 0
+        assert _has_ref(git_repo, "nb.py")
+        assert pair_baseline.read_baseline(py) == canonicalize_py_text(py.read_text(encoding="utf-8"))
+
+    def test_dummy_ipynb_to_py_refreshes_baseline(self, git_repo):
+        nb = _make_ipynb([("code", "x = 1", [])])
+        ipynb = git_repo / "nb.ipynb"
+        nbformat.write(nb, str(ipynb))
+        py = git_repo / "nb.dummy.py"
+
+        result = _invoke("convert", "ipynb-to-py", str(ipynb), str(py))
+
+        assert result.exit_code == 0
+        assert _has_ref(git_repo, "nb.dummy.py")
+
+    def test_noncanonical_ipynb_to_py_does_not_refresh_baseline(self, git_repo):
+        nb = _make_ipynb([("code", "x = 1", [])])
+        ipynb = git_repo / "nb.ipynb"
+        nbformat.write(nb, str(ipynb))
+        py = git_repo / "custom.py"
+
+        result = _invoke("convert", "ipynb-to-py", str(ipynb), str(py))
+
+        assert result.exit_code == 0
+        assert not _has_ref(git_repo, "custom.py")
+
+    def test_ipynb_to_py_non_git_still_succeeds(self, tmp_path):
+        nb = _make_ipynb([("code", "x = 1", [])])
+        ipynb = tmp_path / "nb.ipynb"
+        nbformat.write(nb, str(ipynb))
+        py = tmp_path / "nb.py"
+
+        result = _invoke("convert", "ipynb-to-py", str(ipynb), str(py))
+
+        assert result.exit_code == 0
+        assert py.exists()
+
 
 # ---------------------------------------------------------------------------
 # py-to-ipynb — new file creation
@@ -129,6 +217,37 @@ def test_dummy_py_default_output(self, tmp_path):
         assert result.exit_code == 0
         assert (tmp_path / "foo.ipynb").exists()
 
+    def test_canonical_py_to_ipynb_refreshes_baseline(self, git_repo):
+        py = git_repo / "script.py"
+        py.write_text("# %%\nx = 1\n", encoding="utf-8")
+
+        result = _invoke("convert", "py-to-ipynb", str(py))
+
+        assert result.exit_code == 0
+        assert (git_repo / "script.ipynb").exists()
+        assert _has_ref(git_repo, "script.py")
+        assert pair_baseline.read_baseline(py) == canonicalize_py_text(py.read_text(encoding="utf-8"))
+
+    def test_noncanonical_py_to_ipynb_does_not_refresh_baseline(self, git_repo):
+        py = git_repo / "script.py"
+        py.write_text("# %%\nx = 1\n", encoding="utf-8")
+        out = git_repo / "custom.ipynb"
+
+        result = _invoke("convert", "py-to-ipynb", str(py), str(out))
+
+        assert result.exit_code == 0
+        assert out.exists()
+        assert not _has_ref(git_repo, "script.py")
+
+    def test_py_to_ipynb_non_git_still_succeeds(self, tmp_path):
+        py = tmp_path / "script.py"
+        py.write_text("# %%\nx = 1\n", encoding="utf-8")
+
+        result = _invoke("convert", "py-to-ipynb", str(py))
+
+        assert result.exit_code == 0
+        assert (tmp_path / "script.ipynb").exists()
+
 
 # ---------------------------------------------------------------------------
 # py-to-ipynb — in-place update (preserve outputs)

tests/test_hooks_pair_drift.py

@@ -678,6 +678,40 @@ def test_pre_merge_updates_ref_so_following_post_does_not_conflict(
         assert [cell.source for cell in nb_after.cells if cell.source.strip()] == ["x = 40"]
 
 
+class TestConvertSeededBaseline:
+    def test_convert_then_first_edit_uses_sticky_baseline(self, git_repo: Path):
+        py, ipynb = _make_pair(git_repo, ["x = 1"], ["x = 1"])
+        _git(git_repo, "add", "nb.py")
+        _git(git_repo, "commit", "-m", "init", env=_git_env(100))
+
+        py.write_text(py.read_text(encoding="utf-8").replace("x = 1", "x = 10"), encoding="utf-8")
+
+        runner = CliRunner()
+        convert = runner.invoke(
+            main,
+            ["convert", "py-to-ipynb", str(py), str(ipynb)],
+            catch_exceptions=False,
+        )
+        assert convert.exit_code == 0
+
+        ref_ts = _git(
+            git_repo,
+            "log",
+            "-1",
+            "--format=%ct",
+            pair_baseline._ref_name("nb.py"),
+        ).stdout.strip()
+        assert ref_ts
+
+        py.write_text(py.read_text(encoding="utf-8").replace("x = 10", "x = 20"), encoding="utf-8")
+        code, out = _invoke_post({"tool_name": "Edit", "tool_input": {"file_path": str(py)}})
+
+        assert code == 0
+        assert "Auto-synced" in _additional_context(out)
+        nb_after = nbformat.read(str(ipynb), as_version=4)
+        assert [cell.source for cell in nb_after.cells if cell.source.strip()] == ["x = 20"]
+
+
 class TestGcPairSyncRefsCLI:
     def test_dry_run_reports_without_deleting(self, git_repo: Path, monkeypatch: pytest.MonkeyPatch):
         py, _ipynb = _make_pair(git_repo, ["x = 1"], ["x = 1"])