Add CLI command to update snapshots and refactor directory configuration (#3)

Introduce a new command for updating snapshots without rerunning tests
and refactor the handling of directory names for improved clarity and
consistency. Minor linting adjustments included.

Author: martinmoldrup

CHANGELOG.md

@@ -3,6 +3,9 @@ All notable changes to this project will be documented in this file.
 
 *NOTE:* Version 0.X.X might have breaking changes in bumps of the minor version number. This is because the project is still in early development and the API is not yet stable. It will still be marked clearly in the release notes.
 
+## [unreleased]
+- 🆕 Add new cli command for easy update of snapshots without the need to rerun tests
+
 ## [0.4.0]
 - 🆕 Add support for pandas DataFrame assertions (pandas is an optional dependency, and snappylapy can be used without it)
 - 🆕 Add new simpler and more intuitive Expect API. We can use expect() calls and an overload will select the appropriate Expect class based on the type of the input. This is in contrast with expect.dict(), expect.list(), etc. This is not a breaking change, since the old API is still supported.

scripts/create_new_release.py

@@ -14,7 +14,7 @@ def read_version():
     """Read the version from the pyproject.toml file."""
     with open(PYPROJECT_TOML, "r", encoding="utf-8") as file:
         data = toml.load(file)
-    return data["tool"]["poetry"]["version"]
+    return data["project"]["version"]
 
 def check_version(version: str):
     """Check if the version is newer than the one in pypi."""

snappylapy/_cli.py

@@ -1,14 +1,17 @@
 """Create cli using the typer library."""
+
 import re
 import typer
 import pathlib
-from snappylapy.constants import directory_names
+from enum import Enum
+from snappylapy._utils_directories import DirectoryNamesUtil
+from snappylapy.constants import DIRECTORY_NAMES
 
 app = typer.Typer(
     no_args_is_help=True,
     help="""
-    The CLI provides commands to initialize the repo and to clear test results and snapshots.
-    In the future the future the CLI will be expanded with review and update commands.
+    The CLI provides commands to initialize the repo and to update or clear test results and snapshots.
+    In the future the future the CLI will be expanded with review.
     """,
 )
 
@@ -24,39 +27,48 @@ def init() -> None:
     # Check if already in .gitignore
     with gitignore_path.open("r") as file:
         lines = file.readlines()
-    regex = re.compile(
-        rf"^{re.escape(directory_names.test_results_dir_name)}(/|$)")
+    regex = re.compile(rf"^{re.escape(DIRECTORY_NAMES.test_results_dir_name)}(/|$)")
     if any(regex.match(line) for line in lines):
         typer.echo("Already in .gitignore.")
         return
     # Add to .gitignore to top of file
-    line_to_add = f"# Ignore test results from snappylapy\n{directory_names.test_results_dir_name}/\n\n"
+    line_to_add = f"# Ignore test results from snappylapy\n{DIRECTORY_NAMES.test_results_dir_name}/\n\n"
     with gitignore_path.open("w") as file:
         file.write(line_to_add)
         file.writelines(lines)
-    typer.echo(
-        f"Added {directory_names.test_results_dir_name}/ to .gitignore.")
+    typer.echo(f"Added {DIRECTORY_NAMES.test_results_dir_name}/ to .gitignore.")
 
 
 @app.command()
-def clear(force: bool = typer.Option(
-    False,
-    "--force",
-    "-f",
-    help="Force deletion without confirmation",
-)) -> None:
+def clear(
+    force: bool = typer.Option(
+        False,
+        "--force",
+        "-f",
+        help="Force deletion without confirmation",
+    ),
+) -> None:
     """Clear all test results and snapshots, recursively, using pathlib."""
-    list_of_files_to_delete = get_files_to_delete()
+    directories_to_delete = DirectoryNamesUtil().get_all_directories_created_by_snappylapy()
+    list_of_files_to_delete = DirectoryNamesUtil().get_all_file_paths_created_by_snappylapy()
     if not list_of_files_to_delete:
         typer.echo("No files to delete.")
         return
     if not force:
-        # Ask for confirmation
+        typer.echo("Deleting files:")
+        for file in list_of_files_to_delete:
+            typer.echo(f"- {file}")
+
         typer.secho(
-            "\nAre you sure you want to delete all test results and snapshots?",
-            fg=typer.colors.BRIGHT_BLUE)
-        response = typer.prompt(
-            "Type 'yes' to confirm, anything else to abort.")
+            f"Deleting {len(list_of_files_to_delete)} files from {len(directories_to_delete)} directories:",
+            fg=typer.colors.BRIGHT_BLUE,
+        )
+        for directory in directories_to_delete:
+            typer.echo(f"- {directory}")
+
+        # Ask for confirmation
+        typer.secho("\nAre you sure you want to delete all test results and snapshots?", fg=typer.colors.BRIGHT_BLUE)
+        response = typer.prompt("Type 'yes' to confirm, anything else to abort.", default="no")
         if response.lower() != "yes":
             typer.echo("Aborted.")
             return
@@ -65,19 +77,29 @@ def clear(force: bool = typer.Option(
     typer.echo(f"Deleted {len(list_of_files_to_delete)} files.")
 
 
-def get_files_to_delete() -> list[pathlib.Path]:
-    """Get list of files to delete."""
-    list_of_files_to_delete: list[pathlib.Path] = []
-    for dir_name in [
-            directory_names.test_results_dir_name,
-            directory_names.snapshot_dir_name,
-    ]:
-        for root_dir in pathlib.Path().rglob(dir_name):
-            for file in root_dir.iterdir():
-                if file.is_file():
-                    list_of_files_to_delete.append(file)
-                    typer.echo(f"Found file to delete: {file}")
-    return list_of_files_to_delete
+@app.command()
+def update() -> None:
+    """Update the snapshot files by copying the test results, to the snapshot directory."""
+    files_test_results = DirectoryNamesUtil().get_all_file_paths_test_results()
+    if not files_test_results:
+        typer.echo("No files to update.")
+        return
+    file_statuses = check_file_statuses(files_test_results)
+    files_to_update = [file for file, status in file_statuses.items() if status != FileStatus.UNCHANGED]
+    count_up_to_date_files = len(files_test_results) - len(files_to_update)
+    if not files_to_update:
+        typer.echo(f"All snapshot files are up to date. {count_up_to_date_files} files are up to date.")
+        return
+
+    typer.echo(
+        f"Found {len(files_to_update)} files to update."
+        + (f" {count_up_to_date_files} files are up to date." if count_up_to_date_files > 0 else ""),
+    )
+    for file in files_to_update:
+        snapshot_file = file.parent.parent / DIRECTORY_NAMES.snapshot_dir_name / file.name
+        snapshot_file.parent.mkdir(parents=True, exist_ok=True)
+        snapshot_file.write_bytes(file.read_bytes())
+        typer.echo(f"Updated snapshot: {snapshot_file}")
 
 
 def delete_files(list_of_files_to_delete: list[pathlib.Path]) -> None:
@@ -87,12 +109,40 @@ def delete_files(list_of_files_to_delete: list[pathlib.Path]) -> None:
         file.unlink()
     # Delete directories
     for dir_name in [
-            directory_names.test_results_dir_name,
-            directory_names.snapshot_dir_name,
+        DIRECTORY_NAMES.test_results_dir_name,
+        DIRECTORY_NAMES.snapshot_dir_name,
     ]:
         for root_dir in pathlib.Path().rglob(dir_name):
             root_dir.rmdir()
 
 
+class FileStatus(Enum):
+    """Enum to represent the status of a file."""
+
+    NOT_FOUND = "not_found"
+    CHANGED = "changed"
+    UNCHANGED = "unchanged"
+
+
+def check_file_statuses(
+    file_paths: list[pathlib.Path],
+) -> dict[pathlib.Path, FileStatus]:
+    """Check the status of files in the snapshot directory."""
+    file_statuses: dict[pathlib.Path, FileStatus] = {}
+    for file_path in file_paths:
+        snapshot_file = file_path.parent.parent / DIRECTORY_NAMES.snapshot_dir_name / file_path.name
+        if not snapshot_file.exists():
+            file_statuses[file_path] = FileStatus.NOT_FOUND
+        elif snapshot_file.stat().st_size != file_path.stat().st_size:
+            # TODO: This is not foolproof, does not catch content swaps and byte flips.
+            file_statuses[file_path] = FileStatus.CHANGED
+        elif snapshot_file.read_bytes() != file_path.read_bytes():
+            # TODO: Expensive call, store hashes instead in a data file.
+            file_statuses[file_path] = FileStatus.CHANGED
+        else:
+            file_statuses[file_path] = FileStatus.UNCHANGED
+    return file_statuses
+
+
 if __name__ == "__main__":
     app()

snappylapy/_plugin.py

@@ -9,7 +9,8 @@
 import _pytest.mark
 from collections.abc import Callable
 from snappylapy import Expect, LoadSnapshot
-from snappylapy.constants import DEFEAULT_SNAPSHOT_BASE_DIR
+from snappylapy._utils_directories import DirectoryNamesUtil
+from snappylapy.constants import DEFAULT_SNAPSHOT_BASE_DIR
 from snappylapy.exceptions import TestDirectoryNotParametrizedError
 from snappylapy.fixtures import Settings
 from snappylapy.session import SnapshotSession
@@ -80,7 +81,7 @@ def snappylapy_settings(request: pytest.FixtureRequest) -> Settings:
             path_output_dir = pathlib.Path(input_dir_from_depends)
         settings.depending_test_filename = depends[0].__module__
         settings.depending_test_function = depends[0].__name__
-    settings.depending_snapshots_base_dir = path_output_dir or DEFEAULT_SNAPSHOT_BASE_DIR
+    settings.depending_snapshots_base_dir = path_output_dir or DEFAULT_SNAPSHOT_BASE_DIR
     return settings
 
 
@@ -149,6 +150,9 @@ def pytest_addoption(parser: pytest.Parser) -> None:
 def pytest_sessionstart(session: pytest.Session) -> None:
     """Initialize the snapshot session."""
     session.config.snappylapy_session = SnapshotSession()  # type: ignore[attr-defined]
+    list_of_files_to_delete = DirectoryNamesUtil().get_all_file_paths_test_results()
+    for file in list_of_files_to_delete:
+        file.unlink()
 
 
 class ExceptionDuringTestSetupError(Exception):

snappylapy/_utils_directories.py

@@ -0,0 +1,63 @@
+"""Utility functions for handling directories in snappylapy."""
+import pathlib
+from snappylapy.constants import DIRECTORY_NAMES
+
+
+def find_directories(directory_names: list[str]) -> list[pathlib.Path]:
+    """Find directories with the given names."""
+    found_dirs: list[pathlib.Path] = []
+    for dir_name in directory_names:
+        found_dirs.extend(pathlib.Path().rglob(dir_name))
+    return found_dirs
+
+
+def get_file_paths_from_directories(list_of_directories: list[pathlib.Path]) -> list[pathlib.Path]:
+    """Get file paths from directories."""
+    list_of_files_to_delete: list[pathlib.Path] = []
+    for directory in list_of_directories:
+        if not directory.is_dir():
+            error_msg = f"{directory} is not a directory."
+            raise ValueError(error_msg)
+        list_of_files_to_delete.extend(file for file in directory.iterdir() if file.is_file())
+    return list_of_files_to_delete
+
+
+class DirectoryNamesUtil:
+    """
+    Utility class to handle directory names and operations related to them.
+
+    This class extends the DirectoryNames class to provide methods for finding directories
+    and file paths created by snappylapy.
+    """
+
+    def get_all_directory_names(self) -> list[str]:
+        """Get all directory names."""
+        return [DIRECTORY_NAMES.snapshot_dir_name, DIRECTORY_NAMES.test_results_dir_name]
+
+    def get_all_directories_for_test_results(self) -> list[pathlib.Path]:
+        """Get all directories for test results."""
+        return find_directories([DIRECTORY_NAMES.test_results_dir_name])
+
+    def get_all_directories_for_snapshots(self) -> list[pathlib.Path]:
+        """Get all directories for snapshots."""
+        return find_directories([DIRECTORY_NAMES.snapshot_dir_name])
+
+    def get_all_file_paths_test_results(self) -> list[pathlib.Path]:
+        """Get all file paths in the test results directory."""
+        directories = find_directories([DIRECTORY_NAMES.test_results_dir_name])
+        return get_file_paths_from_directories(directories)
+
+    def get_all_file_paths_snapshots(self) -> list[pathlib.Path]:
+        """Get all file paths in the snapshot directory."""
+        directories = find_directories([DIRECTORY_NAMES.snapshot_dir_name])
+        return get_file_paths_from_directories(directories)
+
+    def get_all_directories_created_by_snappylapy(self) -> list[pathlib.Path]:
+        """Get all directories created by snappylapy."""
+        return find_directories(self.get_all_directory_names())
+
+    def get_all_file_paths_created_by_snappylapy(self) -> list[pathlib.Path]:
+        """Get all file paths created by snappylapy."""
+        return get_file_paths_from_directories(
+            self.get_all_directories_created_by_snappylapy(),
+        )

snappylapy/constants.py

@@ -1,23 +1,19 @@
 """Read-only constants."""
 
 import pathlib
+from typing import NamedTuple
 
-DEFEAULT_SNAPSHOT_BASE_DIR = pathlib.Path()
+DEFAULT_SNAPSHOT_BASE_DIR = pathlib.Path()
 OUTPUT_JSON_INDENTATION_LEVEL = 2
 
 
-class DirectoryNames:
-    """Class to enforce immutable directory names, since there is side effect if they are changed."""
+class DirectoryNames(NamedTuple):
+    """Immutable directory names for snappylapy."""
 
-    @property
-    def snapshot_dir_name(self) -> str:
-        """Snapshot directory name."""
-        return "__snapshots__"
+    snapshot_dir_name: str
+    test_results_dir_name: str
 
-    @property
-    def test_results_dir_name(self) -> str:
-        """Test results directory name."""
-        return "__test_results__"
 
-
-directory_names = DirectoryNames()
+DIRECTORY_NAMES: DirectoryNames = DirectoryNames(
+    snapshot_dir_name="__snapshots__", test_results_dir_name="__test_results__",
+)

snappylapy/fixtures.py

@@ -24,7 +24,7 @@
     JsonPickleSerializer,
     StringSerializer,
 )
-from snappylapy.constants import directory_names
+from snappylapy.constants import DIRECTORY_NAMES
 from snappylapy.session import SnapshotSession
 from typing import Any, Protocol, overload
 
@@ -71,6 +71,7 @@ def __init__(
 
         self.dict = DictExpect(self.settings, snappylapy_session)
         """DictExpect instance for configuring snapshot testing of dictionaries.
+
         The instance is callable with the following parameters:
 
         Parameters
@@ -97,6 +98,7 @@ def __init__(
 
         self.list = ListExpect(self.settings, snappylapy_session)
         """ListExpect instance for configuring snapshot testing of lists.
+
         The instance is callable with the following parameters:
 
         Parameters
@@ -122,6 +124,7 @@ def __init__(
 
         self.string = StringExpect(self.settings, snappylapy_session)
         """StringExpect instance for configuring snapshot testing of strings.
+
         The instance is callable with the following parameters:
 
         Parameters
@@ -147,6 +150,7 @@ def __init__(
 
         self.bytes = BytesExpect(self.settings, snappylapy_session)
         """BytesExpect instance for configuring snapshot testing of bytes.
+
         The instance is callable with the following parameters:
 
         Parameters
@@ -172,7 +176,9 @@ def __init__(
 
         self.dataframe = DataframeExpect(self.settings, snappylapy_session)
         """DataframeExpect instance for configuring snapshot testing of dataframes.
+
         The instance is callable with the following parameters:
+
         Parameters
         ----------
         data_to_snapshot : pd.DataFrame
@@ -278,7 +284,7 @@ def _read_snapshot(self) -> bytes:
             raise ValueError(msg)
         return (
             self.settings.depending_snapshots_base_dir
-            / directory_names.snapshot_dir_name
+            / DIRECTORY_NAMES.snapshot_dir_name
             / self.settings.depending_filename
         ).read_bytes()