conftest.py (9310B)
1 import copy 2 import json 3 import os 4 import ssl 5 import sys 6 import subprocess 7 import urllib 8 9 import html5lib 10 import py 11 import pytest 12 13 from wptserver import WPTServer 14 15 HERE = os.path.dirname(os.path.abspath(__file__)) 16 WPT_ROOT = os.path.normpath(os.path.join(HERE, '..', '..')) 17 HARNESS = os.path.join(HERE, 'harness.html') 18 TEST_TYPES = ('functional', 'unit') 19 20 sys.path.insert(0, os.path.normpath(os.path.join(WPT_ROOT, "tools"))) 21 import localpaths 22 23 sys.path.insert(0, os.path.normpath(os.path.join(WPT_ROOT, "tools", "webdriver"))) 24 import webdriver 25 26 27 def pytest_addoption(parser): 28 parser.addoption("--binary", action="store", default=None, help="path to browser binary") 29 parser.addoption("--headless", action="store_true", default=False, help="run browser in headless mode") 30 31 32 def pytest_collect_file(file_path, path, parent): 33 if file_path.suffix.lower() != '.html': 34 return 35 36 # Tests are organized in directories by type 37 test_type = os.path.relpath(str(file_path), HERE) 38 if os.path.sep not in test_type or ".." in test_type: 39 # HTML files in this directory are not tests 40 return 41 test_type = test_type.split(os.path.sep)[1] 42 43 return HTMLFile.from_parent(parent, path=file_path, test_type=test_type) 44 45 46 def pytest_configure(config): 47 config.proc = subprocess.Popen(["geckodriver"]) 48 config.add_cleanup(config.proc.kill) 49 50 capabilities = {"alwaysMatch": {"acceptInsecureCerts": True, "moz:firefoxOptions": {}}} 51 if config.getoption("--binary"): 52 capabilities["alwaysMatch"]["moz:firefoxOptions"]["binary"] = config.getoption("--binary") 53 if config.getoption("--headless"): 54 capabilities["alwaysMatch"]["moz:firefoxOptions"]["args"] = ["--headless"] 55 56 config.driver = webdriver.Session("localhost", 4444, 57 capabilities=capabilities) 58 config.driver.start() 59 config.add_cleanup(config.driver.end) 60 61 # Although the name of the `_create_unverified_context` method suggests 62 # that it is not intended for external consumption, the standard library's 63 # documentation explicitly endorses its use: 64 # 65 # > To revert to the previous, unverified, behavior 66 # > ssl._create_unverified_context() can be passed to the context 67 # > parameter. 68 # 69 # https://docs.python.org/2/library/httplib.html#httplib.HTTPSConnection 70 config.ssl_context = ssl._create_unverified_context() 71 72 config.server = WPTServer(WPT_ROOT) 73 config.server.start(config.ssl_context) 74 config.add_cleanup(config.server.stop) 75 76 77 def resolve_uri(context, uri): 78 if uri.startswith('/'): 79 base = WPT_ROOT 80 path = uri[1:] 81 else: 82 base = os.path.dirname(context) 83 path = uri 84 85 return os.path.exists(os.path.join(base, path)) 86 87 88 def _summarize(actual): 89 def _scrub_stack(test_obj): 90 copy = dict(test_obj) 91 del copy['stack'] 92 return copy 93 94 def _expand_status(status_obj): 95 for key, value in [item for item in status_obj.items()]: 96 # In "status" and "test" objects, the "status" value enum 97 # definitions are interspersed with properties for unrelated 98 # metadata. The following condition is a best-effort attempt to 99 # ignore non-enum properties. 100 if key != key.upper() or not isinstance(value, int): 101 continue 102 103 del status_obj[key] 104 105 if status_obj['status'] == value: 106 status_obj[u'status_string'] = key 107 108 del status_obj['status'] 109 110 return status_obj 111 112 def _summarize_test(test_obj): 113 del test_obj['index'] 114 115 assert 'phase' in test_obj 116 assert 'phases' in test_obj 117 assert 'COMPLETE' in test_obj['phases'] 118 assert test_obj['phase'] == test_obj['phases']['COMPLETE'] 119 del test_obj['phases'] 120 del test_obj['phase'] 121 122 return _expand_status(_scrub_stack(test_obj)) 123 124 def _summarize_status(status_obj): 125 return _expand_status(_scrub_stack(status_obj)) 126 127 128 summarized = {} 129 130 summarized[u'summarized_status'] = _summarize_status(actual['status']) 131 summarized[u'summarized_tests'] = [ 132 _summarize_test(test) for test in actual['tests']] 133 summarized[u'summarized_tests'].sort(key=lambda test_obj: test_obj.get('name')) 134 summarized[u'summarized_asserts'] = [ 135 {"assert_name": assert_item["assert_name"], 136 "test": assert_item["test"]["name"] if assert_item["test"] else None, 137 "args": assert_item["args"], 138 "status": assert_item["status"]} for assert_item in actual["asserts"]] 139 summarized[u'type'] = actual['type'] 140 141 return summarized 142 143 144 class HTMLFile(pytest.File): 145 def __init__(self, test_type=None, **kwargs): 146 super().__init__(**kwargs) 147 self.test_type = test_type 148 149 def collect(self): 150 url = self.session.config.server.url(self.path) 151 # Some tests are reliant on the WPT servers substitution functionality, 152 # so tests must be retrieved from the server rather than read from the 153 # file system directly. 154 handle = urllib.request.urlopen(url, 155 context=self.parent.session.config.ssl_context) 156 try: 157 markup = handle.read() 158 finally: 159 handle.close() 160 161 if self.test_type not in TEST_TYPES: 162 raise ValueError('Unrecognized test type: "%s"' % self.test_type) 163 164 parsed = html5lib.parse(markup, namespaceHTMLElements=False) 165 name = None 166 expected = None 167 168 for element in parsed.iter(): 169 if not name and element.tag == 'title': 170 name = element.text 171 continue 172 if element.tag == 'script': 173 if element.attrib.get('id') == 'expected': 174 try: 175 expected = json.loads(element.text) 176 except ValueError: 177 print("Failed parsing JSON in %s" % filename) 178 raise 179 180 if not name: 181 raise ValueError('No name found in %s add a <title> element' % filename) 182 elif self.test_type == 'functional': 183 if not expected: 184 raise ValueError('Functional tests must specify expected report data') 185 elif self.test_type == 'unit' and expected: 186 raise ValueError('Unit tests must not specify expected report data') 187 188 yield HTMLItem.from_parent(self, name=name, url=url, expected=expected) 189 190 191 class HTMLItem(pytest.Item): 192 def __init__(self, name, parent=None, config=None, session=None, nodeid=None, test_type=None, url=None, expected=None, **kwargs): 193 super().__init__(name, parent, config, session, nodeid, **kwargs) 194 195 self.test_type = self.parent.test_type 196 self.url = url 197 self.expected = expected 198 199 def reportinfo(self): 200 return self.fspath, None, self.url 201 202 def runtest(self): 203 if self.test_type == 'unit': 204 self._run_unit_test() 205 elif self.test_type == 'functional': 206 self._run_functional_test() 207 else: 208 raise NotImplementedError 209 210 def _run_unit_test(self): 211 driver = self.session.config.driver 212 server = self.session.config.server 213 214 driver.url = server.url(HARNESS) 215 216 actual = driver.execute_async_script( 217 'runTest("%s", "foo", arguments[0])' % self.url 218 ) 219 220 summarized = _summarize(copy.deepcopy(actual)) 221 222 print(json.dumps(summarized, indent=2)) 223 224 assert summarized[u'summarized_status'][u'status_string'] == u'OK', summarized[u'summarized_status'][u'message'] 225 for test in summarized[u'summarized_tests']: 226 msg = "%s\n%s" % (test[u'name'], test[u'message']) 227 assert test[u'status_string'] == u'PASS', msg 228 229 def _run_functional_test(self): 230 driver = self.session.config.driver 231 server = self.session.config.server 232 233 driver.url = server.url(HARNESS) 234 235 test_url = self.url 236 actual = driver.execute_async_script('runTest("%s", "foo", arguments[0])' % test_url) 237 238 print(json.dumps(actual, indent=2)) 239 240 summarized = _summarize(copy.deepcopy(actual)) 241 242 print(json.dumps(summarized, indent=2)) 243 244 # Test object ordering is not guaranteed. This weak assertion verifies 245 # that the indices are unique and sequential 246 indices = [test_obj.get('index') for test_obj in actual['tests']] 247 self._assert_sequence(indices) 248 249 self.expected[u'summarized_tests'].sort(key=lambda test_obj: test_obj.get('name')) 250 251 # Make asserts opt-in for now 252 if "summarized_asserts" not in self.expected: 253 del summarized["summarized_asserts"] 254 else: 255 # We can't be sure of the order of asserts even within the same test 256 # although we could also check for the failing assert being the final 257 # one 258 for obj in [summarized, self.expected]: 259 obj["summarized_asserts"].sort( 260 key=lambda x: (x["test"] or "", x["status"], x["assert_name"], tuple(x["args"]))) 261 262 assert summarized == self.expected 263 264 @staticmethod 265 def _assert_sequence(nums): 266 if nums and len(nums) > 0: 267 assert nums == list(range(nums[-1] + 1))