fdo-mirrors/upower
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/upower/upower.git synced 2026-05-05 15:48:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

linux: Use OutputChecker to inspect daemon log

Browse source 
This also means the daemon log is stored even when succeeding, making it
a bit easier to inspect tests overall.
This commit is contained in:
Benjamin Berg 2022-05-10 10:09:43 +02:00
parent 11fb06d8ae
commit 3c520fb590
2 changed files with 214 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
src/linux/integration-test.py
View file

@ -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()

199
src/linux/output_checker.py Normal file
View file

@ -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