From 772e31abfb30969ecaa9ad49335c3d4be716b3b3 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Tue, 3 Nov 2026 23:59:28 +0000 Subject: [PATCH 1/7] fix(presets): keep user files when removing preset skills Preset skill cleanup removed the whole skill directory, so files the user kept next to a generated SKILL.md (for example references/) were deleted by `preset remove` and `integration switch`. Remove only the generated SKILL.md and drop the directory when it is empty. Assisted-by: Codex CLI (model: gpt-6.1-sol, autonomous) Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- src/specify_cli/presets/_manager_skills.py | 29 +++++++++-------- .../presets/test_command_remove.py | 32 +++++++++++++++++++ 2 files changed, 48 insertions(+), 13 deletions(-) diff --git a/src/specify_cli/presets/_manager_skills.py b/src/specify_cli/presets/_manager_skills.py index 5a4a4fb739..845e1b48f8 100644 --- a/src/specify_cli/presets/_manager_skills.py +++ b/src/specify_cli/presets/_manager_skills.py @@ -2,7 +2,6 @@ import copy import os -import shutil from pathlib import Path from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union @@ -216,7 +215,7 @@ def apply_to_dir( managed_skill_names if managed_names is None else managed_names ) # Restore skills for commands whose winner is non-preset. - # _unregister_skills_in_dir can rmtree the skill dir, so + # _unregister_skills_in_dir can remove the skill file, so # overrides must be handled directly (create dir + write) # without that call. dir_core_ext_names = [ @@ -1275,11 +1274,13 @@ def _delete_agent_preset_skills( if override_source: owned_sources.add(override_source) if source in owned_sources: - from ..integrations._file_changes import after_file_change, before_file_change + from ..integrations._file_changes import unlink - before_file_change(skill_subdir, removal=True) - shutil.rmtree(skill_subdir) - after_file_change(skill_subdir) + unlink(skill_file) + try: + skill_subdir.rmdir() + except OSError: + pass @staticmethod def _warn_unrestored_skill( @@ -1370,7 +1371,7 @@ def _unregister_skills_in_dir( # is_dir() follows symlinks, so a symlinked skill subdirectory # (with a safe, non-symlinked parent) would otherwise slip past # _safe_skills_dir_for_agent's parent-only check and have - # write_text/rmtree operate through it (#2948). + # writes/removals operate through it (#2948). if not self._validate_skill_subdir( skill_subdir, create=False, skills_root=skills_dir ): @@ -1435,7 +1436,7 @@ def _unregister_skills_in_dir( # source cannot produce restored content, so leave the # existing skill untouched rather than leaking a raw # OSError/UnicodeDecodeError out of `preset remove` — and - # rather than falling through to the rmtree below, which + # rather than falling through to the removal below, which # would delete a skill precisely when its replacement # cannot be generated. Matches the `continue` guards above # (unsafe name, missing subdir, foreign owner), which also @@ -1541,12 +1542,14 @@ def _unregister_skills_in_dir( _write_shared_text(skills_dir, skill_file, skill_content) mutated_names.append(skill_name) else: - # No core or extension template — remove the skill entirely - from ..integrations._file_changes import after_file_change, before_file_change + # No core or extension template — remove only the generated file + from ..integrations._file_changes import unlink - before_file_change(skill_subdir, removal=True) - shutil.rmtree(skill_subdir) - after_file_change(skill_subdir) + unlink(skill_file) + try: + skill_subdir.rmdir() + except OSError: + pass mutated_names.append(skill_name) return mutated_names diff --git a/tests/specify_cli/presets/test_command_remove.py b/tests/specify_cli/presets/test_command_remove.py index 50a7e89f7b..1cf416000c 100644 --- a/tests/specify_cli/presets/test_command_remove.py +++ b/tests/specify_cli/presets/test_command_remove.py @@ -5,6 +5,7 @@ from pathlib import Path from unittest.mock import patch +import pytest from typer.testing import CliRunner from specify_cli import app @@ -19,3 +20,34 @@ def test_remove_installed_preset(project_dir, pack_dir): assert result.exit_code == 0, result.output assert not PresetManager(project_dir).registry.is_installed("test-pack") + + +@pytest.mark.parametrize("operation", ["remove", "switch"]) +@pytest.mark.parametrize("user_reference", [False, True]) +def test_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, operation, user_reference): + monkeypatch.chdir(tmp_path) + runner = CliRunner() + preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" + for args in ( + ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], + ["preset", "add", "--dev", str(preset)], + ): + result = runner.invoke(app, args, catch_exceptions=False) + assert result.exit_code == 0, result.output + skill_dir = tmp_path / ".github/skills/speckit-myext-myextcmd" + reference = skill_dir / "references" / "user.md" + original = b"\xef\xbb\xbf# User reference\r\nKeep my project notes." + if user_reference: + reference.parent.mkdir() + reference.write_bytes(original) + + args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] + result = runner.invoke(app, args, catch_exceptions=False) + + assert result.exit_code == 0, result.output + assert not (skill_dir / "SKILL.md").exists() + if user_reference: + assert reference.is_file() + assert reference.read_bytes() == original + else: + assert not skill_dir.exists() From ccf78c7343b6a2c89538ccaa27b905f220fa5f42 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Thu, 5 Nov 2026 17:00:23 +0000 Subject: [PATCH 2/7] fix(presets): restore a removed skill directory's mode on rollback Removing only SKILL.md and then the empty directory left the directory removal out of the lifecycle journal, so a rollback recreated it with default permissions. Journal the directory once it is empty, so its mode comes back without snapshotting any user files. Assisted-by: Codex CLI (model: gpt-6.1-sol, autonomous) Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- src/specify_cli/integrations/_file_changes.py | 7 ++ src/specify_cli/presets/_manager_skills.py | 8 +-- .../presets/test_command_remove.py | 69 +++++++++++++++++++ 3 files changed, 80 insertions(+), 4 deletions(-) diff --git a/src/specify_cli/integrations/_file_changes.py b/src/specify_cli/integrations/_file_changes.py index 9ed0ebc62f..dd9878bee3 100644 --- a/src/specify_cli/integrations/_file_changes.py +++ b/src/specify_cli/integrations/_file_changes.py @@ -45,3 +45,10 @@ def write_bytes(path: Path, content: bytes) -> int: def unlink(path: Path, *, missing_ok: bool = False) -> None: with changing_file(path, removal=True): path.unlink(missing_ok=missing_ok) + + +def rmdir(path: Path) -> None: + """Journal and remove an empty directory without snapshotting user files.""" + if not any(path.iterdir()): + with changing_file(path, removal=True): + path.rmdir() diff --git a/src/specify_cli/presets/_manager_skills.py b/src/specify_cli/presets/_manager_skills.py index 845e1b48f8..18baf05b4b 100644 --- a/src/specify_cli/presets/_manager_skills.py +++ b/src/specify_cli/presets/_manager_skills.py @@ -1274,11 +1274,11 @@ def _delete_agent_preset_skills( if override_source: owned_sources.add(override_source) if source in owned_sources: - from ..integrations._file_changes import unlink + from ..integrations._file_changes import rmdir, unlink unlink(skill_file) try: - skill_subdir.rmdir() + rmdir(skill_subdir) except OSError: pass @@ -1543,11 +1543,11 @@ def _unregister_skills_in_dir( mutated_names.append(skill_name) else: # No core or extension template — remove only the generated file - from ..integrations._file_changes import unlink + from ..integrations._file_changes import rmdir, unlink unlink(skill_file) try: - skill_subdir.rmdir() + rmdir(skill_subdir) except OSError: pass mutated_names.append(skill_name) diff --git a/tests/specify_cli/presets/test_command_remove.py b/tests/specify_cli/presets/test_command_remove.py index 1cf416000c..dddb395248 100644 --- a/tests/specify_cli/presets/test_command_remove.py +++ b/tests/specify_cli/presets/test_command_remove.py @@ -2,6 +2,8 @@ from __future__ import annotations +import os +import stat from pathlib import Path from unittest.mock import patch @@ -51,3 +53,70 @@ def test_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, operat assert reference.read_bytes() == original else: assert not skill_dir.exists() + + +@pytest.mark.parametrize("operation", ["remove", "switch"]) +@pytest.mark.parametrize("user_reference", [False, True]) +def test_preset_skill_cleanup_rollback_preserves_directory_mode( + tmp_path, monkeypatch, capsys, operation, user_reference, +): + from specify_cli.integrations._helpers import _read_integration_json + from specify_cli.integrations._lifecycle import _FileJournal, _transaction + from specify_cli.integrations.installer import read_records + + project = tmp_path / "project" + project.mkdir() + (tmp_path / "home").mkdir() + monkeypatch.chdir(project) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + monkeypatch.setenv("USERPROFILE", str(tmp_path / "home")) + runner = CliRunner() + preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" + for args in ( + ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], + ["preset", "add", "--dev", str(preset)], + ): + result = runner.invoke(app, args, catch_exceptions=False) + assert result.exit_code == 0, result.output + skill_dir = project / ".github/skills/speckit-myext-myextcmd" + skill_file = skill_dir / "SKILL.md" + original = skill_file.read_bytes() + skill_dir.chmod(0o700) + reference = skill_dir / "references/user.md" + if user_reference: + reference.parent.mkdir() + reference.write_bytes(b"User reference before cleanup") + + snapshot = _FileJournal._snapshot + snapshotted = [] + + def snapshot_without_user_files(self, source, destination): + snapshotted.append(source) + assert source != reference and source != reference.parent + if source == skill_dir and user_reference: + assert not any(source.iterdir()) + return snapshot(self, source, destination) + + monkeypatch.setattr(_FileJournal, "_snapshot", snapshot_without_user_files) + args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] + previous_umask = os.umask(0o022) + try: + with pytest.raises(RuntimeError, match="failure after cleanup"): + with _transaction(project, "claude", _read_integration_json(project), read_records(project)): + result = runner.invoke(app, args, catch_exceptions=False) + assert result.exit_code == 0, result.output + assert not skill_file.exists() + assert skill_dir.exists() == user_reference + if user_reference: + reference.write_bytes(b"User reference edited during transaction") + raise RuntimeError("failure after cleanup") + finally: + os.umask(previous_umask) + + assert skill_file.read_bytes() == original + if os.name != "nt": + assert stat.S_IMODE(skill_dir.stat().st_mode) == 0o700 + if user_reference: + assert skill_dir not in snapshotted + assert reference.read_bytes() == b"User reference edited during transaction" + assert "Recovery snapshots retained" not in capsys.readouterr().out From 2e1dcf41ab7ca12e7f280f4ff3374d4a7ff1e9c3 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Thu, 5 Nov 2026 18:37:08 +0000 Subject: [PATCH 3/7] fix(presets): don't follow a replaced skill directory during cleanup Removing SKILL.md by path followed a skill directory that was replaced by a link after validation, so the cleanup could delete a SKILL.md outside the project. Delete it, and the empty directory, relative to directory handles opened without following links below the project (or global skills) root, and keep the journal calls on the same paths. Without dir_fd, re-check each directory below the root before each delete. Assisted-by: Codex CLI (model: gpt-6.1-sol, autonomous) Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- src/specify_cli/integrations/_file_changes.py | 68 +++++- src/specify_cli/presets/_manager_skills.py | 22 +- .../presets/test_command_remove.py | 224 +++++++++++++++++- 3 files changed, 292 insertions(+), 22 deletions(-) diff --git a/src/specify_cli/integrations/_file_changes.py b/src/specify_cli/integrations/_file_changes.py index dd9878bee3..e78fbb0287 100644 --- a/src/specify_cli/integrations/_file_changes.py +++ b/src/specify_cli/integrations/_file_changes.py @@ -2,8 +2,10 @@ from __future__ import annotations +import os +import stat from collections.abc import Callable -from contextlib import contextmanager +from contextlib import ExitStack, contextmanager from contextvars import ContextVar from pathlib import Path @@ -47,8 +49,64 @@ def unlink(path: Path, *, missing_ok: bool = False) -> None: path.unlink(missing_ok=missing_ok) -def rmdir(path: Path) -> None: - """Journal and remove an empty directory without snapshotting user files.""" - if not any(path.iterdir()): +def unlink_in_directory(path: Path, *, root: Path) -> None: + """Remove a generated file and its empty parent beneath a trusted root.""" + directory = path.absolute().parent + root = root.absolute() + try: + relative = directory.relative_to(root) + if not relative.parts or ".." in relative.parts: + return + if ( + getattr(os, "O_DIRECTORY", 0) + and getattr(os, "O_NOFOLLOW", 0) + and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd + and os.listdir in os.supports_fd + ): + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + with ExitStack() as stack: + parent_fd = os.open(root, flags & ~os.O_NOFOLLOW) + stack.callback(os.close, parent_fd) + for part in relative.parts[:-1]: + parent_fd = os.open(part, flags, dir_fd=parent_fd) + stack.callback(os.close, parent_fd) + directory_fd = os.open(directory.name, flags, dir_fd=parent_fd) + stack.callback(os.close, directory_fd) + with changing_file(path, removal=True): + os.unlink(path.name, dir_fd=directory_fd) + if not os.listdir(directory_fd) and os.path.samestat( + os.fstat(directory_fd), + os.stat(directory.name, dir_fd=parent_fd, follow_symlinks=False), + ): + with changing_file(path.parent, removal=True): + os.rmdir(directory.name, dir_fd=parent_fd) + return + + # Without dir_fd (notably Windows), re-check before each path-based + # deletion. This narrows, but cannot eliminate, concurrent swap races. + def safe_directory() -> bool: + parent = root + for part in relative.parts: + parent = parent / part + info = parent.lstat() + if not stat.S_ISDIR(info.st_mode) or ( + getattr(info, "st_file_attributes", 0) + & stat.FILE_ATTRIBUTE_REPARSE_POINT + ): + return False + return True + + if not safe_directory(): + return + original = directory.stat() with changing_file(path, removal=True): - path.rmdir() + if not safe_directory(): + return + path.unlink() + if not any(directory.iterdir()): + with changing_file(path.parent, removal=True): + if safe_directory() and os.path.samestat(original, directory.stat()): + directory.rmdir() + except (OSError, ValueError): + # A replaced/missing directory or a non-empty one is left in place. + return diff --git a/src/specify_cli/presets/_manager_skills.py b/src/specify_cli/presets/_manager_skills.py index 18baf05b4b..169f6d9215 100644 --- a/src/specify_cli/presets/_manager_skills.py +++ b/src/specify_cli/presets/_manager_skills.py @@ -1274,13 +1274,12 @@ def _delete_agent_preset_skills( if override_source: owned_sources.add(override_source) if source in owned_sources: - from ..integrations._file_changes import rmdir, unlink + from ..integrations._file_changes import unlink_in_directory - unlink(skill_file) - try: - rmdir(skill_subdir) - except OSError: - pass + unlink_in_directory( + skill_file, + root=self.project_root if skill_file.is_relative_to(self.project_root) else skills_dir, + ) @staticmethod def _warn_unrestored_skill( @@ -1543,13 +1542,12 @@ def _unregister_skills_in_dir( mutated_names.append(skill_name) else: # No core or extension template — remove only the generated file - from ..integrations._file_changes import rmdir, unlink + from ..integrations._file_changes import unlink_in_directory - unlink(skill_file) - try: - rmdir(skill_subdir) - except OSError: - pass + unlink_in_directory( + skill_file, + root=self.project_root if skill_file.is_relative_to(self.project_root) else skills_dir, + ) mutated_names.append(skill_name) return mutated_names diff --git a/tests/specify_cli/presets/test_command_remove.py b/tests/specify_cli/presets/test_command_remove.py index dddb395248..549ab1587a 100644 --- a/tests/specify_cli/presets/test_command_remove.py +++ b/tests/specify_cli/presets/test_command_remove.py @@ -25,9 +25,22 @@ def test_remove_installed_preset(project_dir, pack_dir): @pytest.mark.parametrize("operation", ["remove", "switch"]) -@pytest.mark.parametrize("user_reference", [False, True]) -def test_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, operation, user_reference): - monkeypatch.chdir(tmp_path) +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("ancestor", [False, True]) +def test_preset_skill_cleanup_preserves_swapped_link_target( + tmp_path, monkeypatch, operation, fallback, ancestor, +): + project = tmp_path / "project" + project.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + probe = tmp_path / "link" + try: + probe.symlink_to(outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + probe.unlink() + monkeypatch.chdir(project) runner = CliRunner() preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" for args in ( @@ -36,12 +49,189 @@ def test_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, operat ): result = runner.invoke(app, args, catch_exceptions=False) assert result.exit_code == 0, result.output - skill_dir = tmp_path / ".github/skills/speckit-myext-myextcmd" + skill_dir = project / ".github/skills/speckit-myext-myextcmd" + original = (skill_dir / "SKILL.md").read_bytes() + swapped_dir = skill_dir.parent if ancestor else skill_dir + outside_skill = outside / skill_dir.name if ancestor else outside + outside_skill.mkdir(exist_ok=True) + outside_file = outside_skill / "SKILL.md" + outside_file.write_bytes(original) + validate = PresetManager._validate_skill_subdir + swapped = False + + def validate_then_swap(self, directory, **kwargs): + nonlocal swapped + valid = validate(self, directory, **kwargs) + if valid and directory == skill_dir and not kwargs["create"] and not swapped: + swapped_dir.rename(project / "retained-skills") + swapped_dir.symlink_to(outside, target_is_directory=True) + swapped = True + return valid + + monkeypatch.setattr(PresetManager, "_validate_skill_subdir", validate_then_swap) + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] + result = runner.invoke(app, args, catch_exceptions=False) + + assert swapped + assert outside_file.is_file(), result.output + assert outside_file.read_bytes() == original + assert swapped_dir.is_symlink() + assert result.exit_code == 0, result.output + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("replacement", ["link", "loop", "file", "outside", "parent", "boundary"]) +def test_preset_skill_cleanup_skips_unsafe_directory(tmp_path, monkeypatch, fallback, replacement): + from specify_cli.integrations import _file_changes + + directory = tmp_path / "skill" + outside = tmp_path / "outside" + outside.mkdir() + outside_file = outside / "SKILL.md" + outside_file.write_bytes(b"Outside skill") + root = tmp_path + if replacement in ("outside", "parent", "boundary"): + directory.mkdir() + (directory / "SKILL.md").write_bytes(b"Retained skill") + root = tmp_path / "root" + root.mkdir() + if replacement == "parent": + directory = root / ".." / directory.name + elif replacement == "boundary": + root = directory + elif replacement == "file": + directory.write_bytes(b"Replacement file") + else: + try: + directory.symlink_to(directory if replacement == "loop" else outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + observed = [] + token = _file_changes.file_change_observer.set(lambda *args: observed.append(args)) + try: + _file_changes.unlink_in_directory(directory / "SKILL.md", root=root) + finally: + _file_changes.file_change_observer.reset(token) + + assert observed == [] + assert outside_file.read_bytes() == b"Outside skill" + if replacement in ("outside", "parent", "boundary"): + assert (directory / "SKILL.md").read_bytes() == b"Retained skill" + elif replacement == "file": + assert directory.is_file() + else: + assert directory.is_symlink() + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("swap_at", ["before_unlink", "after_unlink", "before_rmdir"]) +def test_preset_skill_cleanup_preserves_concurrent_replacements(tmp_path, monkeypatch, fallback, swap_at): + from specify_cli.integrations import _file_changes + + directory = tmp_path / "skills" / "skill" + directory.mkdir(parents=True) + skill_file = directory / "SKILL.md" + skill_file.write_bytes(b"Generated skill") + outside = tmp_path / "outside" + outside.mkdir() + outside_file = outside / "SKILL.md" + outside_file.write_bytes(b"Outside skill") + outside_skill = outside / directory.name + outside_skill.mkdir() + probe = tmp_path / "link" + try: + probe.symlink_to(outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + probe.unlink() + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + descriptor_io = ( + getattr(os, "O_DIRECTORY", 0) + and getattr(os, "O_NOFOLLOW", 0) + and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd + and os.listdir in os.supports_fd + ) + before = _file_changes.before_file_change + after = _file_changes.after_file_change + swapped = False + + def before_change(path, **kwargs): + nonlocal swapped + before(path, **kwargs) + if path == skill_file and swap_at == "before_unlink": + directory.rename(tmp_path / "retained") + directory.symlink_to(outside, target_is_directory=True) + swapped = True + elif path == directory and swap_at == "before_rmdir": + directory.parent.rename(tmp_path / "retained") + directory.parent.symlink_to(outside if descriptor_io else tmp_path / "retained", target_is_directory=True) + swapped = True + + def after_change(path): + nonlocal swapped + after(path) + if path == skill_file and swap_at == "after_unlink": + directory.rename(tmp_path / "retained") + directory.mkdir() + swapped = True + + monkeypatch.setattr(_file_changes, "before_file_change", before_change) + monkeypatch.setattr(_file_changes, "after_file_change", after_change) + _file_changes.unlink_in_directory(skill_file, root=tmp_path) + + assert swapped + assert outside_file.read_bytes() == b"Outside skill" + assert outside_skill.is_dir() + if swap_at == "before_unlink": + assert directory.is_symlink() + if descriptor_io: + assert not (tmp_path / "retained" / "SKILL.md").exists() + elif swap_at == "after_unlink" or not descriptor_io: + assert directory.is_dir() + + +@pytest.mark.parametrize("operation", ["remove", "switch"]) +@pytest.mark.parametrize( + "user_reference,layout", + [(False, "project"), (True, "project"), (False, "linked-parent")], + ids=["empty", "user-files", "linked-parent"], +) +@pytest.mark.parametrize("fallback", [False, True]) +def test_preset_skill_cleanup_preserves_user_files( + tmp_path, monkeypatch, operation, user_reference, layout, fallback, +): + project = tmp_path + if layout == "linked-parent": + (tmp_path / "real" / "project").mkdir(parents=True) + try: + (tmp_path / "link").symlink_to(tmp_path / "real", target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + project = tmp_path / "link" / "project" + monkeypatch.chdir(project) + runner = CliRunner() + preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" + for args in ( + ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], + ["preset", "add", "--dev", str(preset)], + ): + result = runner.invoke(app, args, catch_exceptions=False) + assert result.exit_code == 0, result.output + # getcwd() resolves parent links; retain the user's lexical project path. + monkeypatch.setattr(Path, "cwd", lambda: project) + skill_dir = project / ".github/skills/speckit-myext-myextcmd" reference = skill_dir / "references" / "user.md" original = b"\xef\xbb\xbf# User reference\r\nKeep my project notes." if user_reference: reference.parent.mkdir() reference.write_bytes(original) + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] result = runner.invoke(app, args, catch_exceptions=False) @@ -55,10 +245,32 @@ def test_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, operat assert not skill_dir.exists() +@pytest.mark.parametrize("operation", ["remove", "switch"]) +@pytest.mark.parametrize("fallback", [False, True]) +def test_preset_skill_cleanup_uses_global_boundary(project_dir, tmp_path, monkeypatch, operation, fallback): + home = tmp_path / "home" + monkeypatch.setattr(Path, "home", lambda: home) + skills_root = home / ".hermes/skills" + skill_dir = skills_root / "speckit-custom" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text("---\nmetadata:\n source: preset:global-preset\n---\nGenerated skill\n") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + manager = PresetManager(project_dir) + if operation == "switch": + manager._delete_agent_preset_skills("hermes", [skill_dir.name], "global-preset") + else: + manager._unregister_skills_in_dir([skill_dir.name], skills_root, "hermes", pack_id="global-preset") + + assert not (skill_dir / "SKILL.md").exists() + assert not skill_dir.exists() + + @pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize("user_reference", [False, True]) +@pytest.mark.parametrize("fallback", [False, True]) def test_preset_skill_cleanup_rollback_preserves_directory_mode( - tmp_path, monkeypatch, capsys, operation, user_reference, + tmp_path, monkeypatch, capsys, operation, user_reference, fallback, ): from specify_cli.integrations._helpers import _read_integration_json from specify_cli.integrations._lifecycle import _FileJournal, _transaction @@ -86,6 +298,8 @@ def test_preset_skill_cleanup_rollback_preserves_directory_mode( if user_reference: reference.parent.mkdir() reference.write_bytes(b"User reference before cleanup") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) snapshot = _FileJournal._snapshot snapshotted = [] From c9bb1590f89eb3c4ef0bd2fee1e13868855cb4e1 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Thu, 5 Nov 2026 19:08:10 +0000 Subject: [PATCH 4/7] fix(presets): only skip expected errors in skill-file cleanup unlink_in_directory() caught every OSError and ValueError, which also hid journal failures (IntegrationInstallError is a ValueError) and real delete errors such as PermissionError. Catch only the expected cases at the operation that raises them and let everything else propagate, as it did before this change. Assisted-by: Codex CLI (model: gpt-6.1-sol, autonomous) Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- src/specify_cli/integrations/_file_changes.py | 120 ++++++++++++------ .../presets/test_command_remove.py | 74 +++++++++++ 2 files changed, 154 insertions(+), 40 deletions(-) diff --git a/src/specify_cli/integrations/_file_changes.py b/src/specify_cli/integrations/_file_changes.py index e78fbb0287..28c36ea457 100644 --- a/src/specify_cli/integrations/_file_changes.py +++ b/src/specify_cli/integrations/_file_changes.py @@ -2,6 +2,7 @@ from __future__ import annotations +import errno import os import stat from collections.abc import Callable @@ -55,16 +56,19 @@ def unlink_in_directory(path: Path, *, root: Path) -> None: root = root.absolute() try: relative = directory.relative_to(root) - if not relative.parts or ".." in relative.parts: - return - if ( - getattr(os, "O_DIRECTORY", 0) - and getattr(os, "O_NOFOLLOW", 0) - and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd - and os.listdir in os.supports_fd - ): - flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) - with ExitStack() as stack: + except ValueError: + return + if not relative.parts or ".." in relative.parts: + return + if ( + getattr(os, "O_DIRECTORY", 0) + and getattr(os, "O_NOFOLLOW", 0) + and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd + and os.listdir in os.supports_fd + ): + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + with ExitStack() as stack: + try: parent_fd = os.open(root, flags & ~os.O_NOFOLLOW) stack.callback(os.close, parent_fd) for part in relative.parts[:-1]: @@ -72,41 +76,77 @@ def unlink_in_directory(path: Path, *, root: Path) -> None: stack.callback(os.close, parent_fd) directory_fd = os.open(directory.name, flags, dir_fd=parent_fd) stack.callback(os.close, directory_fd) - with changing_file(path, removal=True): + except OSError as exc: + if exc.errno not in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR): + raise + return + with changing_file(path, removal=True): + try: os.unlink(path.name, dir_fd=directory_fd) - if not os.listdir(directory_fd) and os.path.samestat( + except FileNotFoundError: + pass + try: + empty = not os.listdir(directory_fd) and os.path.samestat( os.fstat(directory_fd), os.stat(directory.name, dir_fd=parent_fd, follow_symlinks=False), - ): - with changing_file(path.parent, removal=True): - os.rmdir(directory.name, dir_fd=parent_fd) - return + ) + except (FileNotFoundError, NotADirectoryError): + return + if empty: + before_file_change(path.parent, removal=True) + try: + os.rmdir(directory.name, dir_fd=parent_fd) + except OSError as exc: + if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST, errno.ENOENT, errno.ENOTDIR): + raise + return + after_file_change(path.parent) + return - # Without dir_fd (notably Windows), re-check before each path-based - # deletion. This narrows, but cannot eliminate, concurrent swap races. - def safe_directory() -> bool: - parent = root - for part in relative.parts: - parent = parent / part + # Without dir_fd (notably Windows), re-check before each path-based + # deletion. This narrows, but cannot eliminate, concurrent swap races. + def safe_directory() -> os.stat_result | None: + parent = root + for part in relative.parts: + parent = parent / part + try: info = parent.lstat() - if not stat.S_ISDIR(info.st_mode) or ( - getattr(info, "st_file_attributes", 0) - & stat.FILE_ATTRIBUTE_REPARSE_POINT - ): - return False - return True - - if not safe_directory(): + except OSError as exc: + if exc.errno not in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR): + raise + return None + if not stat.S_ISDIR(info.st_mode) or ( + getattr(info, "st_file_attributes", 0) + & stat.FILE_ATTRIBUTE_REPARSE_POINT + ): + return None + return info + + original = safe_directory() + if original is None: + return + with changing_file(path, removal=True): + if safe_directory() is None: return - original = directory.stat() - with changing_file(path, removal=True): - if not safe_directory(): - return + try: path.unlink() - if not any(directory.iterdir()): - with changing_file(path.parent, removal=True): - if safe_directory() and os.path.samestat(original, directory.stat()): - directory.rmdir() - except (OSError, ValueError): - # A replaced/missing directory or a non-empty one is left in place. + except FileNotFoundError: + pass + try: + empty = not any(directory.iterdir()) + except OSError as exc: + if exc.errno not in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR): + raise return + if empty: + before_file_change(path.parent, removal=True) + current = safe_directory() + if current is None or not os.path.samestat(original, current): + return + try: + directory.rmdir() + except OSError as exc: + if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST, errno.ENOENT, errno.ENOTDIR): + raise + return + after_file_change(path.parent) diff --git a/tests/specify_cli/presets/test_command_remove.py b/tests/specify_cli/presets/test_command_remove.py index 549ab1587a..b0115b0a38 100644 --- a/tests/specify_cli/presets/test_command_remove.py +++ b/tests/specify_cli/presets/test_command_remove.py @@ -24,6 +24,80 @@ def test_remove_installed_preset(project_dir, pack_dir): assert not PresetManager(project_dir).registry.is_installed("test-pack") +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("failure", ["unlink", "journal"]) +def test_preset_skill_cleanup_propagates_errors(tmp_path, monkeypatch, fallback, failure): + from specify_cli.integrations import _file_changes + from specify_cli.integrations.installer import IntegrationInstallError + + skill_file = tmp_path / "skill" / "SKILL.md" + skill_file.parent.mkdir() + skill_file.write_bytes(b"Generated skill") + error = PermissionError("unlink denied") if failure == "unlink" else IntegrationInstallError("journal failed") + unlink = os.unlink + + def unlink_or_fail(path, **kwargs): + if path == skill_file or path == skill_file.name: + raise error + return unlink(path, **kwargs) + + def observe(path, before, removal): + raise error + + if failure == "unlink": + monkeypatch.setattr(os, "unlink", unlink_or_fail) + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {unlink_or_fail}) + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + token = _file_changes.file_change_observer.set(observe if failure == "journal" else None) + try: + with pytest.raises(type(error)) as caught: + _file_changes.unlink_in_directory(skill_file, root=tmp_path) + finally: + _file_changes.file_change_observer.reset(token) + + assert caught.value is error + assert skill_file.read_bytes() == b"Generated skill" + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("stage", ["open", "unlink", "scan", "rmdir"]) +def test_preset_skill_cleanup_tolerates_expected_errors(tmp_path, monkeypatch, fallback, stage): + from specify_cli.integrations import _file_changes + + directory = tmp_path / "skill" + directory.mkdir() + skill_file = directory / "SKILL.md" + reference = directory / "user.md" + if stage == "open": + directory.rmdir() + elif stage != "unlink": + skill_file.write_bytes(b"Generated skill") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + before = _file_changes.before_file_change + after = _file_changes.after_file_change + + def before_change(path, **kwargs): + before(path, **kwargs) + if stage == "rmdir" and path == directory: + reference.write_bytes(b"Concurrent user file") + + def after_change(path): + after(path) + if stage == "scan" and path == skill_file: + directory.rmdir() + + monkeypatch.setattr(_file_changes, "before_file_change", before_change) + monkeypatch.setattr(_file_changes, "after_file_change", after_change) + _file_changes.unlink_in_directory(skill_file, root=tmp_path) + + assert not skill_file.exists() + assert directory.exists() == (stage == "rmdir") + if stage == "rmdir": + assert reference.read_bytes() == b"Concurrent user file" + + @pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize("fallback", [False, True]) @pytest.mark.parametrize("ancestor", [False, True]) From d1f218351439387abb467c0de9eb5c895f0a94b2 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Fri, 6 Nov 2026 03:12:53 +0000 Subject: [PATCH 5/7] test(presets): move helper and manager tests out of the command suite The four tests that call unlink_in_directory() directly now live in tests/specify_cli/integrations/test_file_changes.py, renamed after the helper, and the global skills root test that calls PresetManager methods moved to test_manager_skills.py. test_command_remove.py keeps the tests that go through the CLI, as design/cli.md asks. No test changed otherwise. Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- .../integrations/test_file_changes.py | 189 ++++++++++++++++ .../presets/test_command_remove.py | 207 ------------------ .../presets/test_manager_skills.py | 21 ++ 3 files changed, 210 insertions(+), 207 deletions(-) create mode 100644 tests/specify_cli/integrations/test_file_changes.py diff --git a/tests/specify_cli/integrations/test_file_changes.py b/tests/specify_cli/integrations/test_file_changes.py new file mode 100644 index 0000000000..1d9086fa3c --- /dev/null +++ b/tests/specify_cli/integrations/test_file_changes.py @@ -0,0 +1,189 @@ +"""Tests for specify_cli.integrations._file_changes.""" + +from __future__ import annotations + +import os + +import pytest + +from specify_cli.integrations import _file_changes +from specify_cli.integrations.installer import IntegrationInstallError + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("failure", ["unlink", "journal"]) +def test_unlink_in_directory_propagates_errors(tmp_path, monkeypatch, fallback, failure): + skill_file = tmp_path / "skill" / "SKILL.md" + skill_file.parent.mkdir() + skill_file.write_bytes(b"Generated skill") + error = PermissionError("unlink denied") if failure == "unlink" else IntegrationInstallError("journal failed") + unlink = os.unlink + + def unlink_or_fail(path, **kwargs): + if path == skill_file or path == skill_file.name: + raise error + return unlink(path, **kwargs) + + def observe(path, before, removal): + raise error + + if failure == "unlink": + monkeypatch.setattr(os, "unlink", unlink_or_fail) + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {unlink_or_fail}) + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + token = _file_changes.file_change_observer.set(observe if failure == "journal" else None) + try: + with pytest.raises(type(error)) as caught: + _file_changes.unlink_in_directory(skill_file, root=tmp_path) + finally: + _file_changes.file_change_observer.reset(token) + + assert caught.value is error + assert skill_file.read_bytes() == b"Generated skill" + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("stage", ["open", "unlink", "scan", "rmdir"]) +def test_unlink_in_directory_tolerates_expected_errors(tmp_path, monkeypatch, fallback, stage): + directory = tmp_path / "skill" + directory.mkdir() + skill_file = directory / "SKILL.md" + reference = directory / "user.md" + if stage == "open": + directory.rmdir() + elif stage != "unlink": + skill_file.write_bytes(b"Generated skill") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + before = _file_changes.before_file_change + after = _file_changes.after_file_change + + def before_change(path, **kwargs): + before(path, **kwargs) + if stage == "rmdir" and path == directory: + reference.write_bytes(b"Concurrent user file") + + def after_change(path): + after(path) + if stage == "scan" and path == skill_file: + directory.rmdir() + + monkeypatch.setattr(_file_changes, "before_file_change", before_change) + monkeypatch.setattr(_file_changes, "after_file_change", after_change) + _file_changes.unlink_in_directory(skill_file, root=tmp_path) + + assert not skill_file.exists() + assert directory.exists() == (stage == "rmdir") + if stage == "rmdir": + assert reference.read_bytes() == b"Concurrent user file" + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("replacement", ["link", "loop", "file", "outside", "parent", "boundary"]) +def test_unlink_in_directory_skips_unsafe_directory(tmp_path, monkeypatch, fallback, replacement): + directory = tmp_path / "skill" + outside = tmp_path / "outside" + outside.mkdir() + outside_file = outside / "SKILL.md" + outside_file.write_bytes(b"Outside skill") + root = tmp_path + if replacement in ("outside", "parent", "boundary"): + directory.mkdir() + (directory / "SKILL.md").write_bytes(b"Retained skill") + root = tmp_path / "root" + root.mkdir() + if replacement == "parent": + directory = root / ".." / directory.name + elif replacement == "boundary": + root = directory + elif replacement == "file": + directory.write_bytes(b"Replacement file") + else: + try: + directory.symlink_to(directory if replacement == "loop" else outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + observed = [] + token = _file_changes.file_change_observer.set(lambda *args: observed.append(args)) + try: + _file_changes.unlink_in_directory(directory / "SKILL.md", root=root) + finally: + _file_changes.file_change_observer.reset(token) + + assert observed == [] + assert outside_file.read_bytes() == b"Outside skill" + if replacement in ("outside", "parent", "boundary"): + assert (directory / "SKILL.md").read_bytes() == b"Retained skill" + elif replacement == "file": + assert directory.is_file() + else: + assert directory.is_symlink() + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("swap_at", ["before_unlink", "after_unlink", "before_rmdir"]) +def test_unlink_in_directory_preserves_concurrent_replacements(tmp_path, monkeypatch, fallback, swap_at): + directory = tmp_path / "skills" / "skill" + directory.mkdir(parents=True) + skill_file = directory / "SKILL.md" + skill_file.write_bytes(b"Generated skill") + outside = tmp_path / "outside" + outside.mkdir() + outside_file = outside / "SKILL.md" + outside_file.write_bytes(b"Outside skill") + outside_skill = outside / directory.name + outside_skill.mkdir() + probe = tmp_path / "link" + try: + probe.symlink_to(outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + probe.unlink() + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + descriptor_io = ( + getattr(os, "O_DIRECTORY", 0) + and getattr(os, "O_NOFOLLOW", 0) + and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd + and os.listdir in os.supports_fd + ) + before = _file_changes.before_file_change + after = _file_changes.after_file_change + swapped = False + + def before_change(path, **kwargs): + nonlocal swapped + before(path, **kwargs) + if path == skill_file and swap_at == "before_unlink": + directory.rename(tmp_path / "retained") + directory.symlink_to(outside, target_is_directory=True) + swapped = True + elif path == directory and swap_at == "before_rmdir": + directory.parent.rename(tmp_path / "retained") + directory.parent.symlink_to(outside if descriptor_io else tmp_path / "retained", target_is_directory=True) + swapped = True + + def after_change(path): + nonlocal swapped + after(path) + if path == skill_file and swap_at == "after_unlink": + directory.rename(tmp_path / "retained") + directory.mkdir() + swapped = True + + monkeypatch.setattr(_file_changes, "before_file_change", before_change) + monkeypatch.setattr(_file_changes, "after_file_change", after_change) + _file_changes.unlink_in_directory(skill_file, root=tmp_path) + + assert swapped + assert outside_file.read_bytes() == b"Outside skill" + assert outside_skill.is_dir() + if swap_at == "before_unlink": + assert directory.is_symlink() + if descriptor_io: + assert not (tmp_path / "retained" / "SKILL.md").exists() + elif swap_at == "after_unlink" or not descriptor_io: + assert directory.is_dir() diff --git a/tests/specify_cli/presets/test_command_remove.py b/tests/specify_cli/presets/test_command_remove.py index b0115b0a38..64db27ce0f 100644 --- a/tests/specify_cli/presets/test_command_remove.py +++ b/tests/specify_cli/presets/test_command_remove.py @@ -24,80 +24,6 @@ def test_remove_installed_preset(project_dir, pack_dir): assert not PresetManager(project_dir).registry.is_installed("test-pack") -@pytest.mark.parametrize("fallback", [False, True]) -@pytest.mark.parametrize("failure", ["unlink", "journal"]) -def test_preset_skill_cleanup_propagates_errors(tmp_path, monkeypatch, fallback, failure): - from specify_cli.integrations import _file_changes - from specify_cli.integrations.installer import IntegrationInstallError - - skill_file = tmp_path / "skill" / "SKILL.md" - skill_file.parent.mkdir() - skill_file.write_bytes(b"Generated skill") - error = PermissionError("unlink denied") if failure == "unlink" else IntegrationInstallError("journal failed") - unlink = os.unlink - - def unlink_or_fail(path, **kwargs): - if path == skill_file or path == skill_file.name: - raise error - return unlink(path, **kwargs) - - def observe(path, before, removal): - raise error - - if failure == "unlink": - monkeypatch.setattr(os, "unlink", unlink_or_fail) - monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {unlink_or_fail}) - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - token = _file_changes.file_change_observer.set(observe if failure == "journal" else None) - try: - with pytest.raises(type(error)) as caught: - _file_changes.unlink_in_directory(skill_file, root=tmp_path) - finally: - _file_changes.file_change_observer.reset(token) - - assert caught.value is error - assert skill_file.read_bytes() == b"Generated skill" - - -@pytest.mark.parametrize("fallback", [False, True]) -@pytest.mark.parametrize("stage", ["open", "unlink", "scan", "rmdir"]) -def test_preset_skill_cleanup_tolerates_expected_errors(tmp_path, monkeypatch, fallback, stage): - from specify_cli.integrations import _file_changes - - directory = tmp_path / "skill" - directory.mkdir() - skill_file = directory / "SKILL.md" - reference = directory / "user.md" - if stage == "open": - directory.rmdir() - elif stage != "unlink": - skill_file.write_bytes(b"Generated skill") - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - before = _file_changes.before_file_change - after = _file_changes.after_file_change - - def before_change(path, **kwargs): - before(path, **kwargs) - if stage == "rmdir" and path == directory: - reference.write_bytes(b"Concurrent user file") - - def after_change(path): - after(path) - if stage == "scan" and path == skill_file: - directory.rmdir() - - monkeypatch.setattr(_file_changes, "before_file_change", before_change) - monkeypatch.setattr(_file_changes, "after_file_change", after_change) - _file_changes.unlink_in_directory(skill_file, root=tmp_path) - - assert not skill_file.exists() - assert directory.exists() == (stage == "rmdir") - if stage == "rmdir": - assert reference.read_bytes() == b"Concurrent user file" - - @pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize("fallback", [False, True]) @pytest.mark.parametrize("ancestor", [False, True]) @@ -155,119 +81,6 @@ def validate_then_swap(self, directory, **kwargs): assert result.exit_code == 0, result.output -@pytest.mark.parametrize("fallback", [False, True]) -@pytest.mark.parametrize("replacement", ["link", "loop", "file", "outside", "parent", "boundary"]) -def test_preset_skill_cleanup_skips_unsafe_directory(tmp_path, monkeypatch, fallback, replacement): - from specify_cli.integrations import _file_changes - - directory = tmp_path / "skill" - outside = tmp_path / "outside" - outside.mkdir() - outside_file = outside / "SKILL.md" - outside_file.write_bytes(b"Outside skill") - root = tmp_path - if replacement in ("outside", "parent", "boundary"): - directory.mkdir() - (directory / "SKILL.md").write_bytes(b"Retained skill") - root = tmp_path / "root" - root.mkdir() - if replacement == "parent": - directory = root / ".." / directory.name - elif replacement == "boundary": - root = directory - elif replacement == "file": - directory.write_bytes(b"Replacement file") - else: - try: - directory.symlink_to(directory if replacement == "loop" else outside, target_is_directory=True) - except (OSError, NotImplementedError): - pytest.skip("Directory links are unavailable") - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - observed = [] - token = _file_changes.file_change_observer.set(lambda *args: observed.append(args)) - try: - _file_changes.unlink_in_directory(directory / "SKILL.md", root=root) - finally: - _file_changes.file_change_observer.reset(token) - - assert observed == [] - assert outside_file.read_bytes() == b"Outside skill" - if replacement in ("outside", "parent", "boundary"): - assert (directory / "SKILL.md").read_bytes() == b"Retained skill" - elif replacement == "file": - assert directory.is_file() - else: - assert directory.is_symlink() - - -@pytest.mark.parametrize("fallback", [False, True]) -@pytest.mark.parametrize("swap_at", ["before_unlink", "after_unlink", "before_rmdir"]) -def test_preset_skill_cleanup_preserves_concurrent_replacements(tmp_path, monkeypatch, fallback, swap_at): - from specify_cli.integrations import _file_changes - - directory = tmp_path / "skills" / "skill" - directory.mkdir(parents=True) - skill_file = directory / "SKILL.md" - skill_file.write_bytes(b"Generated skill") - outside = tmp_path / "outside" - outside.mkdir() - outside_file = outside / "SKILL.md" - outside_file.write_bytes(b"Outside skill") - outside_skill = outside / directory.name - outside_skill.mkdir() - probe = tmp_path / "link" - try: - probe.symlink_to(outside, target_is_directory=True) - except (OSError, NotImplementedError): - pytest.skip("Directory links are unavailable") - probe.unlink() - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - descriptor_io = ( - getattr(os, "O_DIRECTORY", 0) - and getattr(os, "O_NOFOLLOW", 0) - and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd - and os.listdir in os.supports_fd - ) - before = _file_changes.before_file_change - after = _file_changes.after_file_change - swapped = False - - def before_change(path, **kwargs): - nonlocal swapped - before(path, **kwargs) - if path == skill_file and swap_at == "before_unlink": - directory.rename(tmp_path / "retained") - directory.symlink_to(outside, target_is_directory=True) - swapped = True - elif path == directory and swap_at == "before_rmdir": - directory.parent.rename(tmp_path / "retained") - directory.parent.symlink_to(outside if descriptor_io else tmp_path / "retained", target_is_directory=True) - swapped = True - - def after_change(path): - nonlocal swapped - after(path) - if path == skill_file and swap_at == "after_unlink": - directory.rename(tmp_path / "retained") - directory.mkdir() - swapped = True - - monkeypatch.setattr(_file_changes, "before_file_change", before_change) - monkeypatch.setattr(_file_changes, "after_file_change", after_change) - _file_changes.unlink_in_directory(skill_file, root=tmp_path) - - assert swapped - assert outside_file.read_bytes() == b"Outside skill" - assert outside_skill.is_dir() - if swap_at == "before_unlink": - assert directory.is_symlink() - if descriptor_io: - assert not (tmp_path / "retained" / "SKILL.md").exists() - elif swap_at == "after_unlink" or not descriptor_io: - assert directory.is_dir() - @pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize( @@ -319,26 +132,6 @@ def test_preset_skill_cleanup_preserves_user_files( assert not skill_dir.exists() -@pytest.mark.parametrize("operation", ["remove", "switch"]) -@pytest.mark.parametrize("fallback", [False, True]) -def test_preset_skill_cleanup_uses_global_boundary(project_dir, tmp_path, monkeypatch, operation, fallback): - home = tmp_path / "home" - monkeypatch.setattr(Path, "home", lambda: home) - skills_root = home / ".hermes/skills" - skill_dir = skills_root / "speckit-custom" - skill_dir.mkdir(parents=True) - (skill_dir / "SKILL.md").write_text("---\nmetadata:\n source: preset:global-preset\n---\nGenerated skill\n") - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - manager = PresetManager(project_dir) - if operation == "switch": - manager._delete_agent_preset_skills("hermes", [skill_dir.name], "global-preset") - else: - manager._unregister_skills_in_dir([skill_dir.name], skills_root, "hermes", pack_id="global-preset") - - assert not (skill_dir / "SKILL.md").exists() - assert not skill_dir.exists() - @pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize("user_reference", [False, True]) diff --git a/tests/specify_cli/presets/test_manager_skills.py b/tests/specify_cli/presets/test_manager_skills.py index a0176936d2..124e0bb3bb 100644 --- a/tests/specify_cli/presets/test_manager_skills.py +++ b/tests/specify_cli/presets/test_manager_skills.py @@ -1,5 +1,6 @@ """Tests for preset skill artifacts in specify_cli.presets._manager_skills.""" +import os import shutil from pathlib import Path @@ -2747,6 +2748,26 @@ def test_infer_legacy_skill_provenance_rejects_absolute_registry_name( "intended skills subtree (#2948)" ) + @pytest.mark.parametrize("operation", ["remove", "switch"]) + @pytest.mark.parametrize("fallback", [False, True]) + def test_preset_skill_cleanup_uses_global_boundary(self, project_dir, tmp_path, monkeypatch, operation, fallback): + home = tmp_path / "home" + monkeypatch.setattr(Path, "home", lambda: home) + skills_root = home / ".hermes/skills" + skill_dir = skills_root / "speckit-custom" + skill_dir.mkdir(parents=True) + (skill_dir / "SKILL.md").write_text("---\nmetadata:\n source: preset:global-preset\n---\nGenerated skill\n") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + manager = PresetManager(project_dir) + if operation == "switch": + manager._delete_agent_preset_skills("hermes", [skill_dir.name], "global-preset") + else: + manager._unregister_skills_in_dir([skill_dir.name], skills_root, "hermes", pack_id="global-preset") + + assert not (skill_dir / "SKILL.md").exists() + assert not skill_dir.exists() + class TestWrapStrategy: """Skill registration with inherited wrap metadata.""" From d720e48a5c3575550494052280b22f7c3e08a332 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Fri, 6 Nov 2026 17:55:55 +0000 Subject: [PATCH 6/7] fix(presets): recheck the skill directory after journaling it The descriptor path compared the open skill directory with the entry in its parent before journaling the removal, then removed the entry by name. If the directory was replaced with another empty directory while the journal ran, rmdir removed the replacement. Compare again right after journaling, as the fallback path already does, and keep the entry when it is no longer the directory that was emptied. Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- src/specify_cli/integrations/_file_changes.py | 10 ++++++++-- tests/specify_cli/integrations/test_file_changes.py | 8 ++++++-- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/specify_cli/integrations/_file_changes.py b/src/specify_cli/integrations/_file_changes.py index 28c36ea457..76ec954b9f 100644 --- a/src/specify_cli/integrations/_file_changes.py +++ b/src/specify_cli/integrations/_file_changes.py @@ -85,16 +85,22 @@ def unlink_in_directory(path: Path, *, root: Path) -> None: os.unlink(path.name, dir_fd=directory_fd) except FileNotFoundError: pass - try: - empty = not os.listdir(directory_fd) and os.path.samestat( + + def same_directory() -> bool: + return os.path.samestat( os.fstat(directory_fd), os.stat(directory.name, dir_fd=parent_fd, follow_symlinks=False), ) + + try: + empty = not os.listdir(directory_fd) and same_directory() except (FileNotFoundError, NotADirectoryError): return if empty: before_file_change(path.parent, removal=True) try: + if not same_directory(): + return os.rmdir(directory.name, dir_fd=parent_fd) except OSError as exc: if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST, errno.ENOENT, errno.ENOTDIR): diff --git a/tests/specify_cli/integrations/test_file_changes.py b/tests/specify_cli/integrations/test_file_changes.py index 1d9086fa3c..00cd91dbbb 100644 --- a/tests/specify_cli/integrations/test_file_changes.py +++ b/tests/specify_cli/integrations/test_file_changes.py @@ -124,7 +124,7 @@ def test_unlink_in_directory_skips_unsafe_directory(tmp_path, monkeypatch, fallb @pytest.mark.parametrize("fallback", [False, True]) -@pytest.mark.parametrize("swap_at", ["before_unlink", "after_unlink", "before_rmdir"]) +@pytest.mark.parametrize("swap_at", ["before_unlink", "after_unlink", "before_rmdir", "before_rmdir_self"]) def test_unlink_in_directory_preserves_concurrent_replacements(tmp_path, monkeypatch, fallback, swap_at): directory = tmp_path / "skills" / "skill" directory.mkdir(parents=True) @@ -165,6 +165,10 @@ def before_change(path, **kwargs): directory.parent.rename(tmp_path / "retained") directory.parent.symlink_to(outside if descriptor_io else tmp_path / "retained", target_is_directory=True) swapped = True + elif path == directory and swap_at == "before_rmdir_self": + directory.rename(tmp_path / "retained") + directory.mkdir() + swapped = True def after_change(path): nonlocal swapped @@ -185,5 +189,5 @@ def after_change(path): assert directory.is_symlink() if descriptor_io: assert not (tmp_path / "retained" / "SKILL.md").exists() - elif swap_at == "after_unlink" or not descriptor_io: + elif swap_at in ("after_unlink", "before_rmdir_self") or not descriptor_io: assert directory.is_dir() From c87c1119325fbd6dec0ff0f01cc22a0502ba5385 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Sat, 7 Nov 2026 00:33:28 +0000 Subject: [PATCH 7/7] test(presets): run the integration switch cleanup cases in the switch suite The skill cleanup checks ran `preset remove` and `integration switch` from the preset remove suite. Their bodies now live in tests/specify_cli/presets/_skill_cleanup_helpers.py, and each command suite runs them for its own command, as design/cli.md asks. No check changed. Assisted-by: Claude Code (model: claude-opus-5-5, autonomous) --- .../integrations/test_command_switch.py | 32 ++++ .../presets/_skill_cleanup_helpers.py | 169 +++++++++++++++++ .../presets/test_command_remove.py | 177 ++---------------- 3 files changed, 215 insertions(+), 163 deletions(-) create mode 100644 tests/specify_cli/presets/_skill_cleanup_helpers.py diff --git a/tests/specify_cli/integrations/test_command_switch.py b/tests/specify_cli/integrations/test_command_switch.py index 35e3164936..84b8bd52b4 100644 --- a/tests/specify_cli/integrations/test_command_switch.py +++ b/tests/specify_cli/integrations/test_command_switch.py @@ -25,6 +25,12 @@ _write_invalid_manifest, # noqa: F401 runner, # noqa: F401 ) +from tests.specify_cli.presets._skill_cleanup_helpers import ( + INTEGRATION_SWITCH, + check_cleanup_preserves_swapped_link_target, + check_cleanup_preserves_user_files, + check_cleanup_rollback_preserves_directory_mode, +) class TestIntegrationSwitch: @pytest.mark.parametrize("installed", [False, True]) @@ -1188,3 +1194,29 @@ def fail_cleanup(self, integration_key): assert "Failed to clean up extension artifacts for integration 'copilot'" in normalized assert "cleanup exploded" in normalized assert "Switched to integration" in normalized + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("ancestor", [False, True]) +def test_switch_preset_skill_cleanup_preserves_swapped_link_target(tmp_path, monkeypatch, fallback, ancestor): + check_cleanup_preserves_swapped_link_target(tmp_path, monkeypatch, INTEGRATION_SWITCH, fallback, ancestor) + + +@pytest.mark.parametrize( + "user_reference,layout", + [(False, "project"), (True, "project"), (False, "linked-parent")], + ids=["empty", "user-files", "linked-parent"], +) +@pytest.mark.parametrize("fallback", [False, True]) +def test_switch_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, user_reference, layout, fallback): + check_cleanup_preserves_user_files(tmp_path, monkeypatch, INTEGRATION_SWITCH, user_reference, layout, fallback) + + +@pytest.mark.parametrize("user_reference", [False, True]) +@pytest.mark.parametrize("fallback", [False, True]) +def test_switch_preset_skill_cleanup_rollback_preserves_directory_mode( + tmp_path, monkeypatch, capsys, user_reference, fallback, +): + check_cleanup_rollback_preserves_directory_mode( + tmp_path, monkeypatch, capsys, INTEGRATION_SWITCH, user_reference, fallback + ) diff --git a/tests/specify_cli/presets/_skill_cleanup_helpers.py b/tests/specify_cli/presets/_skill_cleanup_helpers.py new file mode 100644 index 0000000000..6695c9b416 --- /dev/null +++ b/tests/specify_cli/presets/_skill_cleanup_helpers.py @@ -0,0 +1,169 @@ +"""Preset skill cleanup checks shared by the `preset remove` and `integration switch` suites.""" + +from __future__ import annotations + +import os +import stat +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from specify_cli import app +from specify_cli.presets import PresetManager + +PRESET_REMOVE = ["preset", "remove", "my-preset"] +INTEGRATION_SWITCH = ["integration", "switch", "claude"] + + +def _install_scaffold_preset() -> CliRunner: + """Initialize Copilot skills in the current directory and add the bundled scaffold preset.""" + runner = CliRunner() + preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" + for args in ( + ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], + ["preset", "add", "--dev", str(preset)], + ): + result = runner.invoke(app, args, catch_exceptions=False) + assert result.exit_code == 0, result.output + return runner + + +def check_cleanup_preserves_swapped_link_target( + tmp_path, monkeypatch, command, fallback, ancestor, +): + project = tmp_path / "project" + project.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + probe = tmp_path / "link" + try: + probe.symlink_to(outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + probe.unlink() + monkeypatch.chdir(project) + runner = _install_scaffold_preset() + skill_dir = project / ".github/skills/speckit-myext-myextcmd" + original = (skill_dir / "SKILL.md").read_bytes() + swapped_dir = skill_dir.parent if ancestor else skill_dir + outside_skill = outside / skill_dir.name if ancestor else outside + outside_skill.mkdir(exist_ok=True) + outside_file = outside_skill / "SKILL.md" + outside_file.write_bytes(original) + validate = PresetManager._validate_skill_subdir + swapped = False + + def validate_then_swap(self, directory, **kwargs): + nonlocal swapped + valid = validate(self, directory, **kwargs) + if valid and directory == skill_dir and not kwargs["create"] and not swapped: + swapped_dir.rename(project / "retained-skills") + swapped_dir.symlink_to(outside, target_is_directory=True) + swapped = True + return valid + + monkeypatch.setattr(PresetManager, "_validate_skill_subdir", validate_then_swap) + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + result = runner.invoke(app, command, catch_exceptions=False) + + assert swapped + assert outside_file.is_file(), result.output + assert outside_file.read_bytes() == original + assert swapped_dir.is_symlink() + assert result.exit_code == 0, result.output + + +def check_cleanup_preserves_user_files( + tmp_path, monkeypatch, command, user_reference, layout, fallback, +): + project = tmp_path + if layout == "linked-parent": + (tmp_path / "real" / "project").mkdir(parents=True) + try: + (tmp_path / "link").symlink_to(tmp_path / "real", target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("Directory links are unavailable") + project = tmp_path / "link" / "project" + monkeypatch.chdir(project) + runner = _install_scaffold_preset() + # getcwd() resolves parent links; retain the user's lexical project path. + monkeypatch.setattr(Path, "cwd", lambda: project) + skill_dir = project / ".github/skills/speckit-myext-myextcmd" + reference = skill_dir / "references" / "user.md" + original = b"\xef\xbb\xbf# User reference\r\nKeep my project notes." + if user_reference: + reference.parent.mkdir() + reference.write_bytes(original) + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + + result = runner.invoke(app, command, catch_exceptions=False) + + assert result.exit_code == 0, result.output + assert not (skill_dir / "SKILL.md").exists() + if user_reference: + assert reference.is_file() + assert reference.read_bytes() == original + else: + assert not skill_dir.exists() + + +def check_cleanup_rollback_preserves_directory_mode( + tmp_path, monkeypatch, capsys, command, user_reference, fallback, +): + from specify_cli.integrations._helpers import _read_integration_json + from specify_cli.integrations._lifecycle import _FileJournal, _transaction + from specify_cli.integrations.installer import read_records + + project = tmp_path / "project" + project.mkdir() + (tmp_path / "home").mkdir() + monkeypatch.chdir(project) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + monkeypatch.setenv("USERPROFILE", str(tmp_path / "home")) + runner = _install_scaffold_preset() + skill_dir = project / ".github/skills/speckit-myext-myextcmd" + skill_file = skill_dir / "SKILL.md" + original = skill_file.read_bytes() + skill_dir.chmod(0o700) + reference = skill_dir / "references/user.md" + if user_reference: + reference.parent.mkdir() + reference.write_bytes(b"User reference before cleanup") + if fallback: + monkeypatch.setattr(os, "supports_dir_fd", set()) + + snapshot = _FileJournal._snapshot + snapshotted = [] + + def snapshot_without_user_files(self, source, destination): + snapshotted.append(source) + assert source != reference and source != reference.parent + if source == skill_dir and user_reference: + assert not any(source.iterdir()) + return snapshot(self, source, destination) + + monkeypatch.setattr(_FileJournal, "_snapshot", snapshot_without_user_files) + previous_umask = os.umask(0o022) + try: + with pytest.raises(RuntimeError, match="failure after cleanup"): + with _transaction(project, "claude", _read_integration_json(project), read_records(project)): + result = runner.invoke(app, command, catch_exceptions=False) + assert result.exit_code == 0, result.output + assert not skill_file.exists() + assert skill_dir.exists() == user_reference + if user_reference: + reference.write_bytes(b"User reference edited during transaction") + raise RuntimeError("failure after cleanup") + finally: + os.umask(previous_umask) + + assert skill_file.read_bytes() == original + if os.name != "nt": + assert stat.S_IMODE(skill_dir.stat().st_mode) == 0o700 + if user_reference: + assert skill_dir not in snapshotted + assert reference.read_bytes() == b"User reference edited during transaction" + assert "Recovery snapshots retained" not in capsys.readouterr().out diff --git a/tests/specify_cli/presets/test_command_remove.py b/tests/specify_cli/presets/test_command_remove.py index 64db27ce0f..0e8f7f21ad 100644 --- a/tests/specify_cli/presets/test_command_remove.py +++ b/tests/specify_cli/presets/test_command_remove.py @@ -2,8 +2,6 @@ from __future__ import annotations -import os -import stat from pathlib import Path from unittest.mock import patch @@ -12,6 +10,12 @@ from specify_cli import app from specify_cli.presets import PresetManager +from tests.specify_cli.presets._skill_cleanup_helpers import ( + PRESET_REMOVE, + check_cleanup_preserves_swapped_link_target, + check_cleanup_preserves_user_files, + check_cleanup_rollback_preserves_directory_mode, +) def test_remove_installed_preset(project_dir, pack_dir): @@ -24,180 +28,27 @@ def test_remove_installed_preset(project_dir, pack_dir): assert not PresetManager(project_dir).registry.is_installed("test-pack") -@pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize("fallback", [False, True]) @pytest.mark.parametrize("ancestor", [False, True]) -def test_preset_skill_cleanup_preserves_swapped_link_target( - tmp_path, monkeypatch, operation, fallback, ancestor, -): - project = tmp_path / "project" - project.mkdir() - outside = tmp_path / "outside" - outside.mkdir() - probe = tmp_path / "link" - try: - probe.symlink_to(outside, target_is_directory=True) - except (OSError, NotImplementedError): - pytest.skip("Directory links are unavailable") - probe.unlink() - monkeypatch.chdir(project) - runner = CliRunner() - preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" - for args in ( - ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], - ["preset", "add", "--dev", str(preset)], - ): - result = runner.invoke(app, args, catch_exceptions=False) - assert result.exit_code == 0, result.output - skill_dir = project / ".github/skills/speckit-myext-myextcmd" - original = (skill_dir / "SKILL.md").read_bytes() - swapped_dir = skill_dir.parent if ancestor else skill_dir - outside_skill = outside / skill_dir.name if ancestor else outside - outside_skill.mkdir(exist_ok=True) - outside_file = outside_skill / "SKILL.md" - outside_file.write_bytes(original) - validate = PresetManager._validate_skill_subdir - swapped = False - - def validate_then_swap(self, directory, **kwargs): - nonlocal swapped - valid = validate(self, directory, **kwargs) - if valid and directory == skill_dir and not kwargs["create"] and not swapped: - swapped_dir.rename(project / "retained-skills") - swapped_dir.symlink_to(outside, target_is_directory=True) - swapped = True - return valid +def test_preset_skill_cleanup_preserves_swapped_link_target(tmp_path, monkeypatch, fallback, ancestor): + check_cleanup_preserves_swapped_link_target(tmp_path, monkeypatch, PRESET_REMOVE, fallback, ancestor) - monkeypatch.setattr(PresetManager, "_validate_skill_subdir", validate_then_swap) - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] - result = runner.invoke(app, args, catch_exceptions=False) - - assert swapped - assert outside_file.is_file(), result.output - assert outside_file.read_bytes() == original - assert swapped_dir.is_symlink() - assert result.exit_code == 0, result.output - - -@pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize( "user_reference,layout", [(False, "project"), (True, "project"), (False, "linked-parent")], ids=["empty", "user-files", "linked-parent"], ) @pytest.mark.parametrize("fallback", [False, True]) -def test_preset_skill_cleanup_preserves_user_files( - tmp_path, monkeypatch, operation, user_reference, layout, fallback, -): - project = tmp_path - if layout == "linked-parent": - (tmp_path / "real" / "project").mkdir(parents=True) - try: - (tmp_path / "link").symlink_to(tmp_path / "real", target_is_directory=True) - except (OSError, NotImplementedError): - pytest.skip("Directory links are unavailable") - project = tmp_path / "link" / "project" - monkeypatch.chdir(project) - runner = CliRunner() - preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" - for args in ( - ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], - ["preset", "add", "--dev", str(preset)], - ): - result = runner.invoke(app, args, catch_exceptions=False) - assert result.exit_code == 0, result.output - # getcwd() resolves parent links; retain the user's lexical project path. - monkeypatch.setattr(Path, "cwd", lambda: project) - skill_dir = project / ".github/skills/speckit-myext-myextcmd" - reference = skill_dir / "references" / "user.md" - original = b"\xef\xbb\xbf# User reference\r\nKeep my project notes." - if user_reference: - reference.parent.mkdir() - reference.write_bytes(original) - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) +def test_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, user_reference, layout, fallback): + check_cleanup_preserves_user_files(tmp_path, monkeypatch, PRESET_REMOVE, user_reference, layout, fallback) - args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] - result = runner.invoke(app, args, catch_exceptions=False) - assert result.exit_code == 0, result.output - assert not (skill_dir / "SKILL.md").exists() - if user_reference: - assert reference.is_file() - assert reference.read_bytes() == original - else: - assert not skill_dir.exists() - - - -@pytest.mark.parametrize("operation", ["remove", "switch"]) @pytest.mark.parametrize("user_reference", [False, True]) @pytest.mark.parametrize("fallback", [False, True]) def test_preset_skill_cleanup_rollback_preserves_directory_mode( - tmp_path, monkeypatch, capsys, operation, user_reference, fallback, + tmp_path, monkeypatch, capsys, user_reference, fallback, ): - from specify_cli.integrations._helpers import _read_integration_json - from specify_cli.integrations._lifecycle import _FileJournal, _transaction - from specify_cli.integrations.installer import read_records - - project = tmp_path / "project" - project.mkdir() - (tmp_path / "home").mkdir() - monkeypatch.chdir(project) - monkeypatch.setenv("HOME", str(tmp_path / "home")) - monkeypatch.setenv("USERPROFILE", str(tmp_path / "home")) - runner = CliRunner() - preset = Path(__file__).resolve().parents[3] / "presets" / "scaffold" - for args in ( - ["init", "--here", "--integration", "copilot", "--script", "sh", "--ignore-agent-tools"], - ["preset", "add", "--dev", str(preset)], - ): - result = runner.invoke(app, args, catch_exceptions=False) - assert result.exit_code == 0, result.output - skill_dir = project / ".github/skills/speckit-myext-myextcmd" - skill_file = skill_dir / "SKILL.md" - original = skill_file.read_bytes() - skill_dir.chmod(0o700) - reference = skill_dir / "references/user.md" - if user_reference: - reference.parent.mkdir() - reference.write_bytes(b"User reference before cleanup") - if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) - - snapshot = _FileJournal._snapshot - snapshotted = [] - - def snapshot_without_user_files(self, source, destination): - snapshotted.append(source) - assert source != reference and source != reference.parent - if source == skill_dir and user_reference: - assert not any(source.iterdir()) - return snapshot(self, source, destination) - - monkeypatch.setattr(_FileJournal, "_snapshot", snapshot_without_user_files) - args = ["preset", "remove", "my-preset"] if operation == "remove" else ["integration", "switch", "claude"] - previous_umask = os.umask(0o022) - try: - with pytest.raises(RuntimeError, match="failure after cleanup"): - with _transaction(project, "claude", _read_integration_json(project), read_records(project)): - result = runner.invoke(app, args, catch_exceptions=False) - assert result.exit_code == 0, result.output - assert not skill_file.exists() - assert skill_dir.exists() == user_reference - if user_reference: - reference.write_bytes(b"User reference edited during transaction") - raise RuntimeError("failure after cleanup") - finally: - os.umask(previous_umask) - - assert skill_file.read_bytes() == original - if os.name != "nt": - assert stat.S_IMODE(skill_dir.stat().st_mode) == 0o700 - if user_reference: - assert skill_dir not in snapshotted - assert reference.read_bytes() == b"User reference edited during transaction" - assert "Recovery snapshots retained" not in capsys.readouterr().out + check_cleanup_rollback_preserves_directory_mode( + tmp_path, monkeypatch, capsys, PRESET_REMOVE, user_reference, fallback + )