fix(persistence): bound exporter encoding

Co-Authored-By: Codex <noreply@openai.com>

Author: percy-raskova

tests/unit/persistence/test_rust_legacy_contract_fixtures.py

@@ -3,9 +3,10 @@
 from __future__ import annotations
 
 import importlib.util
+import os
+import stat
 import subprocess
 import sys
-from collections.abc import Iterator
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -37,15 +38,31 @@ def test_overfull_migration_directory_stops_at_its_sentinel_entry(
 ) -> None:
     """Directory discovery must reject at its bounded sentinel entry."""
 
-    def entries() -> Iterator[Path]:
-        for index in range(exporter.MAX_MIGRATION_DIRECTORY_ENTRIES + 1):
-            yield Path(f"{index:04d}_migration.sql")
-        raise AssertionError("migration discovery consumed beyond its sentinel entry")
+    class ScandirContext:
+        """Bounded fake directory stream."""
 
-    def iterdir(_: Path) -> Iterator[Path]:
-        return entries()
+        def __init__(self) -> None:
+            self.index = 0
 
-    monkeypatch.setattr(Path, "iterdir", iterdir)
+        def __enter__(self) -> ScandirContext:
+            return self
+
+        def __exit__(self, *_: object) -> None:
+            return None
+
+        def __iter__(self) -> ScandirContext:
+            return self
+
+        def __next__(self) -> object:
+            if self.index == exporter.MAX_MIGRATION_DIRECTORY_ENTRIES + 1:
+                raise AssertionError("directory scan consumed beyond its sentinel entry")
+            self.index += 1
+            return object()
+
+    def scandir(_: Path) -> ScandirContext:
+        return ScandirContext()
+
+    monkeypatch.setattr(os, "scandir", scandir)
     with pytest.raises(RuntimeError, match=r"migration directory: entries exceed"):
         exporter._numbered_migrations()
 
@@ -92,6 +109,87 @@ def read_bytes(*_: object, **__: object) -> bytes:
     assert not read_bytes_called
 
 
+def test_frame_rejects_over_budget_text_before_encoding() -> None:
+    """A character-length overflow must not invoke a costly custom encoder."""
+
+    class EncodingTrap(str):
+        """String that fails if the exporter encodes it."""
+
+        def encode(self, *_: object, **__: object) -> bytes:
+            raise AssertionError("over-budget text must not be encoded")
+
+    chunk = EncodingTrap("x" * exporter.MAX_BYTES)
+    with pytest.raises(ValueError, match=r"frame: framed bytes exceed"):
+        exporter._frame([chunk], label="frame")
+
+
+def test_frame_checks_encoded_multibyte_length_before_nul_scan() -> None:
+    """UTF-8 expansion must still respect the remaining byte budget."""
+    chunk = "é" * (exporter.MAX_BYTES // 2)
+    with pytest.raises(ValueError, match=r"frame: framed bytes exceed"):
+        exporter._frame([chunk], label="frame")
+
+
+def test_frame_preserves_empty_chunk_precedence_after_a_full_chunk() -> None:
+    """An empty chunk remains invalid before any aggregate-size error."""
+    full_chunk = "x" * (exporter.MAX_BYTES - 1)
+    with pytest.raises(ValueError, match=r"frame: empty chunk 1"):
+        exporter._frame([full_chunk, ""], label="frame")
+
+
+@pytest.mark.parametrize(
+    ("mode", "size"),
+    [
+        (stat.S_IFREG, exporter.MAX_BYTES + 1),
+        (stat.S_IFDIR, len(b"expected")),
+    ],
+)
+def test_fixture_check_uses_one_descriptor_and_stops_before_read(
+    monkeypatch: pytest.MonkeyPatch,
+    mode: int,
+    size: int,
+) -> None:
+    """Oversize and nonregular fixtures must stop after descriptor metadata."""
+    open_calls = 0
+
+    class Fixture:
+        """Descriptor-backed fixture trap."""
+
+        def __enter__(self) -> Fixture:
+            return self
+
+        def __exit__(self, *_: object) -> None:
+            return None
+
+        def fileno(self) -> int:
+            return 41
+
+        def read(self, _: int = -1) -> bytes:
+            raise AssertionError("metadata-rejected fixture must not be read")
+
+    class FixturePath:
+        """Path-shaped trap for the exporter only."""
+
+        def open(self, *_: object, **__: object) -> Fixture:
+            nonlocal open_calls
+            open_calls += 1
+            return Fixture()
+
+        def stat(self, *_: object, **__: object) -> SimpleNamespace:
+            raise AssertionError("fixture checks must not use Path.stat()")
+
+        def __str__(self) -> str:
+            return "fixture.bin"
+
+    def fstat(descriptor: int) -> SimpleNamespace:
+        assert descriptor == 41
+        return SimpleNamespace(st_mode=mode, st_size=size)
+
+    monkeypatch.setattr(os, "fstat", fstat)
+    assert not exporter._check(FixturePath(), b"expected")
+    assert open_calls == 1
+
+
 def test_numbered_migrations_preserve_unique_missing_and_duplicate_results(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,

tools/export_legacy_postgres_contract.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import argparse
+import os
+import stat
 import sys
 from collections.abc import Sequence
 from itertools import islice
@@ -24,7 +26,8 @@ def _numbered_migrations() -> list[str]:
     """Read exactly one bounded source file for each required migration version."""
     migration_dir = SRC / "babylon/persistence/migrations"
     try:
-        entries = list(islice(migration_dir.iterdir(), MAX_MIGRATION_DIRECTORY_ENTRIES + 1))
+        with os.scandir(migration_dir) as directory:
+            entries = list(islice(directory, MAX_MIGRATION_DIRECTORY_ENTRIES + 1))
     except FileNotFoundError:
         entries = []
     if len(entries) > MAX_MIGRATION_DIRECTORY_ENTRIES:
@@ -48,13 +51,13 @@ def _numbered_migrations() -> list[str]:
     return chunks
 
 
-def _migration_matches(entries: Sequence[Path], version: int) -> list[Path]:
+def _migration_matches(entries: Sequence[os.DirEntry[str]], version: int) -> list[Path]:
     """Return no more than the two source paths needed to detect duplicates."""
     prefix = f"{version:04d}_"
     matches: list[Path] = []
     for entry in islice(entries, MAX_MIGRATION_DIRECTORY_ENTRIES):
         if entry.name.startswith(prefix) and entry.name.endswith(".sql") and entry.is_file():
-            matches.append(entry)
+            matches.append(Path(entry.path))
             if len(matches) == 2:
                 break
     return matches
@@ -78,12 +81,14 @@ def _frame(chunks: Sequence[str], *, label: str) -> bytes:
         raise ValueError(f"{label}: {len(chunks)} chunks exceeds {MAX_CHUNKS}")
     framed = bytearray()
     for index, chunk in enumerate(islice(chunks, MAX_CHUNKS)):
+        if not chunk:
+            raise ValueError(f"{label}: empty chunk {index}")
+        remaining = MAX_BYTES - len(framed) - 1
+        if len(chunk) > remaining:
+            raise ValueError(f"{label}: framed bytes exceed {MAX_BYTES}")
         encoded = chunk.encode("utf-8")
-        framed_length = len(framed) + len(encoded) + 1
-        if framed_length > MAX_BYTES:
+        if len(encoded) > remaining:
             raise ValueError(f"{label}: framed bytes exceed {MAX_BYTES}")
-        if not encoded:
-            raise ValueError(f"{label}: empty chunk {index}")
         if b"\0" in encoded:
             raise ValueError(f"{label}: embedded NUL in chunk {index}")
         framed.extend(encoded)
@@ -111,15 +116,21 @@ def _expected() -> tuple[bytes, bytes]:
 
 def _check(path: Path, expected: bytes) -> bool:
     try:
-        actual_size = path.stat().st_size
+        with path.open("rb") as fixture:
+            metadata = os.fstat(fixture.fileno())
+            if not stat.S_ISREG(metadata.st_mode):
+                print(f"invalid fixture: {path} is not a regular file", file=sys.stderr)
+                return False
+            if metadata.st_size != len(expected):
+                print(f"stale fixture: {path}", file=sys.stderr)
+                return False
+            actual = fixture.read(len(expected) + 1)
     except FileNotFoundError:
         print(f"missing fixture: {path}", file=sys.stderr)
         return False
-    if actual_size != len(expected):
-        print(f"stale fixture: {path}", file=sys.stderr)
+    except OSError as error:
+        print(f"fixture access failed: {path}: {error}", file=sys.stderr)
         return False
-    with path.open("rb") as fixture:
-        actual = fixture.read(len(expected) + 1)
     if actual != expected:
         print(f"stale fixture: {path}", file=sys.stderr)
         return False