Validate cache directory before cleaning it

Author: lagru

src/docstub/_cache.py

@@ -5,10 +5,21 @@
 logger = logging.getLogger(__name__)
 
 
+CACHE_DIR_NAME = ".docstub_cache"
+
+
 CACHEDIR_TAG_CONTENT = """\
-Signature: 8a477f597d28d172789f06886806bc55\
-# This file is a cache directory tag automatically created by docstub.\n"
-# For information about cache directory tags see https://bford.info/cachedir/\n"
+Signature: 8a477f597d28d172789f06886806bc55
+# Mark this directory as a cache [1], created by docstub [2]
+# [1] https://bford.info/cachedir/
+# [2] https://github.com/scientific-python/docstub
+"""
+
+
+GITHUB_IGNORE_CONTENT = """\
+# Make git ignore this cache directory, created by docstub [1]
+# [1] https://github.com/scientific-python/docstub
+*
 """
 
 
@@ -43,22 +54,42 @@ def create_cache(path):
     """
     path.mkdir(parents=True, exist_ok=True)
     cachdir_tag_path = path / "CACHEDIR.TAG"
-    cachdir_tag_content = (
-        "Signature: 8a477f597d28d172789f06886806bc55\n"
-        "# This file is a cache directory tag automatically created by docstub.\n"
-        "# For information about cache directory tags see https://bford.info/cachedir/\n"
-    )
+
     if not cachdir_tag_path.is_file():
         with open(cachdir_tag_path, "w") as fp:
-            fp.write(cachdir_tag_content)
+            fp.write(CACHEDIR_TAG_CONTENT)
 
     gitignore_path = path / ".gitignore"
-    gitignore_content = (
-        "# This file is a cache directory automatically created by docstub.\n" "*\n"
-    )
     if not gitignore_path.is_file():
         with open(gitignore_path, "w") as fp:
-            fp.write(gitignore_content)
+            fp.write(GITHUB_IGNORE_CONTENT)
+
+
+def validate_cache(path):
+    """Make sure the given path is a cache created by docstub.
+
+    Parameters
+    ----------
+    path : Path
+
+    Raises
+    ------
+    FileNotFoundError
+    """
+    if not path.is_dir():
+        raise FileNotFoundError(f"expected '{path}' to be a valid directory")
+
+    if not path.name == CACHE_DIR_NAME:
+        raise FileNotFoundError(
+            f"expected directory '{path}' be named '{CACHE_DIR_NAME}'"
+        )
+
+    cachdir_tag_path = path / "CACHEDIR.TAG"
+    if not cachdir_tag_path.is_file():
+        raise FileNotFoundError(f"expected '{path}' to contain a 'CACHEDIR.TAG' file")
+    gitignore_path = path / ".gitignore"
+    if not gitignore_path.is_file():
+        raise FileNotFoundError(f"expected '{path}' to contain a '.gitignore' file")
 
 
 class FuncSerializer[T](Protocol):

src/docstub/_cli.py

@@ -14,7 +14,7 @@
     TypeMatcher,
     common_known_types,
 )
-from ._cache import FileCache
+from ._cache import CACHE_DIR_NAME, FileCache, validate_cache
 from ._config import Config
 from ._path_utils import (
     STUB_HEADER_COMMENT,
@@ -35,7 +35,7 @@ def _cache_dir_in_cwd():
     -------
     cache_dir : Path
     """
-    return Path.cwd() / ".docstub_cache"
+    return Path.cwd() / CACHE_DIR_NAME
 
 
 def _load_configuration(config_paths=None):
@@ -385,8 +385,19 @@ def clean(verbose):
 
     path = _cache_dir_in_cwd()
     if path.exists():
-        assert path.is_dir()
-        shutil.rmtree(_cache_dir_in_cwd())
-        logger.info("cleaned %s", path)
+        try:
+            validate_cache(path)
+        except (FileNotFoundError, ValueError) as e:
+            logger.error(
+                "'%s' might not be a valid cache or might be corrupted. Not "
+                "removing it out of caution. Manually remove it after checking "
+                "if it is safe to do so.\n\nDetails: %s",
+                path,
+                "\n".join(e.args),
+            )
+            sys.exit(1)
+        else:
+            shutil.rmtree(_cache_dir_in_cwd())
+            logger.info("cleaned %s", path)
     else:
         logger.info("no cache to clean")

tests/test_cache.py

@@ -37,6 +37,15 @@ def test_create_cache(tmp_path):
     _cache.create_cache(cache_dir)
 
 
+def test_create_validate_cache(tmp_path):
+    cache_dir = tmp_path / _cache.CACHE_DIR_NAME
+    _cache.create_cache(cache_dir)
+    _cache.validate_cache(cache_dir)
+
+    with pytest.raises(FileNotFoundError, match="expected directory .* named .*"):
+        _cache.validate_cache(tmp_path)
+
+
 class Test_FileCache:
     def test_basic(self, tmp_path):
 

tests/test_cli.py

@@ -90,3 +90,15 @@ def test_basic(self, tmp_path_cwd, verbosity):
         assert run_result.exception is None
         assert run_result.exit_code == 0
         assert not cache_dir.exists()
+
+    def test_corrupted_cache(self, tmp_path_cwd, caplog):
+        cache_dir = tmp_path_cwd / ".docstub_cache"
+        cache_dir.mkdir()
+
+        runner = CliRunner()
+        run_result = runner.invoke(_cli.clean, args=[])
+        assert run_result.exit_code == 1
+        assert "might not be a valid cache or might be corrupted" in "\n".join(
+            caplog.messages
+        )
+        assert cache_dir.is_dir()