diff --git a/autotests/testWPA2-ext-key-id/connection_test.py b/autotests/testWPA2-ext-key-id/connection_test.py index d77a216b..afd48149 100644 --- a/autotests/testWPA2-ext-key-id/connection_test.py +++ b/autotests/testWPA2-ext-key-id/connection_test.py @@ -50,8 +50,6 @@ class Test(unittest.TestCase): testutil.test_iface_operstate() testutil.test_ifaces_connected() - wd.wait(0.5) - device.disconnect() condition = 'not obj.connected' diff --git a/autotests/util/hostapd.py b/autotests/util/hostapd.py index ff5a97c1..758427fe 100644 --- a/autotests/util/hostapd.py +++ b/autotests/util/hostapd.py @@ -102,23 +102,26 @@ class HostapdCLI(object): return True - def _poll_event(self, event): + def _poll_event(self, event, disallow): # Look through the list (most recent is first) until the even is found. # Once found consume this event and any older ones as to not # accidentally trigger a false positive later on. for idx, e in enumerate(self.events): + for d in disallow: + if d in e: + raise Exception('Event %s found while waiting for %s' % (d, event)) if event in e: self.events = self.events[:idx] return e return False - def wait_for_event(self, event, timeout=10): + def wait_for_event(self, event, timeout=10, disallow=[]): if event == 'AP-ENABLED': if self.enabled: return 'AP-ENABLED' - return ctx.non_block_wait(self._poll_event, timeout, event, + return ctx.non_block_wait(self._poll_event, timeout, event, disallow, exception=TimeoutError("waiting for event")) def _data_available(self): @@ -166,8 +169,8 @@ class HostapdCLI(object): self.events = [] cmd = 'EAPOL_REAUTH ' + client_address self.ctrl_sock.sendall(cmd.encode('utf-8')) - self.wait_for_event('CTRL-EVENT-EAP-STARTED') - self.wait_for_event('CTRL-EVENT-EAP-SUCCESS') + self.wait_for_event('CTRL-EVENT-EAP-STARTED', disallow=['AP-STA-DISCONNECTED']) + self.wait_for_event('CTRL-EVENT-EAP-SUCCESS', disallow=['AP-STA-DISCONNECTED']) def reload(self): # Seemingly all three commands needed for the instance to notice @@ -234,7 +237,8 @@ class HostapdCLI(object): if address: cmd = 'REKEY_PTK %s' % address self.ctrl_sock.sendall(cmd.encode('utf-8')) - self.wait_for_event('EAPOL-4WAY-HS-COMPLETED') + self.events = [] + self.wait_for_event('EAPOL-4WAY-HS-COMPLETED', disallow=['AP-STA-DISCONNECTED']) return cmd = 'REKEY_GTK'