Fix lint

Author: epenet

src/syrupy/report.py

@@ -521,8 +521,8 @@ def serialize(self) -> dict[str, Any]:
                     "nodeid": c.nodeid,
                     "name": c.name,
                     "path": str(c.path),
-                    "modulename": c.obj.__module__,
-                    "methodname": c.obj.__name__,
+                    "modulename": c.obj.__module__,  # type: ignore[attr-defined]
+                    "methodname": c.obj.__name__,  # type: ignore[attr-defined]
                 }
                 for c in list(self.collected_items)
             ],

src/syrupy/session.py

@@ -1,11 +1,11 @@
+import json
+import os
 from collections import defaultdict
 from dataclasses import (
     dataclass,
     field,
 )
 from enum import Enum
-import json
-import os
 from pathlib import Path
 from typing import (
     TYPE_CHECKING,
@@ -47,17 +47,20 @@ class ItemStatus(Enum):
     FAILED = "failed"
     SKIPPED = "skipped"
 
-class _MockPytestObject:
-    def __init__(self, collected_item):
+
+class _FakePytestObject:
+    def __init__(self, collected_item: dict[str, str]) -> None:
         self.__module__ = collected_item["modulename"]
         self.__name__ = collected_item["methodname"]
 
-class _MockPytestItem:
-    def __init__(self, collected_item: dict[str, Any]) -> None:
+
+class _FakePytestItem:
+    def __init__(self, collected_item: dict[str, str]) -> None:
         self.nodeid = collected_item["nodeid"]
         self.name = collected_item["name"]
         self.path = Path(collected_item["path"])
-        self.obj = _MockPytestObject(collected_item)
+        self.obj = _FakePytestObject(collected_item)
+
 
 @dataclass
 class SnapshotSession:
@@ -140,6 +143,24 @@ def ran_item(
             except ValueError:
                 pass  # if we don't understand the outcome, leave the item as "not run"
 
+    def _merge_collected_items(self, collected_items: list[dict[str, str]]) -> None:
+        for collected_item in collected_items:
+            custom_item = _FakePytestItem(collected_item)
+            if not any(
+                t.nodeid == custom_item.nodeid and t.name == custom_item.nodeid
+                for t in self._collected_items
+            ):
+                self._collected_items.add(custom_item)  # type: ignore[arg-type]
+
+    def _merge_selected_items(self, selected_items: dict[str, str]) -> None:
+        for key, selected_item in selected_items.items():
+            if key in self._selected_items:
+                status = ItemStatus(selected_item)
+                if status != ItemStatus.NOT_RUN:
+                    self._selected_items[key] = status
+            else:
+                self._selected_items[key] = ItemStatus(selected_item)
+
     def finish(self) -> int:
         exitstatus = 0
         self.flush_snapshot_write_queue()
@@ -152,12 +173,13 @@ def finish(self) -> int:
         )
 
         if is_xdist_worker():
+            worker_count = os.getenv("PYTEST_XDIST_WORKER_COUNT")
+            with open(".pytest_syrupy_worker_count", "w", encoding="utf-8") as f:
+                f.write(worker_count)  # type: ignore[arg-type]
             with open(
-                f"/workspaces/home-assistant-core/.syrupy.worker_count.txt", "w"
-            ) as f:
-                f.write(os.getenv("PYTEST_XDIST_WORKER_COUNT"))
-            with open(
-                f"/workspaces/home-assistant-core/.syrupy.{os.getenv("PYTEST_XDIST_WORKER")}.txt", "w"
+                f".pytest_syrupy_{os.getenv("PYTEST_XDIST_WORKER")}_result",
+                "w",
+                encoding="utf-8",
             ) as f:
                 json.dump(self.report.serialize(), f, indent=2)
             return exitstatus
@@ -166,31 +188,23 @@ def finish(self) -> int:
             # Until this is implemented, running syrupy with pytest-xdist is only
             # partially functional.
             return exitstatus
-        
+
         worker_count = None
         try:
-            with open(f"/workspaces/home-assistant-core/.syrupy.worker_count.txt", "r") as f:
+            with open(".pytest_syrupy_worker_count", encoding="utf-8") as f:
                 worker_count = f.read()
+            os.remove(".pytest_syrupy_worker_count")
         except FileNotFoundError:
             pass
 
         if worker_count:
             for i in range(int(worker_count)):
-                with open(f"/workspaces/home-assistant-core/.syrupy.gw{i}.txt", "r") as f:
+                with open(f".pytest_syrupy_gw{i}_result", encoding="utf-8") as f:
                     data = json.load(f)
-                    for collected_item in data["_collected_items"]:
-                        custom_item = _MockPytestItem(collected_item)
-                        if not any(t.nodeid == custom_item.nodeid and t.name == custom_item.nodeid for t in self._collected_items):
-                            self._collected_items.add(custom_item)
-                    for key, selected_item in data["_selected_items"].items():
-                        if key in self._selected_items:
-                            status = ItemStatus(selected_item)
-                            if status != ItemStatus.NOT_RUN:
-                                self._selected_items[key] = status
-                        else:
-                            self._selected_items[key] = ItemStatus(selected_item)
-
+                    self._merge_collected_items(data["_collected_items"])
+                    self._merge_selected_items(data["_selected_items"])
                     self.report.merge_serialized(data)
+                os.remove(f".pytest_syrupy_gw{i}_result")
 
         if self.report.num_unused:
             if self.update_snapshots: