"""GTC codec: synthetic round-trips, property tests, corruption handling (spec IN-032, IN-033, EDGE-002).""" from __future__ import annotations import numpy as np import pytest from hypothesis import given, settings from hypothesis import strategies as st from simurg.core.errors import FormatError from simurg.formats.gtc import ( CANONICAL_ORDER_V3, CANONICAL_ORDER_V5, XFORM_DTYPE, Gtc, ScannerData, encode_gtc, read_gtc, write_gtc, ) def make_gtc(n: int = 20, version: int = 5, seed: int = 0) -> Gtc: rng = np.random.default_rng(seed) xf = np.zeros(3, XFORM_DTYPE) xf["version"] = 1 xf["offset_x"] = rng.normal(size=3) xf["theta"] = rng.normal(size=3) g = Gtc( version=version, num_snps=n, sample_name="NA12878 ünicode", cluster_file="c.egt", snp_manifest="m.bpm", imaging_date="12/27/2023 5:53:34 PM", autocall_date="01/01/26 12:00 PM", autocall_version="simurg-0.1", normalization_transforms=xf, controls_x=rng.integers(0, 60000, 8).astype(np.uint16), controls_y=rng.integers(0, 60000, 8).astype(np.uint16), raw_x=rng.integers(0, 60000, n).astype(np.uint16), raw_y=rng.integers(0, 60000, n).astype(np.uint16), genotypes=rng.integers(0, 4, n).astype(np.uint8), base_calls=np.frombuffer(b"AG" * n, np.uint8).reshape(n, 2).copy(), genotype_scores=rng.random(n).astype(np.float32), scanner_data=ScannerData("N103", b"\x01\x00\x00\x00", b"\x02\x00\x00\x00", "4.2.1", "user"), call_rate=0.99, gender=b"F", logr_dev=0.12, gc10=0.4, b_allele_freqs=rng.random(n).astype(np.float32), logr_ratios=rng.normal(size=n).astype(np.float32), percentiles_x=np.array([1, 2, 3], np.uint16), percentiles_y=np.array([4, 5, 6], np.uint16), slide_identifier="200000000001", ) g.sample_data["p50gc"] = 0.7 g.sample_data["num_calls"] = n - 2 g.sample_data["num_no_calls"] = 2 return g @pytest.mark.parametrize("version", [3, 5]) def test_roundtrip_canonical(tmp_path, version) -> None: # type: ignore[no-untyped-def] g = make_gtc(version=version) p = tmp_path / "s.gtc" write_gtc(g, p) h = read_gtc(p) assert h.toc_order == (CANONICAL_ORDER_V5 if version == 5 else CANONICAL_ORDER_V3) assert h.sample_name == g.sample_name assert np.array_equal(h.raw_x, g.raw_x) and np.array_equal(h.genotypes, g.genotypes) assert np.array_equal(h.genotype_scores, g.genotype_scores) assert h.scanner_data == g.scanner_data assert h.gc50 == pytest.approx(0.7) and h.num_calls == 18 assert h.gender_char == "F" assert h.base_call_strings()[0] == "AG" assert encode_gtc(h) == p.read_bytes() # byte-identical round-trip if version == 5: assert np.array_equal(h.logr_ratios, g.logr_ratios) def test_noncanonical_layout_preserved(tmp_path) -> None: # type: ignore[no-untyped-def] g = make_gtc() g.toc_order = list(reversed(CANONICAL_ORDER_V5)) g.data_order = [t for t in CANONICAL_ORDER_V5 if t > 3][::-1] data = encode_gtc(g) p = tmp_path / "s.gtc" p.write_bytes(data) h = read_gtc(p) assert h.toc_order == g.toc_order and h.data_order == g.data_order assert encode_gtc(h) == data @settings(max_examples=25, deadline=None) @given(n=st.integers(0, 200), seed=st.integers(0, 10**6), name=st.text(max_size=300)) def test_property_roundtrip(tmp_path_factory, n, seed, name) -> None: # type: ignore[no-untyped-def] g = make_gtc(n=n, seed=seed) g.sample_name = name data = encode_gtc(g) p = tmp_path_factory.mktemp("g") / "x.gtc" p.write_bytes(data) h = read_gtc(p) assert h.sample_name == name and h.num_snps == n assert encode_gtc(h) == data def test_truncated_and_bad_magic(tmp_path) -> None: # type: ignore[no-untyped-def] data = encode_gtc(make_gtc()) for label, blob in { "trunc": data[: len(data) // 3], "magic": b"xtc" + data[3:], "tiny": data[:5], "version": data[:3] + b"\x09" + data[4:], }.items(): p = tmp_path / f"{label}.gtc" p.write_bytes(blob) with pytest.raises(FormatError) as e: read_gtc(p) assert e.value.code == "E_GTC_CORRUPT", label def test_bad_string_length_is_clean_error(tmp_path) -> None: # type: ignore[no-untyped-def] # EDGE-002 g = make_gtc() data = bytearray(encode_gtc(g)) header = 8 + 6 * len(CANONICAL_ORDER_V5) data[header] = 0xFF # sample-name length prefix now runs past EOF data[header + 1] = 0xFF data[header + 2] = 0xFF data[header + 3] = 0x7F p = tmp_path / "s.gtc" p.write_bytes(bytes(data)) with pytest.raises(FormatError) as e: read_gtc(p) assert e.value.code == "E_GTC_CORRUPT" def test_incomplete_write_detection(tmp_path) -> None: # type: ignore[no-untyped-def] g = make_gtc() g.toc_order = [t for t in CANONICAL_ORDER_V5 if t != 1016] g.data_order = [t for t in g.toc_order if t > 3] p = tmp_path / "s.gtc" p.write_bytes(encode_gtc(g)) with pytest.raises(FormatError): read_gtc(p) assert read_gtc(p, check_write_complete=False).num_snps == 20 def test_length_mismatch_rejected_on_write() -> None: g = make_gtc() g.raw_x = g.raw_x[:-1] with pytest.raises(ValueError): encode_gtc(g)