# vim: ft=python fileencoding=utf-8 sts=4 sw=4 et: # Copyright 2015-2021 Florian Bruhin (The Compiler) # # This file is part of qutebrowser. # # qutebrowser is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # qutebrowser is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with qutebrowser. If not, see . """Fixtures for the server webserver.""" import re import sys import json import pathlib import socket import dataclasses from http import HTTPStatus import pytest from PyQt5.QtCore import pyqtSignal, QUrl from end2end.fixtures import testprocess class Request(testprocess.Line): """A parsed line from the flask log output. Attributes: verb/path/status: Parsed from the log output. """ def __init__(self, data): super().__init__(data) try: parsed = json.loads(data) except ValueError: raise testprocess.InvalidLine(data) assert isinstance(parsed, dict) assert set(parsed.keys()) == {'path', 'verb', 'status'} self.verb = parsed['verb'] path = parsed['path'] self.path = '/' if path == '/' else path.rstrip('/') self.status = parsed['status'] self._check_status() def _check_status(self): """Check if the http status is what we expected.""" path_to_statuses = { '/favicon.ico': [ HTTPStatus.OK, HTTPStatus.PARTIAL_CONTENT, HTTPStatus.NOT_MODIFIED, ], '/does-not-exist': [HTTPStatus.NOT_FOUND], '/does-not-exist-2': [HTTPStatus.NOT_FOUND], '/404': [HTTPStatus.NOT_FOUND], '/redirect-later': [HTTPStatus.FOUND], '/redirect-self': [HTTPStatus.FOUND], '/redirect-to': [HTTPStatus.FOUND], '/relative-redirect': [HTTPStatus.FOUND], '/absolute-redirect': [HTTPStatus.FOUND], '/cookies/set': [HTTPStatus.FOUND], '/cookies/set-custom': [HTTPStatus.FOUND], '/500-inline': [HTTPStatus.INTERNAL_SERVER_ERROR], '/500': [HTTPStatus.INTERNAL_SERVER_ERROR], } for i in range(15): path_to_statuses['/redirect/{}'.format(i)] = [HTTPStatus.FOUND] for suffix in ['', '1', '2', '3', '4', '5', '6']: key = ('/basic-auth/user{suffix}/password{suffix}' .format(suffix=suffix)) path_to_statuses[key] = [HTTPStatus.UNAUTHORIZED, HTTPStatus.OK] default_statuses = [HTTPStatus.OK, HTTPStatus.NOT_MODIFIED] sanitized = QUrl('http://localhost' + self.path).path() # Remove ?foo expected_statuses = path_to_statuses.get(sanitized, default_statuses) if self.status not in expected_statuses: raise AssertionError( "{} loaded with status {} but expected {}".format( sanitized, self.status, ' / '.join(repr(e) for e in expected_statuses))) def __eq__(self, other): return NotImplemented @dataclasses.dataclass(frozen=True) class ExpectedRequest: """Class to compare expected requests easily.""" verb: str path: int @classmethod def from_request(cls, request): """Create an ExpectedRequest from a Request.""" return cls(request.verb, request.path) def __eq__(self, other): if isinstance(other, (Request, ExpectedRequest)): return self.verb == other.verb and self.path == other.path else: return NotImplemented class WebserverProcess(testprocess.Process): """Abstraction over a running Flask server process. Reads the log from its stdout and parses it. Signals: new_request: Emitted when there's a new request received. """ new_request = pyqtSignal(Request) Request = Request # So it can be used from the fixture easily. ExpectedRequest = ExpectedRequest KEYS = ['verb', 'path'] def __init__(self, request, script, parent=None): super().__init__(request, parent) self._script = script self.port = self._random_port() self.new_data.connect(self.new_request) def _random_port(self) -> int: """Get a random free port.""" sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.bind(('localhost', 0)) port = sock.getsockname()[1] sock.close() return port def get_requests(self): """Get the requests to the server during this test.""" requests = self._get_data() return [r for r in requests if r.path != '/favicon.ico'] def _parse_line(self, line): self._log(line) started_re = re.compile(r' \* Running on https?://127\.0\.0\.1:{}/ ' r'\(Press CTRL\+C to quit\)'.format(self.port)) if started_re.fullmatch(line): self.ready.emit() return None return Request(line) def _executable_args(self): if hasattr(sys, 'frozen'): executable = str(pathlib.Path(sys.executable).parent / self._script) args = [] else: executable = sys.executable py_file = (pathlib.Path(__file__).parent / self._script).with_suffix('.py') args = [str(py_file)] return executable, args def _default_args(self): return [str(self.port)] @pytest.fixture(scope='session', autouse=True) def server(qapp, request): """Fixture for an server object which ensures clean setup/teardown.""" server = WebserverProcess(request, 'webserver_sub') server.start() yield server server.terminate() @pytest.fixture(autouse=True) def server_per_test(server, request): """Fixture to clean server request list after each test.""" if not hasattr(request.node, '_server_logs'): request.node._server_logs = [] request.node._server_logs.append(('server', server.captured_log)) server.before_test() yield server.after_test() @pytest.fixture def server2(qapp, request): """Fixture for a second server object for cross-origin tests.""" server = WebserverProcess(request, 'webserver_sub') if not hasattr(request.node, '_server_logs'): request.node._server_logs = [] request.node._server_logs.append(('secondary server', server.captured_log)) server.start() yield server server.terminate() @pytest.fixture def ssl_server(request, qapp): """Fixture for a webserver with a self-signed SSL certificate. This needs to be explicitly used in a test. """ server = WebserverProcess(request, 'webserver_sub_ssl') if not hasattr(request.node, '_server_logs'): request.node._server_logs = [] request.node._server_logs.append(('SSL server', server.captured_log)) server.start() yield server server.after_test() server.terminate()