Skip to content

Instantly share code, notes, and snippets.

@DevilXD
Last active June 9, 2020 12:59
Show Gist options
  • Save DevilXD/afc7f923049a5abedea55ba186e7219c to your computer and use it in GitHub Desktop.
Save DevilXD/afc7f923049a5abedea55ba186e7219c to your computer and use it in GitHub Desktop.
Pytest-dependency reordering support (module and session scopes only)
import warnings
from typing import Optional, List, Dict
from collections import defaultdict, deque
from pytest import Module, Item
def pytest_collection_modifyitems(items: List[Item]):
session_names: Set[str] = set()
module_names: Dict[Module, Set[str]] = defaultdict(set)
# gather dependency names
for item in items:
for marker in item.iter_markers("dependency"):
scope = marker.kwargs.get("scope", "module")
name = marker.kwargs.get("name")
if not name:
nodeid = item.nodeid.replace("::()::", "::")
if scope == "session" or scope == "package":
name = nodeid
elif scope == "module":
name = nodeid.split("::", 1)[1]
elif scope == "class":
name = nodeid.split("::", 2)[2]
original = item.originalname if item.originalname is not None else item.name
# remove the parametrization part at the end
if not name.endswith(original):
index = name.rindex(original) + len(original)
name = name[:index]
if scope == "module":
module_names[item.module].add(name)
elif scope == "session":
session_names.add(name)
final_items: List[Item] = []
session_deps: Dict[str, Item] = {}
module_deps: Dict[Module, Dict[str, Item]] = defaultdict(dict)
# group the dependencies by their scopes
cycles = 0
deque_items = deque(items)
while deque_items:
if cycles > len(deque_items):
# seems like we're stuck in a loop now
# just add the remaining items and finish up
final_items.extend(deque_items)
break
item = deque_items.popleft()
correct_order: Optional[bool] = True
for marker in item.iter_markers("dependency"):
depends = marker.kwargs.get("depends", [])
scope = marker.kwargs.get("scope", "module")
name = marker.kwargs.get("name")
if not name:
nodeid = item.nodeid.replace("::()::", "::")
if scope == "session" or scope == "package":
name = nodeid
elif scope == "module":
name = nodeid.split("::", 1)[1]
elif scope == "class":
name = nodeid.split("::", 2)[2]
original = item.originalname if item.originalname is not None else item.name
# remove the parametrization part at the end
if not name.endswith(original):
index = name.rindex(original) + len(original)
name = name[:index]
# pick a scope
if scope == "module":
scope_deps = module_deps[item.module]
scope_names = module_names[item.module]
elif scope == "session":
scope_deps = session_deps
scope_names = session_names
# check deps
if not all(d in scope_deps for d in depends):
# check to see if we're ever gonna see a dep like that
for d in depends:
if d in scope_names:
if correct_order is not None:
correct_order = False
else:
correct_order = None
warnings.warn(
f"Dependency '{d}' of '{name}' doesn't exist, "
"or has incorrect scope!",
RuntimeWarning,
)
break
# save
if scope == "module":
module_deps[item.module][name] = item
elif scope == "session":
session_deps[name] = item # use 'nodeid' instead of the name
# 'correct_order' possible values:
# None - invalid dependency, add anyway
# True - add it to the final list
# False - missing dependency, add it back to the processing deque
if correct_order is None:
# TODO: Take the config into account here
final_items.append(item)
cycles = 0
elif correct_order:
final_items.append(item)
cycles = 0
else:
deque_items.append(item)
cycles += 1
assert len(items) == len(final_items) and all(i in items for i in final_items)
items[:] = final_items
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment