linux: Use OutputChecker to inspect daemon log

This also means the daemon log is stored even when succeeding, making it
a bit easier to inspect tests overall.

2 files changed, 214 insertions, 31 deletions

diff --git a/src/linux/integration-test.py b/src/linux/integration-test.py
index 8ed2a7b..c33c184 100755
--- a/src/linux/integration-test.py
+++ b/src/linux/integration-test.py
@@ -25,6 +25,7 @@ import shutil
 import subprocess
 import unittest
 import time
+from output_checker import OutputChecker
 from packaging.version import parse as parse_version
 
 try:
@@ -146,7 +147,6 @@ class Tests(dbusmock.DBusTestCase):
         self.testbed = UMockdev.Testbed.new()
 
         self.proxy = None
-        self.log = None
         self.daemon = None
         self.start_logind({'CanHybridSleep' : 'yes'})
 
@@ -167,14 +167,6 @@ class Tests(dbusmock.DBusTestCase):
         except:
             pass
 
-        # on failures, print daemon log
-        errors = [x[1] for x in self._outcome.errors if x[1]]
-        if errors and self.log:
-            with open(self.log.name) as f:
-                sys.stderr.write('\n-------------- daemon log: ----------------\n')
-                sys.stderr.write(f.read())
-                sys.stderr.write('------------------------------\n')
-
     #
     # Daemon control and D-BUS I/O
     #
@@ -199,15 +191,16 @@ class Tests(dbusmock.DBusTestCase):
         # note: Python doesn't propagate the setenv from Testbed.new(), so we
         # have to do that ourselves
         env['UMOCKDEV_DIR'] = self.testbed.get_root_dir()
-        self.log = tempfile.NamedTemporaryFile()
+        self.daemon_log = OutputChecker()
+
         if os.getenv('VALGRIND') != None:
             daemon_path = ['valgrind', self.daemon_path, '-v']
         else:
             daemon_path = [self.daemon_path, '-v']
         self.daemon = subprocess.Popen(daemon_path,
-                                       env=env, stdout=self.log,
+                                       env=env, stdout=self.daemon_log.fd,
                                        stderr=subprocess.STDOUT)
-
+        self.daemon_log.writer_attached()
         # wait until the daemon gets online
         timeout = 100
         while timeout > 0:
@@ -236,6 +229,7 @@ class Tests(dbusmock.DBusTestCase):
             except OSError:
                 pass
             self.daemon.wait()
+        self.daemon_log.assert_closed()
         self.daemon = None
         self.proxy = None
 
@@ -273,13 +267,6 @@ class Tests(dbusmock.DBusTestCase):
                                                                   parameters or {},
                                                                   stdout=subprocess.PIPE)
 
