auto-t: Enable test frwk to wait for the radios

Previously, we had to wait for an arbitrary amount of time after
iwd was started form the python scripts to make sure that the
radio objects are available on the D-bus. This patch allows to
wait inside of list_devices() method and get back a list of the
devices once they are available.
This commit is contained in:
Tim Kourt 2018-05-15 11:43:12 -07:00 committed by Denis Kenzior
parent 79ce3f645f
commit 102a455e00
1 changed files with 51 additions and 12 deletions

View File

@ -8,6 +8,7 @@ import sys
import os import os
import threading import threading
import time import time
import collections
from abc import ABCMeta, abstractmethod from abc import ABCMeta, abstractmethod
from enum import Enum from enum import Enum
@ -602,6 +603,39 @@ class PSKAgent(dbus.service.Object):
return passwd return passwd
class DeviceList(collections.Mapping):
def __init__(self, iwd, objects):
self._dict = {}
iwd._object_manager.connect_to_signal("InterfacesAdded",
self._interfaces_added_handler, IWD_DEVICE_INTERFACE)
iwd._object_manager.connect_to_signal("InterfacesRemoved",
self._interfaces_removed_handler, IWD_DEVICE_INTERFACE)
for path in objects:
for interface in objects[path]:
if interface == IWD_DEVICE_INTERFACE:
self._dict[path] = Device(path, objects[path][interface])
def __getitem__(self, key):
return self._dict.__getitem__(key)
def __iter__(self):
return self._dict.__iter__()
def __len__(self):
return self._dict.__len__()
def __delitem__(self, key):
self._dict.pop(key).remove()
def _interfaces_added_handler(self, path, interfaces):
self._dict[path] = Device(interfaces[IWD_DEVICE_INTERFACE])
def _interfaces_removed_handler(self, path, interfaces):
del _dict[path]
class IWD(AsyncOpAbstract): class IWD(AsyncOpAbstract):
'''''' ''''''
_bus = dbus.SystemBus() _bus = dbus.SystemBus()
@ -610,6 +644,7 @@ class IWD(AsyncOpAbstract):
_agent_manager_if = None _agent_manager_if = None
_known_network_manager_if = None _known_network_manager_if = None
_iwd_proc = None _iwd_proc = None
_devices = None
def __init__(self, start_iwd_daemon = False, def __init__(self, start_iwd_daemon = False,
iwd_config_dir = IWD_CONFIG_DIR): iwd_config_dir = IWD_CONFIG_DIR):
@ -648,6 +683,9 @@ class IWD(AsyncOpAbstract):
tries += 1 tries += 1
time.sleep(0.05) time.sleep(0.05)
self._devices = DeviceList(self,
self._object_manager.GetManagedObjects())
def __del__(self): def __del__(self):
if self._iwd_proc is None: if self._iwd_proc is None:
return return
@ -727,18 +765,19 @@ class IWD(AsyncOpAbstract):
assert not os.path.isabs(source) assert not os.path.isabs(source)
shutil.copy(source, IWD_STORAGE_DIR) shutil.copy(source, IWD_STORAGE_DIR)
def list_devices(self): def list_devices(self, wait_to_appear = False):
objects = self._object_manager.GetManagedObjects() if not wait_to_appear:
devices = None return list(self._devices.values())
for path in objects.keys():
interfaces = objects[path] tries = 0
for interface in interfaces.keys(): while len(self._devices) == 0:
if interface in [IWD_DEVICE_INTERFACE]: if tries > 100:
if devices is None: raise TimeoutError('IWD has no associated devices')
devices = []
device = Device(path) tries += 1
devices.append(device) time.sleep(0.2)
return devices
return list(self._devices.values())
def list_known_networks(self): def list_known_networks(self):
'''Returns a list of KnownNetwork objects.''' '''Returns a list of KnownNetwork objects.'''