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 1/7] 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 2/7] 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 3/7] 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 4/7] 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 5/7] 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 6/7] 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 7/7] 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""