-    def have_text_in_log(self, text):
-        return self.count_text_in_log(text) > 0
-
-    def count_text_in_log(self, text):
-        with open(self.log.name) as f:
-            return f.read().count(text)
-
     def assertEventually(self, condition, message=None, timeout=50, value=True):
         '''Assert that condition function eventually returns True.
 
@@ -955,14 +942,14 @@ class Tests(dbusmock.DBusTestCase):
         self.start_daemon()
 
         self.logind_obj.EmitSignal('', 'PrepareForSleep', 'b', [True])
-        self.assertEventually(lambda: self.have_text_in_log("Polling will be paused"), timeout=10)
+        self.daemon_log.check_line("Polling will be paused", timeout=1)
 
         # simulate some battery drain during sleep for which we then
         # can check after we 'woke up'
         self.testbed.set_attribute(bat0, 'energy_now', '40000000')
 
         self.logind_obj.EmitSignal('', 'PrepareForSleep', 'b', [False])
-        self.assertEventually(lambda: self.have_text_in_log("Polling will be resumed"), timeout=10)
+        self.daemon_log.check_line("Polling will be resumed", timeout=1)
 
         devs = self.proxy.EnumerateDevices()
         self.assertEqual(len(devs), 1)
@@ -1010,8 +997,7 @@ class Tests(dbusmock.DBusTestCase):
         self.assertEventually(lambda: self.get_dbus_display_property('WarningLevel'), value=UP_DEVICE_LEVEL_ACTION)
         self.assertEqual(len(self.logind_obj.ListInhibitors()), 2)
 
-        time.sleep(UP_DAEMON_ACTION_DELAY + 0.5) # wait for UP_DAEMON_ACTION_DELAY
-        self.assertEqual(self.count_text_in_log("About to call logind method Hibernate"), 1)
+        self.daemon_log.check_line("About to call logind method Hibernate", timeout=UP_DAEMON_ACTION_DELAY + 0.5)
 
         # block inhibitor lock is released
         self.assertEqual(len(self.logind_obj.ListInhibitors()), 1)
@@ -1048,8 +1034,7 @@ class Tests(dbusmock.DBusTestCase):
         time.sleep(0.5)
         self.assertEqual(self.get_dbus_display_property('WarningLevel'), UP_DEVICE_LEVEL_ACTION)
 
-        time.sleep(UP_DAEMON_ACTION_DELAY + 0.5) # wait for UP_DAEMON_ACTION_DELAY
-        self.assertEqual(self.count_text_in_log("About to call logind method Hibernate"), 1)
+        self.daemon_log.check_line("About to call logind method Hibernate", timeout=UP_DAEMON_ACTION_DELAY + 0.5)
 
         # simulate that battery was charged to 100% during sleep
         self.testbed.set_attribute(bat0, 'energy_now', '60000000')
@@ -1065,8 +1050,7 @@ class Tests(dbusmock.DBusTestCase):
         time.sleep(0.5)
         self.assertEqual(self.get_dbus_display_property('WarningLevel'), UP_DEVICE_LEVEL_ACTION)
 
-        time.sleep(UP_DAEMON_ACTION_DELAY + 0.5) # wait for UP_DAEMON_ACTION_DELAY
-        self.assertEqual(self.count_text_in_log("About to call logind method Hibernate"), 2)
+        self.daemon_log.check_line("About to call logind method Hibernate", timeout=UP_DAEMON_ACTION_DELAY + 0.5)
 
         self.stop_daemon()
 
@@ -1090,7 +1074,7 @@ class Tests(dbusmock.DBusTestCase):
         self.assertEqual(len(devs), 1)
         bat0_up = devs[0]
 
-        self.assertEventually(lambda: self.have_text_in_log(f"saving in {UP_HISTORY_SAVE_INTERVAL} seconds"), timeout=10)
+        self.daemon_log.check_line(f"saving in {UP_HISTORY_SAVE_INTERVAL} seconds", timeout=1)
 
         # simulate that battery has 1% (less than 10%)
         self.testbed.set_attribute(bat0, 'energy_now', '600000')
@@ -1099,8 +1083,8 @@ class Tests(dbusmock.DBusTestCase):
         time.sleep(0.5)
         self.assertEqual(self.get_dbus_display_property('Percentage'), 1)
 
-        self.assertEqual(self.count_text_in_log("saving to disk earlier due to low power"), 1)
-        self.assertEqual(self.count_text_in_log(f"saving in {UP_HISTORY_SAVE_INTERVAL_LOW_POWER} seconds"), 1)
+        self.daemon_log.check_line("saving to disk earlier due to low power")
+        self.daemon_log.check_line(f"saving in {UP_HISTORY_SAVE_INTERVAL_LOW_POWER} seconds")
 
         # simulate that battery was charged to 100% during sleep
         self.testbed.set_attribute(bat0, 'energy_now', '60000000')
@@ -1110,7 +1094,7 @@ class Tests(dbusmock.DBusTestCase):
         self.assertEqual(self.get_dbus_display_property('Percentage'), 100)
 
         # The 5 seconds were not up yet, and the shorter timeout sticks
-        self.assertEqual(self.count_text_in_log("deferring as earlier timeout is already queued"), 1)
+        self.daemon_log.check_line("deferring as earlier timeout is already queued")
 
         self.stop_daemon()
 
diff --git a/src/linux/output_checker.py b/src/linux/output_checker.py
new file mode 100644
index 0000000..d1a8c4c
--- /dev/null
+++ b/src/linux/output_checker.py
@@ -0,0 +1,199 @@
+#! /usr/bin/env python3
+# Copyright © 2020, RedHat Inc.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This program 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+# Authors:
+#       Benjamin Berg <bberg@redhat.com>
+
+import os
+import sys
+import fcntl
+import io
+import re
+import time
+import threading
+import select
+import errno
+
+class OutputChecker(object):
+
+    def __init__(self, out=sys.stdout):
+        self._output = out
+        self._pipe_fd_r, self._pipe_fd_w = os.pipe()
+        self._partial_buf = b''
+        self._lines_sem = threading.Semaphore()
+        self._lines = []
+        self._reader_io = io.StringIO()
+
+        # Just to be sure, shouldn't be a problem even if we didn't set it
+        fcntl.fcntl(self._pipe_fd_r, fcntl.F_SETFL,
+                    fcntl.fcntl(self._pipe_fd_r, fcntl.F_GETFL) | os.O_CLOEXEC | os.O_NONBLOCK)
+        fcntl.fcntl(self._pipe_fd_w, fcntl.F_SETFL,
+                    fcntl.fcntl(self._pipe_fd_w, fcntl.F_GETFL) | os.O_CLOEXEC)
+
+        # Start copier thread
+        self._thread = threading.Thread(target=self._copy, daemon=True)
+        self._thread.start()
+
+    def _copy(self):
+        p = select.poll()
+        p.register(self._pipe_fd_r)
+        while True:
+            try:
+                # Be lazy and wake up occasionally in case _pipe_fd_r became invalid
+                # The reason to do this is because os.read() will *not* return if the
+                # FD is forcefully closed.
+                p.poll(0.1)
+
+                r = os.read(self._pipe_fd_r, 1024)
+                if not r:
+                    os.close(self._pipe_fd_r)
+                    self._pipe_fd_r = -1
+                    self._lines_sem.release()
+                    return
+            except OSError as e:
+                if e.errno == errno.EWOULDBLOCK:
+                    continue
+
+                # We get a bad file descriptor error when the outside closes the FD
+                if self._pipe_fd_r >= 0:
+                    os.close(self._pipe_fd_r)
+                self._pipe_fd_r = -1
+                self._lines_sem.release()
+                return
+
+            l = r.split(b'\n')
+            l[0] = self._partial_buf + l[0]
+            self._lines.extend(l[:-1])
+            self._partial_buf = l[-1]
+
+            self._lines_sem.release()
+
+            os.write(self._output.fileno(), r)
+
+    def check_line_re(self, needle_re, timeout=0, failmsg=None):
+        deadline = time.time() + timeout
+
+        if isinstance(needle_re, str):
+            needle_re = needle_re.encode('ascii')
+
+        r = re.compile(needle_re)
+        ret = []
+
+        while True:
+            try:
+                l = self._lines.pop(0)
+            except IndexError:
+                # EOF, throw error
+                if self._pipe_fd_r == -1:
+                    if failmsg:
+                        raise AssertionError("No further messages: " % failmsg) from None
+                    else:
+                        raise AssertionError('No client waiting for needle %s' % (str(needle_re))) from None
+
+                # Check if should wake up
+                if not self._lines_sem.acquire(timeout = deadline - time.time()):
+                    if failmsg:
+                        raise AssertionError(failmsg) from None
+                    else:
+                        raise AssertionError('Timed out waiting for needle %s (timeout: %0.2f)' % (str(needle_re), timeout)) from None
+                continue
+
+            ret.append(l)
+            if r.search(l):
+                return ret
+
+    def check_line(self, needle, timeout=0, failmsg=None):
+        if isinstance(needle, str):
+            needle = needle.encode('ascii')
+
+        needle_re = re.escape(needle)
+
+        return self.check_line_re(needle_re, timeout=timeout, failmsg=failmsg)
+
+    def check_no_line_re(self, needle_re, wait=0, failmsg=None):
+        deadline = time.time() + wait
+
+        if isinstance(needle_re, str):
+            needle_re = needle_re.encode('ascii')
+
+        r = re.compile(needle_re)
+        ret = []
+
+        while True:
+            try:
+                l = self._lines.pop(0)
+            except IndexError:
+                # EOF, so everything good
+                if self._pipe_fd_r == -1:
+                    break
+
+                # Check if should wake up
+                if not self._lines_sem.acquire(timeout = deadline - time.time()):
+                    # Timed out, so everything is good
+                    break
+                continue
+
+            ret.append(l)
+            if r.search(l):
+                if failmsg:
+                    raise AssertionError(failmsg)
+                else:
+                    raise AssertionError('Found needle %s but shouldn\'t have been there (timeout: %0.2f)' % (str(needle_re), wait))
+
+        return ret
+
+    def check_no_line(self, needle, wait=0, failmsg=None):
+        if isinstance(needle, str):
+            needle = needle.encode('ascii')
+
+        needle_re = re.escape(needle)
+
+        return self.check_no_line_re(needle_re, wait=wait, failmsg=failmsg)
+
+    def clear(self):
+        ret = self._lines
+        self._lines = []
+        return ret
+
+    def assert_closed(self, timeout=1):
+        self._thread.join(timeout)
+
+        if self._thread.is_alive() != False:
+            raise AssertionError("OutputCheck: Write side has not been closed yet!")
+
+    def force_close(self):
+
+        fd = self._pipe_fd_r
+        self._pipe_fd_r = -1
+        if fd >= 0:
+            os.close(fd)
+
+        self._thread.join()
+
+    @property
+    def fd(self):
+        return self._pipe_fd_w
+
+    def writer_attached(self):
+        os.close(self._pipe_fd_w)
+        self._pipe_fd_w = -1
+
+    def __del__(self):
+        if self._pipe_fd_r >= 0:
+            os.close(self._pipe_fd_r)
+            self._pipe_fd_r = -1
+        if self._pipe_fd_w >= 0:
+            os.close(self._pipe_fd_w)
+            self._pipe_fd_w = -1