diff --git a/loopx/canary/maintainability_ratchet.py b/loopx/canary/maintainability_ratchet.py index ebad2d40c4..498b4dbe44 100644 --- a/loopx/canary/maintainability_ratchet.py +++ b/loopx/canary/maintainability_ratchet.py @@ -6,10 +6,13 @@ from collections import Counter from importlib.util import resolve_name import json +import os from pathlib import Path import subprocess from typing import Any, Mapping, Sequence +from ..paths import display_path_text + MAINTAINABILITY_REPORT_SCHEMA_VERSION = "control_plane_maintainability_report_v0" MODULE_METRIC_BASELINE_SCHEMA_VERSION = "control_plane_module_metric_baseline_v0" @@ -209,18 +212,24 @@ def tracked_python_paths(repository_root: Path) -> set[Path]: # `-z` frames each pathname on NUL, so a path carrying U+0085/U+2028/U+2029 # stays one record even in a repository that sets `core.quotePath=false`, # where `git ls-files` emits it raw and `str.splitlines()` would tear it. + # The read stays in bytes so the record's own identity survives: a text + # read turns a CR inside a legal name into LF and an undecodable record + # into U+FFFD, and the rebuilt path would not name the tracked file, so + # that module would silently leave the ratchet. completed = subprocess.run( ["git", "ls-files", "-z", "*.py"], cwd=repository_root, check=True, - text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE, ) - return { - repository_root / relative_path - for relative_path in completed.stdout.split("\0") - if relative_path and (repository_root / relative_path).is_file() - } + paths: set[Path] = set() + for record in completed.stdout.split(b"\0"): + if not record: + continue + path = repository_root / os.fsdecode(record) + if path.is_file(): + paths.add(path) + return paths def _dependency_finding( @@ -236,12 +245,14 @@ def _dependency_finding( if imported_name: identity += f":{imported_name}" return { - "id": _finding_id("dependency_debt", identity), + "id": _finding_id("dependency_debt", display_path_text(identity)), "category": "dependency_debt", "rule": rule, - "path": path.relative_to(repository_root).as_posix(), - "source": source, - "target": target, + # A name Git can hold but UTF-8 cannot encode would otherwise write + # bytes no JSON reader can parse; the label keeps the record locatable. + "path": display_path_text(path.relative_to(repository_root).as_posix()), + "source": display_path_text(source), + "target": display_path_text(target), **({"imported_name": imported_name} if imported_name else {}), } @@ -459,10 +470,14 @@ def collect_oversized_decision_functions( identity = f"{module}:{qualified_name}" findings.append( { - "id": _finding_id("oversized_decision_function", identity), + "id": _finding_id( + "oversized_decision_function", display_path_text(identity) + ), "category": "oversized_decision_function", - "path": path.relative_to(repository_root).as_posix(), - "module": module, + "path": display_path_text( + path.relative_to(repository_root).as_posix() + ), + "module": display_path_text(module), "symbol": qualified_name, "line": function.lineno, "metrics": { @@ -615,10 +630,10 @@ def collect_module_metric_findings( continue findings.append( { - "id": _finding_id("module_metric_budget", relative), + "id": _finding_id("module_metric_budget", display_path_text(relative)), "category": "module_metric_budget", - "path": relative, - "module": _module_name(path, package_root), + "path": display_path_text(relative), + "module": display_path_text(_module_name(path, package_root)), "metrics": metrics, "metric_ceilings": ceilings, "regressions": regressions, @@ -692,9 +707,11 @@ def diff_scoped_module_ceiling_violations( if crossed_inherited_ceiling and head_ceiling < head_lines: violations.append( { - "id": _finding_id("module_metric_budget", relative), + "id": _finding_id( + "module_metric_budget", display_path_text(relative) + ), "category": "module_metric_budget", - "path": relative, + "path": display_path_text(relative), "base_lines": 0 if base_lines is None else len(base_lines.splitlines()), "base_ceiling": base_ceiling, "head_lines": head_lines, @@ -825,7 +842,7 @@ def build_control_plane_maintainability_report( ) return { "schema_version": MAINTAINABILITY_REPORT_SCHEMA_VERSION, - "repository_root": str(repository_root), + "repository_root": display_path_text(repository_root), "policy": { "decision_statement_limit": DECISION_STATEMENT_LIMIT, "decision_point_limit": DECISION_POINT_LIMIT, diff --git a/loopx/contract.py b/loopx/contract.py index 13c21679c3..48d86ec5c3 100644 --- a/loopx/contract.py +++ b/loopx/contract.py @@ -42,7 +42,13 @@ build_run_history_audit, collect_history, ) -from .paths import DEFAULT_RUNTIME_ROOT, LEGACY_RUNTIME_ROOT, rel_or_abs, resolve_runtime_root +from .paths import ( + DEFAULT_RUNTIME_ROOT, + LEGACY_RUNTIME_ROOT, + display_path_text, + rel_or_abs, + resolve_runtime_root, +) from .public_safe_text import PROVIDER_TOKEN_SHAPE_PATTERN_SOURCE from .registry import inspect_registry, inspect_registry_boundary, registry_goals, resolve_state_file from .state_projection import state_projection_gap_warning @@ -372,7 +378,13 @@ class _GitAnswer: stdout: str -def _retire_owned_probe(process: subprocess.Popen[str]) -> None: +@dataclass(frozen=True) +class _GitBytesAnswer: + returncode: int + stdout: bytes + + +def _retire_owned_probe(process: subprocess.Popen[str] | subprocess.Popen[bytes]) -> None: """Stop this probe's own process group, then drop its pipes unread. A descendant that inherited the pipes keeps them open after the leader @@ -429,6 +441,36 @@ def _git_query(*arguments: str) -> _GitAnswer | None: return _GitAnswer(returncode=process.returncode, stdout=stdout) +def _git_query_bytes(*arguments: str) -> _GitBytesAnswer | None: + """One bounded local Git read that keeps the raw output bytes. + + Pathname listings carry their record identity in these bytes. The text read + above translates a CR inside a legal name into LF and replaces an + undecodable record with U+FFFD, so a pathname rebuilt from it can name a + file that does not exist and the tracked file would leave the scan. + """ + + argv = ["git", *arguments] + try: + process = subprocess.Popen( + argv, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=os.name == "posix", + ) + except OSError: + return None + try: + stdout, _stderr = process.communicate(timeout=GIT_PROBE_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + _retire_owned_probe(process) + return None + except BaseException: + _retire_owned_probe(process) + raise + return _GitBytesAnswer(returncode=process.returncode, stdout=stdout) + + def _git_worktree_root(path: Path) -> Path | None: """Return the worktree root, or None when Git says this is not one. @@ -1071,17 +1113,17 @@ def _tracked_scan_files(scan_root: Path) -> list[Path]: if not rel_root: rel_root = "." - tracked = _git_query("-C", str(repo_root), "ls-files", "-z", "--", rel_root) + tracked = _git_query_bytes("-C", str(repo_root), "ls-files", "-z", "--", rel_root) if tracked is None: raise GitProbeUnavailable("git did not answer the tracked-file enumeration") if tracked.returncode != 0: return [] files: list[Path] = [] - for rel_path in tracked.stdout.split("\0"): - if not rel_path: + for record in tracked.stdout.split(b"\0"): + if not record: continue - path = repo_root / rel_path + path = repo_root / os.fsdecode(record) if path.suffix not in DEFAULT_SCAN_SUFFIXES and not path.is_symlink(): continue path = path.resolve() @@ -1119,6 +1161,19 @@ def _record_unreadable_directory( unreadable_directories.append((directory, _directory_reason(error))) +def _display_scan_path(path: Path, root: Path) -> str: + """Label a scanned path so even an undecodable name reaches the report. + + ``rel_or_abs`` returns ``os.fsdecode``'s text, which keeps a name that is + not valid UTF-8 as lone surrogates. A report that carried those verbatim + wrote bytes no strict UTF-8 reader accepts, and the JSON consumer could not + read the finding that named the file. The escape keeps the record's exact + bytes readable and distinct from every other name. + """ + + return display_path_text(rel_or_abs(path, root)) + + def _unreadable_directory_label( directory: Path, scan_roots: Mapping[Path, Path] ) -> str: @@ -1130,9 +1185,9 @@ def _unreadable_directory_label( """ containing = [root for root in scan_roots if directory.is_relative_to(root)] if not containing: - return str(directory) + return display_path_text(directory) root = min(containing, key=lambda candidate: (len(candidate.parts), str(candidate))) - return str(directory.relative_to(scan_roots[root])) or "." + return display_path_text(directory.relative_to(scan_roots[root])) or "." def _unreadable_directory_labels( @@ -1266,7 +1321,7 @@ def record_hit(hit: str, *, blocking: bool = True) -> None: if not resolved_scan_root.exists(): # A scan root that is not there contributes no files, so a typo in # the caller's path would otherwise report a clean empty boundary. - missing_scan_roots.append(str(scan_root)) + missing_scan_roots.append(display_path_text(scan_root)) continue display_root = resolved_scan_root.parent if resolved_scan_root.is_file() else resolved_scan_root # `file_roots` renders a file relative to the root that asked for it and @@ -1291,7 +1346,10 @@ def record_hit(hit: str, *, blocking: bool = True) -> None: if _is_local_private_state_path(resolved_scan_root, classification_root): # Explicit outgoing selection cannot turn an ignored private file # or a pruned private directory into a successful empty scan. - record_hit(f"{rel_or_abs(resolved_scan_root, classification_root)}: local_private_state") + record_hit( + f"{_display_scan_path(resolved_scan_root, classification_root)}: " + "local_private_state" + ) continue for file_path in iter_scan_files( resolved_scan_root, @@ -1337,11 +1395,11 @@ def public_files() -> Iterator[Path]: skipped_private_state_files.append(rel_or_abs(path, root)) if git.get("inside_worktree") and not git.get("ignored"): private_state_git_warnings.append( - f"{rel_or_abs(path, root)}: private state should be gitignored" + f"{_display_scan_path(path, root)}: private state should be gitignored" ) continue private_file_git[path] = git - record_hit(f"{rel_or_abs(path, root)}: local_private_state") + record_hit(f"{_display_scan_path(path, root)}: local_private_state") yield path for read in iter_utf8_file_reads(public_files()): @@ -1353,7 +1411,7 @@ def public_files() -> Iterator[Path]: read.error.strerror or type(read.error).__name__ if isinstance(read.error, OSError) else "invalid UTF-8" ) - unreadable_files.append(f"{rel_or_abs(path, root)}: {detail}") + unreadable_files.append(f"{_display_scan_path(path, root)}: {detail}") continue text = read.text assert text is not None @@ -1376,7 +1434,7 @@ def public_files() -> Iterator[Path]: ): display_package = package_path or "" record_hit( - f"{rel_or_abs(path, root)}: " + f"{_display_scan_path(path, root)}: " "non_public_package_registry " f"({display_package})" ) @@ -1397,7 +1455,7 @@ def public_files() -> Iterator[Path]: if name == "private_doc_url": scan_line = _PUBLIC_LARK_DEVELOPER_CONSOLE_HOST.sub("", scan_line) if rule.pattern.search(scan_line): - hit = f"{rel_or_abs(path, root)}:{line_no}: {name}" + hit = f"{_display_scan_path(path, root)}:{line_no}: {name}" if name == "credential" and _credential_hits_are_all_references(line): credential_reference_hits.append(hit) continue @@ -1418,10 +1476,12 @@ def public_files() -> Iterator[Path]: or unreadable_directories or unavailable_git_probes ), - "scan_roots": [str(path) for path in scan_roots], + "scan_roots": [display_path_text(path) for path in scan_roots], "files": len(files), "scanned_files": len(files) - len(skipped_private_state_files), - "skipped_private_state_files": skipped_private_state_files, + "skipped_private_state_files": [ + display_path_text(item) for item in skipped_private_state_files + ], "credential_reference_hits": credential_reference_hits, "unreadable_files": unreadable_files, # Identity is the absolute directory, so two roots that each hold an @@ -1432,7 +1492,9 @@ def public_files() -> Iterator[Path]: "missing_scan_roots": missing_scan_roots, # A Git question that was not answered leaves the boundary unknown: # this list keeps a partial answer from being reported as clean. - "unavailable_git_probes": sorted(set(unavailable_git_probes)), + "unavailable_git_probes": sorted( + {display_path_text(item) for item in unavailable_git_probes} + ), "allowed_hits": allowed_hits, "private_state_git_warnings": private_state_git_warnings, "policy": policy, @@ -1691,7 +1753,7 @@ def add_global_error(code: str, message: str) -> None: "ok": not errors, "registry": str(registry_path), "runtime_root": str(runtime_root), - "scan_roots": [str(path) for path in scan_roots], + "scan_roots": [display_path_text(path) for path in scan_roots], "summary": { "errors": len(errors), "warnings": len(warnings), diff --git a/loopx/paths.py b/loopx/paths.py index 9a5a41cf5a..4b48657933 100644 --- a/loopx/paths.py +++ b/loopx/paths.py @@ -377,3 +377,26 @@ def rel_or_abs(path: Path, root: Path) -> str: return str(path.relative_to(root)) except ValueError: return str(path) + + +GIT_BYTE_PATH_PREFIX = "git-bytes:" + + +def display_path_text(path: Path | str) -> str: + """Render a pathname so a UTF-8 reader can always parse the report. + + ``os.fsdecode`` keeps the bytes of a name that is not valid UTF-8 as lone + surrogates (PEP 383). A report that printed such a name verbatim writes + bytes no strict UTF-8 reader accepts, and the JSON consumer then cannot + read the finding that named the file. A decodable name is returned + unchanged; an undecodable one keeps its exact bytes in a ``git-bytes:`` + labelled ``repr``, the marker the issue-fix artifact already uses, so two + distinct byte records never collapse onto one label. + """ + + text = str(path) + try: + text.encode("utf-8") + except UnicodeEncodeError: + return f"{GIT_BYTE_PATH_PREFIX}{os.fsencode(text)!r}" + return text diff --git a/tests/canary/test_git_path_display.py b/tests/canary/test_git_path_display.py new file mode 100644 index 0000000000..94ba446bb2 --- /dev/null +++ b/tests/canary/test_git_path_display.py @@ -0,0 +1,246 @@ +"""A name Git can hold must never make a report's JSON text unparseable.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys + +import pytest + +from loopx.contract import scan_public_boundary +from loopx.canary.maintainability_ratchet import _dependency_finding +from loopx.control_plane.runtime.file_reads import Utf8FileRead +from loopx.paths import display_path_text + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] +RATCHET_COMMAND = ( + REPOSITORY_ROOT / "examples" / "control_plane" + / "control-plane-maintainability-ratchet-smoke.py" +) + + +def _os_path(path: Path, name: bytes) -> str: + return os.path.join(os.fsencode(path), name) + + +def _supports_raw_name(directory: Path) -> bool: + """Report whether this filesystem can hold a name that is not UTF-8.""" + + probe = _os_path(directory, b"probe\xff") + try: + os.close(os.open(probe, os.O_CREAT | os.O_WRONLY, 0o600)) + except OSError: + return False + os.unlink(probe) + return True + + +def _write_raw_name(path: str, text: str) -> None: + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + + +def _git(repo: Path, *args: str) -> None: + subprocess.run( + ["git", "-C", str(repo), *args], + check=True, + capture_output=True, + text=True, + encoding="utf-8", + errors="backslashreplace", + ) + + +def _init_repo(repo: Path) -> None: + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.name", "LoopX Test") + _git(repo, "config", "user.email", "loopx@example.invalid") + + +def test_display_path_text_marks_only_the_undecodable_record() -> None: + assert display_path_text(Path("loopx/plain.py")) == "loopx/plain.py" + assert display_path_text("docs/中文.md") == "docs/中文.md" + + escaped = display_path_text("loopx/bad\udcff.py") + + # The label keeps the record's bytes, so the reader can still tell which + # file the finding names, and JSON written from it stays valid UTF-8. + assert escaped == "git-bytes:" + repr(b"loopx/bad\xff.py") + assert json.dumps({"path": escaped}, ensure_ascii=False).encode("utf-8") + # Two distinct bytes records never collapse onto one label. + assert escaped != display_path_text("loopx/bad\udcfe.py") + # A decodable name that merely looks like the marker keeps its own label. + assert display_path_text("git-bytes:b'x'") == "git-bytes:b'x'" + + +@pytest.mark.skipif( + os.name != "posix", reason="non-UTF-8 Git paths are POSIX-only" +) +def test_maintainability_json_keeps_every_tracked_name( + tmp_path: Path, +) -> None: + """Plain, CR and undecodable module names all reach one parseable report.""" + + repo = tmp_path / "repo" + if not _supports_raw_name(tmp_path): + pytest.skip("this filesystem rejects non-UTF-8 file names") + _init_repo(repo) + control_plane = repo / "loopx" / "control_plane" + control_plane.mkdir(parents=True) + (repo / "loopx" / "quota.py").write_text("", encoding="utf-8") + (repo / "loopx" / "status.py").write_text("", encoding="utf-8") + padding = "\n".join(["# padding"] * 2001) + for name in ("plain.py", "cr\rname.py"): + _write_raw_name(str(control_plane / name), padding) + _write_raw_name(_os_path(control_plane, b"hot\xffspot.py"), padding) + _git(repo, "add", "--", "loopx") + _git(repo, "commit", "-m", "oversized modules, one with an undecodable name") + + command = repo / "examples" / "control_plane" / RATCHET_COMMAND.name + command.parent.mkdir(parents=True) + command.write_bytes(RATCHET_COMMAND.read_bytes()) + result = subprocess.run( + [sys.executable, str(command), "--format", "json"], + cwd=repo, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env={**os.environ, "PYTHONPATH": str(REPOSITORY_ROOT)}, + check=False, + timeout=120, + ) + + assert result.returncode == 1, result.stderr + # The command's own writer is `json.dumps(..., ensure_ascii=False)`; a + # verbatim surrogate name would make this strict decode fail. + payload = json.loads(result.stdout.decode("utf-8")) + assert payload["ok"] is False + findings = [ + item + for item in payload["findings"] + if item["category"] == "module_metric_budget" + ] + paths = {item["path"] for item in findings} + assert paths == { + "loopx/control_plane/plain.py", + "loopx/control_plane/cr\rname.py", + "git-bytes:" + repr(b"loopx/control_plane/hot\xffspot.py"), + } + escaped = next(item for item in findings if item["path"].startswith("git-bytes:")) + assert escaped["id"].startswith("module_metric_budget:git-bytes:") + assert escaped["module"].startswith("git-bytes:") + + +@pytest.mark.skipif( + os.name != "posix", reason="non-UTF-8 Git paths are POSIX-only" +) +def test_public_boundary_json_keeps_every_tracked_name(tmp_path: Path) -> None: + """The check command's JSON names plain, CR and undecodable findings.""" + + repo = tmp_path / "repo" + if not _supports_raw_name(tmp_path): + pytest.skip("this filesystem rejects non-UTF-8 file names") + _init_repo(repo) + outgoing = repo / "outgoing" + outgoing.mkdir() + _write_raw_name(str(outgoing / "plain.md"), "ticket t-" + "20260828123456-example\n") + _write_raw_name( + str(outgoing / "cr\rname.md"), "\nticket t-" + "20260828123456-example\n" + ) + _write_raw_name( + _os_path(outgoing, b"leak\xff.md"), + "\n\nticket t-" + "20260828123456-example\n", + ) + _git(repo, "add", "--", "outgoing") + _git(repo, "commit", "-m", "leaking files, one with an undecodable name") + + result = subprocess.run( + [ + sys.executable, + "-m", + "loopx.cli", + "--format", + "json", + "--registry", + str(tmp_path / "registry.json"), + "--runtime-root", + str(tmp_path / "runtime"), + "check", + "--scan-root", + "outgoing", + ], + cwd=repo, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env={**os.environ, "PYTHONPATH": str(REPOSITORY_ROOT)}, + check=False, + timeout=120, + ) + + assert result.returncode == 1, result.stderr + payload = json.loads(result.stdout.decode("utf-8")) + assert payload["ok"] is False + messages = "\n".join(str(item) for item in payload["errors"]) + expected = { + "plain.md:1: internal_task_id", + "cr\rname.md:2: internal_task_id", + "git-bytes:" + repr(b"leak\xff.md") + ":3: internal_task_id", + } + for label in expected: + assert label in messages, messages + + # The library entry the command wraps keeps the same projection. + boundary = scan_public_boundary([outgoing], registry={}) + assert boundary["ok"] is False + assert set(boundary["hits"]) == expected + assert json.dumps(boundary, ensure_ascii=False).encode("utf-8") + + +def test_dependency_finding_escapes_an_undecodable_source( + tmp_path: Path, +) -> None: + """The ratchet's own projection keeps a record it cannot spell readable.""" + + path = tmp_path / "loopx" / "control_plane" / "bad\udcff.py" + + finding = _dependency_finding( + source="loopx.control_plane.bad\udcff", + target="loopx.cli", + path=path, + repository_root=tmp_path, + rule="control_plane_outward_dependency", + ) + + assert finding["path"] == "git-bytes:" + repr(b"loopx/control_plane/bad\xff.py") + assert finding["source"].startswith("git-bytes:") + assert finding["id"].startswith("dependency_debt:git-bytes:") + assert json.dumps(finding, ensure_ascii=False).encode("utf-8") + + +def test_boundary_report_escapes_an_undecodable_scanned_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The scan's report names a file whose bytes UTF-8 cannot encode.""" + + root = tmp_path / "outgoing" + root.mkdir() + raw = root / "leak\udcff.md" + monkeypatch.setattr( + "loopx.contract.iter_scan_files", lambda *args, **kwargs: [raw] + ) + monkeypatch.setattr( + "loopx.contract.iter_utf8_file_reads", + lambda paths: iter([Utf8FileRead(raw, "ticket t-20260828123456-example", None)]), + ) + + boundary = scan_public_boundary([root], registry={}) + + expected = "git-bytes:" + repr(b"leak\xff.md") + ":1: internal_task_id" + assert boundary["ok"] is False + assert boundary["hits"] == [expected] + assert boundary["blocking_hits"] == [expected] + assert json.dumps(boundary, ensure_ascii=False).encode("utf-8") diff --git a/tests/canary/test_git_path_framing.py b/tests/canary/test_git_path_framing.py index 061923c766..c762702e10 100644 --- a/tests/canary/test_git_path_framing.py +++ b/tests/canary/test_git_path_framing.py @@ -259,3 +259,36 @@ def test_premerge_keeps_explicit_changed_file_normalization(tmp_path: Path) -> N assert response.returncode == 0, response.stderr payload = json.loads(response.stdout) assert payload["changed_files"] == ["docs/notes.py"] + + +def _repo_with_carriage_return_python_path(tmp_path: Path) -> tuple[Path, str]: + """A repository whose tracked Python module name carries a CR. + + A CR is a legal POSIX filename byte and Git frames it inside one NUL + record. A text-mode read applies universal-newline translation to the same + bytes, so ``name\r.py`` arrives as two lines and the rebuilt path names no + file at all. + """ + + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.name", "LoopX Test") + _git(repo, "config", "user.email", "loopx@example.invalid") + (repo / "base.py").write_text("x = 1\n", encoding="utf-8") + _git(repo, "add", "base.py") + _git(repo, "commit", "-m", "base") + name = "carriage\rreturn.py" + (repo / name).write_text("y = 2\n", encoding="utf-8") + _git(repo, "add", name) + _git(repo, "commit", "-m", "add a pathname with a CR") + return repo, name + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX allows CR in a filename") +def test_tracked_python_paths_keeps_a_carriage_return_name(tmp_path: Path) -> None: + repo, name = _repo_with_carriage_return_python_path(tmp_path) + + paths = tracked_python_paths(repo) + + assert repo / name in paths diff --git a/tests/test_contract_public_boundary_prefilter.py b/tests/test_contract_public_boundary_prefilter.py index 02c3e9ec4b..0605fe2fe6 100644 --- a/tests/test_contract_public_boundary_prefilter.py +++ b/tests/test_contract_public_boundary_prefilter.py @@ -1,6 +1,7 @@ from __future__ import annotations from itertools import product +import os from pathlib import Path import subprocess from typing import Any @@ -297,3 +298,28 @@ def test_windows_paths_outside_the_private_roots_stay_public(tmp_path: Path) -> (tmp_path / "sample.md").write_text(line + "\n", encoding="utf-8") assert contract.scan_public_boundary([tmp_path])["ok"] is True + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX allows CR in a filename") +def test_pruned_scan_root_keeps_a_carriage_return_tracked_name(tmp_path: Path) -> None: + """A CR inside a tracked name survives the Git-side enumeration. + + Pruned roots such as a virtualenv are enumerated from Git alone, so a + pathname the text-mode read tore apart would leave the public-boundary + scan without any other way to notice the file. + """ + + repo = tmp_path / "repo" + scan_root = repo / "venv" + scan_root.mkdir(parents=True) + (scan_root / "pyvenv.cfg").write_text("home = /usr/bin\n", encoding="utf-8") + name = "carriage\rreturn.py" + (scan_root / name).write_text("x = 1\n", encoding="utf-8") + subprocess.run(["git", "init", "-q", str(repo)], check=True) + subprocess.run( + ["git", "-C", str(repo), "add", "-f", "--", "venv/pyvenv.cfg", f"venv/{name}"], + check=True, + ) + + assert contract._tracked_scan_files(scan_root) == [scan_root / name] + assert contract.iter_scan_files(scan_root) == [scan_root / name]