Skip to content

Commit

Permalink
Fix lint
Browse files Browse the repository at this point in the history
  • Loading branch information
epenet committed Oct 17, 2024
1 parent 1852015 commit 73185c8
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 29 deletions.
4 changes: 2 additions & 2 deletions src/syrupy/report.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
],
Expand Down
68 changes: 41 additions & 27 deletions src/syrupy/session.py
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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()
Expand All @@ -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
Expand All @@ -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:
Expand Down

0 comments on commit 73185c8

Please sign in to comment.