luc14n0/glib
1
0
Fork 0
mirror of https://gitlab.gnome.org/GNOME/glib.git synced 2025-08-20 23:58:54 +02:00
Code Issues Packages Projects Releases Wiki Activity

tests: Rework assertEventually() to use GLib main context

Browse Source 
Rather than sleeping and blocking everything, use a timeout source on
the main context so that inputs can continue to be handled while waiting
for a timeout.

This introduces no functional changes to the test.

Signed-off-by: Philip Withnall <pwithnall@gnome.org>

Helps: #2887
This commit is contained in:
Philip Withnall
2024-01-22 13:56:05 +00:00
parent 8a4ad5f998
commit e3b81b3f0d
4 changed files with 76 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
gio/tests/memory-monitor-dbus.py.in
View File

@@ -66,21 +66,27 @@ try:
self.p_mock.terminate() self.p_mock.terminate()
self.p_mock.wait() self.p_mock.wait()
def assertEventually(self, condition, message=None, timeout=50): def assertEventually(self, condition, message=None, timeout=5):
'''Assert that condition function eventually returns True. '''Assert that condition function eventually returns True.
Timeout is in deciseconds, defaulting to 50 (5 seconds). message is Timeout is in seconds, defaulting to 5 seconds. message is
printed on failure. printed on failure.
''' '''
while timeout >= 0: if not message:
while self.main_context.iteration(False): message = 'timed out waiting for ' + str(condition)
pass
if condition(): def timed_out_cb(message):
break self.fail(message)
timeout -= 1 return GLib.SOURCE_REMOVE
time.sleep(0.1)
else: timeout_source = GLib.timeout_source_new_seconds(timeout)
self.fail(message or 'timed out waiting for ' + str(condition)) timeout_source.set_callback(timed_out_cb, message)
timeout_source.attach(self.main_context)
while not condition():
self.main_context.iteration(True)
timeout_source.destroy()
def memory_warning_cb(self, monitor, level): def memory_warning_cb(self, monitor, level):
print("Received memory warning signal, level", level) print("Received memory warning signal, level", level)
@@ -99,11 +105,11 @@ try:
self.dbusmock.EmitWarning(100) self.dbusmock.EmitWarning(100)
# Wait 2 seconds or until warning # Wait 2 seconds or until warning
self.assertEventually(lambda: self.last_warning == 100, "'100' low-memory warning not received", 20) self.assertEventually(lambda: self.last_warning == 100, "'100' low-memory warning not received", 2)
self.dbusmock.EmitWarning(255) self.dbusmock.EmitWarning(255)
# Wait 2 seconds or until warning # Wait 2 seconds or until warning
self.assertEventually(lambda: self.last_warning == 255, "'255' low-memory warning not received", 20) self.assertEventually(lambda: self.last_warning == 255, "'255' low-memory warning not received", 2)
except ImportError as e: except ImportError as e:
@unittest.skip("Cannot import %s" % e.name) @unittest.skip("Cannot import %s" % e.name)

33
gio/tests/memory-monitor-portal.py.in
View File

@@ -84,22 +84,27 @@ try:
self.p_mock.terminate() self.p_mock.terminate()
self.p_mock.wait() self.p_mock.wait()
def assertEventually(self, condition, message=None, timeout=50): def assertEventually(self, condition, message=None, timeout=5):
'''Assert that condition function eventually returns True. '''Assert that condition function eventually returns True.
Timeout is in deciseconds, defaulting to 50 (5 seconds). message is Timeout is in seconds, defaulting to 5 seconds. message is
printed on failure. printed on failure.
''' '''
while timeout >= 0: if not message:
context = GLib.MainContext.default() message = 'timed out waiting for ' + str(condition)
while context.iteration(False):
pass def timed_out_cb(message):
if condition(): self.fail(message)
break return GLib.SOURCE_REMOVE
timeout -= 1
time.sleep(0.1) timeout_source = GLib.timeout_source_new_seconds(timeout)
else: timeout_source.set_callback(timed_out_cb, message)
self.fail(message or 'timed out waiting for ' + str(condition)) timeout_source.attach(self.main_context)
while not condition():
self.main_context.iteration(True)
timeout_source.destroy()
def portal_memory_warning_cb(self, monitor, level): def portal_memory_warning_cb(self, monitor, level):
self.last_warning = level self.last_warning = level
@@ -117,11 +122,11 @@ try:
self.dbusmock.EmitWarning(100) self.dbusmock.EmitWarning(100)
# Wait 2 seconds or until warning # Wait 2 seconds or until warning
self.assertEventually(lambda: self.last_warning == 100, "'100' low-memory warning not received", 20) self.assertEventually(lambda: self.last_warning == 100, "'100' low-memory warning not received", 2)
self.dbusmock.EmitWarning(255) self.dbusmock.EmitWarning(255)
# Wait 2 seconds or until warning # Wait 2 seconds or until warning
self.assertEventually(lambda: self.last_warning == 255, "'255' low-memory warning not received", 20) self.assertEventually(lambda: self.last_warning == 255, "'255' low-memory warning not received", 2)
except ImportError as e: except ImportError as e:
@unittest.skip("Cannot import %s" % e.name) @unittest.skip("Cannot import %s" % e.name)

