feat: Windows portability - path separators, LF line endings, UTF-8 decoding and output, msvcrt lock fallback (#152)
Files changed: - .gitattributes - CHANGES.md - VERSION - raw/CONTRACT.md - tools/README.md - tools/chemenu/commands/_util.py - tools/chemenu/commands/dist_cmd.py - tools/chemenu/commands/docs_verify.py - tools/chemenu/commands/doctor.py - tools/chemenu/commands/eval_cmd.py - tools/chemenu/commands/git_publish.py - tools/chemenu/commands/index_build.py - tools/chemenu/commands/lint.py - tools/chemenu/commands/log_append.py - tools/chemenu/commands/migrate_cmd.py - tools/chemenu/commands/provenance_cmd.py - tools/chemenu/commands/raw_cmd.py - tools/chemenu/commands/run_budget.py - tools/chemenu/commands/upstream_cmd.py - tools/chemenu/commands/version_cmd.py - tools/chemenu/commands/work_cmd.py - tools/chemenu/config.py - tools/chemenu/corpus_cache.py - tools/chemenu/filelock.py - tools/chemenu/frontmatter_io.py - tools/chemenu/kb_scan.py - tools/chemenu/kb_state.py - tools/chemenu/lint_core.py - tools/chemenu/prerequisites.py - tools/chemenu/provenance.py - tools/chemenu/search/base.py - tools/chemenu/search/ripgrep.py - tools/chemenu/telemetry/writer.py - tools/chemenu/tests/test_dist_cmd.py - tools/chemenu/tests/test_portability.py - tools/chemenu/tests/test_search.py - tools/chemenu/tests/test_trace_ingest.py - tools/chemenu/type_resolver.py - tools/chemenu/upload.py - tools/chemenu/version.py - tools/run_wikitool.py - tools/trace_ingest.py
This commit is contained in:
torben committed
1 parent
9d06050338
commit
d0f08d1fba
42 files changed
+780
-89
No files matched your search
@@ -41,6 +41,7 @@ def repo(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
(root / "EVALS.md").write_text("# EVALS\n", encoding="utf-8")
|
||||
(root / "CLAUDE.md").write_text("# CLAUDE\n\n@AGENTS.md\n", encoding="utf-8")
|
||||
(root / ".gitignore").write_text("*.pyc\n", encoding="utf-8")
|
||||
(root / ".gitattributes").write_text("* text=auto eol=lf\n", encoding="utf-8")
|
||||
(root / "VERSION").write_text("0.3.1\n", encoding="utf-8")
|
||||
|
||||
for name in config.LICENSE_FILES:
|
||||
@@ -291,6 +292,13 @@ def test_plan_ships_the_environment_template_but_not_the_filled_file(repo):
|
||||
assert config.ENVIRONMENT_FILE not in plan
|
||||
|
||||
|
||||
def test_plan_ships_the_line_ending_rule(repo):
|
||||
"""An instance cloned onto Windows with `core.autocrlf=true` needs it as much
|
||||
as this repository does: without it the sh launcher gets CRLF (Gitea #152)."""
|
||||
plan = dist_cmd.build_plan()
|
||||
assert "eol=lf" in plan[".gitattributes"].content
|
||||
|
||||
|
||||
def test_plan_ships_the_claude_harness_shim(repo):
|
||||
"""Claude Code loads `CLAUDE.md` and not `AGENTS.md`, so a distributed
|
||||
instance running that harness would start every session without the
|
||||
|
||||
@@ -0,0 +1,423 @@
|
||||
"""Windows portability, held on a Linux CI that never runs Windows (Gitea #152).
|
||||
|
||||
Most of what breaks on Windows is code that is right on POSIX by accident: a path
|
||||
stringified with the native separator, a text file written in the native line ending,
|
||||
a pipe decoded in the locale's code page, a lock imported from a module Windows does
|
||||
not have. None of it fails here, so nothing here would notice it coming back.
|
||||
|
||||
The guards below therefore read the source rather than run it. Each guard comes with
|
||||
a test that feeds it the defect it exists for, because a guard that silently matches
|
||||
nothing passes for the wrong reason. The remaining tests simulate the Windows side of
|
||||
what can be simulated: a missing `fcntl`, a `cp1252` pipe.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import types
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from chemenu import filelock
|
||||
|
||||
TOOLS_DIR = Path(__file__).resolve().parents[2]
|
||||
PACKAGE_DIR = TOOLS_DIR / "chemenu"
|
||||
|
||||
# Receivers whose `open` is not a text-file open: `os.open` takes flags, and the
|
||||
# archive modules open binary members.
|
||||
NOT_TEXT_OPEN = {"os", "tarfile", "zipfile", "gzip"}
|
||||
|
||||
# The one module allowed to import the platform lock modules.
|
||||
LOCK_MODULE = PACKAGE_DIR / "filelock.py"
|
||||
|
||||
|
||||
def _source_files() -> list[Path]:
|
||||
"""Every shipped Python file: `tools/chemenu` without its tests, and the scripts
|
||||
beside it in `tools/` - `trace_ingest.py` among them, which a harness runs on
|
||||
every tool call."""
|
||||
package = (p for p in PACKAGE_DIR.rglob("*.py") if "tests" not in p.relative_to(PACKAGE_DIR).parts)
|
||||
return sorted([*package, *TOOLS_DIR.glob("*.py")])
|
||||
|
||||
|
||||
def _call_name(call: ast.Call) -> tuple[str | None, str | None]:
|
||||
"""(function name, receiver name) - `('open', 'os')` for `os.open(...)`."""
|
||||
func = call.func
|
||||
if isinstance(func, ast.Name):
|
||||
return func.id, None
|
||||
if isinstance(func, ast.Attribute):
|
||||
receiver = func.value.id if isinstance(func.value, ast.Name) else None
|
||||
return func.attr, receiver
|
||||
return None, None
|
||||
|
||||
|
||||
def _keywords(call: ast.Call) -> set[str | None]:
|
||||
return {keyword.arg for keyword in call.keywords}
|
||||
|
||||
|
||||
def _open_mode(call: ast.Call, receiver: str | None) -> ast.expr | None:
|
||||
"""The mode argument of an `open`: second for `open`/`io.open`, first for `Path.open`."""
|
||||
for keyword in call.keywords:
|
||||
if keyword.arg == "mode":
|
||||
return keyword.value
|
||||
index = 1 if isinstance(call.func, ast.Name) or receiver == "io" else 0
|
||||
return call.args[index] if len(call.args) > index else None
|
||||
|
||||
|
||||
# --- the four detectors, each a function of source text --------------------------
|
||||
|
||||
|
||||
def stringified_relative_paths(source: str) -> list[int]:
|
||||
"""Lines that turn a `relative_to` result into a string with the native separator.
|
||||
|
||||
`str(p.relative_to(root))` and `f"{p.relative_to(root)}"` both give `kb\\x.md` on
|
||||
Windows; `.as_posix()` is the portable spelling.
|
||||
"""
|
||||
def is_relative_to(node: ast.AST) -> bool:
|
||||
return (
|
||||
isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Attribute)
|
||||
and node.func.attr == "relative_to"
|
||||
)
|
||||
|
||||
lines = []
|
||||
for node in ast.walk(ast.parse(source)):
|
||||
if isinstance(node, ast.Call) and _call_name(node) == ("str", None):
|
||||
if node.args and is_relative_to(node.args[0]):
|
||||
lines.append(node.lineno)
|
||||
if isinstance(node, ast.FormattedValue) and is_relative_to(node.value):
|
||||
lines.append(node.value.lineno)
|
||||
return sorted(lines)
|
||||
|
||||
|
||||
def unportable_file_io(source: str) -> list[tuple[int, str]]:
|
||||
"""Text file access that depends on the platform: a missing `encoding=` on any
|
||||
text open, read or write, and a missing `newline=` on a write.
|
||||
|
||||
Without `encoding=`, Windows reads and writes in the locale's code page; without
|
||||
`newline=`, a write produces CRLF there.
|
||||
"""
|
||||
findings = []
|
||||
for node in ast.walk(ast.parse(source)):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
name, receiver = _call_name(node)
|
||||
keywords = _keywords(node)
|
||||
method = isinstance(node.func, ast.Attribute)
|
||||
if name == "read_text" and method:
|
||||
writes = False
|
||||
elif name == "write_text" and method:
|
||||
writes = True
|
||||
elif name == "open" and receiver not in NOT_TEXT_OPEN:
|
||||
mode = _open_mode(node, receiver)
|
||||
if mode is None:
|
||||
writes = False
|
||||
elif isinstance(mode, ast.Constant) and isinstance(mode.value, str):
|
||||
if "b" in mode.value:
|
||||
continue
|
||||
writes = bool(set(mode.value) & set("wax+"))
|
||||
else:
|
||||
writes = True # a mode this scan cannot read is treated as a write
|
||||
else:
|
||||
continue
|
||||
if "encoding" not in keywords:
|
||||
findings.append((node.lineno, f"{name} without encoding="))
|
||||
if writes and "newline" not in keywords:
|
||||
findings.append((node.lineno, f"{name} writes without newline="))
|
||||
return sorted(findings)
|
||||
|
||||
|
||||
def undecoded_subprocess_text(source: str) -> list[int]:
|
||||
"""Lines where a subprocess call asks for text but not for an encoding.
|
||||
|
||||
`text=True` alone decodes with the locale's code page - cp1252 on Windows - and
|
||||
`git` and `rg` write UTF-8. Only calls spelled `subprocess.<name>(...)` are read:
|
||||
`text=` is an ordinary parameter name elsewhere, and every call site in the
|
||||
package uses that spelling.
|
||||
"""
|
||||
lines = []
|
||||
for node in ast.walk(ast.parse(source)):
|
||||
if not isinstance(node, ast.Call) or _call_name(node)[1] != "subprocess":
|
||||
continue
|
||||
for keyword in node.keywords:
|
||||
if keyword.arg in ("text", "universal_newlines") and not (
|
||||
isinstance(keyword.value, ast.Constant) and keyword.value.value is False
|
||||
):
|
||||
if "encoding" not in _keywords(node):
|
||||
lines.append(node.lineno)
|
||||
return sorted(lines)
|
||||
|
||||
|
||||
def platform_lock_imports(source: str) -> list[int]:
|
||||
"""Lines importing `fcntl` or `msvcrt` - which only `chemenu.filelock` may do."""
|
||||
lines = []
|
||||
for node in ast.walk(ast.parse(source)):
|
||||
if isinstance(node, ast.Import) and any(a.name in ("fcntl", "msvcrt") for a in node.names):
|
||||
lines.append(node.lineno)
|
||||
if isinstance(node, ast.ImportFrom) and node.module in ("fcntl", "msvcrt"):
|
||||
lines.append(node.lineno)
|
||||
return sorted(lines)
|
||||
|
||||
|
||||
def _scan(detector, *, skip: Path | None = None) -> list[str]:
|
||||
findings = []
|
||||
for path in _source_files():
|
||||
if path == skip:
|
||||
continue
|
||||
relative = path.relative_to(TOOLS_DIR).as_posix()
|
||||
for finding in detector(path.read_text(encoding="utf-8")):
|
||||
findings.append(f"{relative}: {finding}")
|
||||
return findings
|
||||
|
||||
|
||||
# --- the guards ------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_guards_scan_the_whole_package():
|
||||
files = _source_files()
|
||||
assert len(files) > 60
|
||||
assert PACKAGE_DIR / "search" / "ripgrep.py" in files
|
||||
assert TOOLS_DIR / "trace_ingest.py" in files
|
||||
assert not any("tests" in p.relative_to(TOOLS_DIR).parts for p in files)
|
||||
|
||||
|
||||
def test_no_relative_path_is_stringified_with_the_native_separator():
|
||||
assert _scan(stringified_relative_paths) == []
|
||||
|
||||
|
||||
def test_every_text_file_access_names_its_encoding_and_every_write_its_newline():
|
||||
assert _scan(unportable_file_io) == []
|
||||
|
||||
|
||||
def test_every_text_subprocess_names_its_encoding():
|
||||
assert _scan(undecoded_subprocess_text) == []
|
||||
|
||||
|
||||
def test_subprocess_is_only_called_through_its_module_name():
|
||||
"""The spelling `undecoded_subprocess_text` relies on - a `from subprocess
|
||||
import run` would take its call sites out of the guard's sight."""
|
||||
def from_imports(source: str) -> list[int]:
|
||||
return [
|
||||
node.lineno for node in ast.walk(ast.parse(source))
|
||||
if isinstance(node, ast.ImportFrom) and node.module == "subprocess"
|
||||
]
|
||||
|
||||
assert _scan(from_imports) == []
|
||||
|
||||
|
||||
def test_only_the_lock_module_imports_a_platform_lock():
|
||||
assert _scan(platform_lock_imports, skip=LOCK_MODULE) == []
|
||||
assert platform_lock_imports(LOCK_MODULE.read_text(encoding="utf-8")) != []
|
||||
|
||||
|
||||
# --- each detector catches the defect it exists for ------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", [
|
||||
"x = str(path.relative_to(root))",
|
||||
"x = str(Path(p).relative_to(config.ROOT))",
|
||||
'x = f"{path.relative_to(root)} is missing"',
|
||||
])
|
||||
def test_a_stringified_relative_path_is_caught(source):
|
||||
assert stringified_relative_paths(source) == [1]
|
||||
|
||||
|
||||
def test_a_posix_relative_path_passes():
|
||||
assert stringified_relative_paths("x = path.relative_to(root).as_posix()") == []
|
||||
assert stringified_relative_paths("x = path.relative_to(root).parts[0]") == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", [
|
||||
'path.write_text(text, encoding="utf-8")',
|
||||
'open(target, "a", encoding="utf-8")',
|
||||
'path.open("w", encoding="utf-8")',
|
||||
'open(target, mode="x", encoding="utf-8")',
|
||||
'open(target, mode, encoding="utf-8")',
|
||||
])
|
||||
def test_a_write_without_newline_is_caught(source):
|
||||
assert unportable_file_io(source) != []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", [
|
||||
"path.read_text()",
|
||||
"open(target)",
|
||||
'path.write_text(text, newline="\\n")',
|
||||
])
|
||||
def test_a_text_access_without_encoding_is_caught(source):
|
||||
assert unportable_file_io(source) != []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", [
|
||||
'write_text(path, text)', # a module's own helper of that name, not Path's
|
||||
'path.write_text(text, encoding="utf-8", newline="\\n")',
|
||||
'open(target, "a", encoding="utf-8", newline="\\n")',
|
||||
'path.read_text(encoding="utf-8")',
|
||||
'open(target, "rb")',
|
||||
'dest.open("wb")',
|
||||
"os.open(path, flags)",
|
||||
'tarfile.open(path, "w:gz")',
|
||||
])
|
||||
def test_portable_file_access_passes(source):
|
||||
assert unportable_file_io(source) == []
|
||||
|
||||
|
||||
def test_text_without_encoding_is_caught_and_with_it_passes():
|
||||
assert undecoded_subprocess_text("subprocess.run(argv, capture_output=True, text=True)") == [1]
|
||||
assert undecoded_subprocess_text("subprocess.run(argv, universal_newlines=True)") == [1]
|
||||
assert undecoded_subprocess_text('subprocess.run(argv, text=True, encoding="utf-8")') == []
|
||||
assert undecoded_subprocess_text("subprocess.run(argv, capture_output=True)") == []
|
||||
|
||||
|
||||
def test_a_platform_lock_import_is_caught():
|
||||
assert platform_lock_imports("import fcntl") == [1]
|
||||
assert platform_lock_imports("def f():\n import msvcrt\n") == [2]
|
||||
assert platform_lock_imports("from fcntl import flock") == [1]
|
||||
|
||||
|
||||
# --- the lock without fcntl ------------------------------------------------------
|
||||
|
||||
|
||||
class FakeMsvcrt(types.ModuleType):
|
||||
"""What `msvcrt.locking` is asked to do, and from which position."""
|
||||
|
||||
LK_UNLCK, LK_LOCK, LK_NBLCK = 0, 1, 2
|
||||
|
||||
def __init__(self, failures: int = 0):
|
||||
super().__init__("msvcrt")
|
||||
self.calls: list[tuple[int, int, int]] = []
|
||||
self.failures = failures
|
||||
|
||||
def locking(self, fd: int, mode: int, nbytes: int) -> None:
|
||||
self.calls.append((mode, os.lseek(fd, 0, os.SEEK_CUR), nbytes))
|
||||
if mode == self.LK_LOCK and self.failures:
|
||||
self.failures -= 1
|
||||
raise OSError(36, "Resource deadlock avoided")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def windows_locking(monkeypatch):
|
||||
"""Hide `fcntl` the way Windows does and hand `chemenu.filelock` a fake `msvcrt`."""
|
||||
fake = FakeMsvcrt()
|
||||
monkeypatch.setitem(sys.modules, "fcntl", None) # `import fcntl` now raises ImportError
|
||||
monkeypatch.setitem(sys.modules, "msvcrt", fake)
|
||||
return fake
|
||||
|
||||
|
||||
def test_without_fcntl_the_lock_falls_back_to_msvcrt(tmp_path, windows_locking):
|
||||
target = tmp_path / "trace.jsonl"
|
||||
with open(target, "a", encoding="utf-8", newline="\n") as handle:
|
||||
handle.write("first\n")
|
||||
handle.flush()
|
||||
before = os.lseek(handle.fileno(), 0, os.SEEK_CUR)
|
||||
with filelock.exclusive(handle):
|
||||
assert windows_locking.calls == [(FakeMsvcrt.LK_LOCK, filelock.WINDOWS_LOCK_OFFSET, 1)]
|
||||
assert os.lseek(handle.fileno(), 0, os.SEEK_CUR) == before
|
||||
handle.write("second\n")
|
||||
assert windows_locking.calls[-1] == (FakeMsvcrt.LK_UNLCK, filelock.WINDOWS_LOCK_OFFSET, 1)
|
||||
# The lock sits past the data, so the lines themselves are untouched.
|
||||
assert target.read_text(encoding="utf-8") == "first\nsecond\n"
|
||||
|
||||
|
||||
def test_a_contended_windows_lock_is_waited_for_like_flock(tmp_path, monkeypatch):
|
||||
fake = FakeMsvcrt(failures=2)
|
||||
monkeypatch.setitem(sys.modules, "fcntl", None)
|
||||
monkeypatch.setitem(sys.modules, "msvcrt", fake)
|
||||
with open(tmp_path / "budget.lock", "a", encoding="utf-8", newline="\n") as handle:
|
||||
with filelock.exclusive(handle):
|
||||
pass
|
||||
modes = [mode for mode, _, _ in fake.calls]
|
||||
assert modes == [FakeMsvcrt.LK_LOCK] * 3 + [FakeMsvcrt.LK_UNLCK]
|
||||
|
||||
|
||||
def test_the_budget_counter_locks_without_fcntl(tmp_path, monkeypatch, windows_locking):
|
||||
from chemenu.commands import run_budget
|
||||
|
||||
monkeypatch.setattr(run_budget, "STATE_DIR", tmp_path)
|
||||
monkeypatch.setattr(run_budget, "LOCK_FILE", tmp_path / "budget.lock")
|
||||
with run_budget._state_lock():
|
||||
pass
|
||||
assert [mode for mode, _, _ in windows_locking.calls] == [FakeMsvcrt.LK_LOCK, FakeMsvcrt.LK_UNLCK]
|
||||
|
||||
|
||||
def test_the_telemetry_writer_locks_without_fcntl(tmp_path, windows_locking):
|
||||
from chemenu.telemetry import writer
|
||||
|
||||
target = tmp_path / "trace.jsonl"
|
||||
with open(target, "a", encoding="utf-8", newline="\n") as handle:
|
||||
writer._locked_write(handle, '{"event":"x"}\n')
|
||||
assert [mode for mode, _, _ in windows_locking.calls] == [FakeMsvcrt.LK_LOCK, FakeMsvcrt.LK_UNLCK]
|
||||
assert target.read_text(encoding="utf-8") == '{"event":"x"}\n'
|
||||
|
||||
|
||||
def test_with_fcntl_the_lock_is_flock(tmp_path):
|
||||
fcntl = pytest.importorskip("fcntl")
|
||||
with open(tmp_path / "lock", "a", encoding="utf-8", newline="\n") as handle:
|
||||
with filelock.exclusive(handle):
|
||||
with open(tmp_path / "lock", "a", encoding="utf-8", newline="\n") as other:
|
||||
with pytest.raises(BlockingIOError):
|
||||
fcntl.flock(other.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
|
||||
|
||||
# --- the launcher's own output ---------------------------------------------------
|
||||
|
||||
|
||||
def _launcher_stderr(**env: str) -> bytes:
|
||||
"""What `run_wikitool.py` writes for an unknown command named `Fußnoten`.
|
||||
|
||||
Typer echoes the command name back, which makes it a non-ASCII line that
|
||||
needs no checkout, no corpus and no preflight.
|
||||
"""
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(TOOLS_DIR / "run_wikitool.py"), "Fußnoten"],
|
||||
capture_output=True,
|
||||
env={**os.environ, **env},
|
||||
)
|
||||
assert result.returncode != 0
|
||||
return result.stderr
|
||||
|
||||
|
||||
def test_output_into_a_pipe_is_utf8():
|
||||
"""Linux and macOS: the bytes are what they always were."""
|
||||
assert "Fußnoten".encode("utf-8") in _launcher_stderr()
|
||||
|
||||
|
||||
def test_output_into_a_pipe_is_utf8_even_under_a_cp1252_locale():
|
||||
"""What Windows does to a pipe, simulated: Python would otherwise write `Fu\\xdfnoten`."""
|
||||
stderr = _launcher_stderr(PYTHONIOENCODING="cp1252")
|
||||
assert "Fußnoten".encode("utf-8") in stderr
|
||||
assert "Fußnoten".encode("cp1252") not in stderr
|
||||
|
||||
|
||||
# --- line endings in a checkout --------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path, eol", [
|
||||
("tools/wikitool", "lf"),
|
||||
("tools/trace-hook", "lf"),
|
||||
("tools/preflight.sh", "lf"),
|
||||
("AGENTS.md", "lf"),
|
||||
])
|
||||
def test_every_text_file_is_checked_out_with_lf(path, eol):
|
||||
"""`core.autocrlf=true` would otherwise give the sh launchers `bash\\r`."""
|
||||
result = subprocess.run(
|
||||
["git", "check-attr", "text", "eol", "--", path],
|
||||
cwd=TOOLS_DIR.parent, capture_output=True, text=True, encoding="utf-8",
|
||||
)
|
||||
if result.returncode != 0:
|
||||
pytest.skip("not a git checkout")
|
||||
assert f"{path}: text: auto" in result.stdout
|
||||
assert f"{path}: eol: {eol}" in result.stdout
|
||||
|
||||
|
||||
def test_sources_are_kept_byte_for_byte():
|
||||
"""raw/ is immutable: a CRLF source must not be normalized on `git add`."""
|
||||
result = subprocess.run(
|
||||
["git", "check-attr", "text", "--", "raw/articles/x.md", "incoming/x.md"],
|
||||
cwd=TOOLS_DIR.parent, capture_output=True, text=True, encoding="utf-8",
|
||||
)
|
||||
if result.returncode != 0:
|
||||
pytest.skip("not a git checkout")
|
||||
assert "raw/articles/x.md: text: unset" in result.stdout
|
||||
assert "incoming/x.md: text: unset" in result.stdout
|
||||
@@ -1,3 +1,4 @@
|
||||
import json
|
||||
import subprocess
|
||||
import time
|
||||
from pathlib import Path
|
||||
@@ -442,3 +443,50 @@ def test_a_page_with_no_frontmatter_at_all_is_reported(kb_dir, tmp_path):
|
||||
(kb_dir / "entities" / "Naked.md").write_text("# Naked\n\nProse only.\n", encoding="utf-8")
|
||||
reported = unreadable_pages(load_pages_by_path(kb_dir, tmp_path))
|
||||
assert [entry["path"] for entry in reported] == ["kb/entities/Naked.md"]
|
||||
|
||||
|
||||
# What `rg --json --path-separator / Contract kb` printed on the Windows target
|
||||
# system (Gitea #152, T3): the flag does not reach JSON output.
|
||||
T3_MATCH = (
|
||||
r'{"type":"match","data":{"path":{"text":"kb\\comparisons\\COLLECTION.md"},'
|
||||
r'"lines":{"text":"Contract\n"},"line_number":1,"absolute_offset":0,"submatches":[]}}'
|
||||
)
|
||||
|
||||
|
||||
def test_a_windows_rg_path_reaches_the_parser_with_forward_slashes(monkeypatch):
|
||||
data = json.loads(T3_MATCH)["data"]
|
||||
monkeypatch.setattr(ripgrep, "NATIVE_SEPARATOR", "\\")
|
||||
assert ripgrep._record_path(data) == "kb/comparisons/COLLECTION.md"
|
||||
|
||||
|
||||
def test_a_posix_rg_path_is_left_as_it_is(monkeypatch):
|
||||
"""On POSIX a backslash is an ordinary character of a name, not a separator."""
|
||||
data = json.loads(T3_MATCH)["data"]
|
||||
monkeypatch.setattr(ripgrep, "NATIVE_SEPARATOR", "/")
|
||||
assert ripgrep._record_path(data) == "kb\\comparisons\\COLLECTION.md"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("separator, found", [("\\", True), ("/", False)])
|
||||
def test_a_windows_rg_match_becomes_a_full_hit(monkeypatch, kb_dir, tmp_path, pages, separator, found):
|
||||
"""The T3 line shape, pointed at a real fixture page: with the Windows separator
|
||||
the hit carries the repo-relative POSIX key, its title and its frontmatter."""
|
||||
page = kb_dir / "entities" / "systems" / "aurora.md"
|
||||
match = json.loads(T3_MATCH)
|
||||
match["data"]["path"]["text"] = str(page).replace("/", "\\")
|
||||
match["data"]["lines"]["text"] = "Hosts things.\n"
|
||||
|
||||
def fake_run(argv, **kwargs):
|
||||
return subprocess.CompletedProcess(argv, 0, json.dumps(match) + "\n", "")
|
||||
|
||||
monkeypatch.setattr(ripgrep.subprocess, "run", fake_run)
|
||||
monkeypatch.setattr(ripgrep, "NATIVE_SEPARATOR", separator)
|
||||
hits = RipgrepBackend(kb_dir, tmp_path).search(SearchQuery(text="Hosts"), pages)
|
||||
if not found:
|
||||
assert hits == []
|
||||
return
|
||||
[hit] = hits
|
||||
assert hit.path == "kb/entities/systems/aurora.md"
|
||||
assert hit.title == "aurora"
|
||||
assert hit.kind == "entity"
|
||||
assert hit.collection == "entities"
|
||||
assert "ZFS" in hit.summary
|
||||
@@ -5,6 +5,7 @@ change in our normalisation shows up here rather than in a silent gap in a
|
||||
scored run.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tomllib
|
||||
@@ -145,6 +146,23 @@ def test_malformed_stdin_never_fails_a_tool_call(tmp_path, payload):
|
||||
assert result.stdout == ""
|
||||
|
||||
|
||||
def test_a_utf8_payload_survives_a_cp1252_locale():
|
||||
"""What Windows does to `sys.stdin`, simulated: the harness writes UTF-8, and a
|
||||
locale-decoded read fails outright on a byte cp1252 leaves undefined - `Ł` is
|
||||
`C5 81` in UTF-8 (Gitea #152). A `ß` alone would not show it: misread on the
|
||||
way in and miswritten on the way out, it comes back intact."""
|
||||
payload = {**VIBE_POST_TOOL, "session_id": "v-Łódź-Fußnoten"}
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--source", "mistral-vibe", "--dry-run"],
|
||||
input=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
|
||||
capture_output=True,
|
||||
env={**os.environ, "PYTHONIOENCODING": "cp1252"},
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
record = json.loads(result.stdout.decode("utf-8"))
|
||||
assert record["session_id"] == "v-Łódź-Fußnoten"
|
||||
|
||||
|
||||
def test_dry_run_prints_exactly_one_valid_event():
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--source", "mistral-vibe", "--dry-run"],
|
||||
|
||||
Reference in new issue
Block a user