Kernel/iwd
3
0
Fork 0
mirror of https://git.kernel.org/pub/scm/network/wireless/iwd.git synced 2024-11-22 06:29:23 +01:00
Code Releases Activity

test-runner: isolate Process/Namespace into utils

Browse Source 
Way too many classes have a dependency on the TestContext class, in
most cases only for is_verbose. This patch removes the dependency from
Process and Namespace classes.

For Process, the test arguments can be parsed in the class itself which
will allow for this class to be completely isolated into its own file.

The Namespace class was already relatively isolated. Both were moved
into utils.py which makes 'run-tests' quite a bit nicer to look at and
more fitting to its name.
This commit is contained in:
James Prestwood 2022-04-06 12:14:44 -07:00 committed by Denis Kenzior
parent 6003ad1527
commit c74ac94c31
2 changed files with 464 additions and 461 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

470
tools/run-tests
View File

@ -2,7 +2,6 @@
import os import os
import shutil import shutil
import fcntl
import sys import sys
import subprocess import subprocess
import atexit import atexit
@ -14,17 +13,16 @@ import multiprocessing
import re import re
import traceback import traceback
from runner import Runner
from configparser import ConfigParser from configparser import ConfigParser
from prettytable import PrettyTable from prettytable import PrettyTable
from termcolor import colored from termcolor import colored
from glob import glob from glob import glob
from collections import namedtuple from collections import namedtuple
from time import sleep
import dbus.mainloop.glib import dbus.mainloop.glib
from gi.repository import GLib from gi.repository import GLib
from weakref import WeakValueDictionary
from runner import Runner
from utils import Process, Namespace
config = None config = None
intf_id = 0 intf_id = 0
@ -81,241 +79,6 @@ def find_binary(list):
return path return path
return None return None
# Partial DBus config. The remainder (<listen>) will be filled in for each
# namespace that is created so each individual dbus-daemon has its own socket
# and address.
dbus_config = '''
<!DOCTYPE busconfig PUBLIC \
"-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN" \
"http://www.freedesktop.org/standards/dbus/1.0/\
busconfig.dtd\">
<busconfig>
<type>system</type>
<limit name=\"reply_timeout\">2147483647</limit>
<auth>ANONYMOUS</auth>
<allow_anonymous/>
<policy context=\"default\">
<allow user=\"*\"/>
<allow own=\"*\"/>
<allow send_type=\"method_call\"/>
<allow send_type=\"signal\"/>
<allow send_type=\"method_return\"/>
<allow send_type=\"error\"/>
<allow receive_type=\"method_call\"/>
<allow receive_type=\"signal\"/>
<allow receive_type=\"method_return\"/>
<allow receive_type=\"error\"/>
<allow send_destination=\"*\" eavesdrop=\"true\"/>
<allow eavesdrop=\"true\"/>
</policy>
'''
class Process(subprocess.Popen):
processes = WeakValueDictionary()
ctx = None
def __new__(cls, *args, **kwargs):
obj = super().__new__(cls)
cls.processes[id(obj)] = obj
return obj
def __init__(self, args, namespace=None, outfile=None, env=None, check=False, cleanup=None):
self.write_fds = []
self.io_watch = None
self.cleanup = cleanup
self.verbose = False
self.out = ''
self.hup = False
self.killed = False
self.namespace = namespace
logfile = args[0]
if not self.ctx:
global config
self.ctx = config.ctx
if self.ctx.is_verbose(args[0], log=False):
self.verbose = True
if namespace:
args = ['ip', 'netns', 'exec', namespace] + args
logfile += '-%s' % namespace
if outfile:
# outfile is only used by iwmon, in which case we don't want
# to append to an existing file.
self._append_outfile(outfile, append=False)
if self.ctx.args.log:
testdir = os.getcwd()
# Special case any processes started prior to a test
# (i.e. from testhome). Put these in the root log directory
if testdir == self.ctx.args.testhome:
testdir = '.'
else:
testdir = os.path.basename(testdir)
logfile = '%s/%s/%s' % (self.ctx.args.log, testdir, logfile)
self._append_outfile(logfile)
super().__init__(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
env=env, cwd=os.getcwd())
# Set as non-blocking so read() in the IO callback doesn't block forever
fl = fcntl.fcntl(self.stdout, fcntl.F_GETFL)
fcntl.fcntl(self.stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)
self.io_watch = GLib.io_add_watch(self.stdout, GLib.IO_IN |
GLib.IO_HUP | GLib.IO_ERR, self.process_io)
print("Starting process {}".format(self.args))
if check:
self.wait(10)
self.killed = True
if self.returncode != 0:
raise subprocess.CalledProcessError(returncode=self.returncode,
cmd=args)
@classmethod
def get_all(cls):
return cls.processes.values()
@classmethod
def kill_all(cls):
for p in cls.processes.values():
if p.args[0] == 'dmesg':
continue
p.kill()
@staticmethod
def _write_io(instance, data, stdout=True):
for f in instance.write_fds:
f.write(data)
# Write out a separator so multiple process calls per
# test are easer to read.
if instance.hup:
f.write("Terminated: {}\n\n".format(instance.args))
f.flush()
if instance.verbose and stdout:
sys.__stdout__.write(data)
sys.__stdout__.flush()
@classmethod
def write_separators(cls, sep):
for proc in cls.processes.values():
if proc.killed:
continue
cls._write_io(proc, sep, stdout=False)
def process_io(self, source, condition):
if condition & GLib.IO_HUP:
self.hup = True
data = source.read()
if not data:
return True
data = data.decode('utf-8')
# Save data away in case the caller needs it (e.g. list_sta)
self.out += data
self._write_io(self, data)
return True
def _append_outfile(self, file, append=True):
gid = int(os.environ.get('SUDO_GID', os.getgid()))
uid = int(os.environ.get('SUDO_UID', os.getuid()))
dir = os.path.dirname(file)
if not path_exists(dir):
os.mkdir(dir)
os.chown(dir, uid, gid)
file = os.path.join(dir,file)
# If the out file exists, append. Useful for processes like
# hostapd_cli where it is called multiple times independently.
if os.path.isfile(file) and append:
mode = 'a'
else:
mode = 'w'
try:
f = open(file, mode)
except Exception as e:
traceback.print_exc()
sys.exit(0)
os.fchown(f.fileno(), uid, gid)
self.write_fds.append(f)
def wait_for_socket(self, socket, wait):
Namespace.non_block_wait(os.path.exists, wait, socket)
# Wait for both process termination and HUP signal
def __wait(self, timeout):
try:
super().wait(timeout)
if not self.hup:
return False
return True
except:
return False
# Override wait() so it can do so non-blocking
def wait(self, timeout=10):
Namespace.non_block_wait(self.__wait, timeout, 1)
self._cleanup()
def _cleanup(self):
if self.cleanup:
self.cleanup()
self.write_fds = []
if self.io_watch:
GLib.source_remove(self.io_watch)
self.io_watch = None
self.cleanup = None
self.killed = True
# Override kill()
def kill(self, force=False):
if self.killed:
return
print("Killing process {}".format(self.args))
if force:
super().kill()
else:
self.terminate()
try:
self.wait(timeout=15)
except:
dbg("Process %s did not complete in 15 seconds!" % self.name)
super().kill()
self._cleanup()
def __str__(self):
return str(self.args) + '\n'
class Interface: class Interface:
def __init__(self, name, config): def __init__(self, name, config):
self.name = name self.name = name
@ -447,9 +210,7 @@ class Hostapd:
A set of running hostapd instances. This is really just a single A set of running hostapd instances. This is really just a single
process since hostapd can be started with multiple config files. process since hostapd can be started with multiple config files.
''' '''
def __init__(self, ctx, radios, configs, radius): def __init__(self, radios, configs, radius):
self.ctx = ctx
if len(configs) != len(radios): if len(configs) != len(radios):
raise Exception("Config (%d) and radio (%d) list length not equal" % \ raise Exception("Config (%d) and radio (%d) list length not equal" % \
(len(configs), len(radios))) (len(configs), len(radios)))
@ -485,7 +246,7 @@ class Hostapd:
if radius: if radius:
args.append(radius) args.append(radius)
if ctx.is_verbose('hostapd'): if Process.is_verbose('hostapd'):
args.append('-d') args.append('-d')
self.process = Process(args) self.process = Process(args)
@ -545,226 +306,13 @@ class Hostapd:
# Hostapd may have already been stopped # Hostapd may have already been stopped
if self.process: if self.process:
self.ctx.stop_process(self.process) self.process.kill()
self.ctx = None
# Hostapd creates simdb sockets for EAP-SIM/AKA tests but does not # Hostapd creates simdb sockets for EAP-SIM/AKA tests but does not
# clean them up. # clean them up.
for f in glob("/tmp/eap_sim_db*"): for f in glob("/tmp/eap_sim_db*"):
os.remove(f) os.remove(f)
dbus_count = 0
class Namespace:
def __init__(self, args, name, radios):
self.dbus_address = None
self.name = name
self.radios = radios
self.args = args
Process(['ip', 'netns', 'add', name]).wait()
for r in radios:
Process(['iw', 'phy', r.name, 'set', 'netns', 'name', name]).wait()
self.start_dbus()
def reset(self):
self._bus = None
for r in self.radios:
r._radio = None
self.radios = []
Process.kill_all()
def __del__(self):
if self.name:
print("Removing namespace %s" % self.name)
Process(['ip', 'netns', 'del', self.name]).wait()
def get_bus(self):
return self._bus
def start_process(self, args, env=None, **kwargs):
if not env:
env = os.environ.copy()
if hasattr(self, "dbus_address"):
# In case this process needs DBus...
env['DBUS_SYSTEM_BUS_ADDRESS'] = self.dbus_address
return Process(args, namespace=self.name, env=env, **kwargs)
def stop_process(self, p, force=False):
p.kill(force)
def is_process_running(self, process):
for p in Process.get_all():
if p.namespace == self.name and p.args[0] == process:
return True
return False
def _cleanup_dbus(self):
try:
os.remove(self.dbus_address.split('=')[1])
except:
pass
os.remove(self.dbus_cfg)
def start_dbus(self):
global dbus_count
self.dbus_address = 'unix:path=/tmp/dbus%d' % dbus_count
self.dbus_cfg = '/tmp/dbus%d.conf' % dbus_count
dbus_count += 1
with open(self.dbus_cfg, 'w+') as f:
f.write(dbus_config)
f.write('<listen>%s</listen>\n' % self.dbus_address)
f.write('</busconfig>\n')
p = self.start_process(['dbus-daemon', '--config-file=%s' % self.dbus_cfg],
cleanup=self._cleanup_dbus)
p.wait_for_socket(self.dbus_address.split('=')[1], 5)
self._bus = dbus.bus.BusConnection(address_or_type=self.dbus_address)
def start_iwd(self, config_dir = '/tmp', storage_dir = '/tmp/iwd'):
args = []
iwd_radios = ','.join([r.name for r in self.radios if r.use == 'iwd'])
if self.args.valgrind:
args.extend(['valgrind', '--leak-check=full', '--track-origins=yes',
'--show-leak-kinds=all',
'--log-file=/tmp/valgrind.log.%p'])
args.extend(['iwd', '-E'])
if iwd_radios != '':
args.extend(['-p', iwd_radios])
if self.is_verbose(args[0]):
args.append('-d')
env = os.environ.copy()
env['CONFIGURATION_DIRECTORY'] = config_dir
env['STATE_DIRECTORY'] = storage_dir
if self.is_verbose('iwd-dhcp'):
env['IWD_DHCP_DEBUG'] = '1'
if self.is_verbose('iwd-tls'):
env['IWD_TLS_DEBUG'] = '1'
if self.is_verbose('iwd-acd'):
env['IWD_ACD_DEBUG'] = '1'
return self.start_process(args, env=env)
def is_verbose(self, process, log=True):
process = os.path.basename(process)
if self.args is None:
return False
# every process is verbose when logging is enabled
if log and self.args.log:
return True
if process in self.args.verbose:
return True
# Special case here to enable verbose output with valgrind running
if process == 'valgrind' and 'iwd' in self.args.verbose:
return True
# Handle any glob matches
for item in self.args.verbose:
if process in glob(item):
return True
return False
@staticmethod
def non_block_wait(func, timeout, *args, exception=True):
'''
Convenience function for waiting in a non blocking
manor using GLibs context iteration i.e. does not block
the main loop while waiting.
'func' will be called at least once and repeatedly until
either it returns success, throws an exception, or the
'timeout' expires.
'timeout' is the ultimate timeout in seconds
'*args' will be passed to 'func'
If 'exception' is an Exception type it will be raised.
If 'exception' is True a generic TimeoutError will be raised.
Any other value will not result in an exception.
'''
# Simple class for signaling the wait timeout
class Bool:
def __init__(self, value):
self.value = value
def wait_timeout_cb(done):
done.value = True
return False
mainloop = GLib.MainLoop()
done = Bool(False)
timeout = GLib.timeout_add_seconds(timeout, wait_timeout_cb, done)
context = mainloop.get_context()
while True:
try:
ret = func(*args)
if ret:
if not done.value:
GLib.source_remove(timeout)
return ret
except Exception as e:
if not done.value:
GLib.source_remove(timeout)
raise e
if done.value == True:
if isinstance(exception, Exception):
raise exception
elif type(exception) == bool and exception:
raise TimeoutError("Timeout on non_block_wait")
else:
return
context.iteration(may_block=True)
def __str__(self):
ret = 'Namespace: %s\n' % self.name
ret += 'Processes:\n'
for p in Process.get_all():
ret += '\t%s' % str(p)
ret += 'Radios:\n'
if len(self.radios) > 0:
for r in self.radios:
ret += '\t%s\n' % str(r)
else:
ret += '\tNo Radios\n'
ret += 'DBus Address: %s\n' % self.dbus_address
ret += '===================================================\n\n'
return ret
class BarChart(): class BarChart():
def __init__(self, height=10, max_width=80): def __init__(self, height=10, max_width=80):
self._height = height self._height = height
@ -828,7 +376,7 @@ class TestContext(Namespace):
self._mem_chart = BarChart() self._mem_chart = BarChart()
def start_dbus_monitor(self): def start_dbus_monitor(self):
if not self.is_verbose('dbus-monitor'): if not Process.is_verbose('dbus-monitor'):
return return
self.start_process(['dbus-monitor', '--address', self.dbus_address]) self.start_process(['dbus-monitor', '--address', self.dbus_address])
@ -933,7 +481,7 @@ class TestContext(Namespace):
radius_config = settings.get('radius_server', None) radius_config = settings.get('radius_server', None)
self.hostapd = Hostapd(self, hapd_radios, hapd_configs, radius_config) self.hostapd = Hostapd(hapd_radios, hapd_configs, radius_config)
self.hostapd.attach_cli() self.hostapd.attach_cli()
def get_frequencies(self): def get_frequencies(self):
@ -1001,7 +549,7 @@ class TestContext(Namespace):
time.sleep(3) time.sleep(3)
ofono_args = ['ofonod', '-n', '--plugin=atmodem,phonesim'] ofono_args = ['ofonod', '-n', '--plugin=atmodem,phonesim']
if self.is_verbose('ofonod'): if Process.is_verbose('ofonod'):
ofono_args.append('-d') ofono_args.append('-d')
self.start_process(ofono_args) self.start_process(ofono_args)

455
tools/utils.py Normal file
View File

@ -0,0 +1,455 @@
import os
import subprocess
import fcntl
import sys
import traceback
import shutil
import dbus
from gi.repository import GLib
from weakref import WeakValueDictionary
from glob import glob
from runner import RunnerCoreArgParse
class Process(subprocess.Popen):
processes = WeakValueDictionary()
testargs = RunnerCoreArgParse().parse_args()
def __new__(cls, *args, **kwargs):
obj = super().__new__(cls)
cls.processes[id(obj)] = obj
return obj
def __init__(self, args, namespace=None, outfile=None, env=None, check=False, cleanup=None):
self.write_fds = []
self.io_watch = None
self.cleanup = cleanup
self.verbose = False
self.out = ''
self.hup = False
self.killed = False
self.namespace = namespace
logfile = args[0]
if Process.is_verbose(args[0], log=False):
self.verbose = True
if namespace:
args = ['ip', 'netns', 'exec', namespace] + args
logfile += '-%s' % namespace
if outfile:
# outfile is only used by iwmon, in which case we don't want
# to append to an existing file.
self._append_outfile(outfile, append=False)
if self.testargs.log:
testdir = os.getcwd()
# Special case any processes started prior to a test
# (i.e. from testhome). Put these in the root log directory
if testdir == self.testargs.testhome:
testdir = '.'
else:
testdir = os.path.basename(testdir)
logfile = '%s/%s/%s' % (self.testargs.log, testdir, logfile)
self._append_outfile(logfile)
super().__init__(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
env=env, cwd=os.getcwd())
# Set as non-blocking so read() in the IO callback doesn't block forever
fl = fcntl.fcntl(self.stdout, fcntl.F_GETFL)
fcntl.fcntl(self.stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)
self.io_watch = GLib.io_add_watch(self.stdout, GLib.IO_IN |
GLib.IO_HUP | GLib.IO_ERR, self.process_io)
print("Starting process {}".format(self.args))
if check:
self.wait(10)
self.killed = True
if self.returncode != 0:
raise subprocess.CalledProcessError(returncode=self.returncode,
cmd=args)
@staticmethod
def is_verbose(process, log=True):
process = os.path.basename(process)
if Process.testargs is None:
return False
# every process is verbose when logging is enabled
if log and Process.testargs.log:
return True
if process in Process.testargs.verbose:
return True
# Special case here to enable verbose output with valgrind running
if process == 'valgrind' and 'iwd' in Process.testargs.verbose:
return True
# Handle any glob matches
for item in Process.testargs.verbose:
if process in glob(item):
return True
return False
@classmethod
def get_all(cls):
return cls.processes.values()
@classmethod
def kill_all(cls):
for p in cls.processes.values():
if p.args[0] == 'dmesg':
continue
p.kill()
@staticmethod
def _write_io(instance, data, stdout=True):
for f in instance.write_fds:
f.write(data)
# Write out a separator so multiple process calls per
# test are easer to read.
if instance.hup:
f.write("Terminated: {}\n\n".format(instance.args))
f.flush()
if instance.verbose and stdout:
sys.__stdout__.write(data)
sys.__stdout__.flush()
@classmethod
def write_separators(cls, sep):
for proc in cls.processes.values():
if proc.killed:
continue
cls._write_io(proc, sep, stdout=False)
def process_io(self, source, condition):
if condition & GLib.IO_HUP:
self.hup = True
data = source.read()
if not data:
return True
data = data.decode('utf-8')
# Save data away in case the caller needs it (e.g. list_sta)
self.out += data
self._write_io(self, data)
return True
def _append_outfile(self, file, append=True):
gid = int(os.environ.get('SUDO_GID', os.getgid()))
uid = int(os.environ.get('SUDO_UID', os.getuid()))
dir = os.path.dirname(file)
if not os.path.exists(dir):
os.mkdir(dir)
os.chown(dir, uid, gid)
file = os.path.join(dir,file)
# If the out file exists, append. Useful for processes like
# hostapd_cli where it is called multiple times independently.
if os.path.isfile(file) and append:
mode = 'a'
else:
mode = 'w'
try:
f = open(file, mode)
except Exception as e:
traceback.print_exc()
sys.exit(0)
os.fchown(f.fileno(), uid, gid)
self.write_fds.append(f)
def wait_for_socket(self, socket, wait):
Namespace.non_block_wait(os.path.exists, wait, socket)
# Wait for both process termination and HUP signal
def __wait(self, timeout):
try:
super().wait(timeout)
if not self.hup:
return False
return True
except:
return False
# Override wait() so it can do so non-blocking
def wait(self, timeout=10):
Namespace.non_block_wait(self.__wait, timeout, 1)
self._cleanup()
def _cleanup(self):
if self.cleanup:
self.cleanup()
self.write_fds = []
if self.io_watch:
GLib.source_remove(self.io_watch)
self.io_watch = None
self.cleanup = None
self.killed = True
# Override kill()
def kill(self, force=False):
if self.killed:
return
print("Killing process {}".format(self.args))
if force:
super().kill()
else:
self.terminate()
try:
self.wait(timeout=15)
except:
print("Process %s did not complete in 15 seconds!" % self.name)
super().kill()
self._cleanup()
def __str__(self):
return str(self.args) + '\n'
dbus_count = 0
# Partial DBus config. The remainder (<listen>) will be filled in for each
# namespace that is created so each individual dbus-daemon has its own socket
# and address.
dbus_config = '''
<!DOCTYPE busconfig PUBLIC \
"-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN" \
"http://www.freedesktop.org/standards/dbus/1.0/\
busconfig.dtd\">
<busconfig>
<type>system</type>
<limit name=\"reply_timeout\">2147483647</limit>
<auth>ANONYMOUS</auth>
<allow_anonymous/>
<policy context=\"default\">
<allow user=\"*\"/>
<allow own=\"*\"/>
<allow send_type=\"method_call\"/>
<allow send_type=\"signal\"/>
<allow send_type=\"method_return\"/>
<allow send_type=\"error\"/>
<allow receive_type=\"method_call\"/>
<allow receive_type=\"signal\"/>
<allow receive_type=\"method_return\"/>
<allow receive_type=\"error\"/>
<allow send_destination=\"*\" eavesdrop=\"true\"/>
<allow eavesdrop=\"true\"/>
</policy>
'''
class Namespace:
def __init__(self, args, name, radios):
self.dbus_address = None
self.name = name
self.radios = radios
self.args = args
Process(['ip', 'netns', 'add', name]).wait()
for r in radios:
Process(['iw', 'phy', r.name, 'set', 'netns', 'name', name]).wait()
self.start_dbus()
def reset(self):
self._bus = None
for r in self.radios:
r._radio = None
self.radios = []
Process.kill_all()
def __del__(self):
if self.name:
print("Removing namespace %s" % self.name)
Process(['ip', 'netns', 'del', self.name]).wait()
def get_bus(self):
return self._bus
def start_process(self, args, env=None, **kwargs):
if not env:
env = os.environ.copy()
if hasattr(self, "dbus_address"):
# In case this process needs DBus...
env['DBUS_SYSTEM_BUS_ADDRESS'] = self.dbus_address
return Process(args, namespace=self.name, env=env, **kwargs)
def stop_process(self, p, force=False):
p.kill(force)
def is_process_running(self, process):
for p in Process.get_all():
if p.namespace == self.name and p.args[0] == process:
return True
return False
def _cleanup_dbus(self):
try:
os.remove(self.dbus_address.split('=')[1])
except:
pass
os.remove(self.dbus_cfg)
def start_dbus(self):
global dbus_count
self.dbus_address = 'unix:path=/tmp/dbus%d' % dbus_count
self.dbus_cfg = '/tmp/dbus%d.conf' % dbus_count
dbus_count += 1
with open(self.dbus_cfg, 'w+') as f:
f.write(dbus_config)
f.write('<listen>%s</listen>\n' % self.dbus_address)
f.write('</busconfig>\n')
p = self.start_process(['dbus-daemon', '--config-file=%s' % self.dbus_cfg],
cleanup=self._cleanup_dbus)
p.wait_for_socket(self.dbus_address.split('=')[1], 5)
self._bus = dbus.bus.BusConnection(address_or_type=self.dbus_address)
def start_iwd(self, config_dir = '/tmp', storage_dir = '/tmp/iwd'):
args = []
iwd_radios = ','.join([r.name for r in self.radios if r.use == 'iwd'])
if self.args.valgrind:
args.extend(['valgrind', '--leak-check=full', '--track-origins=yes',
'--show-leak-kinds=all',
'--log-file=/tmp/valgrind.log.%p'])
args.extend(['iwd', '-E'])
if iwd_radios != '':
args.extend(['-p', iwd_radios])
if Process.is_verbose(args[0]):
args.append('-d')
env = os.environ.copy()
env['CONFIGURATION_DIRECTORY'] = config_dir
env['STATE_DIRECTORY'] = storage_dir
if Process.is_verbose('iwd-dhcp'):
env['IWD_DHCP_DEBUG'] = '1'
if Process.is_verbose('iwd-tls'):
env['IWD_TLS_DEBUG'] = '1'
if Process.is_verbose('iwd-acd'):
env['IWD_ACD_DEBUG'] = '1'
return self.start_process(args, env=env)
@staticmethod
def non_block_wait(func, timeout, *args, exception=True):
'''
Convenience function for waiting in a non blocking
manor using GLibs context iteration i.e. does not block
the main loop while waiting.
'func' will be called at least once and repeatedly until
either it returns success, throws an exception, or the
'timeout' expires.
'timeout' is the ultimate timeout in seconds
'*args' will be passed to 'func'
If 'exception' is an Exception type it will be raised.
If 'exception' is True a generic TimeoutError will be raised.
Any other value will not result in an exception.
'''
# Simple class for signaling the wait timeout
class Bool:
def __init__(self, value):
self.value = value
def wait_timeout_cb(done):
done.value = True
return False
mainloop = GLib.MainLoop()
done = Bool(False)
timeout = GLib.timeout_add_seconds(timeout, wait_timeout_cb, done)
context = mainloop.get_context()
while True:
try:
ret = func(*args)
if ret:
if not done.value:
GLib.source_remove(timeout)
return ret
except Exception as e:
if not done.value:
GLib.source_remove(timeout)
raise e
if done.value == True:
if isinstance(exception, Exception):
raise exception
elif type(exception) == bool and exception:
raise TimeoutError("Timeout on non_block_wait")
else:
return
context.iteration(may_block=True)
def __str__(self):
ret = 'Namespace: %s\n' % self.name
ret += 'Processes:\n'
for p in Process.get_all():
ret += '\t%s' % str(p)
ret += 'Radios:\n'
if len(self.radios) > 0:
for r in self.radios:
ret += '\t%s\n' % str(r)
else:
ret += '\tNo Radios\n'
ret += 'DBus Address: %s\n' % self.dbus_address
ret += '===================================================\n\n'
return ret