tests: Add ensure_not_logged to TestProcess.

tests/integration/features/conftest.py

@@ -131,6 +131,11 @@ def compare_session(quteproc, expected):
     assert utils.partial_compare(data, expected)
 
 
+@bdd.then(bdd.parsers.parse('"{pattern}" should not be logged'))
+def ensure_not_logged(quteproc, pattern):
+    quteproc.ensure_not_logged(message=pattern)
+
+
 @bdd.then("no crash should happen")
 def no_crash():
     """Don't do anything.

tests/integration/test_testprocess.py

@@ -74,13 +74,14 @@ class PythonProcess(testprocess.Process):
         return (sys.executable, ['-c', ';'.join(code)])
 
 
-class TestWaitFor:
+@pytest.yield_fixture
+def pyproc():
+    proc = PythonProcess()
+    yield proc
+    proc.terminate()
 
-    @pytest.yield_fixture
+
-    def pyproc(self):
+class TestWaitFor:
-        proc = PythonProcess()
-        yield proc
-        proc.terminate()
 
     def test_successful(self, pyproc):
         """Using wait_for with the expected text."""
@@ -134,3 +135,36 @@ class TestWaitFor:
         pyproc.wait_for(data="foobar")
         with pytest.raises(testprocess.WaitForTimeout):
             pyproc.wait_for(data="foobar", timeout=100)
+
+
+class TestEnsureNotLogged:
+
+    @pytest.mark.parametrize('message, pattern', [
+        ('blacklisted', 'blacklisted'),
+        ('bl[a]cklisted', 'bl[a]cklisted'),
+        ('blacklisted', 'black*'),
+    ])
+    def test_existing_message(self, pyproc, message, pattern):
+        pyproc.code = "print('{}')".format(message)
+        pyproc.start()
+        with stopwatch(max_ms=1000):
+            with pytest.raises(testprocess.BlacklistedMessageError):
+                pyproc.ensure_not_logged(data=pattern, delay=2000)
+
+    def test_late_message(self, pyproc):
+        pyproc.code = "time.sleep(0.5); print('blacklisted')"
+        pyproc.start()
+        with pytest.raises(testprocess.BlacklistedMessageError):
+            pyproc.ensure_not_logged(data='blacklisted', delay=1000)
+
+    def test_no_matching_message(self, pyproc):
+        pyproc.code = "print('blacklisted... nope!')"
+        pyproc.start()
+        pyproc.ensure_not_logged(data='blacklisted', delay=100)
+
+    def test_wait_for_and_blacklist(self, pyproc):
+        pyproc.code = "print('blacklisted')"
+        pyproc.start()
+        pyproc.wait_for(data='blacklisted')
+        with pytest.raises(testprocess.BlacklistedMessageError):
+            pyproc.ensure_not_logged(data='blacklisted', delay=0)

tests/integration/testprocess.py

@@ -49,6 +49,11 @@ class WaitForTimeout(Exception):
     """Raised when wait_for didn't get the expected message."""
 
 
+class BlacklistedMessageError(Exception):
+
+    """Raised when ensure_not_logged found a message."""
+
+
 class Line:
 
     """Container for a line of data the process emits.
@@ -219,13 +224,18 @@ class Process(QObject):
         else:
             return value == expected
 
-    def wait_for(self, timeout=None, **kwargs):
+    def wait_for(self, timeout=None, *, override_waited_for=False, **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.
 
+        Args:
+            timeout: How long to wait for the message.
+            override_waited_for: If set, gets triggered by previous messages
+                                 again.
+
         Return:
             The matched line.
         """
@@ -246,7 +256,7 @@ class Process(QObject):
                 value = getattr(line, key)
                 matches.append(self._match_data(value, expected))
 
-            if all(matches) and not line.waited_for:
+            if all(matches) and (not line.waited_for or override_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.
@@ -280,3 +290,18 @@ class Process(QObject):
                     # this line again.
                     line.waited_for = True
                     return line
+
+    def ensure_not_logged(self, delay=500, **kwargs):
+        """Make sure the data matching the given arguments is not logged.
+
+        If nothing is found in the log, we wait for delay ms to make sure
+        nothing arrives.
+        """
+        __tracebackhide__ = True
+        try:
+            line = self.wait_for(timeout=delay, override_waited_for=True,
+                                 **kwargs)
+        except WaitForTimeout:
+            return
+        else:
+            raise BlacklistedMessageError(line)