Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion src/skillspector/cleanup.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,41 @@

"""Shared cleanup helpers for SkillSpector."""

import os
import shutil
import stat
from collections.abc import Callable
from pathlib import Path

from skillspector.python_ast import clear_python_ast_cache


def _retry_writable(function: Callable[[str], object], path: str, _error: BaseException) -> None:
"""Clear a read-only bit and retry once; Windows refuses to delete read-only files."""
try:
# chmod follows links, so never touch whatever a link points at.
if not (os.path.islink(path) or os.path.isjunction(path)):
# Add the owner-write bit only; replacing the mode would strip read and
# search permission on POSIX and leave the entry harder to remove.
os.chmod(path, stat.S_IMODE(os.lstat(path).st_mode) | stat.S_IWRITE)
function(path)
except OSError:
pass


def remove_temp_tree(path: str | Path) -> None:
"""Best-effort removal of a scan temp directory, including read-only files.

``git clone`` writes its pack files read-only, so ``ignore_errors=True``
alone leaves every cloned repository behind on Windows.
"""
shutil.rmtree(path, onexc=_retry_writable)


def cleanup_result(result: dict[str, object]) -> None:
"""Release scan-local resources and remove a temp dir if set."""
python_ast_cache_key = result.get("python_ast_cache_key")
clear_python_ast_cache(python_ast_cache_key if isinstance(python_ast_cache_key, str) else None)
temp_dir = result.get("temp_dir_for_cleanup")
if temp_dir and isinstance(temp_dir, str):
shutil.rmtree(temp_dir, ignore_errors=True)
remove_temp_tree(temp_dir)
3 changes: 2 additions & 1 deletion src/skillspector/input_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@

import httpx

from skillspector.cleanup import remove_temp_tree
from skillspector.logging_config import get_logger

logger = get_logger(__name__)
Expand Down Expand Up @@ -738,7 +739,7 @@ def resolve(self, input_path: str) -> tuple[Path, str]:
def cleanup(self) -> None:
"""Clean up temporary files created during resolution."""
if self._temp_dir and self._temp_dir.exists():
shutil.rmtree(self._temp_dir, ignore_errors=True)
remove_temp_tree(self._temp_dir)
self._temp_dir = None

def temp_dir_for_cleanup(self) -> Path | None:
Expand Down
91 changes: 91 additions & 0 deletions tests/unit/test_cleanup.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Tests for scan temp-directory cleanup."""

from __future__ import annotations

import os
import stat
from pathlib import Path

import pytest

from skillspector.cleanup import cleanup_result
from skillspector.input_handler import InputHandler


def _refuse_read_only_unlink(monkeypatch: pytest.MonkeyPatch) -> None:
"""Apply Windows semantics everywhere: a read-only file cannot be unlinked."""
real_unlink = os.unlink

def unlink(path: str, *args: object, dir_fd: int | None = None) -> None:
mode = os.stat(path, dir_fd=dir_fd, follow_symlinks=False).st_mode
if not mode & stat.S_IWRITE:
raise PermissionError(13, "Access is denied", path)
real_unlink(path, *args, dir_fd=dir_fd)

monkeypatch.setattr(os, "unlink", unlink)


def _clone_with_read_only_pack(root: Path) -> Path:
"""Lay out the read-only pack files ``git clone`` leaves in a temp checkout."""
pack_dir = root / "repo" / ".git" / "objects" / "pack"
pack_dir.mkdir(parents=True)
for name in ("pack-1.idx", "pack-1.pack"):
pack = pack_dir / name
pack.write_bytes(b"PACK")
pack.chmod(stat.S_IREAD)
(root / "repo" / "SKILL.md").write_text("# Skill\n", encoding="utf-8")
return root


def test_cleanup_result_removes_read_only_git_objects(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A Git URL scan's temp clone is removed even though Git marks packs read-only."""
temp_dir = _clone_with_read_only_pack(tmp_path / "skillspector_scan")
_refuse_read_only_unlink(monkeypatch)

cleanup_result({"temp_dir_for_cleanup": str(temp_dir)})

assert not temp_dir.exists()


def test_cleanup_result_stays_best_effort_when_a_file_cannot_be_removed(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A file that is still locked is left behind; cleanup never fails the scan."""
temp_dir = tmp_path / "skillspector_locked"
temp_dir.mkdir()
locked = temp_dir / "locked.pack"
locked.write_bytes(b"PACK")
(temp_dir / "SKILL.md").write_text("# Skill\n", encoding="utf-8")
real_unlink = os.unlink

def unlink(path: str, *args: object, dir_fd: int | None = None) -> None:
if os.path.basename(path) == locked.name:
raise PermissionError(32, "The file is in use by another process", path)
real_unlink(path, *args, dir_fd=dir_fd)

monkeypatch.setattr(os, "unlink", unlink)

cleanup_result({"temp_dir_for_cleanup": str(temp_dir)})

assert locked.exists()
assert not (temp_dir / "SKILL.md").exists()


def test_input_handler_cleanup_removes_read_only_git_objects(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""The handler's own cleanup path removes the same read-only clone."""
handler = InputHandler()
handler._temp_dir = _clone_with_read_only_pack(tmp_path / "skillspector_handler")
temp_dir = handler._temp_dir
_refuse_read_only_unlink(monkeypatch)

handler.cleanup()

assert not temp_dir.exists()
assert handler.temp_dir_for_cleanup() is None
Loading