tests: Various cleanups.
This commit is contained in:
Florian Bruhin
parent
7baed5f80f
commit
b8467b8fef
@ -32,19 +32,56 @@ import pytest_bdd as bdd
|
||||
from helpers import utils # pylint: disable=import-error
|
||||
|
||||
|
||||
## Given
|
||||
|
||||
|
||||
@bdd.given(bdd.parsers.parse("I set {sect} -> {opt} to {value}"))
|
||||
def set_setting(quteproc, httpbin, sect, opt, value):
|
||||
def set_setting_given(quteproc, httpbin, sect, opt, value):
|
||||
"""Set a qutebrowser setting.
|
||||
|
||||
This is available as "Given:" step so it can be used as "Background:".
|
||||
"""
|
||||
value = value.replace('(port)', str(httpbin.port))
|
||||
quteproc.set_setting(sect, opt, value)
|
||||
|
||||
|
||||
@bdd.given(bdd.parsers.parse("I open {path}"))
|
||||
def open_path_given(quteproc, path):
|
||||
"""Open a URL.
|
||||
|
||||
This is available as "Given:" step so it can be used as "Background:".
|
||||
|
||||
It always opens a new tab, unlike "When I open ..."
|
||||
"""
|
||||
quteproc.open_path(path, new_tab=True)
|
||||
|
||||
|
||||
@bdd.given(bdd.parsers.parse("I run {command}"))
|
||||
def run_command_given(quteproc, command):
|
||||
"""Run a qutebrowser command.
|
||||
|
||||
This is available as "Given:" step so it can be used as "Background:".
|
||||
"""
|
||||
quteproc.send_cmd(command)
|
||||
|
||||
|
||||
@bdd.given("I have a fresh instance")
|
||||
def fresh_instance(quteproc):
|
||||
"""Restart qutebrowser instance for tests needing a fresh state."""
|
||||
quteproc.terminate()
|
||||
quteproc.start()
|
||||
|
||||
|
||||
## When
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I open {path}"))
|
||||
def open_path_when(quteproc, path):
|
||||
def open_path(quteproc, path):
|
||||
"""Open a URL.
|
||||
|
||||
If used like "When I open ... in a new tab", the URL is opened ina new
|
||||
tab.
|
||||
"""
|
||||
new_tab_suffix = ' in a new tab'
|
||||
if path.endswith(new_tab_suffix):
|
||||
path = path[:-len(new_tab_suffix)]
|
||||
@ -56,25 +93,18 @@ def open_path_when(quteproc, path):
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I set {sect} -> {opt} to {value}"))
|
||||
def set_setting_when(quteproc, httpbin, sect, opt, value):
|
||||
def set_setting(quteproc, httpbin, sect, opt, value):
|
||||
"""Set a qutebrowser setting."""
|
||||
value = value.replace('(port)', str(httpbin.port))
|
||||
quteproc.set_setting(sect, opt, value)
|
||||
|
||||
|
||||
@bdd.given(bdd.parsers.parse("I run {command}"))
|
||||
def run_command_given(quteproc, command):
|
||||
quteproc.send_cmd(command)
|
||||
|
||||
|
||||
@bdd.given("I have a fresh instance")
|
||||
def fresh_instance(quteproc):
|
||||
"""Restart qutebrowser instance for tests needing a fresh state."""
|
||||
quteproc.terminate()
|
||||
quteproc.start()
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I run {command}"))
|
||||
def run_command_when(quteproc, httpbin, command):
|
||||
def run_command(quteproc, httpbin, command):
|
||||
"""Run a qutebrowser command.
|
||||
|
||||
The suffix "with count ..." can be used to pass a count to the command.
|
||||
"""
|
||||
if 'with count' in command:
|
||||
command, count = command.split(' with count ')
|
||||
count = int(count)
|
||||
@ -86,18 +116,25 @@ def run_command_when(quteproc, httpbin, command):
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I reload"))
|
||||
def reload(qtbot, httpbin, quteproc, command):
|
||||
"""Reload and wait until a new request is received."""
|
||||
with qtbot.waitSignal(httpbin.new_request, raising=True):
|
||||
quteproc.send_cmd(':reload')
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I wait until {path} is loaded"))
|
||||
def wait_until_loaded(quteproc, path):
|
||||
"""Wait until the given path is loaded (as per qutebrowser log)."""
|
||||
quteproc.wait_for_load_finished(path)
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.re(r'I wait for (?P<is_regex>regex )?"'
|
||||
r'(?P<pattern>[^"]+)" in the log'))
|
||||
def wait_in_log(quteproc, is_regex, pattern):
|
||||
"""Wait for a given pattern in the qutebrowser log.
|
||||
|
||||
If used like "When I wait for regex ... in the log" the argument is treated
|
||||
as regex. Otherwise, it's treated as a pattern (* can be used as wildcard).
|
||||
"""
|
||||
if is_regex:
|
||||
pattern = re.compile(pattern)
|
||||
quteproc.wait_for(message=pattern)
|
||||
@ -106,16 +143,34 @@ def wait_in_log(quteproc, is_regex, pattern):
|
||||
@bdd.when(bdd.parsers.re(r'I wait for the (?P<category>error|message|warning) '
|
||||
r'"(?P<message>.*)"'))
|
||||
def wait_for_message(quteproc, httpbin, category, message):
|
||||
expect_error(quteproc, httpbin, category, message)
|
||||
"""Wait for a given statusbar message/error/warning."""
|
||||
expect_message(quteproc, httpbin, category, message)
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I wait {delay}s"))
|
||||
def wait_time(quteproc, delay):
|
||||
"""Sleep for the given delay."""
|
||||
time.sleep(float(delay))
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.re('I press the keys? "(?P<keys>[^"]*)"'))
|
||||
def press_keys(quteproc, keys):
|
||||
"""Send the given fake keys to qutebrowser."""
|
||||
quteproc.press_keys(keys)
|
||||
|
||||
|
||||
## Then
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse("{path} should be loaded"))
|
||||
def path_should_be_loaded(httpbin, path):
|
||||
"""Make sure the given path was loaded from the webserver."""
|
||||
httpbin.wait_for(verb='GET', path='/' + path)
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse("The requests should be:\n{pages}"))
|
||||
def list_of_loaded_pages(httpbin, pages):
|
||||
def list_of_requests(httpbin, pages):
|
||||
"""Make sure the given requests were done from the webserver."""
|
||||
expected_requests = [httpbin.ExpectedRequest('GET', '/' + path.strip())
|
||||
for path in pages.split('\n')]
|
||||
actual_requests = httpbin.get_requests()
|
||||
@ -123,20 +178,22 @@ def list_of_loaded_pages(httpbin, pages):
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse("The unordered requests should be:\n{pages}"))
|
||||
def list_of_loaded_pages_unordered(httpbin, pages):
|
||||
def list_of_requests_unordered(httpbin, pages):
|
||||
"""Make sure the given requests were done (in no particular order)."""
|
||||
expected_requests = [httpbin.ExpectedRequest('GET', '/' + path.strip())
|
||||
for path in pages.split('\n')]
|
||||
actual_requests = httpbin.get_requests()
|
||||
# Requests are not hashable, we need to convert to ExpectedRequests
|
||||
actual_requests = map(httpbin.ExpectedRequest.from_request,
|
||||
actual_requests)
|
||||
actual_requests = [httpbin.ExpectedRequest.from_request(req)
|
||||
for req in actual_requests]
|
||||
assert (collections.Counter(actual_requests) ==
|
||||
collections.Counter(expected_requests))
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.re(r'the (?P<category>error|message|warning) '
|
||||
r'"(?P<message>.*)" should be shown.'))
|
||||
def expect_error(quteproc, httpbin, category, message):
|
||||
def expect_message(quteproc, httpbin, category, message):
|
||||
"""Expect the given message in the qutebrowser log."""
|
||||
category_to_loglevel = {
|
||||
'message': logging.INFO,
|
||||
'error': logging.ERROR,
|
||||
@ -148,8 +205,45 @@ def expect_error(quteproc, httpbin, category, message):
|
||||
message=message)
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.re(r'(?P<is_regex>regex )?"(?P<pattern>[^"]+)" should '
|
||||
r'be logged'))
|
||||
def should_be_logged(quteproc, is_regex, pattern):
|
||||
"""Expect the given pattern on regex in the log."""
|
||||
if is_regex:
|
||||
pattern = re.compile(pattern)
|
||||
quteproc.wait_for(message=pattern)
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse('"{pattern}" should not be logged'))
|
||||
def ensure_not_logged(quteproc, pattern):
|
||||
"""Make sure the given pattern was *not* logged."""
|
||||
quteproc.ensure_not_logged(message=pattern)
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse('the javascript message "{message}" should be '
|
||||
'logged'))
|
||||
def javascript_message_logged(quteproc, message):
|
||||
"""Make sure the given message was logged via javascript."""
|
||||
quteproc.wait_for(category='js', function='javaScriptConsoleMessage',
|
||||
message='[*] {}'.format(message))
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse('the javascript message "{message}" should not be '
|
||||
'logged'))
|
||||
def javascript_message_not_logged(quteproc, message):
|
||||
"""Make sure the given message was *not* logged via javascript."""
|
||||
quteproc.ensure_not_logged(category='js',
|
||||
function='javaScriptConsoleMessage',
|
||||
message='[*] {}'.format(message))
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse("The session should look like:\n{expected}"))
|
||||
def compare_session(quteproc, expected):
|
||||
"""Compare the current sessions against the given template.
|
||||
|
||||
partial_compare is used, which means only the keys/values listed will be
|
||||
compared.
|
||||
"""
|
||||
# Translate ... to ellipsis in YAML.
|
||||
loader = yaml.SafeLoader(expected)
|
||||
loader.add_constructor('!ellipsis', lambda loader, node: ...)
|
||||
@ -160,44 +254,6 @@ def compare_session(quteproc, expected):
|
||||
assert utils.partial_compare(data, expected)
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.parse("I wait {delay}s"))
|
||||
def wait_time(quteproc, delay):
|
||||
time.sleep(float(delay))
|
||||
|
||||
|
||||
@bdd.when(bdd.parsers.re('I press the keys? "(?P<keys>[^"]*)"'))
|
||||
def press_keys(quteproc, keys):
|
||||
quteproc.press_keys(keys)
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse('"{pattern}" should not be logged'))
|
||||
def ensure_not_logged(quteproc, pattern):
|
||||
quteproc.ensure_not_logged(message=pattern)
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse('the javascript message "{message}" should be '
|
||||
'logged'))
|
||||
def javascript_message_logged(quteproc, message):
|
||||
quteproc.wait_for(category='js', function='javaScriptConsoleMessage',
|
||||
message='[*] {}'.format(message))
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.parse('the javascript message "{message}" should not be '
|
||||
'logged'))
|
||||
def javascript_message_not_logged(quteproc, message):
|
||||
quteproc.ensure_not_logged(category='js',
|
||||
function='javaScriptConsoleMessage',
|
||||
message='[*] {}'.format(message))
|
||||
|
||||
|
||||
@bdd.then(bdd.parsers.re(r'(?P<is_regex>regex )?"(?P<pattern>[^"]+)" should '
|
||||
r'be logged'))
|
||||
def should_be_logged(quteproc, is_regex, pattern):
|
||||
if is_regex:
|
||||
pattern = re.compile(pattern)
|
||||
quteproc.wait_for(message=pattern)
|
||||
|
||||
|
||||
@bdd.then("no crash should happen")
|
||||
def no_crash():
|
||||
"""Don't do anything.
|
||||
@ -233,6 +289,12 @@ def check_contents(quteproc, filename):
|
||||
|
||||
@bdd.then(bdd.parsers.parse("the following tabs should be open:\n{tabs}"))
|
||||
def check_open_tabs(quteproc, tabs):
|
||||
"""Check the list of open tabs in the session.
|
||||
|
||||
This is a lightweight alternative for "The session should look like: ...".
|
||||
|
||||
It expects a list of URLs, with an optional "(active)" suffix.
|
||||
"""
|
||||
session = quteproc.get_session()
|
||||
active_suffix = ' (active)'
|
||||
tabs = tabs.splitlines()
|
||||
|
||||
@ -119,6 +119,9 @@ class QuteProc(testprocess.Process):
|
||||
_delay: Delay to wait between commands.
|
||||
_ipc_socket: The IPC socket of the started instance.
|
||||
_httpbin: The HTTPBin webserver.
|
||||
|
||||
Signals:
|
||||
got_error: Emitted when there was an error log line.
|
||||
"""
|
||||
|
||||
got_error = pyqtSignal()
|
||||
@ -178,6 +181,11 @@ class QuteProc(testprocess.Process):
|
||||
return executable, args
|
||||
|
||||
def path_to_url(self, path):
|
||||
"""Get a URL based on a filename for the localhost webserver.
|
||||
|
||||
URLs like about:... and qute:... are handled specially and returned
|
||||
verbatim.
|
||||
"""
|
||||
if path.startswith('about:') or path.startswith('qute:'):
|
||||
return path
|
||||
else:
|
||||
@ -195,6 +203,7 @@ class QuteProc(testprocess.Process):
|
||||
pytest.fail(text, pytrace=False)
|
||||
|
||||
def send_cmd(self, command, count=None):
|
||||
"""Send a command to the running qutebrowser instance."""
|
||||
assert self._ipc_socket is not None
|
||||
|
||||
time.sleep(self._delay / 1000)
|
||||
@ -227,6 +236,7 @@ class QuteProc(testprocess.Process):
|
||||
self.set_setting(sect, opt, old_value)
|
||||
|
||||
def open_path(self, path, new_tab=False):
|
||||
"""Open the given path on the local webserver in qutebrowser."""
|
||||
url = self.path_to_url(path)
|
||||
if new_tab:
|
||||
self.send_cmd(':open -t ' + url)
|
||||
|
||||
@ -77,6 +77,11 @@ class Process(QObject):
|
||||
|
||||
Reads the log from its stdout and parses it.
|
||||
|
||||
Attributes:
|
||||
_invalid: A list of lines which could not be parsed.
|
||||
_data: A list of parsed lines.
|
||||
proc: The QProcess for the underlying process.
|
||||
|
||||
Signals:
|
||||
ready: Emitted when the server finished starting up.
|
||||
new_data: Emitted when a new line was parsed.
|
||||
|
||||
@ -40,6 +40,10 @@ _redirect_later_event = None
|
||||
|
||||
@app.route('/data/<path:path>')
|
||||
def send_data(path):
|
||||
"""Send a given data file to qutebrowser.
|
||||
|
||||
If a directory is requested, its index.html is sent.
|
||||
"""
|
||||
if hasattr(sys, 'frozen'):
|
||||
basedir = os.path.realpath(os.path.dirname(sys.executable))
|
||||
data_dir = os.path.join(basedir, 'integration', 'data')
|
||||
@ -54,9 +58,9 @@ def send_data(path):
|
||||
|
||||
@app.route('/custom/redirect-later')
|
||||
def redirect_later():
|
||||
"""302 redirects to / after the given delay.
|
||||
"""302 redirect to / after the given delay.
|
||||
|
||||
If delay is -1, waits until a request on redirect-later-continue is done.
|
||||
If delay is -1, wait until a request on redirect-later-continue is done.
|
||||
"""
|
||||
global _redirect_later_event
|
||||
args = CaseInsensitiveDict(flask.request.args.items())
|
||||
@ -81,6 +85,7 @@ def redirect_later_continue():
|
||||
|
||||
@app.after_request
|
||||
def log_request(response):
|
||||
"""Log a webserver request."""
|
||||
request = flask.request
|
||||
template = '127.0.0.1 - - [{date}] "{verb} {path} {http}" {status} -'
|
||||
print(template.format(
|
||||
@ -95,7 +100,12 @@ def log_request(response):
|
||||
|
||||
class WSGIServer(cherrypy.wsgiserver.CherryPyWSGIServer):
|
||||
|
||||
"""A custom WSGIServer that prints a line on stderr when it's ready."""
|
||||
"""A custom WSGIServer that prints a line on stderr when it's ready.
|
||||
|
||||
Attributes:
|
||||
_ready: Internal state for the 'ready' property.
|
||||
_printed_ready: Whether the initial ready message was printed.
|
||||
"""
|
||||
|
||||
# pylint: disable=no-member
|
||||
# WORKAROUND for https://bitbucket.org/logilab/pylint/issues/702
|
||||
|
||||
Loading…
Reference in New Issue
Block a user