Kernel/iwd
3
0
Fork 0
mirror of https://git.kernel.org/pub/scm/network/wireless/iwd.git synced 2024-12-22 13:02:44 +01:00
Code Releases Activity

auto-t: update wpas.py to use non_block_wait

Browse Source 
This should be used as a replacement for the common GLib
wait loop (internally this is all it does).
This commit is contained in:
James Prestwood 2022-01-11 16:55:49 -08:00 committed by Denis Kenzior
parent 295430a47e
commit f8459427d3
1 changed files with 13 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
autotests/util/wpas.py
View File

@ -58,29 +58,20 @@ class Wpas:
f.close()
return dict([[v.strip() for v in kv] for kv in [l.split('#', 1)[0].split('=', 1) for l in lines] if len(kv) == 2])
def _check_event(self, event):
if not event and len(self._rx_data) >= 1:
print("returning %s" % self._rx_data[0])
return self._rx_data[0]
for e in self._rx_data:
if event in e:
return self._rx_data
return False
def wait_for_event(self, event, timeout=10):
self._wait_timed_out = False
def wait_timeout_cb():
self._wait_timed_out = True
return False
timeout = GLib.timeout_add_seconds(timeout, wait_timeout_cb)
context = ctx.mainloop.get_context()
while True:
context.iteration(may_block=True)
if not event and len(self._rx_data) >= 1:
return self._rx_data[0]
for e in self._rx_data:
if event in e:
GLib.source_remove(timeout)
return self._rx_data
if self._wait_timed_out:
raise TimeoutError('waiting for wpas event timed out')
self._rx_data = []
return ctx.non_block_wait(self._check_event, timeout, event)
def wait_for_result(self, timeout=10):
self._rx_data = []