diff --git a/CHANGELOG.md b/CHANGELOG.md index 953da1c..fccbe16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ SPSS profile. - Import of unencrypted SAV and ZSAV sources into one dedicated wide SQL table and its metadata catalog. - Export of supported dataset semantics to SAV and ZSAV. +- Consistent legacy compatible-name rewriting across type-2, subtype-13, and + VLS subtype-14 records, with fail-closed SAV/ZSAV validation. - SQLite, PostgreSQL, MySQL, and MariaDB profiles, including service-backed PostgreSQL 17/18, MySQL 8.4/9.7, and MariaDB 11.4/11.8/12.3 CI coverage. - Preflight checks for target profile limits, atomic imports, validation, a diff --git a/docs/sav-profile.md b/docs/sav-profile.md index ab9cb5c..ec81335 100755 --- a/docs/sav-profile.md +++ b/docs/sav-profile.md @@ -41,7 +41,7 @@ of this boundary. It records the pinned source commit and installed engine versi | Ordered document text | Supported through a strict type-6 dictionary bridge plus the pinned fork | Import stores normalized document rows; export creates a temporary UTF-8 SAV source and has IBM I/O copy the records into SAV or ZSAV without invalidating ZSAV dictionary offsets. | | Print and write formats independently | Supported as raw IBM I/O tuples | The adapter stores both tuples separately and writes them without collapsing either value. | | Variable sets | Supported through raw IBM I/O | The adapter stores source sets in the extension catalog and writes them through the raw dictionary setter. Invalid target definitions fail before data are written. | -| Legacy compatible variable names | Supported through the strict dictionary bridge | The adapter stores the short-name mapping and rewrites both the fixed variable record and long-name extension before export. | +| Legacy compatible variable names | Supported through the strict dictionary bridge | The adapter atomically rewrites and reparses type-2, subtype-13, and VLS subtype-14 names as one consistent SAV/ZSAV dictionary; malformed or duplicate subtype-14 entries fail closed. | | Source encoding | UTF-8 by default; legacy code page with an explicit OS locale | The caller supplies legacy_locale and the writer verifies that the emitted encoding matches the stored source encoding. Without it, export fails before output creation. | | Custom attributes with one value | Supported | File and variable scalar attributes round trip through the normalized attribute catalog. | | Custom-attribute value arrays | Supported through raw IBM I/O | The adapter represents an array as IBM SPSS `Name[1]`, `Name[2]`, … members and reconstructs the ordered array in the normalized catalog. | diff --git a/src/openstatspec/spss/raw_dictionary.py b/src/openstatspec/spss/raw_dictionary.py index e514064..f77f952 100644 --- a/src/openstatspec/spss/raw_dictionary.py +++ b/src/openstatspec/spss/raw_dictionary.py @@ -5,8 +5,11 @@ """ from __future__ import annotations +import os +import stat from dataclasses import dataclass from pathlib import Path +from tempfile import NamedTemporaryFile from typing import Iterable @@ -70,13 +73,12 @@ def write_document_lines(path: str | Path, lines: Iterable[str], *, encoding: st def write_compatible_names( path: str | Path, names: dict[str, str], *, encoding: str, ) -> None: - """Set exact legacy short names without changing long source variable names. + """Set exact legacy short names consistently in type-2/13/14 records. - IBM I/O has no compatible-name setter. The standard long-name extension - and the fixed 8-byte variable records contain the complete mapping, so this - narrowly rewrites only those two dictionary locations. Every requested - source name must have a long-name record; otherwise the operation fails - instead of silently deriving another name. + IBM I/O has no compatible-name setter. Every requested source name must + have a long-name record, and every very-long-string key must agree with the + type-2 and subtype-13 records. The fully rebuilt dictionary is reparsed + before an atomic replacement is published. """ if not names: return @@ -84,18 +86,18 @@ def write_compatible_names( data = target.read_bytes() byte_order, records = _records(data) terminator = next(record for record in records if record.record_type == 999) - long_names = next( - ( - record for record in records - if record.record_type == 7 and _int(data, record.start + 4, byte_order) == 13 - ), - None, - ) - if long_names is None: + long_name_records = [ + record for record in records + if record.record_type == 7 and _int(data, record.start + 4, byte_order) == 13 + ] + if not long_name_records: raise RawDictionaryError("SAV dictionary has no long-variable-name record.") - payload_start = long_names.start + 16 - payload = data[payload_start : long_names.end] - pairs = _long_name_pairs(payload, encoding) + if len(long_name_records) != 1: + raise RawDictionaryError("SAV dictionary has more than one long-variable-name record.") + long_names = long_name_records[0] + if _int(data, long_names.start + 8, byte_order) != 1: + raise RawDictionaryError("Invalid SAV long-variable-name record dimensions.") + pairs = _long_name_pairs(data[long_names.start + 16 : long_names.end], encoding) replacements: dict[str, str] = {} updated_pairs: list[tuple[str, str]] = [] unresolved = set(names) @@ -112,6 +114,44 @@ def write_compatible_names( raise RawDictionaryError( "Compatible-name update requires a long-name record for: " + missing ) + updated_short_names = [short_name.casefold() for short_name, _ in updated_pairs] + if len(updated_short_names) != len(set(updated_short_names)): + raise RawDictionaryError("Compatible-name update would create duplicate short names.") + + type_2_names = _type_2_names(data, records, byte_order) + for short_name in replacements: + if type_2_names.count(short_name) != 1: + raise RawDictionaryError( + f"Expected exactly one type-2 record for compatible name {short_name!r}." + ) + + long_name_by_short = {short_name: long_name for short_name, long_name in pairs} + vls_records = [ + record for record in records + if record.record_type == 7 and _int(data, record.start + 4, byte_order) == 14 + ] + parsed_vls: dict[int, tuple[list[tuple[str, bytes]], bool]] = {} + seen_vls: set[str] = set() + vls_replacements: dict[str, str] = {} + vls_source_names: set[str] = set() + for record in vls_records: + if _int(data, record.start + 8, byte_order) != 1: + raise RawDictionaryError("Invalid SAV very-long-string record dimensions.") + entries, trailing_tab = _very_long_string_entries(data[record.start + 16 : record.end]) + parsed_vls[record.start] = (entries, trailing_tab) + for short_name, _ in entries: + normalized = short_name.casefold() + if normalized in seen_vls: + raise RawDictionaryError("Duplicate SAV very-long-string entry.") + seen_vls.add(normalized) + if short_name not in long_name_by_short or type_2_names.count(short_name) != 1: + raise RawDictionaryError( + "SAV type-2/subtype-13/subtype-14 names are inconsistent." + ) + replacement = replacements.get(short_name) + if replacement is not None: + vls_replacements[short_name] = replacement + vls_source_names.add(long_name_by_short[short_name]) mutable = bytearray(data) for record in records: @@ -122,22 +162,50 @@ def write_compatible_names( if replacement is not None: mutable[record.start + 24 : record.start + 32] = replacement.encode("ascii").ljust(8, b" ") - new_payload = b" ".join( + new_payload = b"\t".join( short_name.encode("ascii") + b"=" + long_name.encode(encoding) for short_name, long_name in updated_pairs ) - header = bytearray(mutable[long_names.start : long_names.start + 16]) - header[12:16] = _pack(len(new_payload), byte_order) - updated = ( - bytes(mutable[: long_names.start]) + bytes(header) + new_payload - + bytes(mutable[long_names.end :]) - ) - target.write_bytes(_shift_zsav_offsets( + replacement_records = { + long_names.start: _extension_record( + mutable, long_names, new_payload, byte_order=byte_order, + ), + } + for record in vls_records: + entries, trailing_tab = parsed_vls[record.start] + rewritten = [ + (vls_replacements.get(short_name, short_name), width) + for short_name, width in entries + ] + vls_payload = _very_long_string_payload(rewritten, trailing_tab=trailing_tab) + replacement_records[record.start] = _extension_record( + mutable, record, vls_payload, byte_order=byte_order, + ) + + chunks: list[bytes] = [] + cursor = 0 + for record in sorted( + (record for record in records if record.start in replacement_records), + key=lambda item: item.start, + ): + chunks.append(bytes(mutable[cursor : record.start])) + chunks.append(replacement_records[record.start]) + cursor = record.end + chunks.append(bytes(mutable[cursor:])) + updated = b"".join(chunks) + updated = _shift_zsav_offsets( updated, original_data_start=terminator.end, delta=len(updated) - len(data), byte_order=byte_order, - )) + ) + _assert_compatible_name_consistency( + updated, + requested=names, + vls_source_names=vls_source_names, + encoding=encoding, + ) + _atomic_write_bytes(target, updated) def write_extended_mrset_labels( @@ -212,18 +280,181 @@ def write_extended_mrset_labels( def _long_name_pairs(payload: bytes, encoding: str) -> list[tuple[str, str]]: + if not payload: + raise RawDictionaryError("Empty SAV long-variable-name record.") pairs: list[tuple[str, str]] = [] - for raw_pair in payload.split(b" "): + short_names: set[str] = set() + long_names: set[str] = set() + for raw_pair in payload.split(b"\t"): try: raw_short, raw_long = raw_pair.split(b"=", maxsplit=1) - short_name = raw_short.decode("ascii") + short_name = _dictionary_short_name(raw_short) long_name = raw_long.decode(encoding) except (UnicodeDecodeError, ValueError) as error: raise RawDictionaryError("Invalid SAV long-variable-name record.") from error + normalized_short = short_name.casefold() + normalized_long = long_name.casefold() + if not long_name or normalized_short in short_names or normalized_long in long_names: + raise RawDictionaryError("Duplicate or empty SAV long-variable-name entry.") + short_names.add(normalized_short) + long_names.add(normalized_long) pairs.append((short_name, long_name)) return pairs +def _very_long_string_entries(payload: bytes) -> tuple[list[tuple[str, bytes]], bool]: + """Parse subtype-14 entries without accepting ambiguous separators or widths.""" + if not payload: + raise RawDictionaryError("Empty SAV very-long-string record.") + raw_entries = payload.split(b"\t") + trailing_tab = raw_entries[-1] == b"" + if trailing_tab: + raw_entries.pop() + if not raw_entries or any(not entry for entry in raw_entries): + raise RawDictionaryError("Invalid SAV very-long-string separators.") + entries: list[tuple[str, bytes]] = [] + for raw_entry in raw_entries: + if not raw_entry.endswith(b"\x00"): + raise RawDictionaryError("SAV very-long-string entry is missing its NUL terminator.") + try: + raw_short, raw_width = raw_entry[:-1].split(b"=", maxsplit=1) + short_name = _dictionary_short_name(raw_short) + if not raw_width or not raw_width.isdigit(): + raise ValueError + width = int(raw_width.decode("ascii")) + except (UnicodeDecodeError, ValueError) as error: + raise RawDictionaryError("Invalid SAV very-long-string entry.") from error + if not 256 <= width <= 32767: + raise RawDictionaryError("SAV very-long-string width is outside 256..32767.") + entries.append((short_name, raw_width)) + return entries, trailing_tab + + +def _very_long_string_payload( + entries: Iterable[tuple[str, bytes]], *, trailing_tab: bool, +) -> bytes: + payload = b"\t".join( + short_name.encode("ascii") + b"=" + width + b"\x00" + for short_name, width in entries + ) + return payload + (b"\t" if trailing_tab else b"") + + +def _dictionary_short_name(raw_name: bytes) -> str: + try: + name = raw_name.decode("ascii") + except UnicodeDecodeError as error: + raise RawDictionaryError("SAV dictionary short name is not ASCII.") from error + allowed_first = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz@#$" + allowed_rest = allowed_first + b"0123456789_." + if not 1 <= len(raw_name) <= 8 or raw_name[:1] not in allowed_first: + raise RawDictionaryError("Invalid SAV dictionary short name.") + if any(character not in allowed_rest for character in raw_name[1:]): + raise RawDictionaryError("Invalid SAV dictionary short name.") + return name + + +def _type_2_names(data: bytes, records: list[_Record], byte_order: str) -> list[str]: + names: list[str] = [] + for record in records: + if record.record_type != 2 or _int(data, record.start + 4, byte_order) < 0: + continue + raw_name = data[record.start + 24 : record.start + 32].rstrip(b" ") + names.append(_dictionary_short_name(raw_name)) + return names + + +def _extension_record( + data: bytes, record: _Record, payload: bytes, *, byte_order: str, +) -> bytes: + header = bytearray(data[record.start : record.start + 16]) + header[8:12] = _pack(1, byte_order) + header[12:16] = _pack(len(payload), byte_order) + return bytes(header) + payload + + +def _assert_compatible_name_consistency( + data: bytes, + *, + requested: dict[str, str], + vls_source_names: set[str], + encoding: str, +) -> None: + byte_order, records = _records(data) + type_2_names = _type_2_names(data, records, byte_order) + normalized_type_2 = [name.casefold() for name in type_2_names] + long_name_records = [ + record for record in records + if record.record_type == 7 and _int(data, record.start + 4, byte_order) == 13 + ] + if len(long_name_records) != 1: + raise RawDictionaryError("Rewritten SAV has an inconsistent long-variable-name record.") + long_name_record = long_name_records[0] + if _int(data, long_name_record.start + 8, byte_order) != 1: + raise RawDictionaryError("Rewritten SAV has invalid long-variable-name dimensions.") + pairs = _long_name_pairs( + data[long_name_record.start + 16 : long_name_record.end], encoding, + ) + pair_by_long = {long_name: short_name for short_name, long_name in pairs} + normalized_pair_names = {short_name.casefold() for short_name, _ in pairs} + for short_name, _ in pairs: + if normalized_type_2.count(short_name.casefold()) != 1: + raise RawDictionaryError("Rewritten type-2 and subtype-13 names are inconsistent.") + + vls_names: list[str] = [] + for record in records: + if record.record_type != 7 or _int(data, record.start + 4, byte_order) != 14: + continue + if _int(data, record.start + 8, byte_order) != 1: + raise RawDictionaryError("Rewritten SAV has invalid very-long-string dimensions.") + entries, _ = _very_long_string_entries(data[record.start + 16 : record.end]) + vls_names.extend(short_name for short_name, _ in entries) + normalized_vls = [name.casefold() for name in vls_names] + if len(normalized_vls) != len(set(normalized_vls)): + raise RawDictionaryError("Rewritten SAV has duplicate very-long-string entries.") + for short_name in normalized_vls: + if short_name not in normalized_pair_names or normalized_type_2.count(short_name) != 1: + raise RawDictionaryError( + "Rewritten type-2/subtype-13/subtype-14 names are inconsistent." + ) + + for source_name, requested_name in requested.items(): + replacement = _validated_compatible_name(requested_name) + if pair_by_long.get(source_name) != replacement: + raise RawDictionaryError("Requested compatible name is absent after dictionary rewrite.") + if normalized_type_2.count(replacement.casefold()) != 1: + raise RawDictionaryError("Requested compatible name is inconsistent with type-2 records.") + if ( + source_name in vls_source_names + and normalized_vls.count(replacement.casefold()) != 1 + ): + raise RawDictionaryError("Requested compatible name is inconsistent with subtype-14.") + + +def _atomic_write_bytes(target: Path, data: bytes) -> None: + """Publish validated dictionary bytes without exposing a partial replacement.""" + mode = stat.S_IMODE(target.stat().st_mode) + temporary: Path | None = None + try: + with NamedTemporaryFile( + mode="wb", + dir=target.parent, + prefix=f".{target.name}.", + suffix=".tmp", + delete=False, + ) as handle: + temporary = Path(handle.name) + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.chmod(temporary, mode) + os.replace(temporary, target) + temporary = None + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + def _validated_compatible_name(value: str) -> str: encoded = str(value).encode("ascii") if not 1 <= len(encoded) <= 8: diff --git a/tests/conformance.py b/tests/conformance.py index f15352f..677f865 100755 --- a/tests/conformance.py +++ b/tests/conformance.py @@ -7,6 +7,8 @@ import pyspssio from pandas.testing import assert_frame_equal +from openstatspec.spss.raw_dictionary import write_compatible_names + def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any]: """Write a real SAV/ZSAV fixture for every pyspssio-supported core feature. @@ -20,6 +22,7 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any] """ destination = Path(destination) long_text = "Õ🙂漢字" * 90 + compatible_text = "\u00d5\U0001f642\u6f22\u5b57" * 30 frame = pd.DataFrame({ "discrete_missing": [1.0, 2.0, 3.0, 4.0], "range_only": [-1.0, 0.0, 1.0, 2.0], @@ -28,6 +31,7 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any] "code": [3.0, 1.0, 2.0, 1.0], "status": ["NA", "DK", "ok", ""], "comment": [long_text, "", "näide", long_text], + "very_long_compatible_name": [compatible_text, "", "tail", compatible_text], "interview_date": [23123.0, 23124.0, 23125.0, 23126.0], "interview_time": [3661.25, 0.0, 86399.5, 12.0], "interview_datetime": [23123.5, 23124.0, 23125.75, 23126.125], @@ -39,11 +43,14 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any] "resp_b": [0.0, 1.0, 1.0, 0.0], }) metadata = { - "var_types": {"status": 8, "comment": 1024}, + "var_types": { + "status": 8, "comment": 1024, "very_long_compatible_name": 360, + }, "var_formats": { "discrete_missing": "F8.0", "range_only": "F8.0", "lowest_range": "F12.1", "highest_range": "F12.1", "code": "F8.0", - "status": "A8", "comment": "A1024", "interview_date": "DATE11", + "status": "A8", "comment": "A1024", "very_long_compatible_name": "A360", + "interview_date": "DATE11", "interview_time": "TIME8", "interview_datetime": "DATETIME20", "interview_dtime": "DTIME10", "formatted_comma": "COMMA12.2", "formatted_dot": "DOT12.2", "formatted_pct": "PCT8.1", @@ -53,7 +60,8 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any] "range_only": "Range-only numeric user missing", "lowest_range": "LOWEST-style missing range", "highest_range": "HIGHEST-style missing range", "code": "Ordered numeric code", "status": "String missing code", - "comment": "Long UTF-8 comment", "interview_date": "SPSS numeric date", + "comment": "Long UTF-8 comment", "very_long_compatible_name": "Custom VLS name", + "interview_date": "SPSS numeric date", "interview_time": "SPSS numeric time", "interview_datetime": "SPSS numeric datetime", "interview_dtime": "SPSS numeric duration", "formatted_comma": "SPSS comma format", "formatted_dot": "SPSS dot format", "formatted_pct": "SPSS percent format", @@ -74,12 +82,14 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any] "var_measure_levels": { "discrete_missing": "scale", "range_only": "scale", "lowest_range": "scale", "highest_range": "scale", "code": "scale", - "status": "nominal", "comment": "nominal", "interview_date": "scale", + "status": "nominal", "comment": "nominal", + "very_long_compatible_name": "nominal", "interview_date": "scale", }, "var_alignments": {column: "left" for column in frame.columns}, "var_column_widths": { "discrete_missing": 8, "range_only": 8, "lowest_range": 12, "highest_range": 12, - "code": 8, "status": 12, "comment": 48, "interview_date": 11, + "code": 8, "status": 12, "comment": 48, "very_long_compatible_name": 48, + "interview_date": 11, "interview_time": 8, "interview_datetime": 20, "interview_dtime": 10, "formatted_comma": 12, "formatted_dot": 12, "formatted_pct": 8, }, @@ -91,6 +101,9 @@ def write_supported_semantics_fixture(destination: str | Path) -> dict[str, Any] "case_weight_var": "code", } pyspssio.write_sav(str(destination), frame, metadata=metadata) + write_compatible_names( + destination, {"very_long_compatible_name": "VLSTEXT"}, encoding="UTF-8", + ) return {"long_text": long_text, "file_label": metadata["file_label"]} @@ -130,7 +143,7 @@ def compare_sav_semantics(source: str | Path, exported: str | Path) -> dict[str, for attribute in ( "encoding", "file_label", "case_weight_var", "file_attributes", "mrsets", "var_types", "var_formats", "var_labels", "var_alignments", "var_column_widths", "var_measure_levels", "var_roles", - "var_value_labels", "var_attributes", + "var_value_labels", "var_attributes", "var_compat_names", ): if source_metadata.get(attribute) != exported_metadata.get(attribute): failures.append(attribute) diff --git a/tests/test_vls_compatible_names.py b/tests/test_vls_compatible_names.py new file mode 100644 index 0000000..ff1f48c --- /dev/null +++ b/tests/test_vls_compatible_names.py @@ -0,0 +1,143 @@ +import sqlite3 + +import pandas as pd +import pyspssio +import pytest + +import openstatspec +from openstatspec.spss import raw_dictionary +from openstatspec.spss import sav as sav_module + + +_SOURCE_NAME = "a05x_very_long_source_name" +_COMPATIBLE_NAME = "A05XM" +_VALUE = "\u00d5\U0001f642\u6f22\u5b57" * 30 + + +def _write_vls_source(path) -> None: + pyspssio.write_sav( + str(path), + pd.DataFrame({ + "before": [1.0, 2.0, 3.0], + _SOURCE_NAME: [_VALUE, "", "tail"], + "after": [4.0, 5.0, 6.0], + }), + metadata={"var_types": {_SOURCE_NAME: 360}}, + ) + + +def _subtype_14_record(path): + data = path.read_bytes() + byte_order, records = raw_dictionary._records(data) + matches = [ + record for record in records + if record.record_type == 7 + and raw_dictionary._int(data, record.start + 4, byte_order) == 14 + ] + assert len(matches) == 1 + return data, byte_order, matches[0] + + +def _subtype_14_entries(path) -> list[tuple[str, int]]: + data, _, record = _subtype_14_record(path) + entries, _ = raw_dictionary._very_long_string_entries( + data[record.start + 16 : record.end], + ) + return [(name, int(width)) for name, width in entries] + + +def _replace_subtype_14_payload(path, payload: bytes) -> None: + data, byte_order, record = _subtype_14_record(path) + header = bytearray(data[record.start : record.start + 16]) + header[8:12] = raw_dictionary._pack(1, byte_order) + header[12:16] = raw_dictionary._pack(len(payload), byte_order) + path.write_bytes( + data[: record.start] + bytes(header) + payload + data[record.end :], + ) + + +@pytest.mark.parametrize("suffix", [".sav", ".zsav"]) +def test_vls_custom_compatible_name_round_trips_as_one_variable(tmp_path, suffix: str) -> None: + source = tmp_path / f"source{suffix}" + destination = tmp_path / f"destination{suffix}" + database = f"sqlite:///{tmp_path / f'vls-{suffix[1:]}.sqlite'}" + _write_vls_source(source) + raw_dictionary.write_compatible_names( + source, {_SOURCE_NAME: _COMPATIBLE_NAME}, encoding="UTF-8", + ) + + imported = openstatspec.import_sav( + source, database_url=database, dataset_id=f"vls-{suffix[1:]}", + ) + assert imported.diagnostics == () + exported = openstatspec.export_sav( + database_url=database, + dataset_id=f"vls-{suffix[1:]}", + destination=destination, + ) + assert exported.diagnostics == () + + metadata = pyspssio.read_metadata(str(destination)) + frame = pyspssio.read_sav(str(destination))[0] + assert metadata["var_names"] == ["before", _SOURCE_NAME, "after"] + assert metadata["var_types"][_SOURCE_NAME] == 360 + assert metadata["var_compat_names"][_SOURCE_NAME] == _COMPATIBLE_NAME + assert list(frame.columns) == ["before", _SOURCE_NAME, "after"] + assert frame[_SOURCE_NAME].tolist() == [_VALUE, "", "tail"] + assert _subtype_14_entries(destination) == [(_COMPATIBLE_NAME, 360)] + + +@pytest.mark.parametrize("damage", ["malformed", "duplicate"]) +def test_vls_rewrite_rejects_invalid_subtype_14_without_publishing(tmp_path, damage: str) -> None: + source = tmp_path / "invalid.sav" + _write_vls_source(source) + data, _, record = _subtype_14_record(source) + payload = data[record.start + 16 : record.end] + if damage == "malformed": + damaged = payload.replace(b"\x00", b"!", 1) + else: + entry = payload.removesuffix(b"\t") + damaged = entry + b"\t" + entry + _replace_subtype_14_payload(source, damaged) + expected = source.read_bytes() + + with pytest.raises(raw_dictionary.RawDictionaryError): + raw_dictionary.write_compatible_names( + source, {_SOURCE_NAME: _COMPATIBLE_NAME}, encoding="UTF-8", + ) + + assert source.read_bytes() == expected + assert list(tmp_path.glob(f".{source.name}.*.tmp")) == [] + + +def test_malformed_vls_export_removes_output_and_records_no_success(tmp_path, monkeypatch) -> None: + source = tmp_path / "source.sav" + destination = tmp_path / "failed.sav" + database_path = tmp_path / "failed.sqlite" + database = f"sqlite:///{database_path}" + _write_vls_source(source) + raw_dictionary.write_compatible_names( + source, {_SOURCE_NAME: _COMPATIBLE_NAME}, encoding="UTF-8", + ) + openstatspec.import_sav(source, database_url=database, dataset_id="failed-vls") + real_write = sav_module.write_compatible_names + + def malformed_write(path, names, *, encoding): + data, _, record = _subtype_14_record(path) + payload = data[record.start + 16 : record.end].replace(b"\x00", b"!", 1) + _replace_subtype_14_payload(path, payload) + real_write(path, names, encoding=encoding) + + monkeypatch.setattr(sav_module, "write_compatible_names", malformed_write) + with pytest.raises(raw_dictionary.RawDictionaryError): + openstatspec.export_sav( + database_url=database, + dataset_id="failed-vls", + destination=destination, + ) + + assert not destination.exists() + connection = sqlite3.connect(database_path) + assert connection.execute( + "select direction, status from operation_catalog order by created_at" + ).fetchall() == [("import", "succeeded")]