Pin text I/O to UTF-8 and fail CI on locale-dependent reads/writes (#3296)

Co-authored-by: ShuQingDollarVoyager <57471784+ShuQingDollarVoyager@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

scripts/gen_surface_types.py

@@ -124,7 +124,7 @@ HEADER = (
 
 def load_pinned() -> list[dict[str, str]]:
     """Read `schema/PINNED.json` and verify each vendored file's sha256."""
-    entries: list[dict[str, str]] = json.loads((SCHEMA_DIR / "PINNED.json").read_text())
+    entries: list[dict[str, str]] = json.loads((SCHEMA_DIR / "PINNED.json").read_text(encoding="utf-8"))
     for entry in entries:
         path = SCHEMA_DIR / f"{entry['protocol_version']}.json"
         actual = hashlib.sha256(path.read_bytes()).hexdigest()
@@ -189,7 +189,7 @@ def run_codegen(schema_path: Path, output_path: Path) -> None:
             "--type-mappings", "byte=string", "uri=string", "uri-template=string",
             "--disable-timestamp",
         ],
-        capture_output=True, text=True,
+        capture_output=True, encoding="utf-8", errors="replace",
     )
     # fmt: on
     if result.returncode != 0:
@@ -222,16 +222,16 @@ def allow_open_class_extras(source: str, open_classes: frozenset[str]) -> str:
 def build(entry: dict[str, str]) -> str:
     """Generate, post-process, and format one version's surface module text."""
     version = entry["protocol_version"]
-    schema = json.loads((SCHEMA_DIR / f"{version}.json").read_text())
+    schema = json.loads((SCHEMA_DIR / f"{version}.json").read_text(encoding="utf-8"))
     patch_schema(schema, SCHEMA_PATCHES.get(version, []))
     make_server_info_opaque(schema)
 
     with tempfile.TemporaryDirectory() as tmp:
         patched = Path(tmp) / "schema.json"
-        patched.write_text(json.dumps(schema))
+        patched.write_text(json.dumps(schema), encoding="utf-8")
         raw = Path(tmp) / "raw.py"
         run_codegen(patched, raw)
-        source = raw.read_text()
+        source = raw.read_text(encoding="utf-8")
 
     source = re.sub(r"\A# generated by datamodel-codegen:\n#[^\n]*\n", "", source)
     source = re.sub(r"^class Model\(RootModel\[Any\]\):\n {4}root: Any\n+", "", source, count=1, flags=re.MULTILINE)
@@ -253,12 +253,12 @@ def build(entry: dict[str, str]) -> str:
 
     staging = TYPES_DIR / f"_staging_{version}.py"
     try:
-        staging.write_text(source)
+        staging.write_text(source, encoding="utf-8")
         subprocess.run(
             ["uv", "run", "--frozen", "ruff", "format", "--no-cache", str(staging)],
             cwd=REPO_ROOT, capture_output=True, check=True,
         )  # fmt: skip
-        return staging.read_text()
+        return staging.read_text(encoding="utf-8")
     finally:
         staging.unlink(missing_ok=True)
 
@@ -275,10 +275,10 @@ def main(argv: list[str] | None = None) -> int:
         candidate = build(entry)
         if not args.check:
             target.parent.mkdir(parents=True, exist_ok=True)
-            target.write_text(candidate)
+            target.write_text(candidate, encoding="utf-8")
             print(f"{entry['protocol_version']}: wrote {target.relative_to(REPO_ROOT)} ({len(candidate)} bytes)")
             continue
-        committed = target.read_text() if target.is_file() else ""
+        committed = target.read_text(encoding="utf-8") if target.is_file() else ""
         if committed != candidate:
             drift = True
             sys.stderr.writelines(

scripts/test

@@ -3,7 +3,8 @@
 set -ex
 
 uv run --frozen coverage erase
-uv run --frozen coverage run -m pytest -n auto $@
+# PYTHONWARNDEFAULTENCODING=1 mirrors CI: text I/O without encoding= fails under pytest's "error" filter.
+PYTHONWARNDEFAULTENCODING=1 uv run --frozen coverage run -m pytest -n auto "$@"
 uv run --frozen coverage combine
 uv run --frozen coverage report
 # strict-no-cover spawns `uv run coverage json` internally without --frozen;

scripts/update_readme_snippets.py

@@ -43,15 +43,7 @@ def process_snippet_block(match: re.Match[str], check_mode: bool = False) -> str
     file_path = match.group(2)
 
     try:
-        # Read the entire file. A missing source file must be fatal: a "Warning"
-        # that returns the stale block lets --check pass with exit 0, so a
-        # renamed or deleted snippet is invisible to CI. SystemExit deliberately
-        # escapes the `except Exception` below.
-        file = Path(file_path)
-        if not file.exists():
-            sys.exit(f"Error: snippet-source file not found: {file_path}")
-
-        code = file.read_text().rstrip()
+        code = Path(file_path).read_text(encoding="utf-8").rstrip()
         github_url = get_github_url(file_path)
 
         # Build the replacement block
@@ -88,9 +80,9 @@ def process_snippet_block(match: re.Match[str], check_mode: bool = False) -> str
 
         return replacement
 
-    except Exception as e:
-        print(f"Error processing {file_path}: {e}")
-        return full_match
+    except (OSError, UnicodeDecodeError) as e:
+        # Fatal, not "warn and keep the stale block": that would let --check pass with exit 0.
+        sys.exit(f"Error processing {file_path}: {e}")
 
 
 def update_readme_snippets(check_mode: bool = False) -> bool:
@@ -107,7 +99,7 @@ def update_readme_snippets(check_mode: bool = False) -> bool:
         print(f"Error: README file not found: {readme_path}")
         return False
 
-    content = readme_path.read_text()
+    content = readme_path.read_text(encoding="utf-8")
     original_content = content
 
     # Pattern to match snippet-source blocks
@@ -129,14 +121,14 @@ def update_readme_snippets(check_mode: bool = False) -> bool:
             )
             return False
         else:
-            print(f"✓ {readme_path} code snippets are up to date")
+            print(f"{readme_path} code snippets are up to date")
             return True
     else:
         if updated_content != original_content:
-            readme_path.write_text(updated_content)
-            print(f"✓ Updated {readme_path}")
+            readme_path.write_text(updated_content, encoding="utf-8")
+            print(f"Updated {readme_path}")
         else:
-            print(f"✓ {readme_path} already up to date")
+            print(f"{readme_path} already up to date")
         return True