-
Notifications
You must be signed in to change notification settings - Fork 0
/
pytest_sipp.py
355 lines (278 loc) · 11.1 KB
/
pytest_sipp.py
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
import os.path
import re
import pytest
import pysipp
from pytest_exceptional import PytestException
from _pytest import fixtures
from _pytest.python import transfer_markers, Metafunc, PyobjMixin
try:
from shutil import which
except ImportError:
from backports.shutil_which import which
SCENARIO_ROOT = None
class SIPpNotFound(PytestException):
"""Missing dependencies error.
"""
def toterminal(self, longrepr, tw):
markup = dict(red=True, bold=True)
tw.line()
tw.write(" Could not find a suitable SIPp binary.", **markup)
tw.line(" Is it installed properly?")
tw.line(" See http://avina.qa.sangoma.local/sng_qa/sipp.html "
"for instructions.")
tw.line()
class SIPpTest(PyobjMixin, pytest.Item):
def __init__(self, name, parent, obj, config=None, callspec=None,
keywords=None, session=None, fixtureinfo=None,
description=None):
super(SIPpTest, self).__init__(name, parent, config=config,
session=session)
self.obj = obj
self._obj = obj
self.function = obj.function
self.fixturenames = fixtureinfo.names_closure
self._fixtureinfo = fixtureinfo
self.funcargs = {}
self.keywords.update(self.obj.__dict__)
if callspec:
self.callspec = callspec
self.keywords.update(callspec.keywords)
self._genid = callspec.id
if hasattr(callspec, 'param'):
self.param = callspec.param
if keywords:
self.keywords.update(keywords)
self._request = fixtures.FixtureRequest(self)
def runtest(self):
self.ihook.pytest_pyfunc_call(pyfuncitem=self)
def setup(self):
super(SIPpTest, self).setup()
fixtures.fillfixtures(self)
def generate_sipp_tests(metafunc, scen_node, **kwargs):
sipp_conf = getattr(metafunc.function, 'sipp_conf', None)
if sipp_conf:
settings = dict(sipp_conf.kwargs)
if not scen_node:
scen_node = settings.pop('scen_node', None)
settings.update(kwargs)
else:
settings = kwargs
scripts_root = settings.get('scen_root', SCENARIO_ROOT)
plugins = settings.get('pysipp_plugins', [])
exclude_expr = settings.get('exclude_expr')
if not scen_node:
metafunc.parametrize('sippscen',
[pysipp.scenario(autolocalsocks=False)],
ids=['default_sippscen'],
indirect=True)
return
elif os.path.isdir(scen_node):
scen_path = scen_node
elif scripts_root:
scen_path = os.path.join(scripts_root, scen_node)
else:
raise ValueError("Don't know where to find {}".format(scen_node))
if exclude_expr:
class reject_by_pattern(object):
def pysipp_load_scendir(self, path, xmls, confpy):
if re.match(exclude_expr, path):
return False
plugins.append(reject_by_pattern())
# This walks the file system every time we want some
# scenarios, and probably should be doing some caching, but
# this is not the layer to put it.
#
# The problem is we want different scenario objects for every
# test. Caching them will mean we end up sharing.
with pysipp.plugin.register(plugins):
scripts = list(pysipp.walk(scen_path, delay_conf_scen=True,
autolocalsocks=False))
try:
paths, scenarios = zip(*scripts)
except ValueError:
fname = metafunc.function.__name__
raise ValueError(
"No SIPp scenarios could be found under {}\n"
"The test '{}' has asked for scen_node '{}' under "
"scen_root '{}'".format(
scen_path, fname, scen_node, scripts_root
)
)
metafunc.parametrize('sippscen',
scenarios,
ids=[path[len(scen_path)+1:] for path in paths],
indirect=True)
class SIPpTestDescription(object):
def __init__(self, scen_node=None, **kwargs):
self.function = None
self.scen_node = scen_node
self.kwargs = kwargs
def __call__(self, function):
self.function = function
return self
def __getattr__(self, key):
# Pass through attribute access to the underlying method to
# when necessary to keep pytest's introspection happy
if self.function:
return getattr(self.function, key)
raise AttributeError(key)
def generate_tests(self, metafunc):
generate_sipp_tests(metafunc, self.scen_node, **self.kwargs)
@pytest.hookimpl
def pytest_pyfunc_call(pyfuncitem):
testdescription = pyfuncitem.obj
if not isinstance(testdescription, SIPpTestDescription):
return
config = pyfuncitem.config
funcargs = pyfuncitem.funcargs
argnames = pyfuncitem._fixtureinfo.argnames
testargs = {arg: funcargs[arg] for arg in argnames}
sippargs = dict(funcargs)
sippargs.update(testdescription.kwargs)
sippscen = sippargs.pop('sippscen')
# TODO: Hack to get tests inside classes to work. I'm doing
# something wrong if this isn't being handled for me.
func_code = None
func_code = getattr(testdescription.function, 'func_code', func_code)
func_code = getattr(testdescription.function, '__code__', func_code)
if 'self' in func_code.co_varnames:
testargs['self'] = None
testwrapper = testdescription.function(**testargs)
try:
excinfo = next(testwrapper)
except StopIteration:
raise RuntimeError("generator didn't yield")
def run_sippscen():
config.hook.pytest_run_sipp_scenario(item=pyfuncitem,
sippscen=sippscen,
sippargs=sippargs)
try:
if excinfo:
with pytest.raises(excinfo):
run_sippscen()
else:
run_sippscen()
finally:
try:
next(testwrapper)
except StopIteration:
pass
else:
raise RuntimeError("generator didn't stop")
config.hook.pytest_run_sipp_scenario_post(item=pyfuncitem,
sippscen=sippscen)
return True
@pytest.hookimpl
def pytest_run_sipp_scenario(item, sippscen, sippargs):
pytest.log.info('Launching SIPp scenario {}...'.format(sippscen.dirpath))
pytest.log.info('Running commands:\n{}'.format(sippscen.pformat_cmds()))
timeout = sippargs.pop('timeout', 180)
sippscen(timeout=timeout, **sippargs)
def gensipptests(collector, name, testdescription):
funcobj = testdescription.function
assert funcobj is not None
module = collector.getparent(pytest.Module).obj
clscol = collector.getparent(pytest.Class)
cls = clscol and clscol.obj or None
transfer_markers(testdescription, cls, module)
fm = collector.session._fixturemanager
fixtureinfo = fm.getfixtureinfo(collector, testdescription, cls)
if 'sippscen' not in fixtureinfo.names_closure:
fixtureinfo.names_closure.extend(['sippscen'])
metafunc = Metafunc(testdescription, fixtureinfo, collector.config,
cls=cls, module=module)
methods = [testdescription.generate_tests]
if hasattr(module, "pytest_generate_tests"):
methods.append(module.pytest_generate_tests)
if hasattr(cls, "pytest_generate_tests"):
methods.append(cls().pytest_generate_tests)
# SIPp tests will always be parameterized
collector.ihook.pytest_generate_tests.call_extra(methods,
dict(metafunc=metafunc))
fixtures.add_funcarg_pseudo_fixture_def(collector, metafunc, fm)
for callspec in metafunc._calls:
subname = "%s[%s]" % (name, callspec.id)
yield SIPpTest(name=subname,
parent=collector,
obj=testdescription,
callspec=callspec,
fixtureinfo=fixtureinfo,
keywords={callspec.id: True})
def sipp_test(scen_node=None, **kwargs):
# Support direct decoration
if callable(scen_node) and not kwargs:
return SIPpTestDescription()(scen_node)
return SIPpTestDescription(scen_node, **kwargs)
def issipptest(obj):
# Depending on when this is run, either we still have the test
# description class or we have a pytest item
return (isinstance(obj, SIPpTest)
or isinstance(obj, SIPpTestDescription))
@pytest.hookimpl
def pytest_namespace():
return {'sipp_test': sipp_test,
'issipptest': issipptest}
@pytest.hookimpl
def pytest_addoption(parser):
group = parser.getgroup('sipp')
group.addoption('--sipp-scen', '--sippscen', action='append', default=None,
help='path to the sipp scenario/script directory')
group.addoption(
'--sip-port', action='store', default=5060,
help="default port the dut listen's on for sip requests"
" (eg. default sip profile port)"
)
@pytest.hookimpl
def pytest_configure(config):
def set_scenario_root(path):
global SCENARIO_ROOT
SCENARIO_ROOT = path
config.hook.pytest_sipp_scenario_root.call_historic(
set_scenario_root,
kwargs=dict(config=config)
)
@pytest.hookimpl
def pytest_pycollect_makeitem(collector, name, obj):
if collector.funcnamefilter(name) and isinstance(obj, SIPpTestDescription):
return list(gensipptests(collector, name, obj))
@pytest.hookimpl
def pytest_generate_tests(metafunc):
# Handle parameterization of regular tests that use the sippscen
# fixture directly, instead of the sipp_test wrapper
if (not isinstance(metafunc.function, SIPpTestDescription)
and 'sippscen' in metafunc.funcargnames):
generate_sipp_tests(metafunc, None)
@pytest.fixture(scope='session')
def sipp_proxyaddr(request, dut_ip):
'''Return the dut's default sip profile socket.
(can be overridden from 5060 using --sip-port option)
'''
return dut_ip, request.config.getoption('--sip-port')
@pytest.fixture
def scen_db_path(request):
"""The path to the root of the SIPp scenario database
"""
sipp_conf = request.node.get_marker('sipp_conf')
if sipp_conf:
return sipp_conf.kwargs.get('scen_root', SCENARIO_ROOT)
return SCENARIO_ROOT
@pytest.fixture
def sippscen(request):
return request.param
@pytest.hookimpl
def pytest_runtest_protocol(item, nextitem):
if issipptest(item.obj) and not which('sipp'):
SIPpNotFound.makereport(item, when='setup')
return True # Do not continue with this test
@pytest.hookimpl
def pytest_addhooks(pluginmanager):
class SIPpHook:
@pytest.hookspec(historic=True)
def pytest_sipp_scenario_root(config):
"""Specify the default root for sipp scripts"""
@pytest.hookspec(firstresult=True)
def pytest_run_sipp_scenario(item, sippscen, sippargs):
"""Control how sippscen is called"""
def pytest_run_sipp_scenario_post(item, sippscen):
"""Post test hook"""
pluginmanager.add_hookspecs(SIPpHook())