fix: unused snapshot detection for targeting single parameterized test case

src/syrupy/__init__.py

@@ -1,8 +1,8 @@
 import argparse
-import glob
 import sys
 from gettext import gettext
 from typing import (
+    TYPE_CHECKING,
     Any,
     ContextManager,
     List,
@@ -15,7 +15,7 @@
 from .constants import DISABLE_COLOR_ENV_VAR
 from .exceptions import FailedToLoadModuleMember
 from .extensions import DEFAULT_EXTENSION
-from .location import TestLocation
+from .location import PyTestLocation
 from .session import SnapshotSession
 from .terminal import (
     received_style,
@@ -27,6 +27,9 @@
     import_module_member,
 )
 
+if TYPE_CHECKING:
+    from _pytest.config import Config
+
 
 def __default_extension_option(value: str) -> Any:
     try:
@@ -101,37 +104,19 @@ def snapshot_name(name: str) -> str:
     return None
 
 
-def __is_testpath(arg: str) -> bool:
-    return not arg.startswith("-") and bool(glob.glob(arg.split("::")[0]))
-
-
-def __is_testnode(arg: str) -> bool:
-    return __is_testpath(arg) and "::" in arg
-
-
-def __is_testmodule(arg: str) -> bool:
-    return arg == "--pyargs"
-
-
 def pytest_sessionstart(session: Any) -> None:
     """
     Initialize snapshot session before tests are collected and ran.
     https://docs.pytest.org/en/latest/reference.html#_pytest.hookspec.pytest_sessionstart
     """
-    config = session.config
-    config._syrupy = SnapshotSession(
+    config: "Config" = session.config
+    session.config._syrupy = SnapshotSession(
         warn_unused_snapshots=config.option.warn_unused_snapshots,
         update_snapshots=config.option.update_snapshots,
-        base_dir=config.rootdir,
-        is_providing_paths=any(
-            __is_testpath(arg) or __is_testmodule(arg)
-            for arg in config.invocation_params.args
-        ),
-        is_providing_nodes=any(
-            __is_testnode(arg) for arg in config.invocation_params.args
-        ),
+        base_dir=str(config.rootpath),
+        invocation_args=config.invocation_params.args,
     )
-    config._syrupy.start()
+    session.config._syrupy.start()
 
 
 def pytest_collection_modifyitems(
@@ -180,6 +165,6 @@ def snapshot(request: Any) -> "SnapshotAssertion":
     return SnapshotAssertion(
         update_snapshots=request.config.option.update_snapshots,
         extension_class=request.config.option.default_extension,
-        test_location=TestLocation(request.node),
+        test_location=PyTestLocation(request.node),
         session=request.session.config._syrupy,
     )

src/syrupy/assertion.py

@@ -15,7 +15,7 @@
 
 if TYPE_CHECKING:
     from .extensions.base import AbstractSyrupyExtension
-    from .location import TestLocation
+    from .location import PyTestLocation
     from .session import SnapshotSession
     from .types import (
         PropertyFilter,
@@ -47,7 +47,7 @@ class SnapshotAssertion:
     name: str = attr.ib(default="snapshot")
     _session: "SnapshotSession" = attr.ib(kw_only=True)
     _extension_class: Type["AbstractSyrupyExtension"] = attr.ib(kw_only=True)
-    _test_location: "TestLocation" = attr.ib(kw_only=True)
+    _test_location: "PyTestLocation" = attr.ib(kw_only=True)
     _update_snapshots: bool = attr.ib(kw_only=True)
     _exclude: Optional["PropertyFilter"] = attr.ib(init=False, default=None)
     _extension: Optional["AbstractSyrupyExtension"] = attr.ib(init=False, default=None)

src/syrupy/constants.py

@@ -10,3 +10,5 @@
 
 DISABLE_COLOR_ENV_VAR = "ANSI_COLORS_DISABLED"
 DISABLE_COLOR_ENV_VARS = {DISABLE_COLOR_ENV_VAR, "NO_COLOR"}
+
+PYTEST_NODE_SEP = "::"

src/syrupy/extensions/base.py

@@ -44,7 +44,7 @@
 from syrupy.utils import walk_snapshot_dir
 
 if TYPE_CHECKING:
-    from syrupy.location import TestLocation
+    from syrupy.location import PyTestLocation
     from syrupy.types import (
         PropertyFilter,
         PropertyMatcher,
@@ -72,7 +72,7 @@ def serialize(
 class SnapshotFossilizer(ABC):
     @property
     @abstractmethod
-    def test_location(self) -> "TestLocation":
+    def test_location(self) -> "PyTestLocation":
         raise NotImplementedError
 
     def get_snapshot_name(self, *, index: int = 0) -> str:
@@ -83,7 +83,8 @@ def get_snapshot_name(self, *, index: int = 0) -> str:
     def get_location(self, *, index: int) -> str:
         """Returns full location where snapshot data is stored."""
         basename = self._get_file_basename(index=index)
-        return str(Path(self._dirname).joinpath(f"{basename}.{self._file_extension}"))
+        fileext = f".{self._file_extension}" if self._file_extension else ""
+        return str(Path(self._dirname).joinpath(f"{basename}{fileext}"))
 
     def is_snapshot_location(self, *, location: str) -> bool:
         """Checks if supplied location is valid for this snapshot extension"""
@@ -373,9 +374,9 @@ def __strip_ends(self, line: str) -> str:
 
 
 class AbstractSyrupyExtension(SnapshotSerializer, SnapshotFossilizer, SnapshotReporter):
-    def __init__(self, test_location: "TestLocation"):
+    def __init__(self, test_location: "PyTestLocation"):
         self._test_location = test_location
 
     @property
-    def test_location(self) -> "TestLocation":
+    def test_location(self) -> "PyTestLocation":
         return self._test_location

src/syrupy/location.py

@@ -5,8 +5,10 @@
     Optional,
 )
 
+from syrupy.constants import PYTEST_NODE_SEP
 
-class TestLocation:
+
+class PyTestLocation:
     def __init__(self, node: Any):
         self._node = node
         self.filepath = self._node.fspath
@@ -17,8 +19,7 @@ def __init__(self, node: Any):
 
     @property
     def classname(self) -> Optional[str]:
-        _, __, qualname = self._node.location
-        return ".".join(list(self.__valid_ids(qualname))[:-1]) or None
+        return ".".join(self._node.nodeid.split(PYTEST_NODE_SEP)[1:-1]) or None
 
     @property
     def filename(self) -> str:

src/syrupy/report.py

@@ -8,18 +8,22 @@
     Dict,
     Iterator,
     List,
+    Optional,
+    Set,
+    Tuple,
 )
 
 import attr
-import pytest
+from _pytest.mark.expression import Expression
 
+from .constants import PYTEST_NODE_SEP
 from .data import (
     Snapshot,
     SnapshotFossil,
     SnapshotFossils,
     SnapshotUnknownFossil,
 )
-from .location import TestLocation
+from .location import PyTestLocation
 from .terminal import (
     bold,
     error_style,
@@ -29,6 +33,8 @@
 )
 
 if TYPE_CHECKING:
+    import pytest
+
     from .assertion import SnapshotAssertion
 
 
@@ -38,8 +44,6 @@ class SnapshotReport:
     all_items: Dict["pytest.Item", bool] = attr.ib()
     ran_items: Dict["pytest.Item", bool] = attr.ib()
     update_snapshots: bool = attr.ib()
-    is_providing_paths: bool = attr.ib()
-    is_providing_nodes: bool = attr.ib()
     warn_unused_snapshots: bool = attr.ib()
     assertions: List["SnapshotAssertion"] = attr.ib()
     discovered: "SnapshotFossils" = attr.ib(factory=SnapshotFossils)
@@ -48,8 +52,12 @@ class SnapshotReport:
     matched: "SnapshotFossils" = attr.ib(factory=SnapshotFossils)
     updated: "SnapshotFossils" = attr.ib(factory=SnapshotFossils)
     used: "SnapshotFossils" = attr.ib(factory=SnapshotFossils)
+    _invocation_args: Tuple[str, ...] = attr.ib(factory=tuple)
+    _provided_test_paths: Dict[str, List[str]] = attr.ib(factory=dict)
+    _keyword_expressions: Set["Expression"] = attr.ib(factory=set)
 
     def __attrs_post_init__(self) -> None:
+        self.__parse_invocation_args()
         for assertion in self.assertions:
             self.discovered.merge(assertion.extension.discover_snapshots())
             for result in assertion.executions.values():
@@ -67,6 +75,43 @@ def __attrs_post_init__(self) -> None:
                 else:
                     self.failed.update(snapshot_fossil)
 
+    def __parse_invocation_args(self) -> None:
+        arg_groups: List[Tuple[Optional[str], str]] = []
+        path_as_package = False
+        maybe_opt_arg = None
+        for arg in self._invocation_args:
+            if arg.strip() == "--pyargs":
+                path_as_package = True
+            elif arg.startswith("-"):
+                if "=" in arg:
+                    arg0, arg1 = arg.split("=")
+                    arg_groups.append((arg0.strip(), arg1.strip()))
+                elif maybe_opt_arg is None:
+                    maybe_opt_arg = arg
+                    continue  # do not reset maybe_opt_arg
+            else:
+                arg_groups.append((maybe_opt_arg, arg.strip()))
+
+            maybe_opt_arg = None
+
+        for maybe_opt_arg, arg_value in arg_groups:
+            if maybe_opt_arg == "-k":  # or maybe_opt_arg == "-m":
+                self._keyword_expressions.add(Expression.compile(arg_value))
+            elif maybe_opt_arg is None:
+                import importlib
+
+                parts = arg_value.split(PYTEST_NODE_SEP)
+                package_or_filepath = parts[0].strip()
+                filepath = Path(
+                    importlib.import_module(package_or_filepath).__file__
+                    if path_as_package
+                    else package_or_filepath
+                )
+                filepath_abs = str(
+                    filepath if filepath.is_absolute() else filepath.absolute()
+                )
+                self._provided_test_paths[filepath_abs] = parts[1:]
+
     @property
     def num_created(self) -> int:
         return self._count_snapshots(self.created)
@@ -89,7 +134,7 @@ def num_unused(self) -> int:
 
     @property
     def ran_all_collected_tests(self) -> bool:
-        return self.all_items == self.ran_items and not self.is_providing_nodes
+        return self.all_items == self.ran_items
 
     @property
     def unused(self) -> "SnapshotFossils":
@@ -98,23 +143,25 @@ def unused(self) -> "SnapshotFossils":
             self.discovered, self.used
         ):
             snapshot_location = unused_snapshot_fossil.location
-            if self.is_providing_paths and not any(
-                TestLocation(node).matches_snapshot_location(snapshot_location)
-                for node in self.ran_items
+            if self._provided_test_paths and not self._selected_items_match_location(
+                snapshot_location
             ):
+                # Paths/Packages were provided to pytest and the snapshot location
+                # does not match therefore ignore this unused snapshot fossil file
                 continue
 
-            if self.ran_all_collected_tests:
+            provided_nodes = self._get_matching_path_nodes(snapshot_location)
+            if self.ran_all_collected_tests and not any(provided_nodes):
+                # All collected tests were run and files were not filtered by ::node
+                # therefore the snapshot fossil file at this location can be deleted
                 unused_snapshots = {*unused_snapshot_fossil}
                 mark_for_removal = snapshot_location not in self.used
             else:
                 unused_snapshots = {
                     snapshot
                     for snapshot in unused_snapshot_fossil
-                    if any(
-                        TestLocation(node).matches_snapshot_name(snapshot.name)
-                        for node in self.ran_items
-                    )
+                    if self._selected_items_match_name(snapshot.name)
+                    or self._provided_nodes_match_name(snapshot.name, provided_nodes)
                 }
                 mark_for_removal = False
 
@@ -217,3 +264,43 @@ def _diff_snapshot_fossils(
 
     def _count_snapshots(self, snapshot_fossils: "SnapshotFossils") -> int:
         return sum(len(snapshot_fossil) for snapshot_fossil in snapshot_fossils)
+
+    def _is_matching_path(self, snapshot_location: str, provided_path: str) -> bool:
+        path = Path(provided_path)
+        return str(path if path.is_dir() else path.parent) in snapshot_location
+
+    def _get_matching_path_nodes(self, snapshot_location: str) -> List[List[str]]:
+        return [
+            self._provided_test_paths[path]
+            for path in self._provided_test_paths
+            if self._is_matching_path(snapshot_location, path)
+        ]
+
+    def _provided_nodes_match_name(
+        self, snapshot_name: str, provided_nodes: List[List[str]]
+    ) -> bool:
+        return any(snapshot_name in ".".join(node_path) for node_path in provided_nodes)
+
+    def _provided_keywords_match_name(self, snapshot_name: str) -> bool:
+        names = snapshot_name.split(".")
+        return any(
+            expr.evaluate(lambda subname: any(subname in name for name in names))
+            for expr in self._keyword_expressions
+        )
+
+    def _ran_items_match_name(self, snapshot_name: str) -> bool:
+        return any(
+            PyTestLocation(item).matches_snapshot_name(snapshot_name)
+            for item in self.ran_items
+        )
+
+    def _selected_items_match_name(self, snapshot_name: str) -> bool:
+        if self._keyword_expressions:
+            return self._provided_keywords_match_name(snapshot_name)
+        return self._ran_items_match_name(snapshot_name)
+
+    def _selected_items_match_location(self, snapshot_location: str) -> bool:
+        return any(
+            PyTestLocation(item).matches_snapshot_location(snapshot_location)
+            for item in self.ran_items
+        )

src/syrupy/session.py

@@ -5,6 +5,7 @@
     Iterable,
     List,
     Optional,
+    Tuple,
 )
 
 import attr
@@ -24,8 +25,7 @@ class SnapshotSession:
     base_dir: str = attr.ib()
     update_snapshots: bool = attr.ib()
     warn_unused_snapshots: bool = attr.ib()
-    is_providing_paths: bool = attr.ib()
-    is_providing_nodes: bool = attr.ib()
+    _invocation_args: Tuple[str, ...] = attr.ib(factory=tuple)
     report: Optional["SnapshotReport"] = attr.ib(default=None)
     _all_items: Dict["pytest.Item", bool] = attr.ib(factory=dict)
     _ran_items: Dict["pytest.Item", bool] = attr.ib(factory=dict)
@@ -48,8 +48,7 @@ def finish(self) -> int:
             assertions=self._assertions,
             update_snapshots=self.update_snapshots,
             warn_unused_snapshots=self.warn_unused_snapshots,
-            is_providing_paths=self.is_providing_paths,
-            is_providing_nodes=self.is_providing_nodes,
+            invocation_args=self._invocation_args,
         )
         if self.report.num_unused:
             if self.update_snapshots: