1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
import enum
import pathlib
import ijson
import pytest
def _get_available_backends():
backends = []
for backend in ijson.ALL_BACKENDS:
try:
backends.append(ijson.get_backend(backend))
except ImportError:
pass
return backends
_available_backends = _get_available_backends()
class InputType(enum.Enum):
ASYNC_FILE = enum.auto()
ASYNC_TYPES_COROUTINES_FILE = enum.auto()
FILE = enum.auto()
SENDABLE = enum.auto()
class BackendAdaptor:
"""
Ties a backend together with an input type to provide easy access to
calling the backend's methods and retrieving all results in a single call.
"""
def __init__(self, backend, input_type, suffix, get_all):
self.backend = backend
self._input_type = input_type
self._suffix = suffix
self._get_all = get_all
@property
def pytest_parameter_id(self):
return f"{self.backend.backend_name}-{self._input_type.name.lower()}"
def __getattr__(self, name):
routine = getattr(self.backend, name + self._suffix)
def get_all_for_name(*args, **kwargs):
return self._get_all(routine, *args, **kwargs)
return get_all_for_name
from .support.async_ import get_all as get_all_async
from .support.async_types_coroutines import get_all as get_all_async_types_coroutines
from .support.coroutines import get_all as get_all_coro
from .support.generators import get_all as get_all_gen
_pull_backend_adaptors = [
backend_adaptor
for backend in _available_backends
for backend_adaptor in [
BackendAdaptor(backend, InputType.ASYNC_FILE, "_async", get_all_async),
BackendAdaptor(backend, InputType.ASYNC_TYPES_COROUTINES_FILE, "_async", get_all_async_types_coroutines),
BackendAdaptor(backend, InputType.FILE, "_gen", get_all_gen),
]
]
_push_backend_adaptors = [
backend_adaptor
for backend in _available_backends
for backend_adaptor in [
BackendAdaptor(backend, InputType.SENDABLE, "_coro", get_all_coro),
]
]
_all_backend_adaptors = _pull_backend_adaptors + _push_backend_adaptors
BACKEND_PARAM_NAME = "backend"
ADAPTOR_PARAM_NAME = "adaptor"
def pytest_generate_tests(metafunc):
requires_backend = BACKEND_PARAM_NAME in metafunc.fixturenames
requires_adaptor = ADAPTOR_PARAM_NAME in metafunc.fixturenames
assert not (requires_backend and requires_adaptor)
names = []
# if both are required we need to match backend and adaptors correctly
if requires_backend:
names = BACKEND_PARAM_NAME
values = _available_backends
ids = [backend.backend_name for backend in _available_backends]
elif requires_adaptor:
pull_only = bool(list(metafunc.definition.iter_markers('pull_only')))
adaptors = _pull_backend_adaptors if pull_only else _all_backend_adaptors
names = ADAPTOR_PARAM_NAME
values = adaptors
ids = [adaptor.pytest_parameter_id for adaptor in adaptors]
if names:
metafunc.parametrize(names, values, ids=ids)
def pytest_addoption(parser):
group = parser.getgroup("Memory leak tests")
group.addoption("--memleaks", action="store_true", help="include memory leak tests")
group.addoption("--memleaks-only", action="store_true", help="run ONLY memory leak tests")
def pytest_collection_modifyitems(config, items):
if config.option.memleaks_only:
skip_mark = pytest.mark.skip(reason="running only memleak tests")
for item in items:
if pathlib.Path(item.fspath).name != "test_memleaks.py":
item.add_marker(skip_mark)
elif not config.option.memleaks:
skip_mark = pytest.mark.skip(reason="run with --memleaks or --memleaks-only option")
for item in items:
if pathlib.Path(item.fspath).name == "test_memleaks.py":
item.add_marker(skip_mark)
|
