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