[gnome-settings-daemon/benzea/tests-output-checker] tests: Port tests to use new OutputChecker class
- From: Benjamin Berg <bberg src gnome org>
- To: commits-list gnome org
- Cc:
- Subject: [gnome-settings-daemon/benzea/tests-output-checker] tests: Port tests to use new OutputChecker class
- Date: Mon, 15 Feb 2021 11:07:50 +0000 (UTC)
commit 72ab413f928c8307f90f647ae794953618a47844
Author: Benjamin Berg <bberg redhat com>
Date: Wed Dec 2 17:59:09 2020 +0100
tests: Port tests to use new OutputChecker class
This improves readabilty and means that all output is logged to stdout
(almost) immediately instead of splitting it into files which
potentially causes extra confusion.
plugins/power/test.py | 189 +++++++++++++-------------------------------------
tests/gsdtestcase.py | 20 +++---
2 files changed, 57 insertions(+), 152 deletions(-)
---
diff --git a/plugins/power/test.py b/plugins/power/test.py
index 7bb190a1..131bfd6e 100755
--- a/plugins/power/test.py
+++ b/plugins/power/test.py
@@ -22,6 +22,7 @@ sys.path.insert(0, builddir)
import gsdtestcase
import gsdpowerconstants
import gsdpowerenums
+from output_checker import OutputChecker
import dbus
from dbus.mainloop.glib import DBusGMainLoop
@@ -75,24 +76,17 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
'gnome_screensaver', stdout=subprocess.PIPE)
gsdtestcase.set_nonblock(self.screensaver.stdout)
- self.session_log_write = open(os.path.join(self.workdir, 'gnome-session.log'), 'wb', buffering=0)
+ self.session_log = OutputChecker()
self.session = subprocess.Popen(['gnome-session', '-f',
'-a', os.path.join(self.workdir, 'autostart'),
'--session=dummy', '--debug'],
- stdout=self.session_log_write,
+ stdout=self.session_log.fd,
stderr=subprocess.STDOUT)
+ self.session_log.writer_attached()
# wait until the daemon is on the bus
- try:
- self.wait_for_bus_object('org.gnome.SessionManager',
- '/org/gnome/SessionManager')
- except:
- # on failure, print log
- with open(self.session_log_write.name) as f:
- print('----- session log -----\n%s\n------' % f.read())
- raise
-
- self.session_log = open(self.session_log_write.name, 'rb', buffering=0)
+ self.wait_for_bus_object('org.gnome.SessionManager',
+ '/org/gnome/SessionManager')
self.obj_session_mgr = self.session_bus_con.get_object(
'org.gnome.SessionManager', '/org/gnome/SessionManager')
@@ -115,7 +109,6 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
self.settings_gsd_power = Gio.Settings(schema_id='org.gnome.settings-daemon.plugins.power')
Gio.Settings.sync()
- self.plugin_log_write = open(os.path.join(self.workdir, 'plugin_power.log'), 'wb', buffering=0)
# avoid painfully long delays of actions for tests
env = os.environ.copy()
# Disable PulseAudio output from libcanberra
@@ -132,22 +125,16 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
# We need to redirect stdout to grab the debug messages.
# stderr is not needed by the testing infrastructure but is useful to
# see warnings and errors.
+ self.plugin_log = OutputChecker()
self.daemon = subprocess.Popen(
[os.path.join(builddir, 'gsd-power'), '--verbose'],
- stdout=self.plugin_log_write,
+ stdout=self.plugin_log.fd,
+ stderr=subprocess.STDOUT,
env=env)
+ self.plugin_log.writer_attached()
- # you can use this for reading the current daemon log in tests
- self.plugin_log = open(self.plugin_log_write.name, 'rb', buffering=0)
-
- # wait until plugin is ready
- timeout = 100
- while timeout > 0:
- time.sleep(0.1)
- timeout -= 1
- log = self.plugin_log.read()
- if b'System inhibitor fd is' in log:
- break
+ # Store the early-init messages, some tests need them.
+ self.plugin_startup_msgs = self.plugin_log.check_line(b'System inhibitor fd is', timeout=10)
# always start with zero idle time
self.reset_idle_timer()
@@ -161,9 +148,7 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
if daemon_running:
self.daemon.terminate()
self.daemon.wait()
- self.plugin_log.close()
- self.plugin_log_write.flush()
- self.plugin_log_write.close()
+ self.plugin_log.assert_closed()
self.upowerd.terminate()
self.upowerd.wait()
@@ -196,10 +181,10 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
assert self.session
self.session.terminate()
self.session.wait()
-
- self.session_log_write.flush()
- self.session_log_write.close()
- self.session_log.close()
+ # At least the FD may still be opened.
+ # Probably by dummyapp.desktop (i.e. sleep 3600) which seems to not be
+ # killed by gnome-session.
+ #self.session_log.assert_closed()
def check_logind_gnome_session(self):
'''Check that gnome-session is built with logind support'''
@@ -295,29 +280,13 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
Fail after the given timeout.
'''
- # check that it request logout
- while timeout > 0:
- time.sleep(1)
- timeout -= 1
- # check that it requested logout
- log = self.session_log.read()
- if log is None:
- continue
-
- if log and (b'GsmManager: requesting logout' in log):
- break
- else:
- self.fail('timed out waiting for gnome-session logout call')
+ self.session_log.check_line(b'GsmManager: requesting logout', timeout)
def check_no_logout(self, seconds):
'''Check that no logout is requested in the given time'''
# wait for specified time to ensure it didn't do anything
- time.sleep(seconds)
- # check that it did not logout
- log = self.session_log.read()
- if log:
- self.assertFalse(b'GsmManager: requesting logout' in log, 'unexpected logout request')
+ self.session_log.check_no_line(b'GsmManager: requesting logout', seconds)
def check_for_suspend(self, timeout, methods=COMMON_SUSPEND_METHODS):
'''Check that one of the given suspend methods are requested. Default
@@ -327,30 +296,10 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
Fail after the given timeout.
'''
- # Create a list of byte string needles to search for
- needles = [' {} '.format(m).encode('ascii') for m in methods]
+ needle = r'|'.join(' {} '.format(m) for m in methods)
- suspended = False
-
- # check that it request suspend
- while timeout > 0:
- time.sleep(1)
- timeout -= 1
- # check that it requested suspend
- log = self.logind.stdout.read()
- if log is None:
- continue
-
- for n in needles:
- if n in log:
- suspended = True
- break
-
- if suspended:
- break
-
- if not suspended:
- self.fail('timed out waiting for logind suspend call, methods: %s' % ', '.join(methods))
+ self.logind_log.check_line_re(needle, timeout,
+ failmsg='timed out waiting for logind suspend call, methods: %s' % ',
'.join(methods))
def check_for_lid_inhibited(self, timeout=0):
'''Check that the lid inhibitor has been added.
@@ -373,38 +322,25 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
Fail after the given timeout.
'''
- time.sleep(timeout)
- # check that it requested uninhibition
- log = self.plugin_log.read()
-
- if b'uninhibiting lid close' in log:
- self.fail('lid uninhibit should not have happened')
+ self.plugin_log.check_no_line(b'uninhibiting lid close', wait=timeout)
def check_no_suspend(self, seconds, methods=COMMON_SUSPEND_METHODS):
'''Check that no Suspend or Hibernate is requested in the given time'''
- # wait for specified time to ensure it didn't do anything
- time.sleep(seconds)
- # check that it did not suspend or hibernate
- log = self.logind.stdout.read()
- if log is None:
- return
-
- for m in methods:
- needle = ' {} '.format(m).encode('ascii')
+ needle = r'|'.join(' {} '.format(m) for m in methods)
- self.assertFalse(needle in log, 'unexpected %s request' % m)
+ self.logind_log.check_no_line(needle, wait=seconds)
def check_suspend_no_hibernate(self, seconds):
'''Check that Suspend was requested and not Hibernate, in the given time'''
- # wait for specified time to ensure it didn't do anything
- time.sleep(seconds)
- # check that it did suspend and didn't hibernate
- log = self.logind.stdout.read()
- if log:
- self.assertTrue(b' Suspend' in log, 'missing Suspend request')
- self.assertFalse(b' Hibernate' in log, 'unexpected Hibernate request')
+ lines = self.logind_log.check_no_line(b' Hibernate', wait=seconds)
+ # Check that we did suspend
+ for l in lines:
+ if b' Suspend' in l:
+ break
+ else:
+ self.fail('Missing Suspend request')
def check_plugin_log(self, needle, timeout=0, failmsg=None):
'''Check that needle is found in the log within the given timeout.
@@ -412,36 +348,13 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
Fail after the given timeout.
'''
- if type(needle) == str:
- needle = needle.encode('ascii')
- # Fast path if the message was already logged
- log = self.plugin_log.read()
- if needle in log:
- return
-
- while timeout > 0:
- time.sleep(0.5)
- timeout -= 0.5
-
- # read new data (lines) from the log
- log = self.plugin_log.read()
- if needle in log:
- break
- else:
- if failmsg is not None:
- self.fail(failmsg)
- else:
- self.fail('timed out waiting for needle "%s"' % needle)
+ self.plugin_log.check_line(needle, timeout=timeout, failmsg=failmsg)
def check_no_dim(self, seconds):
'''Check that mode is not set to dim in the given time'''
# wait for specified time to ensure it didn't do anything
- time.sleep(seconds)
- # check that we don't dim
- log = self.plugin_log.read()
- if log:
- self.assertFalse(b'Doing a state transition: dim' in log, 'unexpected dim request')
+ self.plugin_log.check_no_line(b'Doing a state transition: dim', wait=seconds)
def check_dim(self, timeout):
'''Check that mode is set to dim in the given time'''
@@ -476,20 +389,12 @@ class PowerPluginBase(gsdtestcase.GSDTestCase):
def check_no_blank(self, seconds):
'''Check that no blank is requested in the given time'''
- # wait for specified time to ensure it didn't blank
- time.sleep(seconds)
- # check that it did not blank
- log = self.plugin_log.read()
- self.assertFalse(b'TESTSUITE: Blanked screen' in log, 'unexpected blank request')
+ self.plugin_log.check_no_line(b'TESTSUITE: Blanked screen', wait=seconds)
def check_no_unblank(self, seconds):
'''Check that no unblank is requested in the given time'''
- # wait for specified time to ensure it didn't unblank
- time.sleep(seconds)
- # check that it did not unblank
- log = self.plugin_log.read()
- self.assertFalse(b'TESTSUITE: Unblanked screen' in log, 'unexpected unblank request')
+ self.plugin_log.check_no_line(b'TESTSUITE: Unblanked screen', wait=seconds)
class PowerPluginTest1(PowerPluginBase):
def test_screensaver(self):
@@ -784,7 +689,7 @@ class PowerPluginTest5(PowerPluginBase):
# Wait and flush log
time.sleep (gsdpowerconstants.LID_CLOSE_SAFETY_TIMEOUT + 1)
- self.plugin_log.read()
+ self.plugin_log.clear()
idle_delay = math.ceil(gsdpowerconstants.MINIMUM_IDLE_DIM_DELAY /
gsdpowerconstants.IDLE_DELAY_TO_IDLE_DIM_MULTIPLIER)
self.reset_idle_timer()
@@ -832,7 +737,7 @@ class PowerPluginTest5(PowerPluginBase):
# Wait and flush log
time.sleep (gsdpowerconstants.LID_CLOSE_SAFETY_TIMEOUT + 1)
- self.plugin_log.read()
+ self.plugin_log.clear()
# Add an external monitor
self.set_has_external_monitor(True)
@@ -1315,19 +1220,21 @@ class PowerPluginTest8(PowerPluginBase):
if self.skip_sysfs_backlight:
self.skipTest("sysfs backlight support required for test")
- # We cannot use check_plugin_log here because the startup check already
- # read the relevant message.
- log = open(self.plugin_log_write.name, 'rb').read()
- self.assertIn(b'Step size for backlight is 5.', log)
+ for l in self.plugin_startup_msgs:
+ if b'Step size for backlight is 5.' in l:
+ break
+ else:
+ self.fail('Step size is not 5')
def test_legacy_brightness_step(self):
if self.skip_sysfs_backlight:
self.skipTest("sysfs backlight support required for test")
- # We cannot use check_plugin_log here because the startup check already
- # read the relevant message.
- log = open(self.plugin_log_write.name, 'rb').read()
- self.assertIn(b'Step size for backlight is 1.', log)
+ for l in self.plugin_startup_msgs:
+ if b'Step size for backlight is 1.' in l:
+ break
+ else:
+ self.fail('Step size is not 1')
def test_legacy_brightness_rounding(self):
if self.skip_sysfs_backlight:
diff --git a/tests/gsdtestcase.py b/tests/gsdtestcase.py
index 28068f6f..6c7168f9 100644
--- a/tests/gsdtestcase.py
+++ b/tests/gsdtestcase.py
@@ -14,6 +14,8 @@ import shutil
import sys
from glob import glob
+from output_checker import OutputChecker
+
from gi.repository import GLib
try:
@@ -166,9 +168,11 @@ class GSDTestCase(X11SessionTestCase):
if parameters is None:
parameters = {}
+ self.logind_log = OutputChecker()
self.logind, self.logind_obj = self.spawn_server_template('logind',
parameters,
- stdout=subprocess.PIPE)
+ stdout=self.logind_log.fd)
+ self.logind_log.writer_attached()
# Monkey patch SuspendThenHibernate functions in for dbusmock <= 0.17.2
# This should be removed once we can depend on dbusmock 0.17.3
@@ -177,24 +181,21 @@ class GSDTestCase(X11SessionTestCase):
self.logind_obj.AddMethod('org.freedesktop.login1.Session', 'SetBrightness', 'ssu', '', '')
- # set log to nonblocking
- set_nonblock(self.logind.stdout)
-
def stop_logind(self):
'''stop mock logind'''
self.logind.terminate()
self.logind.wait()
+ self.logind_log.assert_closed()
def start_mutter(klass):
''' start mutter '''
os.environ['MUTTER_DEBUG_RESET_IDLETIME']='1'
- klass.mutter_log = open(os.path.join(klass.workdir, 'mutter.log'), 'wb', buffering=0)
# See https://gitlab.gnome.org/GNOME/mutter/merge_requests/15
- klass.mutter = subprocess.Popen(['mutter', '--x11'],
- stdout=klass.mutter_log,
- stderr=subprocess.STDOUT)
+ klass.mutter = subprocess.Popen(['mutter', '--x11'])
+ klass.wait_for_bus_object('org.gnome.Mutter.IdleMonitor',
+ '/org/gnome/Mutter/IdleMonitor/Core')
def stop_mutter(klass):
'''stop mutter'''
@@ -203,9 +204,6 @@ class GSDTestCase(X11SessionTestCase):
klass.mutter.terminate()
klass.mutter.wait()
- klass.mutter_log.flush()
- klass.mutter_log.close()
-
@classmethod
def reset_idle_timer(klass):
'''trigger activity to reset idle timer'''
[
Date Prev][
Date Next] [
Thread Prev][
Thread Next]
[
Thread Index]
[
Date Index]
[
Author Index]
