diff --git a/src/pt_snap_cli/core/snapshot_import_backend.py b/src/pt_snap_cli/core/snapshot_import_backend.py index 2317622..546caee 100644 --- a/src/pt_snap_cli/core/snapshot_import_backend.py +++ b/src/pt_snap_cli/core/snapshot_import_backend.py @@ -38,7 +38,15 @@ def dump_to_db( ) except UnsafePickleError as exc: raise ImportExecutionError(f"Snapshot pickle rejected: {exc}") from exc - except (OSError, sqlite3.DatabaseError, pickle.UnpicklingError, ValueError) as exc: + except ( + OSError, + sqlite3.DatabaseError, + pickle.UnpicklingError, + ValueError, + OverflowError, + ) as exc: + # SQLite rejects out-of-range integer bindings with OverflowError, + # rather than DatabaseError. Keep the original numeric failure. raise ImportExecutionError(f"Snapshot import backend failed: {exc}") from exc if not ok: diff --git a/src/pt_snap_cli/query/config.py b/src/pt_snap_cli/query/config.py index 6a343e5..4dd4b22 100644 --- a/src/pt_snap_cli/query/config.py +++ b/src/pt_snap_cli/query/config.py @@ -204,8 +204,16 @@ def from_dict(cls, data: dict[str, Any], default_category: str | None = None) -> Returns: QueryTemplate instance. """ + data = _require_mapping(data, "template") + if "name" in data: + _require_name(data["name"], "template") parameters = {} - for name, param_data in data.get("parameters", {}).items(): + parameter_data = _require_mapping(data.get("parameters", {}), "parameters") + for name, param_data in parameter_data.items(): + _require_name(name, "parameter") + param_data = _require_mapping(param_data, f"parameter '{name}'") + if not isinstance(param_data.get("type", "str"), str): + raise ValueError(f"Parameter '{name}' type must be a string") parameters[name] = QueryParameter( name=name, type=param_data.get("type", "str"), @@ -218,6 +226,8 @@ def from_dict(cls, data: dict[str, Any], default_category: str | None = None) -> template_name = data.get("name", "") query_variants = _parse_query_variants(template_name, data.get("query_variants")) explicit_query = data.get("query") + if explicit_query is not None and not isinstance(explicit_query, str): + raise ValueError(f"query for '{template_name}' must be a SQL string") if query_variants: if isinstance(explicit_query, str) and explicit_query.strip(): raise ValueError( @@ -245,6 +255,17 @@ def from_dict(cls, data: dict[str, Any], default_category: str | None = None) -> ) +def _require_mapping(raw: Any, label: str) -> dict[str, Any]: + if not isinstance(raw, dict): + raise ValueError(f"{label} must be a mapping") + return raw + + +def _require_name(raw: Any, label: str) -> None: + if not isinstance(raw, str) or not raw.strip(): + raise ValueError(f"{label} name must be a non-empty string") + + def _parse_semantics_version(template_name: str, raw: Any) -> int | None: if raw is None: return None @@ -408,23 +429,15 @@ def load_yaml(cls, path: str | Path, default_category: str | None = None) -> Que Raises: FileNotFoundError: If file does not exist. yaml.YAMLError: If YAML is invalid. + ValueError: If the configuration structure or template metadata is invalid. """ path = Path(path) if not path.exists(): raise FileNotFoundError(f"Config file not found: {path}") with open(path, encoding="utf-8") as f: - data = yaml.safe_load(f) or {} - - queries = {} - for name, query_data in data.get("queries", {}).items(): - query_data["name"] = name - queries[name] = QueryTemplate.from_dict(query_data, default_category=default_category) - - return cls( - version=data.get("version", "1.0"), - queries=queries, - ) + data = yaml.safe_load(f) + return cls._from_yaml_data(data, default_category=default_category) @classmethod def load_yaml_from_string(cls, content: str) -> QueryConfig: @@ -435,13 +448,29 @@ def load_yaml_from_string(cls, content: str) -> QueryConfig: Returns: QueryConfig instance. + + Raises: + yaml.YAMLError: If YAML is invalid. + ValueError: If the configuration structure or template metadata is invalid. """ - data = yaml.safe_load(content) or {} + return cls._from_yaml_data(yaml.safe_load(content)) + + @classmethod + def _from_yaml_data(cls, raw: Any, default_category: str | None = None) -> QueryConfig: + # An empty YAML document remains an empty configuration. Other falsy + # values ([], false, 0, "") are not mappings and must not silently pass. + data = _require_mapping({} if raw is None else raw, "configuration") + query_definitions = _require_mapping(data.get("queries", {}), "queries") queries = {} - for name, query_data in data.get("queries", {}).items(): - query_data["name"] = name - queries[name] = QueryTemplate.from_dict(query_data) + for name, query_data in query_definitions.items(): + _require_name(name, "query") + query_data = _require_mapping(query_data, f"query '{name}'") + # Do not overwrite caller data or shared YAML aliases with the + # enclosing name. Each query gets its own top-level mapping. + queries[name] = QueryTemplate.from_dict( + {**query_data, "name": name}, default_category=default_category + ) return cls( version=data.get("version", "1.0"), diff --git a/src/pt_snap_cli/snapshot/PROVENANCE.md b/src/pt_snap_cli/snapshot/PROVENANCE.md index 7b2ebff..4c8b7cd 100644 --- a/src/pt_snap_cli/snapshot/PROVENANCE.md +++ b/src/pt_snap_cli/snapshot/PROVENANCE.md @@ -321,3 +321,21 @@ JSON tuple-to-array normalization; unsupported deep evidence is omitted while the original text stack and snapshot import remain accepted. No allocator, event identity, source mapping, license or executable fixture changed. No external implementation was incorporated. + +## 2026-11-01 — Snapshot input error normalization (issue #280) + +First-party changes in `base/entities.py` validate the mappings, sequences and +required keys consumed by entity construction, finite allocator numeric fields, +string-valued block states, and nonempty trace action strings. Producer block +state strings remain unchanged rather than being restricted to replay constants; +the existing SQL value mapping and missing-column default are unchanged. Failures +of existing trace integer conversion become `ValueError` with the cause preserved. +`tools/adaptors/snapshot2db.py` +checks the root/device-trace container before device selection. Numeric strings +in traces, optional defaults, sentinels, tuple containers, event IDs and successful +replay behavior remain unchanged. The core import adapter also normalizes SQLite +integer-binding overflow at its existing domain-error boundary, without imposing +a new representation-wide numeric range. Generated regressions use trusted +in-memory representations behind the loader seam, not new executable pickle +fixtures. No parser/layout, unknown-field policy or upstream-code change is made; +source mappings, attribution and license records are unchanged. diff --git a/src/pt_snap_cli/snapshot/base/entities.py b/src/pt_snap_cli/snapshot/base/entities.py index c057f70..dd1e284 100644 --- a/src/pt_snap_cli/snapshot/base/entities.py +++ b/src/pt_snap_cli/snapshot/base/entities.py @@ -1,7 +1,32 @@ +import math from dataclasses import dataclass, field from typing import Any, Literal, cast +def _validate_mapping(raw: Any, label: str, required: tuple[str, ...] = ()) -> None: + if not isinstance(raw, dict): + raise ValueError(f"{label} must be a mapping") + for name in required: + if name not in raw: + raise ValueError(f"{label} is missing required field '{name}'") + + +def _validate_sequence(raw: Any, label: str) -> None: + # Tuples already work for trusted in-memory representations; retain that + # compatibility rather than requiring a particular serialized container. + if not isinstance(raw, (list, tuple)): + raise ValueError(f"{label} must be a list or tuple") + + +def _validate_numbers(raw: dict[str, Any], label: str, names: tuple[str, ...]) -> None: + for name in names: + value = raw[name] + if not isinstance(value, (int, float)) or ( + isinstance(value, float) and not math.isfinite(value) + ): + raise ValueError(f"{label} field '{name}' must be a finite number") + + class Frame: filename: str = "" line: int = -1 @@ -11,6 +36,7 @@ class Frame: @classmethod def from_dict(cls, frame_dict: dict[str, Any]): + _validate_mapping(frame_dict, "Frame", ("filename", "line", "name")) frame = cls() frame.filename = frame_dict["filename"] frame.line = frame_dict["line"] @@ -39,10 +65,17 @@ def _format_callstack(frames: list[Frame] | list[dict[str, Any]]) -> str: def _validate_raw_frames(frames: list[dict[str, Any]]) -> None: + _validate_sequence(frames, "Frames") for frame in frames: - _ = frame["filename"] - _ = frame["line"] - _ = frame["name"] + _validate_mapping(frame, "Frame", ("filename", "line", "name")) + + +def _trace_integer(trace_dict: dict[str, Any], name: str, default: int) -> int: + """Retain integer conversion while rejecting malformed numeric metadata.""" + try: + return int(trace_dict.get(name, default)) + except (TypeError, ValueError, OverflowError) as exc: + raise ValueError(f"Trace field '{name}' must be integer-compatible") from exc @dataclass @@ -82,19 +115,24 @@ class TraceEntry: @classmethod def from_dict(cls, trace_dict: dict[str, Any], _raw_frames: bool = False): + _validate_mapping(trace_dict, "Trace entry") + action = trace_dict.get("action") + if not isinstance(action, str) or not action.strip(): + raise ValueError("Trace field 'action' must be a non-empty string") frame_dicts = trace_dict.get("frames", []) + _validate_sequence(frame_dicts, "Frames") if _raw_frames: _validate_raw_frames(frame_dicts) trace_entry = cls( - action=trace_dict["action"], - addr=int(trace_dict.get("addr", -1)), - size=int(trace_dict.get("size", 0)), - stream=int(trace_dict.get("stream", 0)), - device_free=int(trace_dict.get("device_free", -1)), + action=action, + addr=_trace_integer(trace_dict, "addr", -1), + size=_trace_integer(trace_dict, "size", 0), + stream=_trace_integer(trace_dict, "stream", 0), + device_free=_trace_integer(trace_dict, "device_free", -1), _origin=trace_dict, frames=[] if _raw_frames else [Frame.from_dict(frame) for frame in frame_dicts], _raw_frames=frame_dicts if _raw_frames else None, - idx=int(trace_dict["id"]) if "id" in trace_dict else -1, + idx=_trace_integer(trace_dict, "id", -1), ) return trace_entry @@ -146,11 +184,9 @@ class Block: requested_size: int = 0 # size requested during malloc, may be smaller than # size due to rounding address: int = -1 - state: Literal[ - "active_allocated", # used by a tensor - "active_pending_free", # waiting for another stream to finish using this, then it will become free - "inactive", - ] = BlockState.INACTIVE # free for reuse + # Preserve producer state strings; BlockState names the replay states, not + # a closed input enum (PyTorch also emits "active_awaiting_free"). + state: str = BlockState.INACTIVE # free for reuse frames: list[Frame] = field( default_factory=list ) # stack trace from where the allocation occurred @@ -163,14 +199,20 @@ class Block: @classmethod def from_dict(cls, block_dict: dict[str, Any], _raw_frames: bool = False): + _validate_mapping(block_dict, "Block", ("size", "requested_size", "address", "state")) + _validate_numbers(block_dict, "Block", ("size", "requested_size", "address")) + state = block_dict["state"] + if not isinstance(state, str): + raise ValueError("Block field 'state' must be a string") frame_dicts = block_dict.get("frames", []) + _validate_sequence(frame_dicts, "Frames") if _raw_frames: _validate_raw_frames(frame_dicts) block = cls( size=block_dict["size"], requested_size=block_dict["requested_size"], address=block_dict["address"], - state=block_dict["state"], + state=state, frames=[] if _raw_frames else [Frame.from_dict(frame) for frame in frame_dicts], _raw_frames=frame_dicts if _raw_frames else None, ) @@ -231,7 +273,12 @@ def from_dict( ignore_inactive_blocks: bool = False, _raw_frames: bool = False, ): + numeric_fields = ("address", "total_size", "stream", "allocated_size", "active_size") + _validate_mapping(segment_dict, "Segment", (*numeric_fields, "segment_type", "blocks")) + _validate_numbers(segment_dict, "Segment", numeric_fields) + _validate_sequence(segment_dict["blocks"], "Segment blocks") frame_dicts = segment_dict.get("frames", []) + _validate_sequence(frame_dicts, "Frames") if _raw_frames: _validate_raw_frames(frame_dicts) segment = cls( @@ -248,6 +295,7 @@ def from_dict( is_expandable=segment_dict.get("is_expandable", False), ) for block in segment_dict["blocks"]: + _validate_mapping(block, "Block", ("state",)) if ignore_inactive_blocks and block["state"] == BlockState.INACTIVE: continue _block = Block.from_dict(block, _raw_frames=_raw_frames) @@ -328,9 +376,13 @@ def from_dict( ignore_inactive_blocks: bool = False, _raw_frames: bool = False, ): + _validate_mapping(snapshot_dict, "Snapshot") segments_dict = snapshot_dict.get("segments", []) device_traces = snapshot_dict.get("device_traces", []) + _validate_sequence(segments_dict, "Snapshot segments") + _validate_sequence(device_traces, "Snapshot device_traces") device_trace_list = device_traces[device] if 0 <= device < len(device_traces) else [] + _validate_sequence(device_trace_list, "Device traces") snapshot = cls() snapshot.segments = [] snapshot.trace_entries = [] @@ -339,6 +391,7 @@ def from_dict( snapshot.total_activated = 0 # 读取dump_snapshot时内存状态 for segment_dict in segments_dict: + _validate_mapping(segment_dict, "Segment") # 当segment原始数据中缺少device字段明确指向其所属设备时，默认其归属为device0 # 此时如果from_dict指定device为0或未指定而缺省为0，则未知归属的device也会纳入分析 if segment_dict.get("device", 0) != device: diff --git a/src/pt_snap_cli/snapshot/tools/adaptors/snapshot2db.py b/src/pt_snap_cli/snapshot/tools/adaptors/snapshot2db.py index 1f3bcad..3f616a8 100644 --- a/src/pt_snap_cli/snapshot/tools/adaptors/snapshot2db.py +++ b/src/pt_snap_cli/snapshot/tools/adaptors/snapshot2db.py @@ -191,8 +191,15 @@ def dump(pickle_file: str | Path, dump_file: str | Path, device: int | None = No except Exception as e: dump_logger.error(f"Failed to load pickle file: {e}") return False + if not isinstance(data, dict): + raise ValueError("Snapshot must be a mapping") # Snapshot format: one list of trace dictionaries per device. + if not isinstance(data.get("device_traces", []), (list, tuple)): + raise ValueError("Snapshot device_traces must be a list or tuple") device_traces = cast(list[list[object]], data.get("device_traces", [])) + for index, traces in enumerate(device_traces): + if not isinstance(traces, (list, tuple)): + raise ValueError(f"Snapshot device_traces[{index}] must be a list or tuple") # 当指定device为空时dump所有记录了跟踪事件的device，否则仅dump指定device need_dump_devices = [device for device in range(len(device_traces)) if device_traces[device]] dump_logger.info(f"Recognized have trace events devices {need_dump_devices}.") diff --git a/tests/core/test_snapshot_input_errors_properties.py b/tests/core/test_snapshot_input_errors_properties.py new file mode 100644 index 0000000..abddb9c --- /dev/null +++ b/tests/core/test_snapshot_input_errors_properties.py @@ -0,0 +1,544 @@ +"""Trace input failures through real replay/SQLite; never deserialize a pickle.""" + +from __future__ import annotations + +import copy +import importlib +import sqlite3 +from contextlib import closing +from pathlib import Path +from tempfile import TemporaryDirectory + +import pytest +from hypothesis import example, given, settings +from hypothesis import strategies as st + +from pt_snap_cli.core.errors import ImportExecutionError +from pt_snap_cli.core.snapshot_import_backend import SnapshotImportBackend +from pt_snap_cli.snapshot.base import Block, DeviceSnapshot, TraceEntry +from pt_snap_cli.snapshot.tools.adaptors.database import SnapshotDb, block2record + +PROPERTY_SETTINGS = settings( + max_examples=25, derandomize=True, database=None, deadline=None, print_blob=True +) +SCALARS = st.one_of(st.none(), st.booleans(), st.integers(-100, 100), st.text(max_size=12)) +BAD_NUMBERS = st.one_of( + st.none(), + st.lists(SCALARS, max_size=3), + st.dictionaries(st.text(max_size=8), SCALARS, max_size=3), + st.sampled_from([float("nan"), float("inf"), float("-inf")]), +) +SOURCE_BYTES = b"inert placeholder; only the patched in-memory loader may be used" + + +def _valid_representation() -> dict[str, object]: + return { + "segments": [ + { + "address": 4096, + "total_size": 16, + "stream": 0, + "segment_type": "small", + "allocated_size": 16, + "active_size": 16, + "frames": [], + "blocks": [ + { + "size": 16, + "requested_size": 8, + "address": 4096, + "state": "active_allocated", + "frames": [], + } + ], + } + ], + "device_traces": [ + [ + { + "action": "oom", + "size": 1, + "frames": [{"filename": "generated.py", "line": 1, "name": "run"}], + } + ] + ], + } + + +def _assert_representation_rejected( + representation: object, + cause: type[Exception], + existing: bool, + *, + device: int | None = 0, +) -> Exception: + before = copy.deepcopy(representation) + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + source = root / "trusted-in-memory.pkl" + source.write_bytes(SOURCE_BYTES) + output = root / "output" + output.mkdir() + backend = SnapshotImportBackend() + destination = backend.target_db_path(source, output) + previous = b"existing destination must remain byte-for-byte unchanged" + if existing: + destination.write_bytes(previous) + loads = [] + + def load_representation(path: Path) -> object: + assert path == source + loads.append(path) + return representation + + # Older lazy-import tests evict snapshot2db from sys.modules. Always + # patch the current module, not a collection-time reference. + snapshot2db = importlib.import_module("pt_snap_cli.snapshot.tools.adaptors.snapshot2db") + monkeypatch.setattr(snapshot2db, "load_snapshot_representation", load_representation) + with pytest.raises(ImportExecutionError, match="Snapshot import backend failed") as error: + backend.dump_to_db(source, output, device=device) + underlying = error.value.__cause__ + assert isinstance(underlying, cause) + assert loads == [source] + assert representation == before + assert source.read_bytes() == SOURCE_BYTES + if existing: + assert destination.read_bytes() == previous + else: + assert not destination.exists() + assert list(output.iterdir()) == ([destination] if existing else []) + return underlying + + +def _assert_rejected(event: object, cause: type[Exception], existing: bool) -> Exception: + return _assert_representation_rejected( + {"segments": [], "device_traces": [[event]]}, cause, existing + ) + + +@PROPERTY_SETTINGS +@given(size=st.integers(0, 1 << 20), existing=st.booleans()) +def test_generated_missing_action_rejects_without_publication(size: int, existing: bool) -> None: + cause = _assert_rejected({"size": size, "frames": []}, ValueError, existing) + assert "Trace field 'action'" in str(cause) + + +@PROPERTY_SETTINGS +@given( + action=st.one_of( + st.none(), + st.booleans(), + st.integers(-(1 << 65), 1 << 65), + st.lists(SCALARS, max_size=3), + st.dictionaries(st.text(max_size=8), SCALARS, max_size=3), + st.sampled_from(["", " ", "\t\n"]), + ), + existing=st.booleans(), +) +def test_generated_malformed_action_rejects_without_publication( + action: object, existing: bool +) -> None: + cause = _assert_rejected({"action": action, "frames": []}, ValueError, existing) + assert "Trace field 'action'" in str(cause) + + +@PROPERTY_SETTINGS +@given(event=st.one_of(SCALARS, st.lists(SCALARS, max_size=3)), existing=st.booleans()) +def test_generated_non_mapping_trace_rejects_without_publication( + event: object, existing: bool +) -> None: + cause = _assert_rejected(event, ValueError, existing) + assert "Trace entry must be a mapping" in str(cause) + + +@pytest.mark.parametrize("field", ["addr", "size", "stream", "device_free", "id"]) +@PROPERTY_SETTINGS +@given(value=BAD_NUMBERS, existing=st.booleans()) +def test_generated_wrong_numeric_types_and_nonfinite_values_retain_conversion_cause( + field: str, value: object, existing: bool +) -> None: + event = {"action": "oom", "size": 1, "frames": [], field: value} + cause = _assert_rejected(event, ValueError, existing) + assert f"Trace field '{field}'" in str(cause) + assert isinstance(cause.__cause__, (TypeError, ValueError, OverflowError)) + + +@pytest.mark.parametrize("field", ["addr", "size", "stream", "id"]) +@PROPERTY_SETTINGS +@given( + extra=st.integers(0, 1 << 128), + negative=st.booleans(), + as_text=st.booleans(), + existing=st.booleans(), +) +def test_generated_sqlite_integer_overflow_is_a_controlled_import_error( + field: str, extra: int, negative: bool, as_text: bool, existing: bool +) -> None: + # The existing SQLite INTEGER boundary, not a new snapshot-wide upper + # bound: device_free is deliberately absent because it is not stored here. + value = -(1 << 63) - 1 - extra if negative else (1 << 63) + extra + event = {"action": "oom", "size": 1, "frames": [], field: str(value) if as_text else value} + _assert_rejected(event, OverflowError, existing) + + +@pytest.mark.parametrize("device", [None, 0]) +@PROPERTY_SETTINGS +@example(size=0, stream=0, event_id=0, as_text=False) +@example(size=0, stream=0, event_id=-1, as_text=True) +@example(size=(1 << 63) - 1, stream=(1 << 63) - 1, event_id=(1 << 63) - 1, as_text=True) +@given( + size=st.integers(0, (1 << 63) - 1), + stream=st.integers(0, (1 << 63) - 1), + event_id=st.integers(0, (1 << 63) - 1), + as_text=st.booleans(), +) +def test_generated_valid_numeric_strings_and_sentinels_still_import( + device: int | None, size: int, stream: int, event_id: int, as_text: bool +) -> None: + event = { + "action": "oom", + "size": str(size) if as_text else size, + "stream": str(stream) if as_text else stream, + "id": str(event_id) if as_text else event_id, + } + # Valid empty list/tuple entries remain empty devices, not malformed data. + representation = {"segments": [], "device_traces": [[event], [], ()]} + before = copy.deepcopy(representation) + trace = TraceEntry.from_dict(event) + assert (trace.addr, trace.device_free, trace.frames) == (-1, -1, []) + assert (trace.size, trace.stream, trace.idx) == (size, stream, event_id) + assert TraceEntry.from_dict({**event, "device_free": str(1 << 128)}).device_free == 1 << 128 + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + source = root / "trusted-in-memory.pkl" + source.write_bytes(SOURCE_BYTES) + output = root / "output" + loads = [] + + def load_representation(path: Path) -> dict[str, object]: + assert path == source + loads.append(path) + return representation + + snapshot2db = importlib.import_module("pt_snap_cli.snapshot.tools.adaptors.snapshot2db") + monkeypatch.setattr(snapshot2db, "load_snapshot_representation", load_representation) + destination = SnapshotImportBackend().dump_to_db(source, output, device=device) + with closing(sqlite3.connect(destination)) as connection: + assert connection.execute( + "SELECT id, address, size, stream FROM trace_entry_0" + ).fetchall() == [(event_id, -1, size, stream)] + assert loads == [source] + assert representation == before + assert source.read_bytes() == SOURCE_BYTES + assert list(output.iterdir()) == [destination] + + +@pytest.mark.parametrize("error_type", [KeyError, TypeError, AttributeError]) +def test_unrelated_backend_programming_errors_are_not_reclassified(error_type) -> None: + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + source = root / "trusted-in-memory.pkl" + source.write_bytes(SOURCE_BYTES) + output = root / "output" + output.mkdir() + backend = SnapshotImportBackend() + destination = backend.target_db_path(source, output) + destination.write_bytes(b"previous database") + failure = error_type("unrelated implementation error") + + def fail(**kwargs): + raise failure + + monkeypatch.setattr(backend, "_load_run_dump_to_db", lambda: fail) + with pytest.raises(error_type) as error: + backend.dump_to_db(source, output) + assert error.value is failure + assert destination.read_bytes() == b"previous database" + assert source.read_bytes() == SOURCE_BYTES + assert list(output.iterdir()) == [destination] + + +@pytest.mark.parametrize("location", ["root", "segment", "block", "frame"]) +@PROPERTY_SETTINGS +@given(value=st.one_of(SCALARS, st.lists(SCALARS, max_size=3)), existing=st.booleans()) +def test_generated_non_mapping_snapshot_records_are_controlled( + location: str, value: object, existing: bool +) -> None: + representation = _valid_representation() + if location == "root": + representation = value + elif location == "segment": + representation["segments"][0] = value + elif location == "block": + representation["segments"][0]["blocks"][0] = value + else: + representation["device_traces"][0][0]["frames"][0] = value + cause = _assert_representation_rejected(representation, ValueError, existing) + assert "must be a mapping" in str(cause) + + +@pytest.mark.parametrize( + "location", + ["device_traces", "segments", "blocks", "trace_frames", "segment_frames", "block_frames"], +) +@PROPERTY_SETTINGS +@given( + value=st.one_of(SCALARS, st.dictionaries(st.text(max_size=8), SCALARS, max_size=3)), + existing=st.booleans(), +) +def test_generated_non_sequence_snapshot_containers_are_controlled( + location: str, value: object, existing: bool +) -> None: + representation = _valid_representation() + if location in ("device_traces", "segments"): + representation[location] = value + elif location == "blocks": + representation["segments"][0]["blocks"] = value + elif location == "segment_frames": + representation["segments"][0]["frames"] = value + elif location == "block_frames": + representation["segments"][0]["blocks"][0]["frames"] = value + else: + representation["device_traces"][0][0]["frames"] = value + cause = _assert_representation_rejected(representation, ValueError, existing) + assert "must be a list or tuple" in str(cause) + + +@PROPERTY_SETTINGS +@given(value=SCALARS, existing=st.booleans()) +def test_generated_mapping_instead_of_device_trace_sequence_is_controlled( + value: object, existing: bool +) -> None: + representation = {"segments": [], "device_traces": [{"invalid": value}]} + cause = _assert_representation_rejected(representation, ValueError, existing) + assert "Snapshot device_traces[0] must be a list or tuple" in str(cause) + + +@pytest.mark.parametrize("device", [None, 0]) +@pytest.mark.parametrize("position", [0, 1]) +@PROPERTY_SETTINGS +@example(value=None, existing=False) +@example(value={}, existing=True) +@example(value=False, existing=True) +@example(value=0, existing=False) +@example(value="", existing=False) +@given( + value=st.one_of(SCALARS, st.dictionaries(st.text(max_size=8), SCALARS, max_size=3)), + existing=st.booleans(), +) +def test_generated_invalid_inner_device_traces_reject_before_device_selection( + device: int | None, position: int, value: object, existing: bool +) -> None: + # Truthiness must not let malformed empty values masquerade as an empty + # device, either when dumping every device or selecting device zero. + cause = _assert_representation_rejected( + {"segments": [], "device_traces": [[] for _ in range(position)] + [value]}, + ValueError, + existing, + device=device, + ) + assert f"Snapshot device_traces[{position}] must be a list or tuple" in str(cause) + + +@pytest.mark.parametrize( + ("entity", "missing"), + [("frame", key) for key in ("filename", "line", "name")] + + [("block", key) for key in ("size", "requested_size", "address", "state")] + + [ + ("segment", key) + for key in ( + "address", + "total_size", + "stream", + "allocated_size", + "active_size", + "segment_type", + "blocks", + ) + ], +) +@PROPERTY_SETTINGS +@given(line=st.integers(0, 1000), existing=st.booleans()) +def test_generated_missing_required_snapshot_fields_are_controlled( + entity: str, missing: str, line: int, existing: bool +) -> None: + representation = _valid_representation() + frame = representation["device_traces"][0][0]["frames"][0] + frame["line"] = line + record = { + "frame": frame, + "block": representation["segments"][0]["blocks"][0], + "segment": representation["segments"][0], + }[entity] + del record[missing] + cause = _assert_representation_rejected(representation, ValueError, existing) + assert f"missing required field '{missing}'" in str(cause) + + +@pytest.mark.parametrize( + ("entity", "field"), + [("block", key) for key in ("size", "requested_size", "address")] + + [ + ("segment", key) + for key in ("address", "total_size", "stream", "allocated_size", "active_size") + ], +) +@PROPERTY_SETTINGS +@given(value=st.one_of(BAD_NUMBERS, st.text(max_size=16)), existing=st.booleans()) +def test_generated_non_numeric_allocator_metadata_is_controlled( + entity: str, field: str, value: object, existing: bool +) -> None: + representation = _valid_representation() + record = representation["segments"][0] + if entity == "block": + record = record["blocks"][0] + record[field] = value + cause = _assert_representation_rejected(representation, ValueError, existing) + assert f"field '{field}' must be a finite number" in str(cause) + + +@PROPERTY_SETTINGS +@given( + value=st.one_of( + st.none(), + st.booleans(), + st.integers(-100, 100), + st.lists(SCALARS, max_size=3), + st.dictionaries(st.text(max_size=8), SCALARS, max_size=3), + ), + existing=st.booleans(), +) +def test_generated_non_string_block_state_is_controlled(value: object, existing: bool) -> None: + representation = _valid_representation() + representation["segments"][0]["blocks"][0]["state"] = value + cause = _assert_representation_rejected(representation, ValueError, existing) + assert "Block field 'state' must be a string" in str(cause) + + +@pytest.mark.parametrize("raw_frames", [False, True]) +@PROPERTY_SETTINGS +@example(state="active_allocated", size=16) +@example(state="active_pending_free", size=16) +@example(state="inactive", size=16) +@example(state="active_awaiting_free", size=16) +@example(state="unknown", size=16) +@example(state="", size=16) +@given(state=st.text(max_size=24), size=st.integers(1, 4096)) +def test_generated_string_block_states_preserve_eager_and_raw_inputs( + state: str, raw_frames: bool, size: int +) -> None: + block_data = { + "size": size, + "requested_size": size // 2, + "address": 4096, + "state": state, + "frames": [{"filename": "generated.py", "line": 1, "name": "run"}], + } + before = copy.deepcopy(block_data) + block = Block.from_dict(block_data, _raw_frames=raw_frames) + assert block.state == state + assert block.to_dict() == before + assert block_data == before + + +@pytest.mark.parametrize( + ("state", "stored_state"), + [ + ("active_allocated", 1), + ("active_pending_free", 0), + ("inactive", -1), + ("active_awaiting_free", "active_awaiting_free"), + ("unknown", "unknown"), + ("", ""), + ], +) +@PROPERTY_SETTINGS +@given(size=st.integers(1, 4096)) +def test_generated_block_states_preserve_sql_mapping_and_missing_column_default( + state: str, stored_state: str | int, size: int +) -> None: + block = Block.from_dict({"size": size, "requested_size": size, "address": 4096, "state": state}) + block.alloc_event_idx = 0 + record = block2record(block) + assert record["state"] == state + before = copy.deepcopy(record) + with TemporaryDirectory() as directory: + database = SnapshotDb(str(Path(directory) / "states.db")) + try: + database.create_block_table() + table = database.get_block_table() + table.insert_record(database.conn, record) + # Unknown strings pass through the existing map. The SQL default + # 99 applies only when the column is omitted, not to unknown values. + table.insert_record(database.conn, {"id": 1}) + assert database.conn.execute("SELECT state FROM block_0 ORDER BY id").fetchall() == [ + (stored_state,), + (99,), + ] + finally: + database.conn.close() + assert record == before + assert block.state == state + + +@pytest.mark.parametrize("state", ["active_awaiting_free", "unknown", ""]) +@PROPERTY_SETTINGS +@given(size=st.integers(1, 4096)) +def test_generated_producer_block_states_still_import_without_normalization( + state: str, size: int +) -> None: + representation = _valid_representation() + segment = representation["segments"][0] + segment.update(total_size=size, allocated_size=size, active_size=size) + segment["blocks"][0].update(state=state, size=size, requested_size=size) + before = copy.deepcopy(representation) + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + source = root / "trusted-in-memory.pkl" + source.write_bytes(SOURCE_BYTES) + output = root / "output" + loads = [] + + def load_representation(path: Path) -> dict[str, object]: + assert path == source + loads.append(path) + return representation + + snapshot2db = importlib.import_module("pt_snap_cli.snapshot.tools.adaptors.snapshot2db") + monkeypatch.setattr(snapshot2db, "load_snapshot_representation", load_representation) + destination = SnapshotImportBackend().dump_to_db(source, output, device=0) + with closing(sqlite3.connect(destination)) as connection: + assert connection.execute("SELECT state FROM block_0").fetchall() == [(state,)] + assert loads == [source] + assert representation == before + assert source.read_bytes() == SOURCE_BYTES + assert list(output.iterdir()) == [destination] + + +@pytest.mark.parametrize("raw_frames", [False, True]) +@PROPERTY_SETTINGS +@example(size=16.0, use_tuples=False) +@given(size=st.integers(1, 4096), use_tuples=st.booleans()) +def test_generated_valid_entity_containers_preserve_input_and_defaults( + raw_frames: bool, size: int | float, use_tuples: bool +) -> None: + representation = _valid_representation() + segment = representation["segments"][0] + segment.update(total_size=size, allocated_size=size, active_size=size) + segment["blocks"][0].update(size=size, requested_size=size) + if use_tuples: + segment["blocks"] = tuple(segment["blocks"]) + representation["segments"] = tuple(representation["segments"]) + event = representation["device_traces"][0][0] + event["frames"] = tuple(event["frames"]) + representation["device_traces"] = ((event,),) + before = copy.deepcopy(representation) + snapshot = DeviceSnapshot.from_dict(representation, 0, _raw_frames=raw_frames) + assert snapshot.total_reserved == snapshot.total_allocated == snapshot.total_activated == size + assert snapshot.segments[0].blocks[0].size == size + assert snapshot.segments[0].device == 0 + assert snapshot.segments[0].is_expandable is False + trace = snapshot.trace_entries[0] + assert (trace.idx, trace.addr, trace.stream, trace.device_free) == (0, -1, 0, -1) + assert trace.get_callstack() == "generated.py:1 run" + assert representation == before diff --git a/tests/query/test_input_structure_properties.py b/tests/query/test_input_structure_properties.py new file mode 100644 index 0000000..a05d49f --- /dev/null +++ b/tests/query/test_input_structure_properties.py @@ -0,0 +1,215 @@ +"""Bounded loader shape regressions, not a new closed YAML schema.""" + +from __future__ import annotations + +import copy +from pathlib import Path +from tempfile import TemporaryDirectory + +import pytest +import yaml +from hypothesis import given, settings +from hypothesis import strategies as st + +from pt_snap_cli.query import config as config_module +from pt_snap_cli.query.config import QueryConfig, QueryTemplate + +PROPERTY_SETTINGS = settings( + max_examples=25, derandomize=True, database=None, deadline=None, print_blob=True +) +NAMES = st.text(alphabet="abcdefghijklmnopqrstuvwxyz", min_size=1, max_size=16) +SCALARS = st.one_of( + st.none(), st.booleans(), st.integers(-(1 << 65), 1 << 65), st.text(max_size=20) +) +NON_MAPPINGS = st.one_of(SCALARS, st.lists(SCALARS, max_size=4)) +INVALID_NAMES = st.one_of( + st.none(), st.booleans(), st.integers(-(1 << 65), 1 << 65), st.sampled_from(["", " "]) +) + + +def _assert_both_loaders_reject(content: str, label: str) -> None: + with pytest.raises(ValueError, match=f"{label} must be a mapping"): + QueryConfig.load_yaml_from_string(content) + with TemporaryDirectory() as directory: + source = Path(directory) / "template.yaml" + source.write_text(content, encoding="utf-8") + before = source.read_bytes() + with pytest.raises(ValueError, match=f"{label} must be a mapping"): + QueryConfig.load_yaml(source) + assert source.read_bytes() == before + assert list(source.parent.iterdir()) == [source] + + +@PROPERTY_SETTINGS +@given( + value=st.one_of( + st.booleans(), + st.integers(-(1 << 65), 1 << 65), + st.text(max_size=20), + st.lists(SCALARS, max_size=4), + ) +) +def test_generated_non_mapping_roots_reject_including_falsy_values(value: object) -> None: + _assert_both_loaders_reject(yaml.safe_dump(value), "configuration") + + +@pytest.mark.parametrize("location", ["queries", "query", "parameters", "parameter"]) +@PROPERTY_SETTINGS +@given(value=NON_MAPPINGS) +def test_generated_nested_non_mappings_raise_validation_error(location: str, value: object) -> None: + content = { + "queries": {"queries": value}, + "query": {"queries": {"generated": value}}, + "parameters": {"queries": {"generated": {"parameters": value}}}, + "parameter": {"queries": {"generated": {"parameters": {"selected": value}}}}, + }[location] + label = { + "queries": "queries", + "query": "query 'generated'", + "parameters": "parameters", + "parameter": "parameter 'selected'", + }[location] + _assert_both_loaders_reject(yaml.safe_dump(content), label) + + +@pytest.mark.parametrize("location", ["query", "parameter"]) +@PROPERTY_SETTINGS +@given(name=INVALID_NAMES) +def test_generated_invalid_declaration_names_are_rejected(location: str, name: object) -> None: + definitions = ( + {name: {"query": "SELECT 1"}} + if location == "query" + else {"generated": {"query": "SELECT 1", "parameters": {name: {}}}} + ) + content = yaml.safe_dump({"queries": definitions}) + with pytest.raises(ValueError, match=f"{location} name must be a non-empty string"): + QueryConfig.load_yaml_from_string(content) + with TemporaryDirectory() as directory: + source = Path(directory) / "template.yaml" + source.write_text(content, encoding="utf-8") + with pytest.raises(ValueError, match=f"{location} name must be a non-empty string"): + QueryConfig.load_yaml(source) + assert source.read_text(encoding="utf-8") == content + + +@pytest.mark.parametrize("location", ["template", "parameters", "parameter"]) +@PROPERTY_SETTINGS +@given(value=NON_MAPPINGS) +def test_direct_template_parser_uses_the_same_mapping_validation( + location: str, value: object +) -> None: + data = { + "template": value, + "parameters": {"parameters": value}, + "parameter": {"parameters": {"selected": value}}, + }[location] + with pytest.raises(ValueError, match="must be a mapping"): + QueryTemplate.from_dict(data) + + +@PROPERTY_SETTINGS +@given(name=NAMES, value=st.integers(-1000, 1000)) +def test_generated_valid_templates_keep_defaults_names_and_category(name: str, value: int) -> None: + body = { + "description": "generated", + "parameters": {"selected": {"type": "int", "default": value}}, + "query": "SELECT {{ selected }} AS value", + } + content = yaml.safe_dump({"queries": {name: body}}) + from_string = QueryConfig.load_yaml_from_string(content) + with TemporaryDirectory() as directory: + source = Path(directory) / "template.yaml" + source.write_text(content, encoding="utf-8") + from_file = QueryConfig.load_yaml(source) + assert from_file == from_string + categorized = QueryConfig.load_yaml(source, default_category="custom") + assert categorized.queries[name].category == "custom" + assert source.read_text(encoding="utf-8") == content + template = from_string.queries[name] + assert template.name == name + assert template.category == "basic" + assert template.validate_params({}) == {"selected": value} + assert template.output_schema == [] + assert template.query == body["query"] + + +@PROPERTY_SETTINGS +@given(name=NAMES, value=SCALARS) +def test_shared_yaml_alias_data_is_not_mutated_and_unknown_fields_remain_compatible( + name: str, value: object +) -> None: + body = { + "query": "SELECT 1", + "extension": value, + "parameters": {"selected": {"type": "int", "extension": value}}, + } + data = {"queries": {name: body, f"other_{name}": body}, "extension": value} + before = copy.deepcopy(data) + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr(config_module.yaml, "safe_load", lambda stream: data) + config = QueryConfig.load_yaml_from_string("loader result supplied above") + assert data == before + assert data["queries"][name] is data["queries"][f"other_{name}"] + assert config.queries[name].name == name + assert config.queries[f"other_{name}"].name == f"other_{name}" + assert all(template.query == "SELECT 1" for template in config.queries.values()) + + +@pytest.mark.parametrize("content", ["", "# comment only\n", "null", "{}", "queries: {}"]) +def test_empty_configuration_contract_is_preserved(content: str) -> None: + assert QueryConfig.load_yaml_from_string(content) == QueryConfig() + + +@PROPERTY_SETTINGS +@given(name=NAMES) +def test_yaml_syntax_errors_retain_the_documented_parser_error(name: str) -> None: + content = f"queries: [{name}" + with pytest.raises(yaml.YAMLError): + QueryConfig.load_yaml_from_string(content) + with TemporaryDirectory() as directory: + source = Path(directory) / "template.yaml" + source.write_text(content, encoding="utf-8") + with pytest.raises(yaml.YAMLError): + QueryConfig.load_yaml(source) + assert source.read_text(encoding="utf-8") == content + + +@pytest.mark.parametrize("field", ["query", "parameter_type"]) +@PROPERTY_SETTINGS +@given( + value=st.one_of( + st.booleans(), + st.integers(-100, 100), + st.lists(SCALARS, max_size=4), + st.dictionaries(NAMES, SCALARS, max_size=4), + ) +) +def test_generated_invalid_sql_and_parameter_type_metadata_is_rejected( + field: str, value: object +) -> None: + body = {"query": value} if field == "query" else {"parameters": {"selected": {"type": value}}} + content = yaml.safe_dump({"queries": {"generated": body}}) + message = "must be a SQL string" if field == "query" else "type must be a string" + with pytest.raises(ValueError, match=message): + QueryConfig.load_yaml_from_string(content) + with TemporaryDirectory() as directory: + source = Path(directory) / "template.yaml" + source.write_text(content, encoding="utf-8") + with pytest.raises(ValueError, match=message): + QueryConfig.load_yaml(source) + assert source.read_text(encoding="utf-8") == content + + +@PROPERTY_SETTINGS +@given(name=INVALID_NAMES) +def test_direct_template_parser_rejects_invalid_explicit_names(name: object) -> None: + with pytest.raises(ValueError, match="template name must be a non-empty string"): + QueryTemplate.from_dict({"name": name, "query": "SELECT 1"}) + + +def test_null_parameter_type_is_not_a_string_but_null_query_keeps_its_default() -> None: + with pytest.raises(ValueError, match="type must be a string"): + QueryTemplate.from_dict({"parameters": {"selected": {"type": None}}}) + assert QueryTemplate.from_dict({"query": None}).query == "" + legacy = QueryTemplate.from_dict({"parameters": {"selected": {"type": "custom_legacy"}}}) + assert legacy.validate_params({"selected": 42}) == {"selected": "42"} diff --git a/tests/snapshot/test_entities.py b/tests/snapshot/test_entities.py index 6069c29..f9e092a 100644 --- a/tests/snapshot/test_entities.py +++ b/tests/snapshot/test_entities.py @@ -90,7 +90,7 @@ def test_from_dict_loads_oom_without_addr_or_stream(self): self.assertEqual(trace.frames, []) def test_from_dict_requires_action(self): - with self.assertRaises(KeyError): + with self.assertRaisesRegex(ValueError, "Trace field 'action'"): TraceEntry.from_dict({"size": 1024, "frames": []}) def test_get_callstack(self): @@ -149,11 +149,11 @@ def test_raw_frames_preserve_empty_missing_and_incomplete_behavior(self): "", ) incomplete = {**base, "frames": [{"filename": "broken.py", "line": 3}]} - with self.assertRaises(KeyError): + with self.assertRaisesRegex(ValueError, "missing required field 'name'"): TraceEntry.from_dict(incomplete) - with self.assertRaises(KeyError): + with self.assertRaisesRegex(ValueError, "missing required field 'name'"): TraceEntry.from_dict(incomplete, _raw_frames=True) - with self.assertRaises(TypeError): + with self.assertRaisesRegex(ValueError, "Frames must be a list or tuple"): TraceEntry.from_dict({**base, "frames": None}, _raw_frames=True) def test_raw_frames_preserve_positional_constructor_and_synthetic_serialization(self): diff --git a/tests/snapshot/test_snapshot2db_runtime.py b/tests/snapshot/test_snapshot2db_runtime.py index 55bcdf7..4209d55 100644 --- a/tests/snapshot/test_snapshot2db_runtime.py +++ b/tests/snapshot/test_snapshot2db_runtime.py @@ -225,7 +225,7 @@ def pre_undo_event(self, event, current_snapshot): monkeypatch.setattr(snapshot2db.DumpEventHooker, "pre_undo_event", pre_undo_event) - with pytest.raises(KeyError, match="name"): + with pytest.raises(ValueError, match="missing required field 'name'"): snapshot2db.dump(snapshot, database, 0) assert replayed_events == 0 @@ -261,7 +261,7 @@ def test_dump_all_removes_committed_devices_when_later_device_is_malformed(tmp_p with snapshot.open("wb") as stream: pickle.dump(representation, stream) - with pytest.raises(KeyError, match="name"): + with pytest.raises(ValueError, match="missing required field 'name'"): snapshot2db.dump(snapshot, database) assert not database.exists() @@ -291,7 +291,7 @@ def test_failed_dump_preserves_existing_database(tmp_path): connection.execute("INSERT INTO marker VALUES ('preserved')") connection.commit() - with pytest.raises(KeyError, match="name"): + with pytest.raises(ValueError, match="missing required field 'name'"): snapshot2db.dump(snapshot, database, 0) with closing(sqlite3.connect(database)) as connection, connection: