From a96dc494307b59f716379a7c9901fa3ba46b8162 Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 14:51:43 +0800 Subject: [PATCH 01/10] test: harden package removal checks and portable runner (#275, #276) --- .gitignore | 1 + AGENTS.md | 14 +- tests/run_tests.sh | 303 ++++++++++++++++---------------------- tests/test_package.py | 154 ++++++++++++++++++- tests/test_test_runner.py | 264 +++++++++++++++++++++++++++++++++ 5 files changed, 551 insertions(+), 185 deletions(-) create mode 100644 tests/test_test_runner.py diff --git a/.gitignore b/.gitignore index add18f6..2d84903 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ coverage.xml platform-tests.xml platform-partition.json htmlcov/ +test_reports/ .pytest_cache/ .ruff_cache/ .mypy_cache/ diff --git a/AGENTS.md b/AGENTS.md index d9ac08c..bd5ca9d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -75,9 +75,17 @@ may point to an existing Chromium executable instead of downloading one. errors. Ownership/teardown guidance lives in [tests](tests/AGENTS.md). - Coverage is measured with branches enabled. The overall floor and independent critical-file floors are documented with [CI](.github/workflows/AGENTS.md). -- `tests/run_tests.sh` assumes a developer-specific Conda path and writes - `test_reports/`; prefer direct pytest. `.pt-snap/`, `.tmp/`, `.worktrees/`, - and `tmp/` are ignored and excluded from root Ruff/Black checks. +- `bash tests/run_tests.sh` is an optional Bash entry point using the active + `python -m pytest`; set `PYTHON=/path/to/python` to choose another executable. + It does not activate environments or install dependencies. Full mode retains + the configured coverage floor and writes ignored `test_reports/` output; + `--reports-dir DIR` changes that destination. `--quick` omits coverage/report + generation and is **not** a substitute for CI's full coverage gate. Use + `--help` for file/keyword/marker selection and `--` for extra pytest arguments. + Pytest's exit status is preserved, including failure and no-tests-collected. + Without Bash, use the direct Python/pytest commands above. `.pt-snap/`, + `.tmp/`, `.worktrees/`, and `tmp/` are ignored and excluded from root + Ruff/Black checks. - `python -m build` writes `dist/` and build intermediates and may download build requirements. See [packaging](src/AGENTS.md) before using it as verification. diff --git a/tests/run_tests.sh b/tests/run_tests.sh index 096d8a4..03fe73d 100755 --- a/tests/run_tests.sh +++ b/tests/run_tests.sh @@ -1,193 +1,142 @@ -#!/bin/bash +#!/usr/bin/env bash -# 测试执行与报告生成脚本 -# 用于运行所有测试并生成详细的测试报告 +# Use the caller's Python environment; never activate or install an environment. +set -eo pipefail -set -e +PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +PYTHON="${PYTHON:-python}" +REPORTS_DIR="test_reports" -# 颜色定义 -RED='\033[0;31m' -GREEN='\033[0;32m' -YELLOW='\033[1;33m' -BLUE='\033[0;34m' -NC='\033[0m' # No Color +show_help() { + cat <>> 激活 conda 环境...${NC}" - if [ -f "/Users/test1/liuyekang/miniconda3/etc/profile.d/conda.sh" ]; then - . /Users/test1/liuyekang/miniconda3/etc/profile.d/conda.sh - conda activate openclaw - echo -e "${GREEN}✓ Conda 环境激活成功${NC}" - else - echo -e "${RED}✗ Conda 初始化脚本不存在${NC}" - exit 1 - fi -} +Options: + -h, --help Show this help without running Python + -q, --quick Omit coverage and report generation (not a CI gate) + -f, --file FILE Select a test file/node (repeatable); bare names use tests/ + -k, --keyword EXPR Select tests by pytest keyword expression + -m, --mark EXPR Select tests by pytest marker expression + -x, --stop Stop after the first failure + --reports-dir DIR Coverage output directory (default: test_reports/) + -- Forward remaining arguments unchanged to pytest + +With no --file selection, run tests/. Full mode retains the configured coverage +floor and writes HTML/XML reports under the ignored test_reports/ directory. +Quick mode adds no coverage options and creates no report directory. Arbitrary +pytest options belong after --. Select files/nodes with -f: positional selectors +passed after -- are additive to the default tests/ selection. The pytest exit +code is returned unchanged. -# 创建报告目录 -create_report_dir() { - echo -e "${BLUE}>>> 创建报告目录...${NC}" - mkdir -p "${REPORTS_DIR}/htmlcov" - mkdir -p "${REPORTS_DIR}/xml" - echo -e "${GREEN}✓ 报告目录创建成功${NC}" +Examples: + bash tests/run_tests.sh + bash tests/run_tests.sh --quick -f test_package.py + bash tests/run_tests.sh -m 'not slow and not nightly' -- -q + PYTHON=/path/to/venv/bin/python bash tests/run_tests.sh --quick -- --collect-only +HELP } -# 运行测试 -run_tests() { - local test_pattern="${1:-tests/}" - - echo -e "${BLUE}>>> 运行测试：${test_pattern}${NC}" - echo "========================================" - - pytest "${test_pattern}" \ - --cov=pt_snap_cli \ - --cov-report=html:"${REPORTS_DIR}/htmlcov" \ - --cov-report=xml:"${REPORTS_DIR}/xml/coverage.xml" \ - --cov-report=term-missing \ - -v \ - --tb=short \ - "${@:2}" - - local exit_code=$? - - if [ $exit_code -eq 0 ]; then - echo -e "${GREEN}✓ 所有测试通过${NC}" - else - echo -e "${RED}✗ 部分测试失败${NC}" - fi - - return $exit_code +usage_error() { + printf 'Error: %s\nUse --help for usage.\n' "$1" >&2 + exit 2 } -# 生成报告摘要 -generate_summary() { - echo "" - echo "========================================" - echo -e "${BLUE}>>> 测试报告摘要${NC}" - echo "========================================" - - if [ -f "${REPORTS_DIR}/xml/coverage.xml" ]; then - echo -e "${GREEN}✓ 覆盖率报告已生成${NC}" - echo " - HTML 报告：${REPORTS_DIR}/htmlcov/index.html" - echo " - XML 报告：${REPORTS_DIR}/xml/coverage.xml" +require_value() { + if [[ $# -lt 2 || -z "$2" || "$2" == -* ]]; then + usage_error "$1 requires a value" fi - - echo "" } -# 显示帮助信息 -show_help() { - echo "用法：$0 [选项] [测试模式]" - echo "" - echo "选项:" - echo " -h, --help 显示帮助信息" - echo " -q, --quick 快速模式（不生成覆盖率报告）" - echo " -f, --file FILE 运行指定测试文件" - echo " -k, --keyword KEYWORD 运行匹配关键字的测试" - echo " -m, --mark MARK 运行标记的测试（如 slow）" - echo " -x, --stop 遇到失败立即停止" - echo "" - echo "测试模式:" - echo " all 运行所有测试（默认）" - echo " unit 只运行单元测试" - echo " integration 只运行集成测试" - echo "" - echo "示例:" - echo " $0 # 运行所有测试并生成报告" - echo " $0 -q # 快速运行所有测试" - echo " $0 -f test_cli.py # 运行指定测试文件" - echo " $0 -k test_render # 运行包含关键字的测试" - echo " $0 -m slow # 运行标记为 slow 的测试" - echo " $0 -x # 遇到失败立即停止" +normalize_path() { + # Git Bash accepts native drive/UNC paths when their separators are forward + # slashes. Leave POSIX filenames (which can contain backslashes) unchanged. + case "$1" in + [A-Za-z]:\\*|\\\\*) printf '%s\n' "${1//\\//}" ;; + *) printf '%s\n' "$1" ;; + esac } -# 主函数 -main() { - local quick_mode=false - local pytest_args=() - - # 解析命令行参数 - while [[ $# -gt 0 ]]; do - case $1 in - -h|--help) - show_help - exit 0 - ;; - -q|--quick) - quick_mode=true - shift - ;; - -f|--file) - pytest_args+=("tests/$2") - shift 2 - ;; - -k|--keyword) - pytest_args+=("-k" "$2") - shift 2 - ;; - -m|--mark) - pytest_args+=("-m" "$2") - shift 2 - ;; - -x|--stop) - pytest_args+=("-x") - shift - ;; - all) - shift - ;; - unit|integration) - echo -e "${YELLOW}警告：$1 模式暂未实现，将运行所有测试${NC}" - shift - ;; - *) - echo -e "${RED}未知选项：$1${NC}" - echo "使用 -h 查看帮助" - exit 1 - ;; - esac - done - - # 进入项目目录 - cd "${PROJECT_ROOT}" - - echo "========================================" - echo -e "${BLUE} 测试执行与报告生成脚本${NC}" - echo "========================================" - echo "" - - # 激活环境 - activate_conda - - # 创建报告目录 - create_report_dir - - # 运行测试 - if [ "$quick_mode" = true ]; then - echo -e "${YELLOW}>>> 快速模式：跳过覆盖率报告${NC}" - run_tests "${pytest_args[@]}" - else - echo -e "${BLUE}>>> 完整模式：生成覆盖率报告${NC}" - run_tests "${pytest_args[@]}" - fi - - local test_result=$? - - # 生成报告摘要 - generate_summary - - echo -e "${BLUE}========================================${NC}" - echo -e "${BLUE}测试完成！${NC}" - echo -e "${BLUE}========================================${NC}" - - exit $test_result -} +quick_mode=false +test_paths=() +pytest_args=() +while [[ $# -gt 0 ]]; do + case "$1" in + -h|--help) + show_help + exit 0 + ;; + -q|--quick) + quick_mode=true + shift + ;; + -f|--file) + require_value "$@" + file_path="$(normalize_path "$2")" + case "$file_path" in + /*|[A-Za-z]:/*|tests/*|./tests/*) test_paths+=("$file_path") ;; + *) test_paths+=("tests/$file_path") ;; + esac + shift 2 + ;; + -k|--keyword) + require_value "$@" + pytest_args+=("-k" "$2") + shift 2 + ;; + -m|--mark) + require_value "$@" + pytest_args+=("-m" "$2") + shift 2 + ;; + -x|--stop) + pytest_args+=("-x") + shift + ;; + --reports-dir) + require_value "$@" + REPORTS_DIR="$(normalize_path "$2")" + shift 2 + ;; + all) + shift + ;; + --) + shift + pytest_args+=("$@") + break + ;; + *) + usage_error "unknown option or mode: $1" + ;; + esac +done + +cd "$PROJECT_ROOT" +if ! command -v "$PYTHON" >/dev/null 2>&1; then + printf 'Error: Python executable not found: %s\n' "$PYTHON" >&2 + exit 127 +fi + +if [[ ${#test_paths[@]} -eq 0 ]]; then + test_paths=("tests/") +fi + +coverage_args=() +if [[ "$quick_mode" == false ]]; then + mkdir -p "$REPORTS_DIR/htmlcov" "$REPORTS_DIR/xml" + coverage_args=( + --cov=pt_snap_cli + "--cov-report=html:$REPORTS_DIR/htmlcov" + "--cov-report=xml:$REPORTS_DIR/xml/coverage.xml" + --cov-report=term-missing + ) + printf 'Coverage reports: %s\n' "$REPORTS_DIR" +fi -# 执行主函数 -main "$@" +# exec preserves pytest's success, failure, interruption and collection exit codes. +exec "$PYTHON" -m pytest "${test_paths[@]}" "${coverage_args[@]}" \ + -v --tb=short "${pytest_args[@]}" diff --git a/tests/test_package.py b/tests/test_package.py index 86014e2..58e8633 100644 --- a/tests/test_package.py +++ b/tests/test_package.py @@ -2,15 +2,59 @@ from __future__ import annotations -import importlib +from collections.abc import Iterable +from importlib.machinery import PathFinder from importlib.metadata import distribution, requires, version -from pathlib import Path +from pathlib import Path, PurePosixPath import pytest +import pt_snap_cli from pt_snap_cli import __version__ ROOT = Path(__file__).resolve().parents[1] +REMOVED_PACKAGES = ("mcp", "vendor") + + +def _assert_removed_package(package_root: Path, name: str) -> None: + """Allow empty namespace/cache residue, not restored code or package assets. + + Inspect without importing the removed module: a broken or side-effectful + resurrected module must fail this guard without executing its contents. + Legacy adjacent .pyc files remain importable and must not be ignored. + """ + removed_root = package_root / name + candidates = list(removed_root.rglob("*")) + list(package_root.glob(f"{name}.*")) + if removed_root.is_file(): + candidates.append(removed_root) + remaining = [ + path.relative_to(package_root).as_posix() + for path in candidates + if path.is_file() + and not ( + path.parent.name == "__pycache__" + and path.suffix in {".pyc", ".pyo"} + and "." in path.stem + ) + ] + assert not remaining, f"Removed package content remains: {sorted(remaining)}" + + PathFinder.invalidate_caches() + spec = PathFinder.find_spec(f"pt_snap_cli.{name}", [str(package_root)]) + assert spec is None or spec.loader is None, f"Removed package is importable: {spec}" + + +def _removed_distribution_entries(files: Iterable[str]) -> list[str]: + """A shipped removed package is a regression, even if it contains only cache.""" + removed = [] + for file in files: + parts = PurePosixPath(file).parts + if len(parts) >= 2 and parts[0] == "pt_snap_cli": + if any( + parts[1] == name or parts[1].startswith(f"{name}.") for name in REMOVED_PACKAGES + ): + removed.append(file) + return sorted(removed) def _runtime_requirement_names() -> set[str]: @@ -54,6 +98,106 @@ def test_pyproject_does_not_declare_mcp(self) -> None: assert "pt-snap-mcp" not in text assert "mcp>=" not in text - def test_mcp_package_is_removed(self) -> None: - with pytest.raises(ModuleNotFoundError): - importlib.import_module("pt_snap_cli.mcp") + @pytest.mark.parametrize("name", REMOVED_PACKAGES) + def test_removed_packages_have_no_code_or_assets(self, name: str) -> None: + # Check both the selected import location and this checkout. An editable + # install's RECORD does not enumerate the source tree. + package_roots = {Path(path).resolve() for path in pt_snap_cli.__path__} + package_roots.add(ROOT / "src" / "pt_snap_cli") + for package_root in package_roots: + _assert_removed_package(package_root, name) + + def test_distribution_does_not_ship_removed_packages(self) -> None: + files = distribution("pt-snap-cli").files + assert files is not None, "Installed distribution must have a file manifest" + assert not _removed_distribution_entries(str(file) for file in files) + + +@pytest.mark.parametrize("name", REMOVED_PACKAGES) +@pytest.mark.parametrize("layout", ["absent", "empty", "pycache-only"]) +def test_removed_package_guard_accepts_clean_and_cache_only_trees(tmp_path, name, layout): + package_root = tmp_path / "pt_snap_cli" + package_root.mkdir() + removed_root = package_root / name + if layout != "absent": + removed_root.mkdir() + if layout == "pycache-only": + for relative in ( + "__pycache__/__init__.cpython-313.pyc", + "memsnapdump/__pycache__/reader.cpython-313.pyc", + "memsnapdump/__pycache__/reader.cpython-310.opt-1.pyc", + ): + cache = removed_root / relative + cache.parent.mkdir(parents=True, exist_ok=True) + cache.write_bytes(b"stale cache must not be imported or deleted") + + before = {path: path.read_bytes() for path in package_root.rglob("*") if path.is_file()} + spec = PathFinder.find_spec(f"pt_snap_cli.{name}", [str(package_root)]) + if layout == "absent": + assert spec is None + else: + assert spec is not None and spec.loader is None + assert spec.submodule_search_locations is not None + _assert_removed_package(package_root, name) + assert {path: path.read_bytes() for path in package_root.rglob("*") if path.is_file()} == before + + +@pytest.mark.parametrize("name", REMOVED_PACKAGES) +@pytest.mark.parametrize( + "relative", + [ + "__init__.py", + "server.py", + "memsnapdump/reader.py", + "__init__.pyc", + "memsnapdump/schema.json", + "__pycache__/hidden.py", + "__pycache__/importable.pyc", + ], +) +def test_removed_package_guard_rejects_restored_code_and_assets(tmp_path, name, relative): + package_root = tmp_path / "pt_snap_cli" + restored = package_root / name / relative + restored.parent.mkdir(parents=True) + payload = b"raise RuntimeError('must never execute removed package code')\n" + restored.write_bytes(payload) + + with pytest.raises(AssertionError, match="Removed package content remains"): + _assert_removed_package(package_root, name) + assert restored.read_bytes() == payload + + +@pytest.mark.parametrize("name", REMOVED_PACKAGES) +@pytest.mark.parametrize("suffix", [".py", ".pyc", ".so", ".pyd"]) +def test_removed_package_guard_rejects_flat_module_replacements(tmp_path, name, suffix): + restored = tmp_path / f"{name}{suffix}" + restored.write_bytes(b"removed module") + with pytest.raises(AssertionError, match="Removed package content remains"): + _assert_removed_package(tmp_path, name) + + +@pytest.mark.parametrize("name", REMOVED_PACKAGES) +@pytest.mark.parametrize( + "entry", + [ + "{name}/__init__.py", + "{name}/memsnapdump/reader.py", + "{name}.py", + "{name}/__pycache__/old.pyc", + ], +) +def test_distribution_guard_rejects_rediscovered_removed_content(name, entry): + path = "pt_snap_cli/" + entry.format(name=name) + assert _removed_distribution_entries(["pt_snap_cli/cli.py", path]) == [path] + + +def test_distribution_guard_allows_unrelated_modules_and_metadata(): + assert _removed_distribution_entries( + [ + "pt_snap_cli/cli.py", + "pt_snap_cli/mcp_tools.py", + "pt_snap_cli/vendor_tools/reader.py", + "pt_snap_cli-1.0.dist-info/METADATA", + "pt_snap_cli-1.0.dist-info/licenses/LICENSE", + ] + ) == [] diff --git a/tests/test_test_runner.py b/tests/test_test_runner.py new file mode 100644 index 0000000..44730b8 --- /dev/null +++ b/tests/test_test_runner.py @@ -0,0 +1,264 @@ +"""Exercise the real Bash entry point without starting a nested test suite.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +def _bash() -> str: + if os.name == "nt": + # Prefer Git Bash to Windows' WSL launcher. Git-for-Windows supplies the + # POSIX tools used by the entry point on the Windows compatibility jobs. + git = shutil.which("git") + assert git is not None, "Git for Windows is required to test the Bash entry point" + bash = Path(git).resolve().parents[1] / "bin" / "bash.exe" + assert bash.is_file(), f"Git Bash not found at {bash}" + return str(bash) + bash = shutil.which("bash") + assert bash is not None, "Bash is required to test the Bash entry point" + return bash + + +@pytest.fixture +def shell_runner(tmp_path): + root = tmp_path / "repository with spaces" + (root / "tests").mkdir(parents=True) + script = root / "tests" / "run_tests.sh" + shutil.copyfile(ROOT / "tests" / "run_tests.sh", script) + outside = tmp_path / "outside repository" + outside.mkdir() + recorder = tmp_path / "isolated modules" + recorder.mkdir() + log = tmp_path / "invocation.json" + # Python really dispatches `-m pytest`, but this isolated module records the + # call instead of loading pytest/plugins, collecting tests or generating data. + (recorder / "pytest.py").write_text( + "import json, os, sys\n" + "from pathlib import Path\n" + "Path(os.environ['RUNNER_TEST_LOG']).write_text(json.dumps({\n" + " 'argv': sys.argv[1:], 'cwd': os.getcwd(), 'executable': sys.executable,\n" + "}), encoding='utf-8')\n" + "raise SystemExit(int(os.environ['RUNNER_TEST_EXIT']))\n", + encoding="utf-8", + ) + env = os.environ.copy() + env.update( + PYTHONPATH=str(recorder), + PYTHONUTF8="1", + PATH=str(Path(sys.executable).parent) + os.pathsep + env.get("PATH", ""), + RUNNER_TEST_LOG=str(log), + ) + + def run(*args, exit_code=0, python=Path(sys.executable).as_posix()): + selected_env = dict(env, RUNNER_TEST_EXIT=str(exit_code)) + if python is None: + selected_env.pop("PYTHON", None) + else: + selected_env["PYTHON"] = python + return subprocess.run( + [_bash(), "--noprofile", "--norc", script.as_posix(), *args], + cwd=outside, + env=selected_env, + text=True, + capture_output=True, + check=False, + timeout=30, + ) + + def invocation(): + return json.loads(log.read_text(encoding="utf-8")) + + return SimpleNamespace(root=root, log=log, run=run, invocation=invocation) + + +def test_default_runner_uses_active_python_and_keeps_coverage(shell_runner): + result = shell_runner.run(python=None) + assert result.returncode == 0, result.stdout + result.stderr + call = shell_runner.invocation() + assert Path(call["cwd"]) == shell_runner.root + assert Path(call["executable"]).resolve() == Path(sys.executable).resolve() + assert call["argv"] == [ + "tests/", + "--cov=pt_snap_cli", + "--cov-report=html:test_reports/htmlcov", + "--cov-report=xml:test_reports/xml/coverage.xml", + "--cov-report=term-missing", + "-v", + "--tb=short", + ] + assert (shell_runner.root / "test_reports" / "htmlcov").is_dir() + assert (shell_runner.root / "test_reports" / "xml").is_dir() + + +def test_quick_mode_forwards_selection_and_pytest_arguments_without_reports(shell_runner): + result = shell_runner.run( + "--quick", + "-f", + "test_package.py::TestPackage::test_version", + "--file", + "tests/query/test_builder.py", + "--keyword", + "version or builder", + "-m", + "not slow and not nightly", + "--stop", + "--", + "-q", + "--maxfail=2", + ) + assert result.returncode == 0, result.stdout + result.stderr + assert shell_runner.invocation()["argv"] == [ + "tests/test_package.py::TestPackage::test_version", + "tests/query/test_builder.py", + "-v", + "--tb=short", + "-k", + "version or builder", + "-m", + "not slow and not nightly", + "-x", + "-q", + "--maxfail=2", + ] + assert not (shell_runner.root / "test_reports").exists() + + +def test_short_aliases_and_all_mode_preserve_option_only_invocations(shell_runner): + result = shell_runner.run("-q", "all", "-k", "test_version", "--mark", "not slow", "-x") + assert result.returncode == 0, result.stdout + result.stderr + assert shell_runner.invocation()["argv"] == [ + "tests/", + "-v", + "--tb=short", + "-k", + "test_version", + "-m", + "not slow", + "-x", + ] + assert not (shell_runner.root / "test_reports").exists() + + +def test_explicit_interpreter_and_absolute_test_path_are_respected(shell_runner): + target = shell_runner.root / "tests" / "test with spaces.py" + result = shell_runner.run("--quick", "--file", target.as_posix()) + assert result.returncode == 0, result.stdout + result.stderr + call = shell_runner.invocation() + assert Path(call["executable"]).resolve() == Path(sys.executable).resolve() + assert Path(call["argv"][0]) == target + assert call["argv"][1:] == ["-v", "--tb=short"] + + +def test_custom_report_directory_preserves_spaces_and_repo_relative_paths(shell_runner): + result = shell_runner.run("--reports-dir", ".tmp/reports with spaces") + assert result.returncode == 0, result.stdout + result.stderr + args = shell_runner.invocation()["argv"] + assert "--cov-report=html:.tmp/reports with spaces/htmlcov" in args + assert "--cov-report=xml:.tmp/reports with spaces/xml/coverage.xml" in args + assert (shell_runner.root / ".tmp/reports with spaces/htmlcov").is_dir() + assert (shell_runner.root / ".tmp/reports with spaces/xml").is_dir() + assert not (shell_runner.root / "test_reports").exists() + + +@pytest.mark.parametrize( + ("supplied", "expected"), + [ + (r"C:\example tests\test_case.py", "C:/example tests/test_case.py"), + ("C:/example tests/test_case.py", "C:/example tests/test_case.py"), + (r"\\server\share\test_case.py", "//server/share/test_case.py"), + ("//server/share/test_case.py", "//server/share/test_case.py"), + ("./tests/test_package.py", "./tests/test_package.py"), + ], +) +def test_file_selection_accepts_native_drive_and_unc_path_syntax(shell_runner, supplied, expected): + # These selectors are recorded only; no network share or external file is + # accessed. Windows Bash may translate forward slashes back to native ones. + result = shell_runner.run("--quick", "--file", supplied) + assert result.returncode == 0, result.stdout + result.stderr + assert shell_runner.invocation()["argv"][0].replace("\\", "/") == expected + + +@pytest.mark.parametrize("native_separators", [False, True]) +def test_report_directory_accepts_absolute_temp_paths(shell_runner, tmp_path, native_separators): + report = tmp_path / "absolute reports" + supplied = str(report) if native_separators else report.as_posix() + result = shell_runner.run("--reports-dir", supplied) + assert result.returncode == 0, result.stdout + result.stderr + args = [arg.replace("\\", "/") for arg in shell_runner.invocation()["argv"]] + assert f"--cov-report=html:{report.as_posix()}/htmlcov" in args + assert f"--cov-report=xml:{report.as_posix()}/xml/coverage.xml" in args + assert (report / "htmlcov").is_dir() + assert (report / "xml").is_dir() + assert not (shell_runner.root / "test_reports").exists() + + +def test_quick_mode_does_not_create_custom_reports(shell_runner): + result = shell_runner.run("--quick", "--reports-dir", "custom reports") + assert result.returncode == 0, result.stdout + result.stderr + assert not any(arg.startswith("--cov") for arg in shell_runner.invocation()["argv"]) + assert not (shell_runner.root / "custom reports").exists() + assert not (shell_runner.root / "test_reports").exists() + + +@pytest.mark.parametrize("exit_code", [0, 1, 2, 3, 4, 5]) +@pytest.mark.parametrize("options", [(), ("--quick",)]) +def test_runner_preserves_pytest_exit_codes(shell_runner, exit_code, options): + result = shell_runner.run(*options, exit_code=exit_code) + assert result.returncode == exit_code, result.stdout + result.stderr + assert shell_runner.log.is_file() + + +@pytest.mark.parametrize("option", ["-h", "--help"]) +def test_help_needs_no_python_and_creates_no_reports(shell_runner, option): + result = shell_runner.run(option, python="missing-python-for-runner-test") + assert result.returncode == 0 + assert "PYTHON" in result.stdout and "--quick" in result.stdout + assert "not a CI gate" in result.stdout + assert not shell_runner.log.exists() + assert not (shell_runner.root / "test_reports").exists() + + +@pytest.mark.parametrize( + "args", + [ + ("-f",), + ("--file", ""), + ("-k",), + ("--keyword", "--quick"), + ("-m",), + ("--mark", ""), + ("--reports-dir",), + ("--reports-dir", "--quick"), + ("--unknown",), + ("unit",), + ("integration",), + ], +) +def test_invalid_options_fail_before_invoking_python(shell_runner, args): + result = shell_runner.run(*args) + assert result.returncode == 2 + assert "Error:" in result.stderr + assert not shell_runner.log.exists() + assert not (shell_runner.root / "test_reports").exists() + + +def test_missing_interpreter_is_an_actionable_failure(shell_runner): + result = shell_runner.run(python="missing-python-for-runner-test") + assert result.returncode == 127 + assert "Python executable not found: missing-python-for-runner-test" in result.stderr + assert not shell_runner.log.exists() + assert not (shell_runner.root / "test_reports").exists() + + +def test_default_report_directory_is_ignored(): + assert "test_reports/" in (ROOT / ".gitignore").read_text(encoding="utf-8").splitlines() From f59fe08fc33d6489cb9154809cdf58e341703028 Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 14:53:20 +0800 Subject: [PATCH 02/10] test: add bounded generated pagination and validation properties (#280) --- pyproject.toml | 1 + tests/core/test_import_metadata_properties.py | 202 ++++++++++++++++++ .../test_snapshot_validation_properties.py | 64 ++++++ tests/query/test_pagination_properties.py | 198 +++++++++++++++++ tests/query/test_template_properties.py | 178 +++++++++++++++ 5 files changed, 643 insertions(+) create mode 100644 tests/core/test_import_metadata_properties.py create mode 100644 tests/core/test_snapshot_validation_properties.py create mode 100644 tests/query/test_pagination_properties.py create mode 100644 tests/query/test_template_properties.py 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..18a8602 --- /dev/null +++ b/tests/core/test_snapshot_validation_properties.py @@ -0,0 +1,64 @@ +"""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 +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 +from pt_snap_cli.snapshot.tools.adaptors import snapshot2db + +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" + 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 + + 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 not source.exists() + 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" From d98c95b97c98326d3ee6fcc7d6fdee257648295f Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 14:55:06 +0800 Subject: [PATCH 03/10] test: keep runner fixture defaults and package formatting lint-clean --- tests/test_package.py | 21 ++++++++++++--------- tests/test_test_runner.py | 4 +++- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/tests/test_package.py b/tests/test_package.py index 58e8633..edb5e64 100644 --- a/tests/test_package.py +++ b/tests/test_package.py @@ -192,12 +192,15 @@ def test_distribution_guard_rejects_rediscovered_removed_content(name, entry): def test_distribution_guard_allows_unrelated_modules_and_metadata(): - assert _removed_distribution_entries( - [ - "pt_snap_cli/cli.py", - "pt_snap_cli/mcp_tools.py", - "pt_snap_cli/vendor_tools/reader.py", - "pt_snap_cli-1.0.dist-info/METADATA", - "pt_snap_cli-1.0.dist-info/licenses/LICENSE", - ] - ) == [] + assert ( + _removed_distribution_entries( + [ + "pt_snap_cli/cli.py", + "pt_snap_cli/mcp_tools.py", + "pt_snap_cli/vendor_tools/reader.py", + "pt_snap_cli-1.0.dist-info/METADATA", + "pt_snap_cli-1.0.dist-info/licenses/LICENSE", + ] + ) + == [] + ) diff --git a/tests/test_test_runner.py b/tests/test_test_runner.py index 44730b8..768f71e 100644 --- a/tests/test_test_runner.py +++ b/tests/test_test_runner.py @@ -59,7 +59,9 @@ def shell_runner(tmp_path): RUNNER_TEST_LOG=str(log), ) - def run(*args, exit_code=0, python=Path(sys.executable).as_posix()): + current_python = Path(sys.executable).as_posix() + + def run(*args, exit_code=0, python=current_python): selected_env = dict(env, RUNNER_TEST_EXIT=str(exit_code)) if python is None: selected_env.pop("PYTHON", None) From 9da523d0417b198c4ee6a21dfa2a40faa526f3f0 Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:08:01 +0800 Subject: [PATCH 04/10] test: resolve the current lazy snapshot loader in each property example --- tests/core/test_snapshot_validation_properties.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/core/test_snapshot_validation_properties.py b/tests/core/test_snapshot_validation_properties.py index 18a8602..175f616 100644 --- a/tests/core/test_snapshot_validation_properties.py +++ b/tests/core/test_snapshot_validation_properties.py @@ -9,6 +9,7 @@ from __future__ import annotations import copy +import importlib from pathlib import Path from tempfile import TemporaryDirectory @@ -18,7 +19,6 @@ from pt_snap_cli.core.errors import ImportExecutionError from pt_snap_cli.core.snapshot_import_backend import SnapshotImportBackend -from pt_snap_cli.snapshot.tools.adaptors import snapshot2db PROPERTY_SETTINGS = settings( max_examples=25, derandomize=True, database=None, deadline=None, print_blob=True @@ -39,6 +39,8 @@ def test_generated_bad_numeric_snapshot_values_reject_without_publication( 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() @@ -52,6 +54,9 @@ def load_representation(path: Path) -> dict[str, object]: 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) @@ -59,6 +64,6 @@ def load_representation(path: Path) -> dict[str, object]: assert isinstance(error.value.__cause__, ValueError) assert loads == [source] assert representation == before - assert not source.exists() + assert source.read_bytes() == source_bytes assert destination.read_bytes() == previous assert list(output.iterdir()) == [destination] From 8432100e682ada528db256b437d91119d84d2943 Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:09:14 +0800 Subject: [PATCH 05/10] test: keep retired-package fixtures independent of historical branding --- tests/test_package.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/test_package.py b/tests/test_package.py index edb5e64..957d480 100644 --- a/tests/test_package.py +++ b/tests/test_package.py @@ -122,10 +122,12 @@ def test_removed_package_guard_accepts_clean_and_cache_only_trees(tmp_path, name if layout != "absent": removed_root.mkdir() if layout == "pycache-only": + # The nested name is deliberately arbitrary: removal covers every child + # of these namespaces, not a blacklist of historical vendor identities. for relative in ( "__pycache__/__init__.cpython-313.pyc", - "memsnapdump/__pycache__/reader.cpython-313.pyc", - "memsnapdump/__pycache__/reader.cpython-310.opt-1.pyc", + "retired_package/__pycache__/reader.cpython-313.pyc", + "retired_package/__pycache__/reader.cpython-310.opt-1.pyc", ): cache = removed_root / relative cache.parent.mkdir(parents=True, exist_ok=True) @@ -148,9 +150,9 @@ def test_removed_package_guard_accepts_clean_and_cache_only_trees(tmp_path, name [ "__init__.py", "server.py", - "memsnapdump/reader.py", + "retired_package/reader.py", "__init__.pyc", - "memsnapdump/schema.json", + "retired_package/schema.json", "__pycache__/hidden.py", "__pycache__/importable.pyc", ], @@ -181,7 +183,7 @@ def test_removed_package_guard_rejects_flat_module_replacements(tmp_path, name, "entry", [ "{name}/__init__.py", - "{name}/memsnapdump/reader.py", + "{name}/retired_package/reader.py", "{name}.py", "{name}/__pycache__/old.pyc", ], From 5db7f9866e2b7c4cc7c1630893c511f606ea1fd1 Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 22:32:37 +0800 Subject: [PATCH 06/10] fix: close failed focus checkpoints before verified cleanup --- src/pt_snap_cli/core/focus_service.py | 33 ++- tests/core/test_focus_checkpoint_cleanup.py | 308 ++++++++++++++++++++ 2 files changed, 330 insertions(+), 11 deletions(-) create mode 100644 tests/core/test_focus_checkpoint_cleanup.py diff --git a/src/pt_snap_cli/core/focus_service.py b/src/pt_snap_cli/core/focus_service.py index f6fd29c..054e959 100644 --- a/src/pt_snap_cli/core/focus_service.py +++ b/src/pt_snap_cli/core/focus_service.py @@ -88,8 +88,11 @@ def _safe_focus_path(path: Path) -> None: def _checkpoint_file(path: Path, content: bytes, mode: int) -> tuple[Path, tuple[int, int]]: fd, name = tempfile.mkstemp(dir=path.parent, prefix=".focus.pt-snap-", suffix=".recovery") backup = Path(name) - identity = os.fstat(fd) + identity: tuple[int, int] | None = None + close_attempted = False try: + info = os.fstat(fd) + identity = (info.st_dev, info.st_ino) remaining = memoryview(content) while remaining: written = os.write(fd, remaining) @@ -99,20 +102,28 @@ def _checkpoint_file(path: Path, content: bytes, mode: int) -> tuple[Path, tuple if hasattr(os, "fchmod"): os.fchmod(fd, mode) else: - FocusService._verify_checkpoint(backup, (identity.st_dev, identity.st_ino)) + FocusService._verify_checkpoint(backup, identity) os.chmod(backup, mode) os.fsync(fd) - FocusService._verify_checkpoint(backup, (identity.st_dev, identity.st_ino)) + FocusService._verify_checkpoint(backup, identity) + close_attempted = True + os.close(fd) except BaseException: - try: - FocusService._verify_checkpoint(backup, (identity.st_dev, identity.st_ino)) - backup.unlink() - except OSError: - pass + # Windows cannot unlink an open checkpoint. Do not retry a failed + # close: the descriptor may already be closed and reused. + if not close_attempted: + try: + os.close(fd) + except OSError: + pass + if identity is not None: + try: + FocusService._verify_checkpoint(backup, identity) + backup.unlink() + except OSError: + pass raise - finally: - os.close(fd) - return backup, (identity.st_dev, identity.st_ino) + return backup, identity @staticmethod def _verify_checkpoint(path: Path, identity: tuple[int, int]) -> None: diff --git a/tests/core/test_focus_checkpoint_cleanup.py b/tests/core/test_focus_checkpoint_cleanup.py new file mode 100644 index 0000000..ed1a106 --- /dev/null +++ b/tests/core/test_focus_checkpoint_cleanup.py @@ -0,0 +1,308 @@ +"""Checkpoint failure cleanup closes owned descriptors before verified deletion.""" + +import copy +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import pt_snap_cli.core.focus_service as focus_module +from pt_snap_cli.config import Config +from pt_snap_cli.core.focus_service import FocusService + + +@pytest.fixture +def focus(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("PT_SNAP_DB_PATH", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + service = FocusService(Config()) + path = service._config.project_focus_path() + path.parent.mkdir() + original = b'{ "db_path" : "previous.db", "extra" : [1,2] }\n' + path.write_bytes(original) + service._config._config = {"nested": {"value": "original"}} + return service, path, original + + +@pytest.fixture +def checkpoint_io(monkeypatch): + probe = SimpleNamespace(fd=None, path=None, events=[]) + mkstemp = focus_module.tempfile.mkstemp + close = os.close + fstat = os.fstat + unlink = Path.unlink + verify = FocusService._verify_checkpoint + + def create(*args, **kwargs): + probe.fd, name = mkstemp(*args, **kwargs) + probe.path = Path(name) + return probe.fd, name + + def close_checkpoint(fd): + if fd == probe.fd: + probe.events.append("close") + close(fd) + + def verify_checkpoint(path, identity): + if path == probe.path: + probe.events.append("verify") + verify(path, identity) + + def unlink_checkpoint(path, *args, **kwargs): + if path == probe.path: + # Enforce Windows' close-before-unlink requirement on every runner. + with pytest.raises(OSError): + fstat(probe.fd) + assert probe.events[-1] == "verify" + probe.events.append("unlink") + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(focus_module.tempfile, "mkstemp", create) + monkeypatch.setattr(focus_module.os, "close", close_checkpoint) + monkeypatch.setattr(FocusService, "_verify_checkpoint", staticmethod(verify_checkpoint)) + monkeypatch.setattr(Path, "unlink", unlink_checkpoint) + yield probe + if probe.fd is not None: + try: + fstat(probe.fd) + except OSError: + pass + else: + close(probe.fd) + pytest.fail("checkpoint descriptor leaked") + assert probe.events.count("close") == 1 + + +@pytest.mark.parametrize( + "failure", ["zero", "short-zero", "write", "fchmod", "permissions", "fsync", "verify"] +) +def test_checkpoint_failure_closes_before_cleanup(focus, checkpoint_io, monkeypatch, failure): + service, path, original = focus + state = copy.deepcopy(service._config._config) + error = OSError("injected checkpoint failure") + write = os.write + verify = FocusService._verify_checkpoint + calls = [] + + def fail(*args): + raise error + + def partial_write(fd, content): + calls.append(len(content)) + if failure == "short-zero" and len(calls) == 1: + return write(fd, content[:3]) + return 0 + + def fail_preparation_verify(target, identity): + verify(target, identity) + if "close" not in checkpoint_io.events: + raise error + + if failure in {"zero", "short-zero"}: + monkeypatch.setattr(focus_module.os, "write", partial_write) + elif failure == "permissions": + # Exercise the portable pathname fallback, including real identity checks. + monkeypatch.delattr(focus_module.os, "fchmod", raising=False) + monkeypatch.setattr(focus_module.os, "chmod", fail) + elif failure == "fchmod": + monkeypatch.setattr(focus_module.os, "fchmod", fail, raising=False) + elif failure == "verify": + monkeypatch.setattr( + FocusService, "_verify_checkpoint", staticmethod(fail_preparation_verify) + ) + else: + monkeypatch.setattr(focus_module.os, failure, fail) + + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("checkpoint failure must prevent focus publication") + + if failure in {"zero", "short-zero"}: + assert str(caught.value) == "Could not write focus recovery checkpoint" + assert calls == ( + [len(original)] if failure == "zero" else [len(original), len(original) - 3] + ) + else: + assert caught.value is error + assert checkpoint_io.events[-3:] == ["close", "verify", "unlink"] + assert path.read_bytes() == original + assert service._config._config == state + assert list(path.parent.iterdir()) == [path] + + +@pytest.mark.parametrize("rollback", [False, True], ids=["commit", "rollback"]) +def test_short_writes_keep_successful_transaction_semantics( + focus, checkpoint_io, monkeypatch, rollback +): + service, path, original = focus + state = copy.deepcopy(service._config._config) + write = os.write + writes = [] + error = RuntimeError("focus mutation failed") + + def short_write(fd, content): + writes.append(len(content)) + return write(fd, content[:3]) + + def mutate(): + with service.project_focus_transaction(): + assert checkpoint_io.events[-1] == "close" + assert checkpoint_io.path.read_bytes() == original + path.write_bytes(b"committed focus") + service._config._config["nested"]["value"] = "changed" + if rollback: + raise error + + monkeypatch.setattr(focus_module.os, "write", short_write) + if rollback: + with pytest.raises(RuntimeError) as caught: + mutate() + assert caught.value is error + assert path.read_bytes() == original + assert service._config._config == state + else: + mutate() + assert path.read_bytes() == b"committed focus" + assert service._config._config == {"nested": {"value": "changed"}} + assert checkpoint_io.events[-2:] == ["verify", "unlink"] + assert len(writes) > 1 + assert list(path.parent.iterdir()) == [path] + + +@pytest.mark.parametrize("cleanup_failure", ["close", "unlink", "verify"]) +def test_cleanup_error_keeps_original_checkpoint_error( + focus, checkpoint_io, monkeypatch, cleanup_failure +): + service, path, original = focus + error = OSError("original checkpoint write failure") + close = os.close + verify = FocusService._verify_checkpoint + + def fail_write(*args): + raise error + + def close_then_fail(fd): + close(fd) + raise OSError("late close error") + + def fail_unlink(target, *args, **kwargs): + assert target == checkpoint_io.path + with pytest.raises(OSError): + os.fstat(checkpoint_io.fd) + assert checkpoint_io.events[-2:] == ["close", "verify"] + raise OSError("unlink denied") + + def fail_verify(target, identity): + verify(target, identity) + raise OSError("verification denied") + + monkeypatch.setattr(focus_module.os, "write", fail_write) + if cleanup_failure == "close": + monkeypatch.setattr(focus_module.os, "close", close_then_fail) + elif cleanup_failure == "unlink": + monkeypatch.setattr(Path, "unlink", fail_unlink) + else: + monkeypatch.setattr(FocusService, "_verify_checkpoint", staticmethod(fail_verify)) + + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("checkpoint failure must prevent focus publication") + assert caught.value is error + assert path.read_bytes() == original + assert checkpoint_io.path.exists() is (cleanup_failure != "close") + + +def test_close_failure_after_checkpoint_write_does_not_retry_close( + focus, checkpoint_io, monkeypatch +): + service, path, original = focus + close = os.close + error = OSError("late checkpoint close error") + + def close_then_fail(fd): + close(fd) + raise error + + monkeypatch.setattr(focus_module.os, "close", close_then_fail) + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("checkpoint close failure must prevent focus publication") + assert caught.value is error + assert checkpoint_io.events[-3:] == ["close", "verify", "unlink"] + assert path.read_bytes() == original + assert list(path.parent.iterdir()) == [path] + + +def test_unknown_checkpoint_identity_closes_without_unverified_deletion( + focus, checkpoint_io, monkeypatch +): + service, path, original = focus + error = OSError("checkpoint identity unavailable") + + def fail_fstat(fd): + assert fd == checkpoint_io.fd + raise error + + monkeypatch.setattr(focus_module.os, "fstat", fail_fstat) + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("unverified checkpoint must prevent focus publication") + assert caught.value is error + assert checkpoint_io.events == ["close"] + assert checkpoint_io.path.read_bytes() == b"" + assert path.read_bytes() == original + + +def test_replaced_checkpoint_after_close_is_not_deleted(focus, checkpoint_io, monkeypatch): + service, path, original = focus + close = os.close + replacement = path.parent / "replacement" + replacement.write_bytes(b"independent file") + saved = path.parent / "owned-checkpoint" + error = OSError("original checkpoint write failure") + + def fail_write(*args): + raise error + + def close_then_replace(fd): + close(fd) + # Keep both real files alive, so their distinct identities cannot be reused. + checkpoint_io.path.rename(saved) + replacement.rename(checkpoint_io.path) + + monkeypatch.setattr(focus_module.os, "write", fail_write) + monkeypatch.setattr(focus_module.os, "close", close_then_replace) + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("checkpoint failure must prevent focus publication") + assert caught.value is error + assert checkpoint_io.events == ["close", "verify"] + assert checkpoint_io.path.read_bytes() == b"independent file" + assert saved.read_bytes() == b"" + assert path.read_bytes() == original + + +def test_symlink_metadata_at_cleanup_is_rejected(focus, checkpoint_io, monkeypatch): + service, path, original = focus + is_symlink = Path.is_symlink + error = OSError("original checkpoint write failure") + + def fail_write(*args): + raise error + + def checkpoint_is_symlink(target): + # Inject only the metadata observation; no Windows symlink privilege is needed. + return target == checkpoint_io.path or is_symlink(target) + + monkeypatch.setattr(focus_module.os, "write", fail_write) + monkeypatch.setattr(Path, "is_symlink", checkpoint_is_symlink) + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("checkpoint failure must prevent focus publication") + assert caught.value is error + assert checkpoint_io.events == ["close", "verify"] + assert checkpoint_io.path.read_bytes() == b"" + assert path.read_bytes() == original From 203950ad5a706284d05fd2ee1d9d480c5a00055e Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 22:50:47 +0800 Subject: [PATCH 07/10] test: scope checkpoint identity failure injection to owned descriptor --- tests/core/test_focus_checkpoint_cleanup.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/tests/core/test_focus_checkpoint_cleanup.py b/tests/core/test_focus_checkpoint_cleanup.py index ed1a106..1e3c6f4 100644 --- a/tests/core/test_focus_checkpoint_cleanup.py +++ b/tests/core/test_focus_checkpoint_cleanup.py @@ -241,15 +241,18 @@ def test_unknown_checkpoint_identity_closes_without_unverified_deletion( ): service, path, original = focus error = OSError("checkpoint identity unavailable") + fstat = os.fstat def fail_fstat(fd): - assert fd == checkpoint_io.fd - raise error + if fd == checkpoint_io.fd: + raise error + return fstat(fd) - monkeypatch.setattr(focus_module.os, "fstat", fail_fstat) - with pytest.raises(OSError) as caught: - with service.project_focus_transaction(): - pytest.fail("unverified checkpoint must prevent focus publication") + with monkeypatch.context() as patch: + patch.setattr(focus_module.os, "fstat", fail_fstat) + with pytest.raises(OSError) as caught: + with service.project_focus_transaction(): + pytest.fail("unverified checkpoint must prevent focus publication") assert caught.value is error assert checkpoint_io.events == ["close"] assert checkpoint_io.path.read_bytes() == b"" From 8df94284b2eb19bc1e6a3442a145d8244d55446d Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 22:53:34 +0800 Subject: [PATCH 08/10] fix: normalize malformed YAML and snapshot input failures --- .../core/snapshot_import_backend.py | 10 +- src/pt_snap_cli/query/config.py | 61 ++- src/pt_snap_cli/snapshot/PROVENANCE.md | 15 + src/pt_snap_cli/snapshot/base/entities.py | 72 ++- .../snapshot/tools/adaptors/snapshot2db.py | 7 + .../test_snapshot_input_errors_properties.py | 434 ++++++++++++++++++ .../query/test_input_structure_properties.py | 215 +++++++++ tests/snapshot/test_entities.py | 8 +- tests/snapshot/test_snapshot2db_runtime.py | 6 +- 9 files changed, 795 insertions(+), 33 deletions(-) create mode 100644 tests/core/test_snapshot_input_errors_properties.py create mode 100644 tests/query/test_input_structure_properties.py 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 40df201..5ae4c9e 100644 --- a/src/pt_snap_cli/snapshot/PROVENANCE.md +++ b/src/pt_snap_cli/snapshot/PROVENANCE.md @@ -299,3 +299,18 @@ Read-only queries preserve path identity, classify absent/corrupt evidence, and conserve live block, requested-byte and occupancy totals. No upstream code or new executable fixtures were incorporated; source mappings and licenses remain unchanged. This supersedes the unmerged implementation proposed in PR #207. + +## 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, +and nonempty trace action strings. 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..13fea1c 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 @@ -163,7 +201,12 @@ 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")) + if not isinstance(block_dict["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( @@ -231,7 +274,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 +296,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 +377,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 +392,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..54620d6 --- /dev/null +++ b/tests/core/test_snapshot_input_errors_properties.py @@ -0,0 +1,434 @@ +"""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 DeviceSnapshot, TraceEntry + +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.integers(-100, 100), st.lists(SCALARS, max_size=3))) +def test_generated_non_string_block_state_is_controlled(value: object) -> None: + representation = _valid_representation() + representation["segments"][0]["blocks"][0]["state"] = value + cause = _assert_representation_rejected(representation, ValueError, True) + assert "Block field 'state' must be a string" in str(cause) + + +@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: From 3fbab2a0f60d3d67a0ae38012cb707cf6e641cde Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Sun, 1 Nov 2026 23:04:22 +0800 Subject: [PATCH 09/10] fix: validate declared block states with precise literal types --- src/pt_snap_cli/snapshot/PROVENANCE.md | 5 ++- src/pt_snap_cli/snapshot/base/entities.py | 21 ++++++++-- .../test_snapshot_input_errors_properties.py | 38 ++++++++++++++++++- 3 files changed, 58 insertions(+), 6 deletions(-) diff --git a/src/pt_snap_cli/snapshot/PROVENANCE.md b/src/pt_snap_cli/snapshot/PROVENANCE.md index 5ae4c9e..729618e 100644 --- a/src/pt_snap_cli/snapshot/PROVENANCE.md +++ b/src/pt_snap_cli/snapshot/PROVENANCE.md @@ -304,8 +304,9 @@ unchanged. This supersedes the unmerged implementation proposed in PR #207. First-party changes in `base/entities.py` validate the mappings, sequences and required keys consumed by entity construction, finite allocator numeric fields, -and nonempty trace action strings. Failures of existing trace integer conversion -become `ValueError` with the cause preserved. `tools/adaptors/snapshot2db.py` +the declared three-state block enum, and nonempty trace action strings. 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 diff --git a/src/pt_snap_cli/snapshot/base/entities.py b/src/pt_snap_cli/snapshot/base/entities.py index 13fea1c..8dac641 100644 --- a/src/pt_snap_cli/snapshot/base/entities.py +++ b/src/pt_snap_cli/snapshot/base/entities.py @@ -176,6 +176,22 @@ class BlockState: INACTIVE = "inactive" +def _parse_block_state( + raw: object, +) -> Literal["active_allocated", "active_pending_free", "inactive"]: + if not isinstance(raw, str): + raise ValueError("Block field 'state' must be a string") + if raw == "active_allocated": + return "active_allocated" + if raw == "active_pending_free": + return "active_pending_free" + if raw == "inactive": + return "inactive" + raise ValueError( + "Block field 'state' must be one of: active_allocated, active_pending_free, inactive" + ) + + @dataclass class Block: # A piece of memory returned from the allocator, or @@ -203,8 +219,7 @@ class Block: 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")) - if not isinstance(block_dict["state"], str): - raise ValueError("Block field 'state' must be a string") + state = _parse_block_state(block_dict["state"]) frame_dicts = block_dict.get("frames", []) _validate_sequence(frame_dicts, "Frames") if _raw_frames: @@ -213,7 +228,7 @@ def from_dict(cls, block_dict: dict[str, Any], _raw_frames: bool = False): 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, ) diff --git a/tests/core/test_snapshot_input_errors_properties.py b/tests/core/test_snapshot_input_errors_properties.py index 54620d6..c183da7 100644 --- a/tests/core/test_snapshot_input_errors_properties.py +++ b/tests/core/test_snapshot_input_errors_properties.py @@ -15,7 +15,7 @@ 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 DeviceSnapshot, TraceEntry +from pt_snap_cli.snapshot.base import Block, DeviceSnapshot, TraceEntry PROPERTY_SETTINGS = settings( max_examples=25, derandomize=True, database=None, deadline=None, print_blob=True @@ -405,6 +405,42 @@ def test_generated_non_string_block_state_is_controlled(value: object) -> None: assert "Block field 'state' must be a string" in str(cause) +@PROPERTY_SETTINGS +@given( + state=st.one_of( + st.sampled_from(["", "unknown", "99", "active", "ACTIVE_ALLOCATED"]), + st.text(max_size=24).map(lambda value: f"unsupported:{value}"), + ), + existing=st.booleans(), +) +def test_generated_unknown_block_states_are_controlled(state: str, existing: bool) -> None: + representation = _valid_representation() + representation["segments"][0]["blocks"][0]["state"] = state + cause = _assert_representation_rejected(representation, ValueError, existing) + assert "Block field 'state' must be one of:" in str(cause) + + +@pytest.mark.parametrize("state", ["active_allocated", "active_pending_free", "inactive"]) +@pytest.mark.parametrize("raw_frames", [False, True]) +@PROPERTY_SETTINGS +@given(size=st.integers(1, 4096)) +def test_generated_declared_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("raw_frames", [False, True]) @PROPERTY_SETTINGS @example(size=16.0, use_tuples=False) From b9c8599865cc959e9c2b3c25c4c6a10aeadf1f55 Mon Sep 17 00:00:00 2001 From: codeasier <48463012+codeasier@users.noreply.github.com> Date: Mon, 2 Nov 2026 00:06:51 +0800 Subject: [PATCH 10/10] fix: preserve producer block state strings without narrowing compatibility --- src/pt_snap_cli/snapshot/PROVENANCE.md | 4 +- src/pt_snap_cli/snapshot/base/entities.py | 28 +---- .../test_snapshot_input_errors_properties.py | 110 +++++++++++++++--- 3 files changed, 101 insertions(+), 41 deletions(-) diff --git a/src/pt_snap_cli/snapshot/PROVENANCE.md b/src/pt_snap_cli/snapshot/PROVENANCE.md index 729618e..0da45d7 100644 --- a/src/pt_snap_cli/snapshot/PROVENANCE.md +++ b/src/pt_snap_cli/snapshot/PROVENANCE.md @@ -304,7 +304,9 @@ unchanged. This supersedes the unmerged implementation proposed in PR #207. First-party changes in `base/entities.py` validate the mappings, sequences and required keys consumed by entity construction, finite allocator numeric fields, -the declared three-state block enum, and nonempty trace action strings. Failures +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 diff --git a/src/pt_snap_cli/snapshot/base/entities.py b/src/pt_snap_cli/snapshot/base/entities.py index 8dac641..dd1e284 100644 --- a/src/pt_snap_cli/snapshot/base/entities.py +++ b/src/pt_snap_cli/snapshot/base/entities.py @@ -176,22 +176,6 @@ class BlockState: INACTIVE = "inactive" -def _parse_block_state( - raw: object, -) -> Literal["active_allocated", "active_pending_free", "inactive"]: - if not isinstance(raw, str): - raise ValueError("Block field 'state' must be a string") - if raw == "active_allocated": - return "active_allocated" - if raw == "active_pending_free": - return "active_pending_free" - if raw == "inactive": - return "inactive" - raise ValueError( - "Block field 'state' must be one of: active_allocated, active_pending_free, inactive" - ) - - @dataclass class Block: # A piece of memory returned from the allocator, or @@ -200,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 @@ -219,7 +201,9 @@ class Block: 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 = _parse_block_state(block_dict["state"]) + 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: diff --git a/tests/core/test_snapshot_input_errors_properties.py b/tests/core/test_snapshot_input_errors_properties.py index c183da7..abddb9c 100644 --- a/tests/core/test_snapshot_input_errors_properties.py +++ b/tests/core/test_snapshot_input_errors_properties.py @@ -16,6 +16,7 @@ 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 @@ -396,35 +397,34 @@ def test_generated_non_numeric_allocator_metadata_is_controlled( assert f"field '{field}' must be a finite number" in str(cause) -@PROPERTY_SETTINGS -@given(value=st.one_of(st.none(), st.integers(-100, 100), st.lists(SCALARS, max_size=3))) -def test_generated_non_string_block_state_is_controlled(value: object) -> None: - representation = _valid_representation() - representation["segments"][0]["blocks"][0]["state"] = value - cause = _assert_representation_rejected(representation, ValueError, True) - assert "Block field 'state' must be a string" in str(cause) - - @PROPERTY_SETTINGS @given( - state=st.one_of( - st.sampled_from(["", "unknown", "99", "active", "ACTIVE_ALLOCATED"]), - st.text(max_size=24).map(lambda value: f"unsupported:{value}"), + 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_unknown_block_states_are_controlled(state: str, existing: bool) -> None: +def test_generated_non_string_block_state_is_controlled(value: object, existing: bool) -> None: representation = _valid_representation() - representation["segments"][0]["blocks"][0]["state"] = state + representation["segments"][0]["blocks"][0]["state"] = value cause = _assert_representation_rejected(representation, ValueError, existing) - assert "Block field 'state' must be one of:" in str(cause) + assert "Block field 'state' must be a string" in str(cause) -@pytest.mark.parametrize("state", ["active_allocated", "active_pending_free", "inactive"]) @pytest.mark.parametrize("raw_frames", [False, True]) @PROPERTY_SETTINGS -@given(size=st.integers(1, 4096)) -def test_generated_declared_block_states_preserve_eager_and_raw_inputs( +@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 = { @@ -441,6 +441,80 @@ def test_generated_declared_block_states_preserve_eager_and_raw_inputs( 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)