Kernel/iwd
3
0
Fork 0
mirror of https://git.kernel.org/pub/scm/network/wireless/iwd.git synced 2024-10-04 02:18:49 +02:00
Code Releases Activity

auto-t: hostapd.py: use non_block_wait

Browse Source
This commit is contained in:
James Prestwood 2021-08-13 09:24:39 -07:00 committed by Denis Kenzior
parent 732871b601
commit 5ebbe07fa6
1 changed files with 12 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
autotests/util/hostapd.py
View File

@ -86,30 +86,19 @@ class HostapdCLI:
def __init__(self, config=None): def __init__(self, config=None):
self._init_hostapd(config) self._init_hostapd(config)
def wait_for_event(self, event, timeout=10): def _poll_event(self, event):
global mainloop if not self._data_available(0.25):
self._wait_timed_out = False
def wait_timeout_cb():
self._wait_timed_out = True
return False return False
timeout = GLib.timeout_add_seconds(timeout, wait_timeout_cb)
context = mainloop.get_context()
while True:
context.iteration(may_block=False)
while self._data_available(0.25):
data = self.ctrl_sock.recv(4096).decode('utf-8') data = self.ctrl_sock.recv(4096).decode('utf-8')
if event in data: if event in data:
GLib.source_remove(timeout)
return data return data
if self._wait_timed_out: return False
raise TimeoutError('waiting for hostapd event timed out')
return None def wait_for_event(self, event, timeout=10):
return ctx.non_block_wait(self._poll_event, timeout, event,
exception=TimeoutError("waiting for event"))
def _data_available(self, timeout=2): def _data_available(self, timeout=2):
[r, w, e] = select.select([self.ctrl_sock], [], [], timeout) [r, w, e] = select.select([self.ctrl_sock], [], [], timeout)
@ -123,10 +112,10 @@ class HostapdCLI:
self.ctrl_sock.send(bytes(command)) self.ctrl_sock.send(bytes(command))
if self._data_available(timeout): ctx.non_block_wait(self._data_available, timeout,
return self.ctrl_sock.recv(4096).decode('utf-8') exception=TimeoutError("waiting for control response"))
raise Exception('timeout waiting for control response') return self.ctrl_sock.recv(4096).decode('utf-8')
def _del_hostapd(self, force=False): def _del_hostapd(self, force=False):
if not self.ctrl_sock: if not self.ctrl_sock: