/
runner.py
135 lines (105 loc) · 4.18 KB
/
runner.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
"""
Provides interface to deal with pytest.
Usage::
session = webdriver.client.Session("127.0.0.1", "4444", "/")
harness_result = ("OK", None)
subtest_results = pytestrunner.run("/path/to/test", session.url)
return (harness_result, subtest_results)
"""
import errno
import json
import os
import shutil
import tempfile
pytest = None
def do_delayed_imports():
global pytest
import pytest
def run(path, server_config, session_config, timeout=0):
"""
Run Python test at ``path`` in pytest. The provided ``session``
is exposed as a fixture available in the scope of the test functions.
:param path: Path to the test file.
:param session_config: dictionary of host, port,capabilities parameters
to pass through to the webdriver session
:param timeout: Duration before interrupting potentially hanging
tests. If 0, there is no timeout.
:returns: (<harness result>, [<subtest result>, ...]),
where <subtest result> is (test id, status, message, stacktrace).
"""
if pytest is None:
do_delayed_imports()
os.environ["WD_HOST"] = session_config["host"]
os.environ["WD_PORT"] = str(session_config["port"])
os.environ["WD_CAPABILITIES"] = json.dumps(session_config["capabilities"])
os.environ["WD_SERVER_CONFIG"] = json.dumps(server_config)
harness = HarnessResultRecorder()
subtests = SubtestResultRecorder()
with TemporaryDirectory() as cache:
try:
pytest.main(["--strict", # turn warnings into errors
"--verbose", # show each individual subtest
"--capture", "no", # enable stdout/stderr from tests
"--basetemp", cache, # temporary directory
"--showlocals", # display contents of variables in local scope
"-p", "no:mozlog", # use the WPT result recorder
"-p", "no:cacheprovider", # disable state preservation across invocations
path],
plugins=[harness, subtests])
except Exception as e:
harness.outcome = ("ERROR", str(e))
return (harness.outcome, subtests.results)
class HarnessResultRecorder(object):
outcomes = {
"failed": "ERROR",
"passed": "OK",
"skipped": "SKIP",
}
def __init__(self):
# we are ok unless told otherwise
self.outcome = ("OK", None)
def pytest_collectreport(self, report):
harness_result = self.outcomes[report.outcome]
self.outcome = (harness_result, None)
class SubtestResultRecorder(object):
def __init__(self):
self.results = []
def pytest_runtest_logreport(self, report):
if report.passed and report.when == "call":
self.record_pass(report)
elif report.failed:
if report.when != "call":
self.record_error(report)
else:
self.record_fail(report)
elif report.skipped:
self.record_skip(report)
def record_pass(self, report):
self.record(report.nodeid, "PASS")
def record_fail(self, report):
self.record(report.nodeid, "FAIL", stack=report.longrepr)
def record_error(self, report):
# error in setup/teardown
if report.when != "call":
message = "%s error" % report.when
self.record(report.nodeid, "ERROR", message, report.longrepr)
def record_skip(self, report):
self.record(report.nodeid, "ERROR",
"In-test skip decorators are disallowed, "
"please use WPT metadata to ignore tests.")
def record(self, test, status, message=None, stack=None):
if stack is not None:
stack = str(stack)
new_result = (test.split("::")[-1], status, message, stack)
self.results.append(new_result)
class TemporaryDirectory(object):
def __enter__(self):
self.path = tempfile.mkdtemp(prefix="pytest-")
return self.path
def __exit__(self, *args):
try:
shutil.rmtree(self.path)
except OSError as e:
# no such file or directory
if e.errno != errno.ENOENT:
raise