Fix(guard,#19475): etendre check-subprocess-encoding aux cellules .ipynb

.pre-commit-config.yaml

@@ -166,19 +166,24 @@ repos:
       - id: check-subprocess-encoding
         name: Refuse NEW text=True without encoding= (subprocess, cp1252 crash)
         description: |
-          Ratchet gate (#13140, generalising the #12813/#12811 fix): a staged
-          .py file must not contain a subprocess call setting text=True (or
-          universal_newlines=True) without encoding= in the same call. On a
-          cp1252 host that raises UnicodeDecodeError on UTF-8 payloads — the
-          exact class that killed the lane-claim guard in production (#12811).
+          Ratchet gate (#13140, generalising the #12813/#12811 fix, extended
+          to .ipynb by #19475 / tapis #15629): a staged .py file or a staged
+          Python-kernel .ipynb file must not contain a subprocess call setting
+          text=True (or universal_newlines=True) without encoding= in the
+          same call. On a cp1252 host the moment a child writes UTF-8 bytes
+          undefined in cp1252 (0x81/0x8D/0x8F/0x90/0x9D, frequent in the ICT
+          symbols and French prose) the parent crashes with
+          UnicodeDecodeError — the exact class that killed the lane-claim
+          guard in production (#12811) and recurred across 19 notebook
+          cellules of the Lean series during the tapis #15629 sweep.
           Retroactively-clean-by-design: only files the commit touches are
           scanned, so the gate is green on main while the historical sweep
           lands tranche by tranche. No CI workflow counterpart on purpose:
           #13097 CI famine — the check runs at commit time, zero queue slots.
         entry: python scripts/check_subprocess_encoding.py
         language: system
         pass_filenames: true
-        files: '\.py$'
+        files: '\.(py|ipynb)$'
 
       - id: cell-source-parses
         name: '#13326 — refuse un-compilable cell source (markdown-typed-code or syntax error)'

scripts/check_subprocess_encoding.py

@@ -21,17 +21,26 @@
 -- including when ``encoding=`` sits on a separate line of a multiline call.
 
 Usage (two modes):
-    python check_subprocess_encoding.py <file.py> [file2.py ...]
+    python check_subprocess_encoding.py <file.{py,ipynb}> [file2 ...]
         Scan the given files (pre-commit mode: pre-commit passes the staged
-        filenames).
+        filenames). For .ipynb, only code cells whose kernelspec is Python
+        are scanned; .NET Interactive / Lean notebooks are skipped (the
+        subprocess / text=True pair is Python-only).
 
     python check_subprocess_encoding.py --base origin/main
-        Scan every .py file changed between merge-base(REF, HEAD) and HEAD,
-        reading the working tree (in CI the checkout IS the head).
+        Scan every .py and Python-kernel .ipynb file changed between
+        merge-base(REF, HEAD) and HEAD, reading the working tree (in CI
+        the checkout IS the head).
 
 Exit code 1 iff at least one violation is reported. Vendored / external
 subtrees (see EXCLUDE_MARKERS) are out of scope in both modes.
 
+Issue #19475 (extension to notebooks, the gap measured by the tapis #15629):
+the original guard refused NEW .py offenses, but the same defect class
+(UnicodeDecodeError cp1252 on UTF-8 payload) was measured almost
+exclusively in .ipynb cells -- the ratchet must catch the same defect in
+notebook source, not just module-level Python.
+
 Known-good fix forms: ``encoding="utf-8", errors="replace"`` -- or the
 single-quote variant ``encoding='utf-8', errors='replace'`` when the call
 lives inside an f-string expression, where nested double quotes are a
@@ -42,6 +51,7 @@
 
 import argparse
 import io
+import json
 import re
 import subprocess
 import sys
@@ -132,6 +142,62 @@ def scan_source(src: str) -> list[tuple[int, str]]:
     return findings
 
 
+def _is_python_kernel(metadata: dict | None) -> bool:
+    """A notebook with kernel Python (any minor) parses for scan_source. .NET
+    Interactive notebooks are skipped (their 'source' is C#/F#, not Python)."""
+    if not metadata:
+        return False
+    ks = metadata.get("kernelspec", {}) or {}
+    name = (ks.get("name") or "").lower()
+    language = (ks.get("language") or "").lower()
+    return name.startswith("python") or language == "python"
+
+
+def _cell_source(cell: dict) -> str:
+    """Concatenate the source field of a code cell (nbformat: list of str or str)."""
+    src = cell.get("source", "")
+    if isinstance(src, list):
+        return "".join(src)
+    return src or ""
+
+
+def scan_notebook(path: str) -> list[tuple[int, int, str]]:
+    """Return (1-based notebook line, 1-based cell index, snippet) per violation.
+
+    The notebook line is the absolute line offset in the concatenated cell
+    sources (each cell separated by a single '\\n' if it does not already end
+    in one). The cell index lets a reviewer jump straight to the offending
+    cell in the Jupyter UI.
+
+    Non-Python kernels are skipped (return []). Malformed JSON returns []
+    (the guard is best-effort; the cell-source-parses guard has its own
+    coverage of the malformed-notebook class).
+    """
+    findings: list[tuple[int, int, str]] = []
+    try:
+        nb = json.loads(Path(path).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return findings
+    if not _is_python_kernel(nb.get("metadata", {})):
+        return findings
+    cumulative_lines = 0
+    for cell_idx, cell in enumerate(nb.get("cells", []) or []):
+        if cell.get("cell_type") != "code":
+            continue
+        src = _cell_source(cell)
+        if not src:
+            cumulative_lines += 0
+            continue
+        for rel_line, snippet in scan_source(src):
+            findings.append((cumulative_lines + rel_line, cell_idx, snippet))
+        # cells separated by a single newline if not already ending in one
+        line_count = src.count("\n")
+        if not src.endswith("\n"):
+            line_count += 1
+        cumulative_lines += line_count + 1
+    return findings
+
+
 def git_out(*args: str) -> str | None:
     try:
         proc = subprocess.run(
@@ -144,7 +210,7 @@ def git_out(*args: str) -> str | None:
 
 
 def changed_python_files(base: str) -> list[str]:
-    """Files changed between merge-base(base, HEAD) and HEAD (paths, .py only)."""
+    """Files changed between merge-base(base, HEAD) and HEAD (paths, .py + .ipynb)."""
     mb = git_out("merge-base", base, "HEAD")
     if mb is None:
         # No common ancestor (orphan branch): fall back to base tip.
@@ -153,27 +219,39 @@ def changed_python_files(base: str) -> list[str]:
     if out is None:
         return []
     return [l.strip() for l in out.splitlines()
-            if l.strip().endswith(".py") and not excluded(l.strip())]
+            if (l.strip().endswith(".py") or l.strip().endswith(".ipynb"))
+            and not excluded(l.strip())]
 
 
 def main(argv: list[str] | None = None) -> int:
     p = argparse.ArgumentParser(
         description="Refuse NEW subprocess text=True calls without encoding=")
     p.add_argument("files", nargs="*", help="files to scan (pre-commit mode)")
     p.add_argument("--base", default=None, metavar="REF",
-                   help="scan .py files changed since merge-base(REF, HEAD)")
+                   help="scan .py and .ipynb files changed since merge-base(REF, HEAD)")
     args = p.parse_args(argv)
 
     if args.base:
         targets = changed_python_files(args.base)
     else:
-        targets = [f for f in args.files if f.endswith(".py") and not excluded(f)]
+        targets = [f for f in args.files
+                   if (f.endswith(".py") or f.endswith(".ipynb"))
+                   and not excluded(f)]
 
     violations = 0
     for f in targets:
         path = Path(f)
         if not path.is_file():
             continue
+        if f.endswith(".ipynb"):
+            try:
+                for line, cell_idx, snippet in scan_notebook(f):
+                    violations += 1
+                    print(f"{f}:cell[{cell_idx}]:{line}: "
+                          f"text=True without encoding= :: {snippet}")
+            except (OSError, ValueError):
+                continue
+            continue
         try:
             src = path.read_text(encoding="utf-8")
         except (OSError, UnicodeDecodeError):
@@ -189,7 +267,10 @@ def main(argv: list[str] | None = None) -> int:
               "inside f-string expressions).")
         return 1
     if args.base:
-        print(f"subprocess-encoding ratchet: {len(targets)} changed .py file(s), 0 violation(s)")
+        n_py = sum(1 for f in targets if f.endswith(".py"))
+        n_ipynb = sum(1 for f in targets if f.endswith(".ipynb"))
+        print(f"subprocess-encoding ratchet: {n_py} .py + {n_ipynb} .ipynb "
+              f"changed file(s), 0 violation(s)")
     return 0
 
 

scripts/tests/test_check_subprocess_encoding.py

@@ -1,11 +1,15 @@
-"""Tests for check_subprocess_encoding (ratchet gate, #13140).
+"""Tests for check_subprocess_encoding (ratchet gate, #13140, extended to
+.ipynb cells by #19475 / tapis #15629).
 
 Unit tests target the pure scan_source() parser (balanced-paren call spans,
 the text/universal_newlines/encoding discrimination, multiline kwargs); the
 integration tests drive main(argv) directly (per the cli-surface lesson:
 running _run_check or scan_source through a bypassed argv hides CLI bugs).
+The .ipynb extension adds scan_notebook(), which iterates code cells of a
+Python-kernel notebook and reuses scan_source on each cell.
 """
 
+import json
 import sys
 from pathlib import Path
 
@@ -167,4 +171,119 @@ def test_main_base_mode(monkeypatch, capsys):
     # No files on disk with those names -> 0 violations, exit 0, paths filtered.
     assert cse.main(["--base", "origin/main"]) == 0
     out = capsys.readouterr().out
-    assert "1 changed .py file(s)" in out
+    # #19475: 1 .py (scripts/a.py) + 1 .ipynb (notebook.ipynb) -- the .py under
+    # _peters is filtered by the EXCLUDE_MARKERS check.
+    assert "1 .py + 1 .ipynb" in out
+    assert "0 violation(s)" in out
+
+
+# ---------------------------------------------------------------------------
+# .ipynb coverage (Issue #19475, extension du ratchet aux cellules)
+# ---------------------------------------------------------------------------
+
+def _write_notebook(path, cells, kernel="python3", language="python"):
+    """Helper: write a minimal nbformat-4 notebook with the given cells."""
+    nb = {
+        "metadata": {
+            "kernelspec": {"name": kernel, "language": language},
+        },
+        "cells": cells,
+    }
+    path.write_text(json.dumps(nb), encoding="utf-8")
+
+
+def test_scan_notebook_violation_in_first_cell(tmp_path):
+    nb = tmp_path / "victim.ipynb"
+    _write_notebook(nb, [
+        {"cell_type": "code",
+         "source": "import subprocess\nsubprocess.run(['x'], text=True)\n"},
+    ])
+    findings = cse.scan_notebook(str(nb))
+    assert len(findings) == 1
+    abs_line, cell_idx, snippet = findings[0]
+    assert cell_idx == 0
+    assert abs_line == 2  # second line of the cell
+    assert "text=True" in snippet
+
+
+def test_scan_notebook_clean_when_encoding_present(tmp_path):
+    nb = tmp_path / "clean.ipynb"
+    _write_notebook(nb, [
+        {"cell_type": "code",
+         "source": ('import subprocess\nsubprocess.run(["x"], text=True, '
+                    'encoding="utf-8", errors="replace")\n')},
+    ])
+    assert cse.scan_notebook(str(nb)) == []
+
+
+def test_scan_notebook_skips_non_python_kernels(tmp_path):
+    nb = tmp_path / "dotnet.ipynb"
+    _write_notebook(
+        nb,
+        [{"cell_type": "code",
+          "source": "// C# code -- this is .NET Interactive, no subprocess.\n"}],
+        kernel=".net-csharp",
+        language="csharp",
+    )
+    # C# subprocess.run is a non-Python construct, but the kernelspec filter
+    # short-circuits before any source scan.
+    assert cse.scan_notebook(str(nb)) == []
+
+
+def test_scan_notebook_skips_markdown_cells(tmp_path):
+    nb = tmp_path / "mixed.ipynb"
+    _write_notebook(nb, [
+        {"cell_type": "markdown",
+         "source": "Reference to subprocess.run(text=True) is just prose.\n"},
+        {"cell_type": "code", "source": "x = 1\n"},
+    ])
+    assert cse.scan_notebook(str(nb)) == []
+
+
+def test_scan_notebook_line_offset_continues_across_cells(tmp_path):
+    nb = tmp_path / "multi.ipynb"
+    _write_notebook(nb, [
+        {"cell_type": "code", "source": "x = 1\n"},  # 1 line
+        {"cell_type": "code",
+         "source": "import subprocess\nsubprocess.run(['x'], text=True)\n"},  # 2 lines
+    ])
+    findings = cse.scan_notebook(str(nb))
+    assert len(findings) == 1
+    abs_line, cell_idx, _ = findings[0]
+    assert cell_idx == 1
+    # Cell 0 occupies 1 line, +1 separator => cell 1 starts at line 3.
+    # The call line in cell 1 is the second line of that cell => abs 4.
+    assert abs_line == 4
+
+
+def test_scan_notebook_malformed_json_returns_empty(tmp_path):
+    nb = tmp_path / "broken.ipynb"
+    nb.write_text("{ this is not json", encoding="utf-8")
+    assert cse.scan_notebook(str(nb)) == []
+
+
+def test_scan_notebook_list_source_is_joined(tmp_path):
+    """nbformat 4.x stores cell source as a list of strings; the parser
+    must join them before scanning, otherwise line offsets are off."""
+    nb = tmp_path / "list_src.ipynb"
+    _write_notebook(nb, [
+        {"cell_type": "code",
+         "source": ["import subprocess\n", "subprocess.run(['x'], text=True)\n"]},
+    ])
+    findings = cse.scan_notebook(str(nb))
+    assert len(findings) == 1
+    _, cell_idx, snippet = findings[0]
+    assert cell_idx == 0
+    assert "text=True" in snippet
+
+
+def test_main_files_mode_picks_up_ipynb(tmp_path, capsys):
+    nb = tmp_path / "bad.ipynb"
+    _write_notebook(nb, [
+        {"cell_type": "code",
+         "source": "import subprocess\nsubprocess.run(['x'], text=True)\n"},
+    ])
+    assert cse.main([str(nb)]) == 1
+    out = capsys.readouterr().out
+    assert "cell[0]" in out
+    assert "text=True without encoding=" in out