qutebrowser/tests/integration/testprocess.py

# 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/>.

"""Base class for a subprocess run for tests.."""

import re
import time
import fnmatch

import pytestqt.plugin  # pylint: disable=import-error
from PyQt5.QtCore import pyqtSlot, pyqtSignal, QProcess, QObject, QElapsedTimer
from PyQt5.QtTest import QSignalSpy


class InvalidLine(Exception):

    """Raised when the process prints a line which is not parsable."""

    pass


class ProcessExited(Exception):

    """Raised when the child process did exit."""

    pass


class WaitForTimeout(Exception):

    """Raised when wait_for didn't get the expected message."""


class Line:

    """Container for a line of data the process emits.

    Attributes:
        data: The raw data passed to the constructor.
        waited_for: If Process.wait_for was used on this line already.
    """

    def __init__(self, data):
        self.data = data
        self.waited_for = False

    def __repr__(self):
        return '{}({!r})'.format(self.__class__.__name__, self.data)


class Process(QObject):

    """Abstraction over a running test subprocess process.

    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):
        super().__init__(parent)
        self._invalid = []
        self._data = []
        self.proc = QProcess()
        self.proc.setReadChannel(QProcess.StandardError)

    def _parse_line(self, line):
        """Parse the given line from the log.

        Return:
            A self.ParseResult member.
        """
        raise NotImplementedError

    def _executable_args(self):
        """Get the executable and arguments to pass to it as a tuple."""
        raise NotImplementedError

    def _get_data(self):
        """Get the parsed data for this test.

        Also waits for 0.5s to make sure any new data is received.

        Subprocesses are expected to alias this to a public method with a
        better name.
        """
        self.proc.waitForReadyRead(500)
        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."""
        if not hasattr(self, 'proc'):
            # I have no idea how this happens, but it does...
            return
        while self.proc.canReadLine():
            line = self.proc.readLine()
            line = bytes(line).decode('utf-8', errors='ignore').rstrip('\r\n')

            try:
                parsed = self._parse_line(line)
            except InvalidLine:
                self._invalid.append(line)
                print("INVALID: {}".format(line))
                continue

            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."""
        with self._wait_signal(self.ready, timeout=60000):
            self._start()

    def _start(self):
        """Actually start the process."""
        executable, args = self._executable_args()
        self.proc.readyRead.connect(self.read_log)
        self.proc.start(executable, args)
        ok = self.proc.waitForStarted()
        assert ok
        assert self.is_running()

    def before_test(self):
        """Restart process before a test if it exited before."""
        self._invalid = []
        if not self.is_running():
            self.start()

    def after_test(self):
        """Clean up data after each test.

        Also checks self._invalid so the test counts as failed if there were
        unexpected output lines earlier.
        """
        if self._invalid:
            # Wait for a bit so the full error has a chance to arrive
            time.sleep(1)
            # Exit the process to make sure we're in a defined state again
            self.terminate()
            self._data.clear()
            raise InvalidLine(self._invalid)

        self._data.clear()
        if not self.is_running():
            raise ProcessExited

    def terminate(self):
        """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

    def _match_data(self, value, expected):
        """Helper for wait_for to match a given value.

        The behavior of this method is slightly different depending on the
        types of the filtered values:

        - If expected is None, the filter always matches.
        - If the value is a string or bytes object and the expected value is
          too, the pattern is treated as a fnmatch glob pattern.
        - If the value is a string or bytes object and the expected value is a
          compiled regex, it is used for matching.
        - If the value is any other type, == is used.

        Return:
            A bool
        """
        regex_type = type(re.compile(''))
        if expected is None:
            return True
        elif isinstance(expected, regex_type):
            return expected.match(value)
        elif isinstance(value, (bytes, str)):
            return fnmatch.fnmatchcase(value, expected)
        else:
            return value == expected

    def wait_for(self, timeout=15000, **kwargs):
        """Wait until a given value is found in the data.

        Keyword arguments to this function get interpreted as attributes of the
        searched data. Every given argument is treated as a pattern which
        the attribute has to match against.

        Return:
            The matched line.
        """
        # Search existing messages
        for line in self._data:
            matches = []

            for key, expected in kwargs.items():
                value = getattr(line, key)
                matches.append(self._match_data(value, expected))

            if all(matches) and not line.waited_for:
                # If we waited for this line, chances are we don't mean the
                # same thing the next time we use wait_for and it matches
                # this line again.
                line.waited_for = True
                return line

        # If there is none, wait for the message
        spy = QSignalSpy(self.new_data)
        elapsed_timer = QElapsedTimer()
        elapsed_timer.start()

        while True:
            got_signal = spy.wait(timeout)
            if not got_signal or elapsed_timer.hasExpired(timeout):
                raise WaitForTimeout("Timed out after {}ms waiting for "
                                     "{!r}.".format(timeout, kwargs))

            for args in spy:
                assert len(args) == 1
                line = args[0]

                matches = []

                for key, expected in kwargs.items():
                    value = getattr(line, key)
                    matches.append(self._match_data(value, expected))

                if all(matches):
                    # If we waited for this line, chances are we don't mean the
                    # same thing the next time we use wait_for and it matches
                    # this line again.
                    line.waited_for = True
                    return line