33
gio/tests/power-profile-monitor-dbus.py.in
View File

@@ -16,7 +16,6 @@ import sys
import subprocess import subprocess
import fcntl import fcntl
import os import os
import time
import taptestrunner import taptestrunner
@@ -66,21 +65,27 @@ try:
self.p_mock.terminate() self.p_mock.terminate()
self.p_mock.wait() self.p_mock.wait()
def assertEventually(self, condition, message=None, timeout=50): def assertEventually(self, condition, message=None, timeout=5):
'''Assert that condition function eventually returns True. '''Assert that condition function eventually returns True.
Timeout is in deciseconds, defaulting to 50 (5 seconds). message is Timeout is in seconds, defaulting to 5 seconds. message is
printed on failure. printed on failure.
''' '''
while timeout >= 0: if not message:
while self.main_context.iteration(False): message = 'timed out waiting for ' + str(condition)
pass
if condition(): def timed_out_cb(message):
break self.fail(message)
timeout -= 1 return GLib.SOURCE_REMOVE
time.sleep(0.1)
else: timeout_source = GLib.timeout_source_new_seconds(timeout)
self.fail(message or 'timed out waiting for ' + str(condition)) timeout_source.set_callback(timed_out_cb, message)
timeout_source.attach(self.main_context)
while not condition():
self.main_context.iteration(True)
timeout_source.destroy()
def power_saver_enabled_cb(self, spec, data): def power_saver_enabled_cb(self, spec, data):
self.power_saver_enabled = self.power_profile_monitor.get_power_saver_enabled() self.power_saver_enabled = self.power_profile_monitor.get_power_saver_enabled()
@@ -91,10 +96,10 @@ try:
self.assertEqual(self.power_profile_monitor.get_power_saver_enabled(), False) self.assertEqual(self.power_profile_monitor.get_power_saver_enabled(), False)
self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('power-saver', variant_level=1)) self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('power-saver', variant_level=1))
self.assertEventually(lambda: self.power_saver_enabled == True, "power-saver didn't become enabled", 10) self.assertEventually(lambda: self.power_saver_enabled == True, "power-saver didn't become enabled", 1)
self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('balanced', variant_level=1)) self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('balanced', variant_level=1))
self.assertEventually(lambda: self.power_saver_enabled == False, "power-saver didn't become disabled", 10) self.assertEventually(lambda: self.power_saver_enabled == False, "power-saver didn't become disabled", 1)
except ImportError as e: except ImportError as e:
@unittest.skip("Cannot import %s" % e.name) @unittest.skip("Cannot import %s" % e.name)

34
gio/tests/power-profile-monitor-portal.py.in
View File

@@ -16,7 +16,6 @@ import sys
import subprocess import subprocess
import fcntl import fcntl
import os import os
import time
import taptestrunner import taptestrunner
@@ -90,22 +89,27 @@ try:
self.p_mock.terminate() self.p_mock.terminate()
self.p_mock.wait() self.p_mock.wait()
def assertEventually(self, condition, message=None, timeout=50): def assertEventually(self, condition, message=None, timeout=5):
'''Assert that condition function eventually returns True. '''Assert that condition function eventually returns True.
Timeout is in deciseconds, defaulting to 50 (5 seconds). message is Timeout is in seconds, defaulting to 5 seconds. message is
printed on failure. printed on failure.
''' '''
while timeout >= 0: if not message:
context = GLib.MainContext.default() message = 'timed out waiting for ' + str(condition)
while context.iteration(False):
pass def timed_out_cb(message):
if condition(): self.fail(message)
break return GLib.SOURCE_REMOVE
timeout -= 1
time.sleep(0.1) timeout_source = GLib.timeout_source_new_seconds(timeout)
else: timeout_source.set_callback(timed_out_cb, message)
self.fail(message or 'timed out waiting for ' + str(condition)) timeout_source.attach(self.main_context)
while not condition():
self.main_context.iteration(True)
timeout_source.destroy()
def power_saver_enabled_cb(self, spec, data): def power_saver_enabled_cb(self, spec, data):
self.power_saver_enabled = self.power_profile_monitor.get_power_saver_enabled() self.power_saver_enabled = self.power_profile_monitor.get_power_saver_enabled()
@@ -116,10 +120,10 @@ try:
self.assertEqual(self.power_profile_monitor.get_power_saver_enabled(), False) self.assertEqual(self.power_profile_monitor.get_power_saver_enabled(), False)
self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('power-saver', variant_level=1)) self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('power-saver', variant_level=1))
self.assertEventually(lambda: self.power_saver_enabled == True, "power-saver didn't become enabled", 10) self.assertEventually(lambda: self.power_saver_enabled == True, "power-saver didn't become enabled", 1)
self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('balanced', variant_level=1)) self.dbus_props.Set('net.hadess.PowerProfiles', 'ActiveProfile', dbus.String('balanced', variant_level=1))
self.assertEventually(lambda: self.power_saver_enabled == False, "power-saver didn't become disabled", 10) self.assertEventually(lambda: self.power_saver_enabled == False, "power-saver didn't become disabled", 1)
def test_power_profile_power_saver_enabled_portal_default(self): def test_power_profile_power_saver_enabled_portal_default(self):
'''power-saver-enabled property default value''' '''power-saver-enabled property default value'''