mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(io): use UTF-8 with locale fallback and preserve line endings on config/text I/O (#1498)
## Description On non-UTF-8 Windows locales (e.g. GBK/cp936 on zh-CN, cp1252 on Western installs) `headroom wrap codex` corrupts `~/.codex/config.toml`. Two root causes, both in how we read/write text: - `Path.read_text()` / bare `open()` default to the **system locale** encoding, so a UTF-8 config fails to decode as the locale codec (and a locale-written file fails to decode as UTF-8) — raising `UnicodeDecodeError`. - `Path.write_text()` / text-mode `open()` translate `\n` → `os.linesep` on write, so an existing `\r\n` becomes `\r\r\n`, which TOML parsers reject with *"carriage return must be followed by newline"*. This adds one small helper module and routes the unsafe config/text I/O through it. Closes #733 ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - New `headroom/fsutil.py` with `read_text` / `write_text` / `append_text`: - `read_text`: decode UTF-8 → fall back to `locale.getpreferredencoding()` (for files a tool wrote in the locale encoding before this fix) → final UTF-8 with `errors="replace"` so it never raises on content. Line endings normalise to `\n`, so callers that search/rewrite the text see one ending and a later `write_text` can't re-double an existing `\r\n`. Supports `default=` for missing files. - `write_text` / `append_text`: UTF-8 with `newline=""` so the bytes written match the content exactly and existing `\r\n` endings are never doubled. - Routed the unsafe config/text I/O across the package through `fsutil` (or added an explicit `encoding="utf-8"` where only decode safety was missing): `mcp_registry/codex.py` (TOML read/write + `_load_toml` via `tomllib.loads`), `mcp_registry/opencode.py`, `mcp_registry/claude.py`, `cli/wrap.py`, `cli/mcp.py`, `cli/memory.py`, `install/providers.py`, `providers/anthropic.py`, `providers/openai.py`, `providers/opencode/config.py`, `providers/opencode/install.py`. - Tests: new `tests/test_fsutil.py` (CRLF preservation, no LF translation, CRLF normalisation on read, UTF-8 non-ASCII round trip, locale-decode fallback, never-raise replace fallback, missing-file default/raise, append preserves endings) and two `test_codex_registrar.py` regression tests (register doesn't double CRLF; non-ASCII values survive a register). ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text $ pytest tests/test_fsutil.py tests/test_mcp_registry/test_codex_registrar.py -q tests\test_fsutil.py ......... [ 25%] tests\test_mcp_registry\test_codex_registrar.py ........................ [100%] 36 passed in 0.32s $ ruff check <changed files> All checks passed! $ ruff format --check <changed files> 14 files already formatted $ mypy headroom --ignore-missing-imports # (run with --python-version 3.12 to # parse the local numpy stub) Success: no issues in changed files ``` Note: locally, the two suites `tests/test_mcp_registry` + `tests/test_cli` share a pre-existing cross-test state leak that flakes `test_wrap_codex_..._serena...` and `test_dead_client_marker...`; both reproduce identically on `main` (changes stashed) and are unrelated to this PR. CI shards run them isolated. ## Real Behavior Proof - Environment: Windows 11, Python 3.13.11, `locale.getpreferredencoding()` = `cp1252` (a non-UTF-8 locale — the exact condition that triggers #733). - Exact command / steps: pre-seed a `~/.codex/config.toml` the way Codex writes it on Windows — CRLF endings plus a non-ASCII value `project = "比赛/机器人"` — then call `CodexRegistrar.register_server(headroom)` and re-parse with `tomllib`. - Observed result: register status REGISTERED, no doubled CRLF, `tomllib` parses, and the non-ASCII value is preserved. Full output: ```text python: 3.13.11 | locale preferred encoding: cp1252 register status: RegisterStatus.REGISTERED doubled CRLF present: False tomllib parsed OK: True non-ASCII project value preserved: True headroom in mcp_servers: True ``` Before this change the same flow produced `\r\r\n` and a `tomllib` "carriage return must be followed by newline" error. - Not tested: a real zh-CN GBK/cp936 Windows install (no such host available); the GBK-specific decode path is covered by `test_read_text_falls_back_to_locale_encoding` which monkeypatches the preferred encoding to `gbk`. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Additional Notes - Purely-binary I/O and sites already using `encoding="utf-8"`+`errors="replace"` (e.g. `learn/analyzer.py`) and the ASCII-only PID file (`install/runtime.py`) were intentionally left untouched. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Parideboy
GitHub
parent
80fa086660
commit
1baa04ef65
14 changed files with 233 additions and 37 deletions
|
|
@ -35,7 +35,7 @@ def load_mcp_config() -> dict[str, Any]:
|
|||
"""Load existing MCP config or return empty structure."""
|
||||
if MCP_CONFIG_PATH.exists():
|
||||
try:
|
||||
with open(MCP_CONFIG_PATH) as f:
|
||||
with open(MCP_CONFIG_PATH, encoding="utf-8") as f:
|
||||
result: dict[str, Any] = json.load(f)
|
||||
return result
|
||||
except (json.JSONDecodeError, OSError):
|
||||
|
|
@ -46,7 +46,7 @@ def load_mcp_config() -> dict[str, Any]:
|
|||
def save_mcp_config(config: dict) -> None:
|
||||
"""Save MCP config, creating directory if needed."""
|
||||
CLAUDE_CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
with open(MCP_CONFIG_PATH, "w") as f:
|
||||
with open(MCP_CONFIG_PATH, "w", encoding="utf-8") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
f.write("\n") # Trailing newline
|
||||
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ from typing import Any
|
|||
|
||||
import click
|
||||
|
||||
from .. import fsutil
|
||||
from ..memory.adapters.sqlite import SQLiteMemoryStore
|
||||
from ..memory.models import Memory, ScopeLevel
|
||||
from ..memory.ports import MemoryFilter
|
||||
|
|
@ -874,7 +875,7 @@ def export_memories(ctx: click.Context, db_path: str, output: str | None) -> Non
|
|||
|
||||
if output:
|
||||
output_path = Path(output)
|
||||
output_path.write_text(json_output)
|
||||
fsutil.write_text(output_path, json_output)
|
||||
print_success(f"Exported {len(memories)} memory(ies) to {output_path}")
|
||||
else:
|
||||
click.echo(json_output)
|
||||
|
|
@ -904,7 +905,7 @@ def import_memories(ctx: click.Context, db_path: str, file: str, force: bool) ->
|
|||
|
||||
try:
|
||||
# Read and parse file
|
||||
content = file_path.read_text()
|
||||
content = fsutil.read_text(file_path)
|
||||
memories_data = json.loads(content)
|
||||
|
||||
if not isinstance(memories_data, list):
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ if sys.platform == "win32" and hasattr(sys.stdout, "buffer"):
|
|||
|
||||
import click
|
||||
|
||||
from headroom import fsutil
|
||||
from headroom._version import __version__ as _HEADROOM_VERSION
|
||||
from headroom.agent_savings import (
|
||||
apply_agent_savings_env_defaults,
|
||||
|
|
@ -128,19 +129,18 @@ from .main import main
|
|||
|
||||
|
||||
def _read_text(path: Path) -> str:
|
||||
"""Read a text file with explicit UTF-8 encoding."""
|
||||
return path.read_text(encoding="utf-8")
|
||||
"""Read a text file as UTF-8, falling back to the system locale encoding."""
|
||||
return fsutil.read_text(path)
|
||||
|
||||
|
||||
def _write_text(path: Path, content: str) -> None:
|
||||
"""Write a text file with explicit UTF-8 encoding."""
|
||||
path.write_text(content, encoding="utf-8")
|
||||
"""Write a text file as UTF-8 without translating line endings (preserves CRLF)."""
|
||||
fsutil.write_text(path, content)
|
||||
|
||||
|
||||
def _append_text(path: Path, content: str) -> None:
|
||||
"""Append to a text file with explicit UTF-8 encoding."""
|
||||
with open(path, "a", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
"""Append to a text file as UTF-8 without translating line endings."""
|
||||
fsutil.append_text(path, content)
|
||||
|
||||
|
||||
_CONTEXT_TOOL_ENV = "HEADROOM_CONTEXT_TOOL"
|
||||
|
|
@ -556,7 +556,7 @@ def _patch_rtk_hook_absolute_path(rtk_path: Path, hook_script_path: Path | None
|
|||
if not hook_script_path.exists():
|
||||
return False
|
||||
|
||||
original = hook_script_path.read_text()
|
||||
original = _read_text(hook_script_path)
|
||||
|
||||
# Quote the absolute path safely for POSIX shells. This matters because
|
||||
# paths containing spaces or other shell-special characters (e.g.
|
||||
|
|
@ -576,7 +576,7 @@ def _patch_rtk_hook_absolute_path(rtk_path: Path, hook_script_path: Path | None
|
|||
)
|
||||
|
||||
if count and patched != original:
|
||||
hook_script_path.write_text(patched)
|
||||
_write_text(hook_script_path, patched)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
|
|
|||
81
headroom/fsutil.py
Normal file
81
headroom/fsutil.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
"""Encoding- and newline-safe text file I/O.
|
||||
|
||||
``Path.read_text()`` / ``Path.write_text()`` and the builtin ``open()`` default
|
||||
to the *system locale* encoding and, in text mode, translate ``\\n`` to
|
||||
``os.linesep`` on write. On non-UTF-8 Windows locales (e.g. GBK / cp936 on
|
||||
zh-CN) this corrupts config files two ways:
|
||||
|
||||
1. Reading a UTF-8 file as GBK — or a GBK file as UTF-8 — raises
|
||||
``UnicodeDecodeError``.
|
||||
2. Writing re-translates ``\\n`` to ``\\r\\n``; content that already has
|
||||
``\\r\\n`` becomes ``\\r\\r\\n``, which TOML parsers reject with
|
||||
"carriage return must be followed by newline".
|
||||
|
||||
These helpers always use UTF-8, fall back to the locale encoding when a file
|
||||
predates the fix (tools may have written it in the locale encoding), and write
|
||||
with ``newline=""`` so existing line endings pass through unchanged.
|
||||
|
||||
See issue #733.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import locale
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
# Sentinel so ``default=None`` can be a real return value if a caller wants it.
|
||||
_RAISE = object()
|
||||
|
||||
|
||||
def read_text(path: str | os.PathLike[str], *, default: object = _RAISE) -> str:
|
||||
"""Read text, preferring UTF-8 and falling back to the locale encoding.
|
||||
|
||||
Decoding order: UTF-8 (strict) → locale preferred encoding (strict) →
|
||||
UTF-8 with ``errors="replace"`` (never raises on content). If the file
|
||||
cannot be opened (missing/unreadable) and ``default`` is given, it is
|
||||
returned; otherwise the ``OSError`` propagates.
|
||||
|
||||
Line endings are normalised to ``\\n`` (universal-newline semantics,
|
||||
matching the stdlib text-mode default) so callers that search or rewrite
|
||||
the text work on a single ending, and a later :func:`write_text` cannot
|
||||
re-double an existing ``\\r\\n``.
|
||||
"""
|
||||
try:
|
||||
raw = Path(path).read_bytes()
|
||||
except OSError:
|
||||
if default is not _RAISE:
|
||||
return default # type: ignore[return-value]
|
||||
raise
|
||||
|
||||
text: str | None = None
|
||||
try:
|
||||
text = raw.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
loc = locale.getpreferredencoding(False)
|
||||
if loc and loc.lower().replace("-", "") != "utf8":
|
||||
try:
|
||||
text = raw.decode(loc)
|
||||
except (UnicodeDecodeError, LookupError):
|
||||
text = None
|
||||
if text is None:
|
||||
text = raw.decode("utf-8", errors="replace")
|
||||
|
||||
return text.replace("\r\n", "\n").replace("\r", "\n")
|
||||
|
||||
|
||||
def write_text(path: str | os.PathLike[str], content: str) -> None:
|
||||
"""Write text as UTF-8 without translating line endings.
|
||||
|
||||
``newline=""`` disables the platform ``\\n`` → ``\\r\\n`` rewrite, so the
|
||||
bytes written match ``content`` exactly and existing ``\\r\\n`` endings are
|
||||
never doubled.
|
||||
"""
|
||||
with Path(path).open("w", encoding="utf-8", newline="") as f:
|
||||
f.write(content)
|
||||
|
||||
|
||||
def append_text(path: str | os.PathLike[str], content: str) -> None:
|
||||
"""Append text as UTF-8 without translating line endings (see ``write_text``)."""
|
||||
with Path(path).open("a", encoding="utf-8", newline="") as f:
|
||||
f.write(content)
|
||||
|
|
@ -7,6 +7,7 @@ import re
|
|||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from headroom import fsutil
|
||||
from headroom._subprocess import run
|
||||
from headroom.providers.install_registry import (
|
||||
apply_provider_scope_mutations,
|
||||
|
|
@ -29,7 +30,7 @@ _ENV_PATTERN = re.compile(
|
|||
|
||||
def _merge_marker_block(file_path: Path, block: str, pattern: re.Pattern[str], marker: str) -> str:
|
||||
if file_path.exists():
|
||||
existing = file_path.read_text()
|
||||
existing = fsutil.read_text(file_path)
|
||||
if marker in existing:
|
||||
return pattern.sub(block, existing)
|
||||
return existing.rstrip() + "\n\n" + block + "\n"
|
||||
|
|
@ -66,7 +67,7 @@ def _apply_unix_env_scope(manifest: DeploymentManifest) -> list[ManagedMutation]
|
|||
for path in targets:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
merged = _merge_marker_block(path, block, _ENV_PATTERN, _ENV_MARKER_START)
|
||||
path.write_text(merged)
|
||||
fsutil.write_text(path, merged)
|
||||
mutations.append(ManagedMutation(target="env", kind="shell-block", path=str(path)))
|
||||
return mutations
|
||||
|
||||
|
|
@ -78,10 +79,10 @@ def _remove_unix_env_scope(mutations: list[ManagedMutation]) -> None:
|
|||
path = Path(mutation.path)
|
||||
if not path.exists():
|
||||
continue
|
||||
content = path.read_text()
|
||||
content = fsutil.read_text(path)
|
||||
if _ENV_MARKER_START not in content:
|
||||
continue
|
||||
path.write_text(_ENV_PATTERN.sub("", content).strip() + "\n")
|
||||
fsutil.write_text(path, _ENV_PATTERN.sub("", content).strip() + "\n")
|
||||
|
||||
|
||||
def _apply_windows_env_scope(manifest: DeploymentManifest) -> list[ManagedMutation]:
|
||||
|
|
|
|||
|
|
@ -213,7 +213,7 @@ def _read_json(path: Path) -> dict[str, Any]:
|
|||
if not path.exists():
|
||||
return {}
|
||||
try:
|
||||
with open(path) as f:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return {}
|
||||
|
|
@ -224,7 +224,7 @@ def _read_json(path: Path) -> dict[str, Any]:
|
|||
|
||||
def _write_json(path: Path, data: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(path, "w") as f:
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
f.write("\n")
|
||||
|
||||
|
|
|
|||
|
|
@ -17,6 +17,8 @@ import sys
|
|||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from headroom import fsutil
|
||||
|
||||
from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
|
|
@ -132,7 +134,7 @@ class CodexRegistrar(MCPRegistrar):
|
|||
else:
|
||||
new_content = (before or after).rstrip("\n") + ("\n" if (before or after) else "")
|
||||
try:
|
||||
self._config_file.write_text(new_content)
|
||||
fsutil.write_text(self._config_file, new_content)
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
|
@ -145,17 +147,16 @@ class CodexRegistrar(MCPRegistrar):
|
|||
if not self._config_file.exists():
|
||||
return {}
|
||||
try:
|
||||
with open(self._config_file, "rb") as f:
|
||||
data = tomllib.load(f)
|
||||
# Read via fsutil (UTF-8 with locale fallback) so a config that a
|
||||
# tool wrote in the system locale (e.g. GBK) still parses instead
|
||||
# of failing tomllib's UTF-8 requirement. See #733.
|
||||
data = tomllib.loads(fsutil.read_text(self._config_file))
|
||||
except (tomllib.TOMLDecodeError, OSError):
|
||||
return {}
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
def _read_text(self) -> str:
|
||||
try:
|
||||
return self._config_file.read_text()
|
||||
except OSError:
|
||||
return ""
|
||||
return fsutil.read_text(self._config_file, default="")
|
||||
|
||||
def _write_block(self, spec: ServerSpec) -> RegisterResult:
|
||||
block = _render_block(spec)
|
||||
|
|
@ -178,7 +179,7 @@ class CodexRegistrar(MCPRegistrar):
|
|||
content = content.rstrip("\n") + "\n\n" + block + "\n"
|
||||
else:
|
||||
content = block + "\n"
|
||||
self._config_file.write_text(content)
|
||||
fsutil.write_text(self._config_file, content)
|
||||
except OSError as exc:
|
||||
return RegisterResult(
|
||||
RegisterStatus.FAILED, f"could not write {self._config_file}: {exc}"
|
||||
|
|
|
|||
|
|
@ -39,7 +39,7 @@ def _read_json(path: Path) -> dict[str, Any]:
|
|||
if not path.exists():
|
||||
return {}
|
||||
try:
|
||||
with open(path) as f:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return {}
|
||||
|
|
@ -50,7 +50,7 @@ def _read_json(path: Path) -> dict[str, Any]:
|
|||
|
||||
def _write_json(path: Path, data: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(path, "w") as f:
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
f.write("\n")
|
||||
|
||||
|
|
|
|||
|
|
@ -199,7 +199,7 @@ def _load_custom_model_config() -> dict[str, Any]:
|
|||
try:
|
||||
# Check if it's a file path
|
||||
if os.path.isfile(env_config):
|
||||
with open(env_config) as f:
|
||||
with open(env_config, encoding="utf-8") as f:
|
||||
loaded = json.load(f)
|
||||
else:
|
||||
# Try to parse as JSON string
|
||||
|
|
@ -225,7 +225,7 @@ def _load_custom_model_config() -> dict[str, Any]:
|
|||
config_file = legacy_models
|
||||
if config_file.exists():
|
||||
try:
|
||||
with open(config_file) as f:
|
||||
with open(config_file, encoding="utf-8") as f:
|
||||
loaded = json.load(f)
|
||||
|
||||
# Only load anthropic-specific config
|
||||
|
|
|
|||
|
|
@ -157,7 +157,7 @@ def _load_custom_model_config() -> dict[str, Any]:
|
|||
try:
|
||||
# Check if it's a file path
|
||||
if os.path.isfile(env_config):
|
||||
with open(env_config) as f:
|
||||
with open(env_config, encoding="utf-8") as f:
|
||||
loaded = json.load(f)
|
||||
else:
|
||||
# Try to parse as JSON string
|
||||
|
|
@ -184,7 +184,7 @@ def _load_custom_model_config() -> dict[str, Any]:
|
|||
config_file = legacy_models
|
||||
if config_file.exists():
|
||||
try:
|
||||
with open(config_file) as f:
|
||||
with open(config_file, encoding="utf-8") as f:
|
||||
loaded = json.load(f)
|
||||
|
||||
openai_config = loaded.get("openai", {})
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ from typing import Any
|
|||
|
||||
import click
|
||||
|
||||
from headroom import fsutil
|
||||
from headroom.install.paths import opencode_config_path
|
||||
from headroom.mcp_registry.install import DEFAULT_PROXY_URL
|
||||
|
||||
|
|
@ -58,7 +59,7 @@ def snapshot_opencode_config_if_unwrapped(config_file: Path, backup_file: Path)
|
|||
if not config_file.exists():
|
||||
return
|
||||
try:
|
||||
content = config_file.read_text()
|
||||
content = fsutil.read_text(config_file)
|
||||
except OSError:
|
||||
return
|
||||
if _PROVIDER_MARKER_START in content or _MCP_MARKER_START in content:
|
||||
|
|
@ -190,7 +191,7 @@ def inject_opencode_provider_config(port: int) -> None:
|
|||
snapshot_opencode_config_if_unwrapped(config_file, backup_file)
|
||||
|
||||
if config_file.exists():
|
||||
content = config_file.read_text()
|
||||
content = fsutil.read_text(config_file)
|
||||
data = _parse_json_loose(content)
|
||||
else:
|
||||
content = ""
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import json
|
|||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
from headroom import fsutil
|
||||
from headroom.install.models import ConfigScope, DeploymentManifest, ManagedMutation, ToolTarget
|
||||
from headroom.install.paths import opencode_config_path
|
||||
|
||||
|
|
@ -38,7 +39,7 @@ def apply_provider_scope(manifest: DeploymentManifest) -> ManagedMutation | None
|
|||
)
|
||||
|
||||
if config_file.exists():
|
||||
content = config_file.read_text()
|
||||
content = fsutil.read_text(config_file)
|
||||
data = _parse_json_loose(content)
|
||||
else:
|
||||
data = {}
|
||||
|
|
@ -80,7 +81,7 @@ def revert_provider_scope(mutation: ManagedMutation, manifest: DeploymentManifes
|
|||
pass
|
||||
if not path.exists():
|
||||
return
|
||||
content = path.read_text()
|
||||
content = fsutil.read_text(path)
|
||||
cleaned = strip_opencode_headroom_blocks(content)
|
||||
if cleaned:
|
||||
path.write_text(cleaned + "\n", encoding="utf-8")
|
||||
|
|
|
|||
73
tests/test_fsutil.py
Normal file
73
tests/test_fsutil.py
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
"""Tests for headroom.fsutil — encoding- and newline-safe text I/O (#733)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from headroom import fsutil
|
||||
|
||||
|
||||
def test_write_text_does_not_double_existing_crlf(tmp_path):
|
||||
"""A string containing \\r\\n must be written verbatim, never as \\r\\r\\n."""
|
||||
p = tmp_path / "config.toml"
|
||||
fsutil.write_text(p, 'model = "gpt-5"\r\nport = 8787\r\n')
|
||||
raw = p.read_bytes()
|
||||
assert b"\r\r\n" not in raw
|
||||
assert raw == b'model = "gpt-5"\r\nport = 8787\r\n'
|
||||
|
||||
|
||||
def test_write_text_does_not_translate_lf(tmp_path):
|
||||
"""\\n must stay \\n on every platform (no \\r\\n rewrite)."""
|
||||
p = tmp_path / "hook.sh"
|
||||
fsutil.write_text(p, "#!/bin/sh\necho hi\n")
|
||||
assert p.read_bytes() == b"#!/bin/sh\necho hi\n"
|
||||
|
||||
|
||||
def test_read_text_normalises_crlf(tmp_path):
|
||||
"""read_text returns universal-newline (\\n) text, so a round trip can't double CRLF."""
|
||||
p = tmp_path / "config.toml"
|
||||
p.write_bytes(b"a = 1\r\nb = 2\r\n")
|
||||
text = fsutil.read_text(p)
|
||||
assert text == "a = 1\nb = 2\n"
|
||||
fsutil.write_text(p, text)
|
||||
assert b"\r" not in p.read_bytes()
|
||||
|
||||
|
||||
def test_read_text_roundtrips_utf8_non_ascii(tmp_path):
|
||||
p = tmp_path / "config.toml"
|
||||
fsutil.write_text(p, 'project = "比赛/机器人"\n')
|
||||
assert fsutil.read_text(p) == 'project = "比赛/机器人"\n'
|
||||
|
||||
|
||||
def test_read_text_falls_back_to_locale_encoding(tmp_path, monkeypatch):
|
||||
"""A file a tool wrote in the locale encoding (e.g. GBK) still decodes."""
|
||||
monkeypatch.setattr(fsutil.locale, "getpreferredencoding", lambda *_: "gbk")
|
||||
p = tmp_path / "config.toml"
|
||||
p.write_bytes('path = "模型"\n'.encode("gbk")) # not valid UTF-8
|
||||
assert fsutil.read_text(p) == 'path = "模型"\n'
|
||||
|
||||
|
||||
def test_read_text_replace_fallback_never_raises(tmp_path, monkeypatch):
|
||||
"""When neither UTF-8 nor the locale encoding decodes, fall back to replace."""
|
||||
monkeypatch.setattr(fsutil.locale, "getpreferredencoding", lambda *_: "ascii")
|
||||
p = tmp_path / "config.toml"
|
||||
p.write_bytes(b"\xff\xfe bad bytes")
|
||||
# Must not raise; returns *something* decodable.
|
||||
assert isinstance(fsutil.read_text(p), str)
|
||||
|
||||
|
||||
def test_read_text_missing_returns_default(tmp_path):
|
||||
p = tmp_path / "nope.toml"
|
||||
assert fsutil.read_text(p, default="") == ""
|
||||
|
||||
|
||||
def test_read_text_missing_raises_without_default(tmp_path):
|
||||
with pytest.raises(OSError):
|
||||
fsutil.read_text(tmp_path / "nope.toml")
|
||||
|
||||
|
||||
def test_append_text_preserves_endings(tmp_path):
|
||||
p = tmp_path / "AGENTS.md"
|
||||
fsutil.write_text(p, "line1\n")
|
||||
fsutil.append_text(p, "line2\n")
|
||||
assert p.read_bytes() == b"line1\nline2\n"
|
||||
|
|
@ -359,3 +359,40 @@ def test_round_trip(tmp_path: Path, spec: ServerSpec) -> None:
|
|||
assert got.command == spec.command
|
||||
assert got.args == spec.args
|
||||
assert got.env == spec.env
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #733 — encoding / line-ending safety on GBK / CRLF Windows configs
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_register_does_not_double_crlf(tmp_path: Path) -> None:
|
||||
"""A pre-existing CRLF config must not gain ``\\r\\r\\n`` after register."""
|
||||
import tomllib
|
||||
|
||||
cfg = _config_path(tmp_path)
|
||||
cfg.parent.mkdir(parents=True, exist_ok=True)
|
||||
cfg.write_bytes(b'model = "gpt-5"\r\nworkers = 1\r\n')
|
||||
|
||||
result = _make_registrar(tmp_path).register_server(_spec())
|
||||
|
||||
assert result.status == RegisterStatus.REGISTERED
|
||||
raw = cfg.read_bytes()
|
||||
assert b"\r\r\n" not in raw
|
||||
tomllib.loads(raw.decode("utf-8")) # still valid TOML
|
||||
|
||||
|
||||
def test_register_preserves_non_ascii_values(tmp_path: Path) -> None:
|
||||
"""A config with non-ASCII (Chinese) values survives register and stays parseable."""
|
||||
import tomllib
|
||||
|
||||
cfg = _config_path(tmp_path)
|
||||
cfg.parent.mkdir(parents=True, exist_ok=True)
|
||||
cfg.write_text('model = "gpt-5"\nproject = "比赛/机器人"\n', encoding="utf-8")
|
||||
|
||||
result = _make_registrar(tmp_path).register_server(_spec())
|
||||
|
||||
assert result.status == RegisterStatus.REGISTERED
|
||||
data = tomllib.loads(cfg.read_text(encoding="utf-8"))
|
||||
assert data["project"] == "比赛/机器人"
|
||||
assert "headroom" in data.get("mcp_servers", {})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue