diff --git a/pyproject.toml b/pyproject.toml index f65fa0e..5e777e2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ Issues = "https://github.com/codeasier/pt-snap-cli/issues" dev = [ "pytest>=7.0.0", "pytest-cov>=4.0.0", + "hypothesis>=6.100.0,<7", "playwright>=1.40.0", "black>=23.0.0", "ruff>=0.0.280", diff --git a/tests/core/test_import_metadata_properties.py b/tests/core/test_import_metadata_properties.py new file mode 100644 index 0000000..e904918 --- /dev/null +++ b/tests/core/test_import_metadata_properties.py @@ -0,0 +1,202 @@ +"""Generated import-metadata validation, without deserializing pickle bytes. + +Mappings are trusted test-owned values. SQLite fixtures deliberately omit type +coercion/constraints so malformed stored values reach the actual read validator. +""" + +from __future__ import annotations + +import sqlite3 +from contextlib import closing +from dataclasses import asdict +from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from pt_snap_cli.core.import_metadata import ( + IMPORT_FORMAT_VERSION, + METADATA_SCHEMA_VERSION, + ImportMetadataService, +) +from pt_snap_cli.core.import_metadata_contract import metadata_from_mapping + +PROPERTY_SETTINGS = settings( + max_examples=35, derandomize=True, database=None, deadline=None, print_blob=True +) +INTEGER_FIELDS = ( + "metadata_schema_version", + "import_format_version", + "source_size", + "requested_device", +) +STRING_FIELDS = ( + "source_sha256", + "source_name", + "importer_name", + "importer_version", + "completed_at", +) +FIELDS = INTEGER_FIELDS + STRING_FIELDS +TEXT = st.text(alphabet="abcdefghijklmnopqrstuvwxyz0123456789_中文", min_size=1, max_size=24) + + +@st.composite +def _valid_metadata(draw: st.DrawFn) -> dict[str, object]: + return { + "metadata_schema_version": METADATA_SCHEMA_VERSION, + "import_format_version": IMPORT_FORMAT_VERSION, + "source_sha256": draw(st.binary(min_size=32, max_size=32)).hex(), + "source_size": draw(st.integers(0, 2**63 - 1)), + "source_name": draw(TEXT) + ".pkl", + "requested_device": draw(st.one_of(st.none(), st.integers(0, 1024))), + "importer_name": draw(TEXT), + "importer_version": draw(TEXT), + "completed_at": draw( + st.datetimes( + min_value=datetime(2000, 1, 1), + max_value=datetime(2030, 12, 31), + timezones=st.just(timezone.utc), + ) + ).isoformat(), + } + + +def _database(path: Path) -> None: + with closing(sqlite3.connect(path)) as conn, conn: + conn.execute( + "CREATE TABLE dictionary (`table` TEXT, `column` TEXT, `key` TEXT, `value` TEXT)" + ) + conn.execute("CREATE TABLE trace_entry_0 (id INTEGER PRIMARY KEY)") + + +def _store_unvalidated(path: Path, values: dict[str, object]) -> None: + # Field names are fixed test constants, never generated SQL identifiers. + columns = ("id",) + tuple(field for field in FIELDS if field in values) + quoted = ", ".join(f'"{field}"' for field in columns) + placeholders = ", ".join("?" for _ in columns) + with closing(sqlite3.connect(path)) as conn, conn: + conn.execute(f"CREATE TABLE pt_snap_metadata ({quoted})") + conn.execute( + f"INSERT INTO pt_snap_metadata VALUES ({placeholders})", + [1] + [values[field] for field in columns[1:]], + ) + + +def _assert_invalid_read_is_nonmutating(values: dict[str, object]) -> None: + with TemporaryDirectory() as directory: + path = Path(directory) / "invalid.db" + _database(path) + _store_unvalidated(path, values) + before = path.read_bytes() + inspection = ImportMetadataService().inspect(path) + assert inspection.status == "invalid" + assert inspection.reason == "metadata_invalid" + assert inspection.metadata is None + assert path.read_bytes() == before + + +@PROPERTY_SETTINGS +@given(values=_valid_metadata()) +def test_generated_valid_metadata_round_trips_mapping_and_sqlite(values: dict[str, object]) -> None: + metadata = metadata_from_mapping(values) + assert asdict(metadata) == values + with TemporaryDirectory() as directory: + path = Path(directory) / "valid.db" + _database(path) + service = ImportMetadataService() + service.write(path, metadata) + before = path.read_bytes() + inspection = service.inspect(path) + assert inspection.status == "available" + assert inspection.reason is None + assert inspection.metadata == metadata + decision = service.evaluate_cache(path, metadata.source_sha256, metadata.requested_device) + assert decision.reused is True + assert decision.metadata == metadata + assert decision.reason is None + assert path.read_bytes() == before + + +@PROPERTY_SETTINGS +@given(values=_valid_metadata(), missing=st.sets(st.sampled_from(FIELDS), min_size=1)) +def test_generated_missing_metadata_fields_have_controlled_inspection( + values: dict[str, object], missing: set[str] +) -> None: + incomplete = {field: value for field, value in values.items() if field not in missing} + with pytest.raises(KeyError) as error: + metadata_from_mapping(incomplete) + assert error.value.args[0] in missing + _assert_invalid_read_is_nonmutating(incomplete) + + +@pytest.mark.parametrize("field", INTEGER_FIELDS) +@PROPERTY_SETTINGS +@given( + values=_valid_metadata(), + wrong_type=st.one_of( + st.booleans(), + st.floats(min_value=-1000, max_value=1000, allow_nan=False, allow_infinity=False), + st.text(max_size=20), + st.lists(st.integers(-10, 10), max_size=3), + ), +) +def test_generated_integer_metadata_does_not_coerce_wrong_types( + field: str, values: dict[str, object], wrong_type: object +) -> None: + values[field] = wrong_type + # This is the pure mapping contract: in particular bool is not int. SQLite + # binds bool as an integer, so testing this only through SQL would hide it. + with pytest.raises(TypeError, match="Expected integer metadata value"): + metadata_from_mapping(values) + + +@pytest.mark.parametrize("field", INTEGER_FIELDS) +@PROPERTY_SETTINGS +@given(values=_valid_metadata(), magnitude=st.integers(1, 2**63 - 1)) +def test_generated_negative_metadata_numbers_have_controlled_inspection( + field: str, values: dict[str, object], magnitude: int +) -> None: + values[field] = -magnitude + with pytest.raises(ValueError): + metadata_from_mapping(values) + _assert_invalid_read_is_nonmutating(values) + + +@pytest.mark.parametrize("field", STRING_FIELDS) +@PROPERTY_SETTINGS +@given(values=_valid_metadata(), empty=st.sampled_from([None, ""])) +def test_generated_missing_string_values_have_controlled_inspection( + field: str, values: dict[str, object], empty: object +) -> None: + values[field] = empty + with pytest.raises(TypeError, match="Expected non-empty metadata string"): + metadata_from_mapping(values) + _assert_invalid_read_is_nonmutating(values) + + +@PROPERTY_SETTINGS +@given( + values=_valid_metadata(), suffix=st.text(alphabet="0123456789abcdef", min_size=0, max_size=70) +) +def test_generated_malformed_digest_has_controlled_inspection( + values: dict[str, object], suffix: str +) -> None: + values["source_sha256"] = "g" + suffix + with pytest.raises(ValueError, match="lowercase SHA-256 digest"): + metadata_from_mapping(values) + _assert_invalid_read_is_nonmutating(values) + + +@PROPERTY_SETTINGS +@given(values=_valid_metadata(), offset_hours=st.integers(1, 23)) +def test_generated_non_utc_timestamp_has_controlled_inspection( + values: dict[str, object], offset_hours: int +) -> None: + values["completed_at"] = f"2026-01-01T00:00:00+{offset_hours:02d}:00" + with pytest.raises(ValueError, match="completed_at must use UTC"): + metadata_from_mapping(values) + _assert_invalid_read_is_nonmutating(values) diff --git a/tests/core/test_snapshot_validation_properties.py b/tests/core/test_snapshot_validation_properties.py new file mode 100644 index 0000000..175f616 --- /dev/null +++ b/tests/core/test_snapshot_validation_properties.py @@ -0,0 +1,69 @@ +"""Trusted in-memory numeric metadata rejection through the real import boundary. + +Only the representation-loading seam is replaced: no pickle is created or +unpickled, and conversion, validation, error normalization and cleanup stay real. +Missing keys and arbitrary structural/type errors are a separate runtime gap; +this module covers the existing controlled ValueError conversion contract. +""" + +from __future__ import annotations + +import copy +import importlib +from pathlib import Path +from tempfile import TemporaryDirectory + +import pytest +from hypothesis import 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 + +PROPERTY_SETTINGS = settings( + max_examples=25, derandomize=True, database=None, deadline=None, print_blob=True +) + + +@pytest.mark.parametrize("field", ["addr", "size", "stream", "device_free", "id"]) +@PROPERTY_SETTINGS +@given(suffix=st.text(alphabet="abcdefghijklmnopqrstuvwxyz0123456789 +-", max_size=40)) +def test_generated_bad_numeric_snapshot_values_reject_without_publication( + field: str, suffix: str +) -> None: + event = {"action": "alloc", "addr": 4096, "size": 1, "stream": 0, "frames": []} + event[field] = f"not-an-integer:{suffix}" + representation = {"segments": [], "device_traces": [[event]]} + before = copy.deepcopy(representation) + + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + source = root / "trusted-in-memory.pkl" + source_bytes = b"inert test placeholder; the representation loader must be replaced" + 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 must survive validation failure") + previous = destination.read_bytes() + loads = [] + + def load_representation(path: Path) -> dict[str, object]: + assert path == source + loads.append(path) + return representation + + # The lazy-import test removes this module from sys.modules. Resolve + # the current object here, not a stale reference captured at collection. + 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=0) + + assert isinstance(error.value.__cause__, ValueError) + assert loads == [source] + assert representation == before + assert source.read_bytes() == source_bytes + assert destination.read_bytes() == previous + assert list(output.iterdir()) == [destination] diff --git a/tests/query/test_pagination_properties.py b/tests/query/test_pagination_properties.py new file mode 100644 index 0000000..b712a9c --- /dev/null +++ b/tests/query/test_pagination_properties.py @@ -0,0 +1,198 @@ +"""Bounded properties for the existing standalone event/block page contract. + +Every generated example owns its database, focus isolation and service lifetime. +The oracle sorts independent input records, not another query's returned rows. +""" + +from __future__ import annotations + +import copy +import sqlite3 +from collections.abc import Iterator +from contextlib import closing, contextmanager +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.query_limits import QueryLimits +from pt_snap_cli.core.query_service import QueryService +from pt_snap_cli.query import registry +from pt_snap_cli.query.config import QueryConfig + +# Deterministic for a fixed Hypothesis version; failing examples are shrunk and +# print a reproduction blob. No example database or machine-speed deadline. +PROPERTY_SETTINGS = settings( + max_examples=40, derandomize=True, database=None, deadline=None, print_blob=True +) +RECORDS = st.lists( + st.tuples(st.integers(-20, 100), st.integers(0, 4)), + unique_by=lambda row: row[0], + max_size=24, +) +LIMITS = st.one_of(st.just(-1), st.integers(0, 30)) +ROW_CAPS = st.one_of(st.none(), st.integers(0, 30)) + + +@contextmanager +def _query_database(records: list[tuple[int, int]]) -> Iterator[tuple[QueryService, Path]]: + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + monkeypatch.chdir(root) + monkeypatch.setattr(Path, "home", lambda: root) + monkeypatch.delenv("PT_SNAP_DB_PATH", raising=False) + monkeypatch.delenv("PT_SNAP_QUERY_TIMEOUT", raising=False) + # Restore the exact original registry, including other tests' factories. + # Only load the two actual packaged contracts being exercised here. + monkeypatch.setattr(registry, "_registry", copy.deepcopy(registry._registry)) + for name in ("event", "block"): + config = QueryConfig.load_yaml(registry.get_template_dir() / "basic" / f"{name}.yaml") + registry.register_query(config.queries[name]) + + db_path = root / "pages.db" + with closing(sqlite3.connect(db_path)) as conn, conn: + conn.execute( + "CREATE TABLE dictionary (`table` TEXT, `column` TEXT, `key` TEXT, `value` TEXT)" + ) + conn.execute( + "CREATE TABLE trace_entry_0 (id INTEGER PRIMARY KEY, action INTEGER, " + "address INTEGER, size INTEGER, stream INTEGER, allocated INTEGER, " + "active INTEGER, reserved INTEGER, callstack TEXT)" + ) + conn.execute( + "CREATE TABLE block_0 (id INTEGER PRIMARY KEY, address INTEGER, size INTEGER, " + "requestedSize INTEGER, state INTEGER, allocEventId INTEGER, freeEventId INTEGER)" + ) + conn.executemany( + "INSERT INTO trace_entry_0 VALUES (?, 4, 4096, ?, 0, 0, 0, 0, 'frame')", + records, + ) + conn.executemany( + "INSERT INTO block_0 VALUES (?, 4096, ?, ?, 1, ?, -1)", + [(event_id, size, size, event_id) for event_id, size in records], + ) + before = db_path.read_bytes() + with closing(QueryService(query_limits=QueryLimits())) as service: + yield service, db_path + assert db_path.read_bytes() == before + assert not (root / ".pt-snap").exists() + + +@pytest.mark.parametrize("template", ["event", "block"]) +@PROPERTY_SETTINGS +@given( + records=RECORDS, + offset=st.integers(0, 30), + limit=LIMITS, + max_rows=ROW_CAPS, + direction=st.sampled_from(["ASC", "DESC"]), + exact_total=st.booleans(), +) +@example(records=[], offset=0, limit=0, max_rows=None, direction="ASC", exact_total=False) +@example( + records=[(7, 2), (-3, 2), (5, 2), (11, 0)], + offset=1, + limit=3, + max_rows=1, + direction="DESC", + exact_total=True, +) +@example( + records=[(7, 2), (-3, 2)], + offset=0, + limit=0, + max_rows=3, + direction="ASC", + exact_total=True, +) +@example( + records=[(7, 2), (-3, 2)], + offset=4, + limit=-1, + max_rows=0, + direction="ASC", + exact_total=False, +) +def test_generated_pages_match_ordered_slice_and_completeness( + template: str, + records: list[tuple[int, int]], + offset: int, + limit: int, + max_rows: int | None, + direction: str, + exact_total: bool, +) -> None: + ordered = sorted(records, key=lambda row: (row[1], row[0]), reverse=direction == "DESC") + available = ordered[offset:] + # max_rows=0 disables the output cap; a template limit=0 is an empty page. + caps = [limit] if limit >= 0 else [] + if max_rows is not None and max_rows > 0: + caps.append(max_rows) + cap = min(caps) if caps else len(available) + expected = available[:cap] + has_more = len(available) > len(expected) + + with _query_database(records) as (service, db_path): + result = service.execute_query( + template, + db_path=db_path, + device_id=0, + params={"order_by": "size", "order_dir": direction, "offset": offset, "limit": limit}, + max_rows=max_rows, + exact_total=exact_total, + ) + + assert [(row["id"], row["size"]) for row in result.rows] == expected + assert result.returned == len(expected) + assert result.has_more is has_more + assert result.total == (len(records) if exact_total else len(expected)) + assert result.total_is_exact is (exact_total or (not has_more and offset == 0)) + assert result.truncated is ( + has_more or offset > 0 or (exact_total and len(records) > len(expected)) + ) + + +@pytest.mark.parametrize("template", ["event", "block"]) +@PROPERTY_SETTINGS +@given( + records=RECORDS, + page_size=st.integers(1, 12), + max_rows=st.one_of(st.none(), st.integers(1, 12)), + direction=st.sampled_from(["ASC", "DESC"]), +) +@example(records=[(5, 3), (1, 3), (8, 3), (2, 1)], page_size=2, max_rows=1, direction="DESC") +@example(records=[], page_size=1, max_rows=None, direction="ASC") +def test_generated_page_concatenation_matches_unlimited_query( + template: str, + records: list[tuple[int, int]], + page_size: int, + max_rows: int | None, + direction: str, +) -> None: + expected = sorted(records, key=lambda row: (row[1], row[0]), reverse=direction == "DESC") + step = min(page_size, max_rows) if max_rows is not None else page_size + common = {"order_by": "size", "order_dir": direction} + with _query_database(records) as (service, db_path): + whole = service.execute_query(template, params=common, db_path=db_path, device_id=0) + pages = [] + # Arithmetic bounds, not has_more, control termination. Query the empty + # trailing page too, so broken completeness cannot hide missing rows. + for offset in range(0, len(records) + step, step): + page = service.execute_query( + template, + params={**common, "offset": offset, "limit": page_size}, + db_path=db_path, + device_id=0, + max_rows=max_rows, + ) + expected_page = expected[offset : offset + step] + assert [(row["id"], row["size"]) for row in page.rows] == expected_page + assert page.has_more is (offset + len(expected_page) < len(expected)) + pages.extend(page.rows) + assert pages == whole.rows + assert [(row["id"], row["size"]) for row in whole.rows] == expected + assert len({row["id"] for row in pages}) == len(records) + assert whole.total_is_exact is True + assert whole.has_more is whole.truncated is False diff --git a/tests/query/test_template_properties.py b/tests/query/test_template_properties.py new file mode 100644 index 0000000..65f8dfe --- /dev/null +++ b/tests/query/test_template_properties.py @@ -0,0 +1,178 @@ +"""Generated YAML contracts at their existing validation/rejection boundaries. + +The registry promises warning-and-skip for malformed files. The lower-level +QueryConfig loader does not currently normalize arbitrary structural YAML into +one domain error; these tests do not claim that stronger, unsupported contract. +""" + +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 registry +from pt_snap_cli.query.config import QueryConfig + +PROPERTY_SETTINGS = settings( + max_examples=35, derandomize=True, database=None, deadline=None, print_blob=True +) +IDENTIFIERS = st.text(alphabet="abcdefghijklmnopqrstuvwxyz", min_size=1, max_size=16) +VALUES = st.recursive( + st.one_of(st.none(), st.booleans(), st.integers(-100, 100), st.text(max_size=16)), + lambda children: st.one_of( + st.lists(children, max_size=4), st.dictionaries(IDENTIFIERS, children, max_size=4) + ), + max_leaves=8, +) + + +def _load_template(body: dict[str, object]): + return QueryConfig.load_yaml_from_string( + yaml.safe_dump({"queries": {"generated": body}}) + ).queries["generated"] + + +@PROPERTY_SETTINGS +@given( + column=IDENTIFIERS, + choices=st.lists(st.integers(-1000, 1000), min_size=1, max_size=8, unique=True), + version=st.integers(1, 100), + sentinel=st.integers(-1000, 1000), + units=st.sampled_from( + ["bytes", "gib", "percent", "event_id", "count", "address", "flag", "text"] + ), +) +def test_generated_yaml_preserves_schema_and_validates_declared_choices( + column: str, choices: list[int], version: int, sentinel: int, units: str +) -> None: + schema = [{"column": column, "type": "int", "units": units, "sentinel": sentinel}] + template = _load_template( + { + "query": "SELECT {{ selected }} AS value", + "semantics_version": version, + "output_schema": schema, + "parameters": { + "selected": {"type": "int", "choices": choices, "default": choices[0]}, + "required_text": {"type": "str", "required": True}, + }, + } + ) + assert template.name == "generated" + assert template.output_schema == schema + assert template.semantics_version == version + assert template.parameters["selected"].choices == choices + for choice in choices: + assert template.validate_params({"selected": str(choice), "required_text": column}) == { + "selected": choice, + "required_text": column, + } + assert template.validate_params({"required_text": column})["selected"] == choices[0] + with pytest.raises(ValueError, match="Required parameter 'required_text' is missing"): + template.validate_params({}) + with pytest.raises(ValueError, match="'selected' must be one of"): + template.validate_params({"selected": max(choices) + 1, "required_text": column}) + + +@PROPERTY_SETTINGS +@given(unknown=IDENTIFIERS, value=VALUES) +def test_generated_unknown_schema_fields_and_query_parameters_are_rejected( + unknown: str, value: object +) -> None: + key = f"unknown_{unknown}" + with pytest.raises(ValueError, match="unsupported keys") as error: + _load_template( + {"query": "SELECT 1", "output_schema": [{"column": "id", "type": "int", key: value}]} + ) + assert key in str(error.value) + template = _load_template( + {"query": "SELECT 1", "parameters": {"limit": {"type": "int", "default": -1}}} + ) + with pytest.raises(ValueError, match=r"Unknown parameter\(s\)") as error: + template.validate_params({key: value}) + assert key in str(error.value) + + +@pytest.mark.parametrize("missing", ["column", "type"]) +@PROPERTY_SETTINGS +@given(column=IDENTIFIERS) +def test_generated_output_columns_require_both_declared_fields(missing: str, column: str) -> None: + schema = {"column": column, "type": "int"} + del schema[missing] + with pytest.raises(ValueError, match="must declare a non-empty"): + _load_template({"query": "SELECT 1", "output_schema": [schema]}) + + +@pytest.mark.parametrize("field", ["units", "metric_semantics", "scope"]) +@PROPERTY_SETTINGS +@given(value=VALUES) +def test_generated_invalid_enum_metadata_is_rejected(field: str, value: object) -> None: + # A prefix constructs an invalid enum without filtering or discarding cases. + malformed = f"unsupported:{value!r}" if isinstance(value, str) else value + with pytest.raises(ValueError, match=f"{field} must be one of"): + _load_template( + { + "query": "SELECT 1", + "output_schema": [{"column": "id", "type": "int", field: malformed}], + } + ) + + +@PROPERTY_SETTINGS +@given( + value=st.one_of( + st.booleans(), + st.integers(-1000, 0), + st.text(max_size=20), + st.lists(st.integers(-1000, 1000), max_size=4), + ) +) +def test_generated_invalid_semantics_versions_are_rejected(value: object) -> None: + with pytest.raises(ValueError, match="semantics_version.*positive integer"): + _load_template({"query": "SELECT 1", "semantics_version": value}) + + +@pytest.mark.parametrize( + "defect", ["root_sequence", "queries_sequence", "scalar_entry", "parameters_sequence", "syntax"] +) +@PROPERTY_SETTINGS +@given(payload=IDENTIFIERS) +def test_generated_malformed_yaml_is_skipped_without_losing_valid_neighbors( + defect: str, payload: str +) -> None: + malformed = { + "root_sequence": yaml.safe_dump([payload]), + "queries_sequence": yaml.safe_dump({"queries": [payload]}), + "scalar_entry": yaml.safe_dump({"queries": {"invalid": payload}}), + "parameters_sequence": yaml.safe_dump({"queries": {"invalid": {"parameters": [payload]}}}), + "syntax": f"queries: [{payload}", + }[defect] + with TemporaryDirectory() as directory, pytest.MonkeyPatch.context() as monkeypatch: + root = Path(directory) + # A detached registry snapshot keeps generated registrations local to + # this example and restores both normal entries and lazy factories. + monkeypatch.setattr(registry, "_registry", copy.deepcopy(registry._registry)) + invalid_name = f"property_invalid_{payload}" + valid_name = f"property_valid_{payload}" + registry._registry.unregister(invalid_name) + registry._registry.unregister(valid_name) + malformed = malformed.replace("invalid:", f"{invalid_name}:") + bad = root / "a_bad.yaml" + bad.write_text(malformed, encoding="utf-8") + (root / "z_good.yaml").write_text( + yaml.safe_dump({"queries": {valid_name: {"query": "SELECT 1 AS id"}}}), + encoding="utf-8", + ) + with pytest.warns(UserWarning, match="Failed to load query template") as warnings: + registry._load_yaml_templates(root) + assert len(warnings) == 1 + assert str(bad) in str(warnings[0].message) + assert registry.get_query(invalid_name) is None + valid = registry.get_query(valid_name) + assert valid is not None + assert valid.query == "SELECT 1 AS id"