refactor: do not instantiate extension with test_location

src/syrupy/assertion.py

@@ -94,7 +94,7 @@ def __post_init__(self) -> None:
     def __init_extension(
         self, extension_class: Type["AbstractSyrupyExtension"]
     ) -> "AbstractSyrupyExtension":
-        return extension_class(test_location=self.test_location)
+        return extension_class()
 
     @property
     def extension(self) -> "AbstractSyrupyExtension":
@@ -268,6 +268,7 @@ def _assert(self, data: "SerializableData") -> bool:
             if not matches and self.update_snapshots:
                 self.session.queue_snapshot_write(
                     extension=self.extension,
+                    test_location=self.test_location,
                     data=serialized_data,
                     index=self.index,
                 )

src/syrupy/extensions/base.py

@@ -3,15 +3,13 @@
     ABC,
     abstractmethod,
 )
-from collections import defaultdict
 from difflib import ndiff
 from gettext import gettext
 from itertools import zip_longest
 from pathlib import Path
 from typing import (
     TYPE_CHECKING,
     Callable,
-    DefaultDict,
     Dict,
     Iterator,
     List,
@@ -79,11 +77,6 @@ def serialize(
 class SnapshotCollectionStorage(ABC):
     _file_extension = ""
 
-    @property
-    @abstractmethod
-    def test_location(self) -> "PyTestLocation":
-        raise NotImplementedError
-
     @classmethod
     def get_snapshot_name(
         cls, *, test_location: "PyTestLocation", index: "SnapshotIndex" = 0
@@ -160,63 +153,58 @@ def read_snapshot(
     def write_snapshot(
         cls,
         *,
-        test_location: "PyTestLocation",
-        snapshots: List[Tuple["SerializedData", "SnapshotIndex"]],
+        snapshot_location: str,
+        snapshots: List[Tuple["SerializedData", "PyTestLocation", "SnapshotIndex"]],
     ) -> None:
         """
         This method is _final_, do not override. You can override
         `_write_snapshot_collection` in a subclass to change behaviour.
         """
+        if not snapshots:
+            return
+
         # First we group by location since it'll let us batch by file on disk.
         # Not as useful for single file snapshots, but useful for the standard
         # Amber extension.
-        locations: DefaultDict[str, List["Snapshot"]] = defaultdict(list)
-        for data, index in snapshots:
-            location = cls.get_location(test_location=test_location, index=index)
+        snapshot_collection = SnapshotCollection(location=snapshot_location)
+        for data, test_location, index in snapshots:
             snapshot_name = cls.get_snapshot_name(
                 test_location=test_location, index=index
             )
-            locations[location].append(Snapshot(name=snapshot_name, data=data))
-
-            # Ensures the folder path for the snapshot file exists.
-            try:
-                Path(
-                    cls.get_location(test_location=test_location, index=index)
-                ).parent.mkdir(parents=True)
-            except FileExistsError:
-                pass
+            snapshot = Snapshot(name=snapshot_name, data=data)
+            snapshot_collection.add(snapshot)
 
-        for location, location_snapshots in locations.items():
-            snapshot_collection = SnapshotCollection(location=location)
-
-            if not test_location.matches_snapshot_location(location):
+            if not test_location.matches_snapshot_location(snapshot_location):
                 warning_msg = gettext(
                     "{line_end}Can not relate snapshot location '{}' "
                     "to the test location.{line_end}"
                     "Consider adding '{}' to the generated location."
                 ).format(
-                    location,
+                    snapshot_location,
                     test_location.basename,
                     line_end="\n",
                 )
                 warnings.warn(warning_msg)
 
-            for snapshot in location_snapshots:
-                snapshot_collection.add(snapshot)
-
-                if not test_location.matches_snapshot_name(snapshot.name):
-                    warning_msg = gettext(
-                        "{line_end}Can not relate snapshot name '{}' "
-                        "to the test location.{line_end}"
-                        "Consider adding '{}' to the generated name."
-                    ).format(
-                        snapshot.name,
-                        test_location.testname,
-                        line_end="\n",
-                    )
-                    warnings.warn(warning_msg)
+            if not test_location.matches_snapshot_name(snapshot.name):
+                warning_msg = gettext(
+                    "{line_end}Can not relate snapshot name '{}' "
+                    "to the test location.{line_end}"
+                    "Consider adding '{}' to the generated name."
+                ).format(
+                    snapshot.name,
+                    test_location.testname,
+                    line_end="\n",
+                )
+                warnings.warn(warning_msg)
+
+        # Ensures the folder path for the snapshot file exists.
+        try:
+            Path(snapshot_location).parent.mkdir(parents=True)
+        except FileExistsError:
+            pass
 
-            cls._write_snapshot_collection(snapshot_collection=snapshot_collection)
+        cls._write_snapshot_collection(snapshot_collection=snapshot_collection)
 
     @abstractmethod
     def delete_snapshots(
@@ -432,9 +420,4 @@ def matches(
 class AbstractSyrupyExtension(
     SnapshotSerializer, SnapshotCollectionStorage, SnapshotReporter, SnapshotComparator
 ):
-    def __init__(self, test_location: "PyTestLocation"):
-        self._test_location = test_location
-
-    @property
-    def test_location(self) -> "PyTestLocation":
-        return self._test_location
+    pass

src/syrupy/session.py

@@ -14,10 +14,13 @@
     Optional,
     Set,
     Tuple,
+    Type,
 )
 
 import pytest
 
+from syrupy.location import PyTestLocation
+
 from .constants import EXIT_STATUS_FAIL_UNUSED
 from .data import SnapshotCollections
 from .report import SnapshotReport
@@ -49,24 +52,33 @@ class SnapshotSession:
     )
 
     _queued_snapshot_writes: Dict[
-        "AbstractSyrupyExtension", List[Tuple["SerializedData", "SnapshotIndex"]]
+        Tuple[Type["AbstractSyrupyExtension"], str],
+        List[Tuple["SerializedData", "PyTestLocation", "SnapshotIndex"]],
     ] = field(default_factory=dict)
 
     def queue_snapshot_write(
         self,
         extension: "AbstractSyrupyExtension",
+        test_location: "PyTestLocation",
         data: "SerializedData",
         index: "SnapshotIndex",
     ) -> None:
-        queue = self._queued_snapshot_writes.get(extension, [])
-        queue.append((data, index))
-        self._queued_snapshot_writes[extension] = queue
+        snapshot_location = extension.get_location(
+            test_location=test_location, index=index
+        )
+        key = (extension.__class__, snapshot_location)
+        queue = self._queued_snapshot_writes.get(key, [])
+        queue.append((data, test_location, index))
+        self._queued_snapshot_writes[key] = queue
 
     def flush_snapshot_write_queue(self) -> None:
-        for extension, queued_write in self._queued_snapshot_writes.items():
+        for (
+            extension_class,
+            snapshot_location,
+        ), queued_write in self._queued_snapshot_writes.items():
             if queued_write:
-                extension.write_snapshot(
-                    test_location=extension.test_location, snapshots=queued_write
+                extension_class.write_snapshot(
+                    snapshot_location=snapshot_location, snapshots=queued_write
                 )
         self._queued_snapshot_writes = {}