[arrow] improve test coverage and error handling

masque/test/test_gdsii_arrow.py

@@ -1,11 +1,15 @@
 from pathlib import Path
+import subprocess
+import sys
+import textwrap
 
+import klamath
 import numpy
 import pytest
 
 pytest.importorskip('pyarrow')
 
-from .. import Ref, Label
+from .. import Ref, Label, PatternError
 from ..library import Library
 from ..pattern import Pattern
 from ..repetition import Grid
@@ -175,6 +179,26 @@ def _make_arrow_test_library() -> Library:
     return lib
 
 
+def _write_invalid_path_type_fixture(path: Path) -> None:
+    with path.open('wb') as stream:
+        header = klamath.library.FileHeader(
+            name=b'test',
+            user_units_per_db_unit=1.0,
+            meters_per_db_unit=1e-9,
+        )
+        header.write(stream)
+        elem = klamath.elements.Path(
+            layer=(1, 0),
+            path_type=3,
+            width=10,
+            extension=(0, 0),
+            xy=numpy.array([[0, 0], [10, 0]], dtype=numpy.int32),
+            properties={},
+        )
+        klamath.library.write_struct(stream, name=b'top', elements=[elem])
+        klamath.records.ENDLIB.write(stream, None)
+
+
 def test_gdsii_arrow_matches_gdsii_readfile(tmp_path: Path) -> None:
     lib = _make_arrow_test_library()
     gds_file = tmp_path / 'arrow_roundtrip.gds'
@@ -238,6 +262,28 @@ def test_gdsii_arrow_removed_raw_mode_arg(tmp_path: Path) -> None:
         gdsii_arrow.read_arrow(libarr, raw_mode=False)
 
 
+def test_gdsii_arrow_invalid_input_raises_klamath_error(tmp_path: Path) -> None:
+    gds_file = tmp_path / 'invalid.gds'
+    gds_file.write_bytes(b'not-a-gds')
+
+    script = textwrap.dedent(f"""
+        from masque.file import gdsii_arrow
+        try:
+            gdsii_arrow.readfile({str(gds_file)!r})
+        except Exception as exc:
+            print(type(exc).__module__)
+            print(type(exc).__qualname__)
+            print(exc)
+        else:
+            raise SystemExit('expected gdsii_arrow.readfile() to fail')
+        """)
+    result = subprocess.run([sys.executable, '-c', script], capture_output=True, text=True, check=False)
+
+    assert result.returncode == 0, result.stderr
+    assert 'klamath.basic' in result.stdout
+    assert 'KlamathError' in result.stdout
+
+
 def test_gdsii_arrow_reads_small_perf_fixture(tmp_path: Path) -> None:
     gds_file = tmp_path / 'many_cells_smoke.gds'
     manifest = write_fixture(gds_file, preset='many_cells', scale=0.001)
@@ -435,6 +481,17 @@ def test_gdsii_arrow_ref_schema(tmp_path: Path) -> None:
     assert aref_props[0]['properties'][0]['value'] == 'fanout-aref'
 
 
+def test_gdsii_arrow_invalid_path_type_matches_gdsii(tmp_path: Path) -> None:
+    gds_file = tmp_path / 'invalid_path_type.gds'
+    _write_invalid_path_type_fixture(gds_file)
+
+    with pytest.raises(PatternError, match='Unrecognized path type: 3'):
+        gdsii.readfile(gds_file)
+
+    with pytest.raises(PatternError, match='Unrecognized path type: 3'):
+        gdsii_arrow.readfile(gds_file)
+
+
 def test_raw_ref_grid_label_constructors_match_public() -> None:
     raw_grid = Grid._from_raw(
         a_vector=numpy.array([20, 0]),

masque/test/test_gdsii_lazy_arrow.py

@@ -1,10 +1,15 @@
 from pathlib import Path
+import subprocess
+import sys
+import textwrap
 
+import klamath
 import numpy
 import pytest
 
 pytest.importorskip('pyarrow')
 
+from .. import PatternError
 from ..library import Library
 from ..pattern import Pattern
 from ..repetition import Grid
@@ -76,6 +81,26 @@ def _make_complex_ref_library() -> Library:
     return lib
 
 
+def _write_invalid_path_type_fixture(path: Path) -> None:
+    with path.open('wb') as stream:
+        header = klamath.library.FileHeader(
+            name=b'test',
+            user_units_per_db_unit=1.0,
+            meters_per_db_unit=1e-9,
+        )
+        header.write(stream)
+        elem = klamath.elements.Path(
+            layer=(1, 0),
+            path_type=3,
+            width=10,
+            extension=(0, 0),
+            xy=numpy.array([[0, 0], [10, 0]], dtype=numpy.int32),
+            properties={},
+        )
+        klamath.library.write_struct(stream, name=b'top', elements=[elem])
+        klamath.records.ENDLIB.write(stream, None)
+
+
 def _transform_rows_key(values: numpy.ndarray) -> tuple[tuple[object, ...], ...]:
     arr = numpy.asarray(values, dtype=float)
     arr = numpy.atleast_2d(arr)
@@ -147,6 +172,38 @@ def test_gdsii_lazy_arrow_ref_queries_match_eager_reader(tmp_path: Path) -> None
         assert _global_refs_key(lazy.find_refs_global(name)) == _global_refs_key(eager.find_refs_global(name))
 
 
+def test_gdsii_lazy_arrow_invalid_input_raises_klamath_error(tmp_path: Path) -> None:
+    gds_file = tmp_path / 'invalid.gds'
+    gds_file.write_bytes(b'not-a-gds')
+
+    script = textwrap.dedent(f"""
+        from masque.file import gdsii_lazy_arrow
+        try:
+            gdsii_lazy_arrow.readfile({str(gds_file)!r})
+        except Exception as exc:
+            print(type(exc).__module__)
+            print(type(exc).__qualname__)
+            print(exc)
+        else:
+            raise SystemExit('expected gdsii_lazy_arrow.readfile() to fail')
+        """)
+    result = subprocess.run([sys.executable, '-c', script], capture_output=True, text=True, check=False)
+
+    assert result.returncode == 0, result.stderr
+    assert 'klamath.basic' in result.stdout
+    assert 'KlamathError' in result.stdout
+
+
+def test_gdsii_lazy_arrow_invalid_path_type_raises_pattern_error(tmp_path: Path) -> None:
+    gds_file = tmp_path / 'invalid_path_type.gds'
+    _write_invalid_path_type_fixture(gds_file)
+
+    lib, _ = gdsii_lazy_arrow.readfile(gds_file)
+
+    with pytest.raises(PatternError, match='Unrecognized path type: 3'):
+        lib['top']
+
+
 def test_gdsii_lazy_arrow_untouched_write_is_copy_through(tmp_path: Path) -> None:
     gds_file = tmp_path / 'copy_source.gds'
     src = _make_small_library()
@@ -165,6 +222,24 @@ def test_gdsii_lazy_arrow_untouched_write_is_copy_through(tmp_path: Path) -> Non
     assert out_file.read_bytes() == gds_file.read_bytes()
 
 
+def test_gdsii_lazy_arrow_gzipped_copy_through(tmp_path: Path) -> None:
+    gds_file = tmp_path / 'copy_source.gds.gz'
+    src = _make_small_library()
+    gdsii.writefile(src, gds_file, meters_per_unit=1e-9, library_name='copy-through-gz')
+
+    lib, info = gdsii_lazy_arrow.readfile(gds_file)
+    out_file = tmp_path / 'copy_out.gds.gz'
+    gdsii_lazy_arrow.writefile(
+        lib,
+        out_file,
+        meters_per_unit=info['meters_per_unit'],
+        logical_units_per_unit=info['logical_units_per_unit'],
+        library_name=info['name'],
+    )
+
+    assert out_file.read_bytes() == gds_file.read_bytes()
+
+
 def test_gdsii_lazy_overlay_merge_and_write(tmp_path: Path) -> None:
     base_a = Library()
     leaf_a = Pattern()