fix: detect portable inventory path aliases

Author: renecannao

scripts/pg_compat/extract_statements.py

@@ -5,6 +5,7 @@
 import json
 import os
 from pathlib import Path
+import unicodedata
 
 if __package__:
     from .common import atomic_write_text, read_jsonl, statement_id
@@ -43,6 +44,13 @@ def _validate_row_object(row, row_index):
             raise ValueError(
                 f"row {row_index}: field name {field!r} must be a string"
             )
+        try:
+            field.encode("utf-8")
+        except UnicodeError as error:
+            raise ValueError(
+                f"row {row_index}: field name {field!r} is not UTF-8 "
+                f"encodable: {error}"
+            ) from error
 
 
 def _validate_required_fields(row, row_index, required_fields):
@@ -273,6 +281,30 @@ def _validate_distinct_paths(input_path, inventory_path, diagnostics_path):
                     aliases = os.path.samefile(left_path, right_path)
                 except OSError:
                     aliases = False
+
+            if not aliases:
+                left_parent = left_resolved.parent
+                right_parent = right_resolved.parent
+                same_parent = left_parent == right_parent
+                if not same_parent:
+                    try:
+                        same_parent = os.path.samefile(
+                            left_parent,
+                            right_parent,
+                        )
+                    except OSError:
+                        same_parent = False
+
+                left_name = unicodedata.normalize(
+                    "NFC",
+                    left_resolved.name,
+                ).casefold()
+                right_name = unicodedata.normalize(
+                    "NFC",
+                    right_resolved.name,
+                ).casefold()
+                aliases = same_parent and left_name == right_name
+
             if aliases:
                 raise ValueError(
                     f"{left_option} and {right_option} must identify distinct files"

tests/pg_compat/test_extract_statements.py

@@ -232,6 +232,19 @@ def test_rejects_non_string_row_keys_with_context(self):
         with self.assertRaisesRegex(ValueError, r"row 0.*field.*1.*string"):
             build_inventory([row])
 
+    def test_rejects_non_utf8_top_level_field_names_with_context(self):
+        row = accepted_row()
+        row["\ud800"] = "invalid key"
+
+        with self.assertRaises(ValueError) as raised:
+            build_inventory([row])
+
+        message = str(raised.exception)
+        self.assertIn("row 0", message)
+        self.assertIn("field name", message)
+        self.assertIn("\\ud800", message)
+        self.assertIn("UTF-8", message)
+
     def test_rejects_non_finite_and_non_utf8_metadata(self):
         invalid_metadata = (
             (float("nan"), "JSON"),
@@ -467,6 +480,50 @@ def test_rejects_same_path_aliases_before_reading_or_writing(self):
             self.assertEqual(input_path.read_text(encoding="utf-8"), "not JSON\n")
             self.assertFalse(shared_output_path.exists())
 
+    def test_rejects_nonexistent_output_names_differing_only_by_case(self):
+        with tempfile.TemporaryDirectory() as directory:
+            directory = Path(directory)
+            input_path = directory / "raw.jsonl"
+            inventory_path = directory / "Inventory.jsonl"
+            diagnostics_path = directory / "inventory.jsonl"
+            input_path.write_text("", encoding="utf-8")
+
+            result = self.run_cli(
+                "--input",
+                input_path,
+                "--inventory",
+                inventory_path,
+                "--diagnostics",
+                diagnostics_path,
+            )
+
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn("distinct files", result.stderr)
+            self.assertFalse(inventory_path.exists())
+            self.assertFalse(diagnostics_path.exists())
+
+    def test_rejects_nonexistent_output_names_with_equivalent_unicode(self):
+        with tempfile.TemporaryDirectory() as directory:
+            directory = Path(directory)
+            input_path = directory / "raw.jsonl"
+            inventory_path = directory / "caf\u00e9.jsonl"
+            diagnostics_path = directory / "cafe\u0301.jsonl"
+            input_path.write_text("", encoding="utf-8")
+
+            result = self.run_cli(
+                "--input",
+                input_path,
+                "--inventory",
+                inventory_path,
+                "--diagnostics",
+                diagnostics_path,
+            )
+
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn("distinct files", result.stderr)
+            self.assertFalse(inventory_path.exists())
+            self.assertFalse(diagnostics_path.exists())
+
     @unittest.skipUnless(hasattr(os, "symlink"), "symlinks are unavailable")
     def test_rejects_existing_symlink_path_aliases(self):
         with tempfile.TemporaryDirectory() as directory: