rnhmjoj/qutebrowser
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests: First steps towards end-to-end tests.

Browse Source
This commit is contained in:
Florian Bruhin 2015-10-10 19:21:12 +02:00
parent e6ee0c08cf
commit 76fcec4e4c
6 changed files with 166 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tests/integration/conftest.py
View File

@ -17,6 +17,9 @@
# You should have received a copy of the GNU General Public License
# along with qutebrowser. If not, see <http://www.gnu.org/licenses/>.
# pylint: disable=unused-import
"""Things needed for integration testing."""
from webserver import httpbin, httpbin_clean # pylint: disable=unused-import
from webserver import httpbin, httpbin_after_test
from quteprocess import quteproc, quteproc_after_test

137
tests/integration/quteprocess.py Normal file
View File

@ -0,0 +1,137 @@
# vim: ft=python fileencoding=utf-8 sts=4 sw=4 et:
# Copyright 2015 Florian Bruhin (The Compiler) <mail@qutebrowser.org>
#
# 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 <http://www.gnu.org/licenses/>.
# pylint doesn't understand the testprocess import
# pylint: disable=no-member
"""Fixtures to run qutebrowser in a QProcess and communicate."""
import re
import sys
import time
import os.path
import collections
import pytest
from PyQt5.QtCore import pyqtSignal
import testprocess # pylint: disable=import-error
from qutebrowser.misc import ipc
LogLine = collections.namedtuple('LogLine', [
'timestamp', 'loglevel', 'category', 'module', 'function', 'line',
'message'])
class QuteProc(testprocess.Process):
"""A running qutebrowser process used for tests.
Attributes:
_ipc_socket: The IPC socket of the started instance.
"""
LOG_RE = re.compile(r"""
(?P<timestamp>\d\d:\d\d:\d\d)
\ (?P<loglevel>VDEBUG|DEBUG|INFO|WARNING|ERROR)
\ +(?P<category>\w+)
\ +(?P<module>\w+):(?P<function>\w+):(?P<line>\d+)
\ (?P<message>.+)
""", re.VERBOSE)
executing_command = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self._ipc_socket = None
def _parse_line(self, line):
match = self.LOG_RE.match(line)
if match is None:
if line.startswith(' '):
# Multiple lines in some log output...
return None
elif not line.strip():
return None
else:
raise testprocess.InvalidLine
log_line = LogLine(**match.groupdict())
start_okay_message = ("load status for "
"<qutebrowser.browser.webview.WebView tab_id=0 "
"url='about:blank'>: LoadStatus.success")
if (log_line.category == 'ipc' and
log_line.message.startswith("Listening as ")):
self._ipc_socket = log_line.message.split(' ', maxsplit=2)[2]
elif (log_line.category == 'webview' and
log_line.message == start_okay_message):
self.ready.emit()
elif (log_line.category == 'commands' and
log_line.module =='command' and log_line.function == 'run' and
log_line.message.startswith('Calling ')):
self.executing_command.emit()
return log_line
def _executable_args(self):
if hasattr(sys, 'frozen'):
executable = os.path.join(os.path.dirname(sys.executable),
'qutebrowser')
args = []
else:
executable = sys.executable
args = ['-m', 'qutebrowser']
args += ['--debug', '--no-err-windows', '--temp-basedir',
'about:blank']
return executable, args
def after_test(self):
bad_msgs = [msg for msg in self._data
if msg.loglevel not in ['VDEBUG', 'DEBUG', 'INFO']]
super().after_test()
if bad_msgs:
text = 'Logged unexpected errors:\n\n' + '\n'.join(
str(e) for e in bad_msgs)
pytest.fail(text, pytrace=False)
def send_cmd(self, command):
assert self._ipc_socket is not None
with self._wait_signal(self.executing_command):
ipc.send_to_running_instance(self._ipc_socket, [':' + command],
target_arg='')
# Wait a bit in cause the command triggers any error.
time.sleep(0.5)
@pytest.yield_fixture(scope='session', autouse=True)
def quteproc(qapp):
"""Fixture for qutebrowser process."""
proc = QuteProc()
proc.start()
yield proc
proc.cleanup()
@pytest.yield_fixture(autouse=True)
def quteproc_after_test(quteproc):
"""Fixture to check the status of and restart the qutebrowser process."""
yield
quteproc.after_test()

4
tests/integration/test_smoke.py
View File

@ -33,3 +33,7 @@ def test_smoke():
argv += ['--debug', '--no-err-windows', '--nowindow', '--temp-basedir',
'about:blank', ':later 500 quit']
subprocess.check_call(argv)
def test_smoke_quteproc(quteproc):
pass

24
tests/integration/testprocess.py
View File

@ -44,9 +44,11 @@ class Process(QObject):
Reads the log from its stdout and parses it.
Signals:
ready: Emitted when the server finished starting up.
new_data: Emitted when a new line was parsed.
"""
ready = pyqtSignal()
new_data = pyqtSignal(object)
def __init__(self, parent=None):
@ -80,6 +82,16 @@ class Process(QObject):
self.read_log()
return self._data
def _wait_signal(self, signal, timeout=5000, raising=True):
"""Wait for a signal to be emitted.
Should be used in a contextmanager.
"""
blocker = pytestqt.plugin.SignalBlocker(
timeout=timeout, raising=raising)
blocker.connect(signal)
return blocker
@pyqtSlot()
def read_log(self):
"""Read the log from the process' stdout."""
@ -95,16 +107,13 @@ class Process(QObject):
print("INVALID: {}".format(line))
continue
print('parsed: {}'.format(parsed))
if parsed is not None:
self._data.append(parsed)
self.new_data.emit(parsed)
def start(self):
"""Start the process and wait until it started."""
blocker = pytestqt.plugin.SignalBlocker(timeout=5000, raising=True)
blocker.connect(self.ready)
with blocker:
with self._wait_signal(self.ready):
self._start()
def _start(self):
@ -113,6 +122,7 @@ class Process(QObject):
self.proc.start(executable, args)
ok = self.proc.waitForStarted()
assert ok
assert self.is_running()
self.proc.readyRead.connect(self.read_log)
def after_test(self):
@ -122,7 +132,7 @@ class Process(QObject):
unexpected output lines earlier.
"""
self._data.clear()
if self.proc.state() != QProcess.Running:
if not self.is_running():
print("Restarting process...")
self.start()
raise ProcessExited
@ -133,3 +143,7 @@ class Process(QObject):
"""Clean up and shut down the process."""
self.proc.terminate()
self.proc.waitForFinished()
def is_running(self):
"""Check if the process is currently running."""
return self.proc.state() == QProcess.Running

4
tests/integration/webserver.py
View File

@ -47,11 +47,9 @@ class HTTPBin(testprocess.Process):
LOG_RE: Used to parse the CLF log which httpbin outputs.
Signals:
ready: Emitted when the server finished starting up.
new_request: Emitted when there's a new request received.
"""
ready = pyqtSignal()
new_request = pyqtSignal(Request)
LOG_RE = re.compile(r"""
@ -127,7 +125,7 @@ def httpbin(qapp):
@pytest.yield_fixture(autouse=True)
def httpbin_clean(httpbin):
def httpbin_after_test(httpbin):
"""Fixture to clean httpbin request list after each test."""
yield
httpbin.after_test()

2
tox.ini
View File

@ -11,7 +11,7 @@ envlist = py34,py35,misc,vulture,pep257,pyflakes,pep8,mccabe,pylint,pyroma,check
setenv =
QT_QPA_PLATFORM_PLUGIN_PATH={envdir}/Lib/site-packages/PyQt5/plugins/platforms
PYTEST_QT_API=pyqt5
passenv = PYTHON DISPLAY XAUTHORITY HOME USERNAME USER CI
passenv = PYTHON DISPLAY XAUTHORITY HOME USERNAME USER CI XDG_*
deps =
-r{toxinidir}/requirements.txt
wheel==0.26.0