1import json 2from urllib.parse import urlunsplit 3 4from .api_handler import ApiHandler 5from ...utils.serializer import serialize_session 6from ...data.session import PAUSED, COMPLETED, ABORTED, PENDING, RUNNING 7 8DEFAULT_LAST_COMPLETED_TESTS_COUNT = 5 9DEFAULT_LAST_COMPLETED_TESTS_STATUS = ["ALL"] 10 11 12class TestsApiHandler(ApiHandler): 13 def __init__( 14 self, 15 wpt_port, 16 wpt_ssl_port, 17 tests_manager, 18 sessions_manager, 19 hostname, 20 web_root, 21 test_loader 22 ): 23 super(TestsApiHandler, self).__init__(web_root) 24 self._tests_manager = tests_manager 25 self._sessions_manager = sessions_manager 26 self._wpt_port = wpt_port 27 self._wpt_ssl_port = wpt_ssl_port 28 self._hostname = hostname 29 self._web_root = web_root 30 self._test_loader = test_loader 31 32 def read_tests(self, response): 33 tests = self._tests_manager.read_tests() 34 self.send_json(tests, response) 35 36 def read_session_tests(self, request, response): 37 uri_parts = self.parse_uri(request) 38 token = uri_parts[2] 39 session = self._sessions_manager.read_session(token) 40 41 if session is None: 42 response.status = 404 43 return 44 45 data = serialize_session(session) 46 tests = { 47 "token": token, 48 "pending_tests": data["pending_tests"], 49 "running_tests": data["running_tests"] 50 } 51 self.send_json(tests, response) 52 53 def read_next_test(self, request, response): 54 try: 55 uri_parts = self.parse_uri(request) 56 token = uri_parts[2] 57 58 hostname = self._hostname 59 60 session = self._sessions_manager.read_session(token) 61 if session is None: 62 response.status = 404 63 return 64 65 if session.status == PAUSED: 66 url = self._generate_wave_url( 67 hostname=hostname, 68 uri="pause.html", 69 token=token 70 ) 71 self.send_json({"next_test": url}, response) 72 return 73 if session.status == COMPLETED or session.status == ABORTED: 74 url = self._generate_wave_url( 75 hostname=hostname, 76 uri="finish.html", 77 token=token 78 ) 79 self.send_json({"next_test": url}, response) 80 return 81 if session.status == PENDING: 82 url = self._generate_wave_url( 83 hostname=hostname, 84 uri="newsession.html", 85 token=token 86 ) 87 self.send_json({"next_test": url}, response) 88 return 89 90 test = self._tests_manager.next_test(session) 91 92 if test is None: 93 if session.status != RUNNING: 94 return 95 url = self._generate_wave_url( 96 hostname=hostname, 97 uri="finish.html", 98 token=token 99 ) 100 self.send_json({"next_test": url}, response) 101 self._sessions_manager.complete_session(token) 102 return 103 104 test_timeout = self._tests_manager.get_test_timeout( 105 test=test, session=session) 106 url = self._generate_test_url( 107 test=test, 108 token=token, 109 test_timeout=test_timeout, 110 hostname=hostname) 111 112 self.send_json({ 113 "next_test": url 114 }, response) 115 except Exception: 116 self.handle_exception("Failed to read next test") 117 response.status = 500 118 119 def read_last_completed(self, request, response): 120 try: 121 uri_parts = self.parse_uri(request) 122 token = uri_parts[2] 123 query = self.parse_query_parameters(request) 124 count = None 125 if "count" in query: 126 count = query["count"] 127 else: 128 count = DEFAULT_LAST_COMPLETED_TESTS_COUNT 129 130 status = None 131 if "status" in query: 132 status = query["status"].split(",") 133 else: 134 status = DEFAULT_LAST_COMPLETED_TESTS_STATUS 135 136 completed_tests = self._tests_manager.read_last_completed_tests( 137 token, count) 138 tests = {} 139 for one_status in status: 140 one_status = one_status.lower() 141 if one_status == "pass": 142 tests["pass"] = completed_tests["pass"] 143 continue 144 if one_status == "fail": 145 tests["fail"] = completed_tests["fail"] 146 continue 147 if one_status == "timeout": 148 tests["timeout"] = completed_tests["timeout"] 149 continue 150 if one_status == "all": 151 tests["pass"] = completed_tests["pass"] 152 tests["fail"] = completed_tests["fail"] 153 tests["timeout"] = completed_tests["timeout"] 154 break 155 self.send_json(data=tests, response=response) 156 except Exception: 157 self.handle_exception("Failed to read last completed tests") 158 response.status = 500 159 160 def read_malfunctioning(self, request, response): 161 try: 162 uri_parts = self.parse_uri(request) 163 token = uri_parts[2] 164 tm = self._tests_manager 165 malfunctioning_tests = tm.read_malfunctioning_tests(token) 166 167 self.send_json(data=malfunctioning_tests, response=response) 168 except Exception: 169 self.handle_exception("Failed to read malfunctioning tests") 170 response.status = 500 171 172 def update_malfunctioning(self, request, response): 173 try: 174 uri_parts = self.parse_uri(request) 175 token = uri_parts[2] 176 177 data = None 178 body = request.body.decode("utf-8") 179 if body != "": 180 data = json.loads(body) 181 182 self._tests_manager.update_malfunctioning_tests(token, data) 183 except Exception: 184 self.handle_exception("Failed to update malfunctioning tests") 185 response.status = 500 186 187 def read_available_apis(self, request, response): 188 try: 189 apis = self._test_loader.get_apis() 190 self.send_json(apis, response) 191 except Exception: 192 self.handle_exception("Failed to read available APIs") 193 response.status = 500 194 195 def handle_request(self, request, response): 196 method = request.method 197 uri_parts = self.parse_uri(request) 198 199 # /api/tests 200 if len(uri_parts) == 2: 201 if method == "GET": 202 self.read_tests(response) 203 return 204 205 # /api/tests/<token> 206 if len(uri_parts) == 3: 207 if method == "GET": 208 if uri_parts[2] == "apis": 209 self.read_available_apis(request, response) 210 return 211 self.read_session_tests(request, response) 212 return 213 214 # /api/tests/<token>/<function> 215 if len(uri_parts) == 4: 216 function = uri_parts[3] 217 if method == "GET": 218 if function == "next": 219 self.read_next_test(request, response) 220 return 221 if function == "last_completed": 222 self.read_last_completed(request, response) 223 return 224 if function == "malfunctioning": 225 self.read_malfunctioning(request, response) 226 return 227 if method == "PUT": 228 if function == "malfunctioning": 229 self.update_malfunctioning(request, response) 230 return 231 232 response.status = 404 233 234 def _generate_wave_url(self, hostname, uri, token): 235 if self._web_root is not None: 236 uri = self._web_root + uri 237 238 return self._generate_url( 239 hostname=hostname, 240 uri=uri, 241 port=self._wpt_port, 242 query="token=" + token 243 ) 244 245 def _generate_test_url(self, hostname, test, token, test_timeout): 246 protocol = "http" 247 port = self._wpt_port 248 249 if "https" in test: 250 protocol = "https" 251 port = self._wpt_ssl_port 252 253 query = "token={}&timeout={}&https_port={}&web_root={}".format( 254 token, 255 test_timeout, 256 self._wpt_ssl_port, 257 self._web_root 258 ) 259 260 return self._generate_url( 261 protocol=protocol, 262 hostname=hostname, 263 port=port, 264 uri=test, 265 query=query 266 ) 267 268 def _generate_url(self, 269 hostname, 270 port=None, 271 uri=None, 272 query=None, 273 protocol=None): 274 if port is None: 275 port = 80 276 if uri is None: 277 uri = "/" 278 if query is None: 279 query = "" 280 if protocol is None: 281 protocol = "http" 282 return urlunsplit([protocol, "{}:{}".format(hostname, port), uri, query, '']) 283