tests: Add wait_connected() and wait_disconnected() helpers

Signed-off-by: Jouni Malinen <j@w1.fi>
This commit is contained in:
Jouni Malinen 2014-12-20 11:51:30 +02:00
parent e0cccf26a4
commit 5f35a5e27f
34 changed files with 231 additions and 615 deletions

View file

@ -54,9 +54,8 @@ def test_ap_cipher_tkip_countermeasures_ap(dev, apdev):
with open(testfile, "w") as f:
f.write("ff:ff:ff:ff:ff:ff")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("No disconnection after two Michael MIC failures")
ev = dev[0].wait_disconnected(timeout=10,
error="No disconnection after two Michael MIC failures")
if "reason=14" not in ev:
raise Exception("Unexpected disconnection reason: " + ev)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=1)
@ -88,9 +87,8 @@ def test_ap_cipher_tkip_countermeasures_sta(dev, apdev):
with open(testfile, "w") as f:
f.write("ff:ff:ff:ff:ff:ff")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("No disconnection after two Michael MIC failures")
ev = dev[0].wait_disconnected(timeout=10,
error="No disconnection after two Michael MIC failures")
if "reason=14 locally_generated=1" not in ev:
raise Exception("Unexpected disconnection reason: " + ev)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=1)

View file

@ -303,17 +303,13 @@ def test_ap_enable_disable_reenable(dev, apdev):
ev = hapd.wait_event(["AP-DISABLED"], timeout=30)
if ev is None:
raise Exception("AP disabling timed out")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("STA disconnect event timed out")
dev[0].wait_disconnected(timeout=10)
hapd.enable()
ev = hapd.wait_event(["AP-ENABLED"], timeout=30)
if ev is None:
raise Exception("AP startup timed out")
dev[1].connect("dynamic", key_mgmt="NONE", scan_freq="2412")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("STA connect event timed out")
dev[0].wait_connected(timeout=10)
def test_ap_double_disable(dev, apdev):
"""Double DISABLE regression test"""

View file

@ -62,9 +62,7 @@ def eap_check_auth(dev, method, initial, rsn=True, sha256=False,
ev = dev.wait_event(["CTRL-EVENT-EAP-FAILURE"])
if ev is None:
raise Exception("EAP failure timed out")
ev = dev.wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Disconnection timed out")
ev = dev.wait_disconnected(timeout=10)
if not local_error_report:
if "reason=23" not in ev:
raise Exception("Proper reason code for disconnection not reported")
@ -696,9 +694,7 @@ def test_ap_wpa2_eap_aka_prime(dev, apdev):
identity="6555444333222111@both",
password="5122250214c33e723a5dd523fc145fc0:981d464c7c52eb6e5036234984ad0bcf:000000000123",
wait_connect=False, scan_freq="2412")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection with the AP timed out")
dev[1].wait_connected(timeout=15)
logger.info("Negative test with incorrect key")
dev[0].request("REMOVE_NETWORK all")
@ -1109,9 +1105,7 @@ def test_ap_wpa2_eap_tls_pkcs12(dev, apdev):
raise Exception("Request for private key passphrase timed out")
id = ev.split(':')[0].split('-')[-1]
dev[0].request("CTRL-RSP-PASSPHRASE-" + id + ":whatever")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[0].wait_connected(timeout=10)
def test_ap_wpa2_eap_tls_pkcs12_blob(dev, apdev):
"""WPA2-Enterprise connection using EAP-TLS and PKCS#12 from configuration blob"""
@ -1209,9 +1203,7 @@ def test_ap_wpa2_eap_tls_diff_ca_trust(dev, apdev):
if ev is None:
raise Exception("EAP-TTLS not re-started")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
ev = dev[0].wait_disconnected(timeout=15)
if "reason=23" not in ev:
raise Exception("Proper reason code for disconnection not reported")
@ -1237,9 +1229,7 @@ def test_ap_wpa2_eap_tls_diff_ca_trust2(dev, apdev):
if ev is None:
raise Exception("EAP-TTLS not re-started")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
ev = dev[0].wait_disconnected(timeout=15)
if "reason=23" not in ev:
raise Exception("Proper reason code for disconnection not reported")
@ -1261,9 +1251,7 @@ def test_ap_wpa2_eap_tls_diff_ca_trust3(dev, apdev):
if ev is None:
raise Exception("EAP-TTLS not re-started")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
ev = dev[0].wait_disconnected(timeout=15)
if "reason=23" not in ev:
raise Exception("Proper reason code for disconnection not reported")
@ -1455,9 +1443,7 @@ def test_ap_wpa2_eap_ttls_server_cert_hash(dev, apdev):
raise Exception("EAP result timed out")
if "Server certificate chain probe" not in ev:
raise Exception("Server certificate probe not reported")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection event not seen")
dev[0].wait_disconnected(timeout=10)
dev[0].request("REMOVE_NETWORK all")
dev[0].connect("test-wpa2-eap", key_mgmt="WPA-EAP", eap="TTLS",
@ -1473,9 +1459,7 @@ def test_ap_wpa2_eap_ttls_server_cert_hash(dev, apdev):
raise Exception("EAP result timed out")
if "Server certificate mismatch" not in ev:
raise Exception("Server certificate mismatch not reported")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection event not seen")
dev[0].wait_disconnected(timeout=10)
dev[0].request("REMOVE_NETWORK all")
eap_connect(dev[0], apdev[0], "TTLS", "DOMAIN\mschapv2 user",
@ -1581,9 +1565,7 @@ def test_ap_wpa2_eap_gpsk(dev, apdev):
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=10)
logger.info("Test failed algorithm negotiation")
dev[0].set_network_quoted(id, "phase1", "cipher=9")
@ -1627,9 +1609,7 @@ def test_ap_wpa2_eap_eke(dev, apdev):
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=10)
logger.info("Test failed algorithm negotiation")
dev[0].set_network_quoted(id, "phase1", "dhgroup=9 encr=9 prf=9 mac=9")
@ -1760,9 +1740,7 @@ def test_ap_wpa2_eap_interactive(dev, apdev):
id = ev.split(':')[0].split('-')[-1]
type = "OTP" if "CTRL-REQ-OTP" in ev else "PASSWORD"
dev[0].request("CTRL-RSP-" + type + "-" + id + ":" + req_pw)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[0].wait_connected(timeout=10)
dev[0].request("REMOVE_NETWORK all")
def test_ap_wpa2_eap_vendor_test(dev, apdev):

View file

@ -52,9 +52,7 @@ def hs20_ap_params(ssid="test-hs20"):
def check_auto_select(dev, bssid):
dev.scan_for_bss(bssid, freq="2412")
dev.request("INTERWORKING_SELECT auto freq=2412")
ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
ev = dev.wait_connected(timeout=15)
if bssid not in ev:
raise Exception("Connected to incorrect network")
dev.request("REMOVE_NETWORK all")
@ -133,9 +131,7 @@ def interworking_ext_sim_auth(dev, method):
resp = res.split(' ')[2].rstrip()
dev.request("CTRL-RSP-SIM-" + id + ":GSM-AUTH:" + resp)
ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
dev.wait_connected(timeout=15)
def interworking_connect(dev, bssid, method):
dev.request("INTERWORKING_CONNECT " + bssid)
@ -148,9 +144,7 @@ def interworking_auth(dev, method):
if "(" + method + ")" not in ev:
raise Exception("Unexpected EAP method selection")
ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
dev.wait_connected(timeout=15)
def check_probe_resp(wt, bssid_unexpected, bssid_expected):
if bssid_unexpected:
@ -551,9 +545,7 @@ def test_ap_hs20_auto_interworking(dev, apdev):
'domain': "example.com",
'update_identifier': "1234" })
dev[0].request("REASSOCIATE")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
dev[0].wait_connected(timeout=15)
check_sp_type(dev[0], "home")
status = dev[0].get_status()
if status['pairwise_cipher'] != "CCMP":
@ -1028,9 +1020,7 @@ def policy_test(dev, ap, values, only_one=True):
raise Exception("Selected incorrect BSS")
break
ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
ev = dev.wait_connected(timeout=15)
if bssid and bssid not in ev:
raise Exception("Connected to incorrect BSS")
@ -1632,9 +1622,7 @@ def _test_ap_hs20_deauth_req_ess(dev, apdev):
if "1 120 http://example.com/" not in ev:
raise Exception("Unexpected deauth imminent notice: " + ev)
hapd.request("DEAUTHENTICATE " + addr)
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection")
dev[0].wait_disconnected(timeout=10)
if "[TEMP-DISABLED]" not in dev[0].list_networks()[0]['flags']:
raise Exception("Network not marked temporarily disabled")
ev = dev[0].wait_event(["SME: Trying to authenticate",
@ -1663,9 +1651,7 @@ def _test_ap_hs20_deauth_req_bss(dev, apdev):
if "0 120 http://example.com/" not in ev:
raise Exception("Unexpected deauth imminent notice: " + ev)
hapd.request("DEAUTHENTICATE " + addr + " reason=4")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection")
ev = dev[0].wait_disconnected(timeout=10)
if "reason=4" not in ev:
raise Exception("Unexpected disconnection reason")
if "[TEMP-DISABLED]" not in dev[0].list_networks()[0]['flags']:
@ -1702,9 +1688,7 @@ def _test_ap_hs20_deauth_req_from_radius(dev, apdev):
raise Exception("Timeout on deauth imminent notice")
if " 1 100" not in ev:
raise Exception("Unexpected deauth imminent contents")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=3)
if ev is None:
raise Exception("Timeout on disconnection")
dev[0].wait_disconnected(timeout=3)
def test_ap_hs20_remediation_required(dev, apdev):
"""Hotspot 2.0 connection and remediation required from RADIUS"""
@ -1854,9 +1838,7 @@ def test_ap_hs20_network_preference(dev, apdev):
dev[0].scan_for_bss(bssid, freq="2412")
dev[0].request("INTERWORKING_SELECT auto freq=2412")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
ev = dev[0].wait_connected(timeout=15)
if bssid not in ev:
raise Exception("Unexpected network selected")
@ -1896,9 +1878,7 @@ def test_ap_hs20_network_preference2(dev, apdev):
dev[0].scan_for_bss(bssid2, freq="2412")
dev[0].request("INTERWORKING_SELECT auto freq=2412")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
ev = dev[0].wait_connected(timeout=15)
if bssid2 not in ev:
raise Exception("Unexpected network selected")
@ -1942,9 +1922,7 @@ def test_ap_hs20_network_preference3(dev, apdev):
dev[0].scan_for_bss(bssid, freq="2412")
dev[0].scan_for_bss(bssid2, freq="2412")
dev[0].request("INTERWORKING_SELECT auto freq=2412")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
ev = dev[0].wait_connected(timeout=15)
if bssid not in ev:
raise Exception("Unexpected network selected")
@ -1986,9 +1964,7 @@ def test_ap_hs20_network_preference4(dev, apdev):
dev[0].scan_for_bss(bssid, freq="2412")
dev[0].scan_for_bss(bssid2, freq="2412")
dev[0].request("INTERWORKING_SELECT auto freq=2412")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Connection timed out")
ev = dev[0].wait_connected(timeout=15)
if bssid not in ev:
raise Exception("Unexpected network selected")
@ -2330,9 +2306,7 @@ def test_ap_hs20_multi_network_and_cred_removal(dev, apdev):
dev[0].add_network()
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection")
dev[0].wait_disconnected(timeout=10)
hapd.disable()
hapd.set("ssid", "another ssid")
@ -2348,9 +2322,7 @@ def test_ap_hs20_multi_network_and_cred_removal(dev, apdev):
dev[0].remove_cred(id)
if len(dev[0].list_networks()) != 3:
raise Exception("Unexpected number of networks after to remove_crec")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection")
dev[0].wait_disconnected(timeout=10)
def _test_ap_hs20_proxyarp(dev, apdev):
bssid = apdev[0]['bssid']

View file

@ -39,9 +39,7 @@ def test_ap_open_packet_loss(dev, apdev):
dev[i].connect("open", key_mgmt="NONE", scan_freq="2412",
wait_connect=False)
for i in range(0, 3):
ev = dev[i].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Association with the AP timed out")
dev[i].wait_connected(timeout=20)
def test_ap_open_unknown_action(dev, apdev):
"""AP with open mode configuration and unknown Action frame"""
@ -62,12 +60,8 @@ def test_ap_open_reconnect_on_inactivity_disconnect(dev, apdev):
hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "open" })
dev[0].connect("open", key_mgmt="NONE", scan_freq="2412")
hapd.request("DEAUTHENTICATE " + dev[0].p2p_interface_addr() + " reason=4")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Timeout on disconnection")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=2)
if ev is None:
raise Exception("Timeout on reconnection")
dev[0].wait_disconnected(timeout=5)
dev[0].wait_connected(timeout=2, error="Timeout on reconnection")
def test_ap_open_assoc_timeout(dev, apdev):
"""AP timing out association"""
@ -107,18 +101,14 @@ def test_ap_open_assoc_timeout(dev, apdev):
if assoc != 3:
raise Exception("Association Request frames not received: assoc=%d" % assoc)
hapd.set("ext_mgmt_frame_handling", "0")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Timeout on connection")
dev[0].wait_connected(timeout=15)
def test_ap_open_id_str(dev, apdev):
"""AP with open mode and id_str"""
hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "open" })
dev[0].connect("open", key_mgmt="NONE", scan_freq="2412", id_str="foo",
wait_connect=False)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Association with the AP timed out")
ev = dev[0].wait_connected(timeout=10)
if "id_str=foo" not in ev:
raise Exception("CTRL-EVENT-CONNECT did not have matching id_str: " + ev)
if dev[0].get_status_field("id_str") != "foo":
@ -137,34 +127,28 @@ def test_ap_open_select_any(dev, apdev):
raise Exception("Unexpected connection")
dev[0].select_network("any")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=10)
def test_ap_open_unexpected_assoc_event(dev, apdev):
"""AP with open mode and unexpected association event"""
hapd = hostapd.add_ap(apdev[0]['ifname'], { "ssid": "open" })
dev[0].connect("open", key_mgmt="NONE", scan_freq="2412")
dev[0].request("DISCONNECT")
dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
dev[0].wait_disconnected(timeout=15)
dev[0].dump_monitor()
# This will be accepted due to matching network
subprocess.call(['iw', 'dev', dev[0].ifname, 'connect', 'open', "2412",
apdev[0]['bssid']])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
dev[0].dump_monitor()
dev[0].request("REMOVE_NETWORK all")
dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
dev[0].wait_disconnected(timeout=5)
dev[0].dump_monitor()
# This will result in disconnection due to no matching network
subprocess.call(['iw', 'dev', dev[0].ifname, 'connect', 'open', "2412",
apdev[0]['bssid']])
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection with the AP timed out")
dev[0].wait_disconnected(timeout=15)
def test_ap_bss_load(dev, apdev):
"""AP with open mode (no security) configuration"""

View file

@ -137,14 +137,10 @@ def test_ap_pmf_assoc_comeback(dev, apdev):
scan_freq="2412")
hapd.set("ext_mgmt_frame_handling", "1")
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection")
dev[0].wait_disconnected(timeout=10)
hapd.set("ext_mgmt_frame_handling", "0")
dev[0].request("REASSOCIATE")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout on re-connection")
dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
if wt.get_sta_counter("assocresp_comeback", apdev[0]['bssid'],
dev[0].p2p_interface_addr()) < 1:
raise Exception("AP did not use association comeback request")
@ -164,9 +160,7 @@ def test_ap_pmf_assoc_comeback2(dev, apdev):
if "OK" not in dev[0].request("DROP_SA"):
raise Exception("DROP_SA failed")
dev[0].request("REASSOCIATE")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout on re-connection")
dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
if wt.get_sta_counter("reassocresp_comeback", apdev[0]['bssid'],
dev[0].p2p_interface_addr()) < 1:
raise Exception("AP did not use reassociation comeback request")

View file

@ -395,9 +395,7 @@ def test_ap_wpa2_psk_ext(dev, apdev):
if ev is None:
raise Exception("Timeout on EAPOL-TX from hostapd")
if "AP-STA-CONNECTED" in ev:
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Timeout on connection event from wpa_supplicant")
dev[0].wait_connected(timeout=15)
break
res = dev[0].request("EAPOL_RX " + bssid + " " + ev.split(' ')[2])
if "OK" not in res:

View file

@ -43,15 +43,11 @@ def test_ap_reassociation_to_same_bss(dev, apdev):
dev[0].connect("test-open", key_mgmt="NONE")
dev[0].request("REASSOCIATE")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Reassociation with the AP timed out")
dev[0].wait_connected(timeout=10, error="Reassociation timed out")
hwsim_utils.test_connectivity(dev[0], hapd)
dev[0].request("REATTACH")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Reassociation (reattach) with the AP timed out")
dev[0].wait_connected(timeout=10, error="Reattach timed out")
hwsim_utils.test_connectivity(dev[0], hapd)
def test_ap_roam_set_bssid(dev, apdev):

View file

@ -44,9 +44,7 @@ def test_ap_wps_init(dev, apdev):
dev[0].request("ENABLE_NETWORK %s no-connect" % id)
dev[0].request("WPS_PBC")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
status = dev[0].get_status()
if status['wpa_state'] != 'COMPLETED' or status['bssid'] != apdev[0]['bssid']:
raise Exception("Not fully connected")
@ -103,9 +101,7 @@ def test_ap_wps_init_2ap_pbc(dev, apdev):
raise Exception("WPS cred event not seen")
if "100e" not in ev:
raise Exception("WPS attributes not included in the cred event")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
dev[1].scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
dev[1].scan_for_bss(apdev[1]['bssid'], freq="2412")
@ -136,9 +132,7 @@ def test_ap_wps_init_2ap_pin(dev, apdev):
raise Exception("WPS-AUTH flag missing from AP2")
dev[0].dump_monitor()
dev[0].request("WPS_PIN any " + pin)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
dev[1].scan_for_bss(apdev[0]['bssid'], freq="2412", force_scan=True)
dev[1].scan_for_bss(apdev[1]['bssid'], freq="2412")
@ -181,9 +175,7 @@ def test_ap_wps_conf(dev, apdev):
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].dump_monitor()
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
status = dev[0].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")
@ -214,9 +206,7 @@ def test_ap_wps_conf_5ghz(dev, apdev):
hapd.request("WPS_PBC")
dev[0].scan_for_bss(apdev[0]['bssid'], freq="5180")
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
sta = hapd.get_sta(dev[0].p2p_interface_addr())
if 'wpsDeviceName' not in sta or sta['wpsDeviceName'] != "Device A":
@ -241,9 +231,7 @@ def test_ap_wps_conf_chan14(dev, apdev):
logger.info("WPS provisioning step")
hapd.request("WPS_PBC")
dev[0].request("WPS_PBC")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
sta = hapd.get_sta(dev[0].p2p_interface_addr())
if 'wpsDeviceName' not in sta or sta['wpsDeviceName'] != "Device A":
@ -268,9 +256,7 @@ def test_ap_wps_twice(dev, apdev):
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].dump_monitor()
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
dev[0].request("DISCONNECT")
logger.info("Restart AP with different passphrase and re-run WPS")
@ -283,9 +269,7 @@ def test_ap_wps_twice(dev, apdev):
hapd.request("WPS_PBC")
dev[0].dump_monitor()
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
networks = dev[0].list_networks()
if len(networks) > 1:
raise Exception("Unexpected duplicated network block present")
@ -311,9 +295,7 @@ def test_ap_wps_incorrect_pin(dev, apdev):
raise Exception("Incorrect config_error reported")
if "msg=8" not in ev:
raise Exception("PIN error detected on incorrect message")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection event")
dev[0].wait_disconnected(timeout=10)
dev[0].request("WPS_CANCEL")
# if a scan was in progress, wait for it to complete before trying WPS again
ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], 5)
@ -333,9 +315,7 @@ def test_ap_wps_incorrect_pin(dev, apdev):
raise Exception("Incorrect config_error reported")
if "msg=10" not in ev:
raise Exception("PIN error detected on incorrect message")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection event")
dev[0].wait_disconnected(timeout=10)
def test_ap_wps_conf_pin(dev, apdev):
"""WPS PIN provisioning with configured AP"""
@ -351,9 +331,7 @@ def test_ap_wps_conf_pin(dev, apdev):
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].dump_monitor()
dev[0].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
status = dev[0].get_status()
if status['wpa_state'] != 'COMPLETED' or status['bssid'] != apdev[0]['bssid']:
raise Exception("Not fully connected")
@ -376,9 +354,7 @@ def test_ap_wps_conf_pin(dev, apdev):
if "WPS-M2D" not in ev:
raise Exception("Unexpected WPS operation started")
hapd.request("WPS_PIN any " + pin)
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=30)
def test_ap_wps_conf_pin_v1(dev, apdev):
"""WPS PIN provisioning with configured WPS v1.0 AP"""
@ -403,10 +379,8 @@ def test_ap_wps_conf_pin_v1(dev, apdev):
raise Exception("WPS-PIN flag not seen in scan results")
dev[0].dump_monitor()
dev[0].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
dev[0].wait_connected(timeout=30)
hapd.request("SET wps_version_number 0x20")
if ev is None:
raise Exception("Association with the AP timed out")
def test_ap_wps_conf_pin_2sta(dev, apdev):
"""Two stations trying to use WPS PIN at the same time"""
@ -427,12 +401,8 @@ def test_ap_wps_conf_pin_2sta(dev, apdev):
dev[1].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
dev[1].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
dev[1].wait_connected(timeout=30)
def test_ap_wps_conf_pin_timeout(dev, apdev):
"""WPS PIN provisioning with configured AP timing out PIN"""
@ -460,9 +430,7 @@ def test_ap_wps_conf_pin_timeout(dev, apdev):
hapd.request("WPS_PIN any " + pin + " 20 " + addr)
dev[0].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
def test_ap_wps_reg_connect(dev, apdev):
"""WPS registrar using AP PIN to connect"""
@ -532,8 +500,8 @@ def test_ap_wps_random_ap_pin(dev, apdev):
dev[1].wps_reg(apdev[0]['bssid'], appin)
dev[0].request("REMOVE_NETWORK all")
dev[1].request("REMOVE_NETWORK all")
dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"])
dev[0].wait_disconnected(timeout=10)
dev[1].wait_disconnected(timeout=10)
logger.info("WPS provisioning step after AP PIN timeout")
hapd.request("WPS_AP_PIN disable")
@ -612,9 +580,7 @@ def test_ap_wps_reg_config_ext_processing(dev, apdev):
hapd.request("SET wps_cred_processing 0")
if "FAIL" in hapd.request("WPS_CONFIG " + new_ssid.encode("hex") + " WPA2PSK CCMP " + new_passphrase.encode("hex")):
raise Exception("WPS_CONFIG command failed")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
def test_ap_wps_reg_config_tkip(dev, apdev):
"""WPS registrar configuring AP to use TKIP and AP upgrading to TKIP+CCMP"""
@ -676,9 +642,7 @@ def test_ap_wps_setup_locked(dev, apdev):
ap_setup_locked=True
elif "config_error=18" not in ev:
raise Exception("config_error=18 not reported")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection event")
dev[0].wait_disconnected(timeout=10)
time.sleep(0.1)
if not ap_setup_locked:
raise Exception("AP setup was not locked")
@ -700,9 +664,7 @@ def test_ap_wps_setup_locked(dev, apdev):
ev = dev[0].wait_event(["WPS-SUCCESS"], timeout=30)
if ev is None:
raise Exception("WPS success was not reported")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
appin = hapd.request("WPS_AP_PIN random")
if "FAIL" in appin:
@ -741,9 +703,7 @@ def test_ap_wps_setup_locked_timeout(dev, apdev):
break
elif "config_error=18" not in ev:
raise Exception("config_error=18 not reported")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on disconnection event")
dev[0].wait_disconnected(timeout=10)
time.sleep(0.1)
if not ap_setup_locked:
raise Exception("AP setup was not locked")
@ -917,9 +877,7 @@ def _test_ap_wps_er_add_enrollee(dev, apdev):
ev = dev[1].wait_event(["WPS-SUCCESS"], timeout=30)
if ev is None:
raise Exception("Enrollee did not report success")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=15)
ev = dev[0].wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("WPS ER did not report success")
@ -938,9 +896,7 @@ def _test_ap_wps_er_add_enrollee(dev, apdev):
if addr2 not in ev:
raise Exception("Unexpected Enrollee MAC address")
dev[0].request("WPS_ER_PIN " + addr2 + " " + pin + " " + addr2)
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[2].wait_connected(timeout=30)
ev = dev[0].wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("WPS ER did not report success")
@ -948,7 +904,7 @@ def _test_ap_wps_er_add_enrollee(dev, apdev):
logger.info("Verify registrar selection behavior")
dev[0].request("WPS_ER_PIN any " + pin + " " + dev[1].p2p_interface_addr())
dev[1].request("DISCONNECT")
dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"])
dev[1].wait_disconnected(timeout=10)
dev[1].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[1].scan(freq="2412")
bss = dev[1].get_bss(apdev[0]['bssid'])
@ -1046,9 +1002,7 @@ def _test_ap_wps_er_add_enrollee_pbc(dev, apdev):
ev = dev[1].wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("Enrollee did not report success")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=15)
ev = dev[0].wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("WPS ER did not report success")
@ -1162,9 +1116,7 @@ def _test_ap_wps_er_v10_add_enrollee_pin(dev, apdev):
dev[1].scan_for_bss(apdev[0]['bssid'], freq=2412)
dev[1].dump_monitor()
dev[1].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=30)
ev = dev[0].wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("WPS ER did not report success")
@ -1208,7 +1160,7 @@ def _test_ap_wps_er_config_ap(dev, apdev):
ev = dev[0].wait_event(["WPS-SUCCESS"])
if ev is None:
raise Exception("WPS ER configuration operation timed out")
dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
dev[0].wait_disconnected(timeout=10)
dev[0].connect(ssid, psk="1234567890", scan_freq="2412")
logger.info("WPS ER restart")
@ -1240,9 +1192,7 @@ def test_ap_wps_fragmentation(dev, apdev):
dev[0].dump_monitor()
dev[0].request("SET wps_fragment_size 50")
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
status = dev[0].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")
@ -1257,9 +1207,7 @@ def test_ap_wps_fragmentation(dev, apdev):
dev[1].scan_for_bss(apdev[0]['bssid'], freq=2412)
dev[1].request("SET wps_fragment_size 50")
dev[1].request("WPS_PIN %s %s" % (apdev[0]['bssid'], pin))
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=30)
status = dev[1].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")
@ -1295,9 +1243,7 @@ def test_ap_wps_new_version_sta(dev, apdev):
dev[0].request("SET wps_version_number 0x43")
dev[0].request("SET wps_vendor_ext_m1 000137100100020001")
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
def test_ap_wps_new_version_ap(dev, apdev):
"""WPS compatibility with new version number on the AP"""
@ -1314,10 +1260,8 @@ def test_ap_wps_new_version_ap(dev, apdev):
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].dump_monitor()
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
dev[0].wait_connected(timeout=30)
hapd.request("SET wps_version_number 0x20")
if ev is None:
raise Exception("Association with the AP timed out")
def test_ap_wps_check_pin(dev, apdev):
"""Verify PIN checking through control interface"""
@ -1407,9 +1351,7 @@ def test_ap_wps_ie_fragmentation(dev, apdev):
hapd.request("WPS_PBC")
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
bss = dev[0].get_bss(apdev[0]['bssid'])
if "wps_device_name" not in bss or bss['wps_device_name'] != "1234567890abcdef1234567890abcdef":
logger.info("Device Name not received correctly")
@ -1464,17 +1406,13 @@ def test_ap_wps_per_station_psk(dev, apdev):
hapd.request("WPS_PBC")
dev[0].scan_for_bss(apdev[0]['bssid'], freq=2412)
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out (1)")
dev[0].wait_connected(timeout=30)
logger.info("Second enrollee")
hapd.request("WPS_PBC")
dev[1].scan_for_bss(apdev[0]['bssid'], freq=2412)
dev[1].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out (2)")
dev[1].wait_connected(timeout=30)
logger.info("External registrar")
dev[2].scan_for_bss(apdev[0]['bssid'], freq=2412)
@ -1536,17 +1474,13 @@ def test_ap_wps_per_station_psk_failure(dev, apdev):
hapd.request("WPS_PBC")
dev[0].scan_for_bss(apdev[0]['bssid'], freq=2412)
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out (1)")
dev[0].wait_connected(timeout=30)
logger.info("Second enrollee")
hapd.request("WPS_PBC")
dev[1].scan_for_bss(apdev[0]['bssid'], freq=2412)
dev[1].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out (2)")
dev[1].wait_connected(timeout=30)
logger.info("External registrar")
dev[2].scan_for_bss(apdev[0]['bssid'], freq=2412)
@ -1614,9 +1548,7 @@ def test_ap_wps_auto_setup_with_config_file(dev, apdev):
hapd.request("WPS_PBC")
dev[0].scan_for_bss(apdev[0]['bssid'], freq="2412")
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
with open(conffile, "r") as f:
lines = f.read().splitlines()
vals = dict()
@ -2297,9 +2229,7 @@ def test_ap_wps_while_connected(dev, apdev):
hapd.request("WPS_PBC")
dev[0].dump_monitor()
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
status = dev[0].get_status()
if status['bssid'] != apdev[0]['bssid']:
raise Exception("Unexpected BSSID")
@ -2323,9 +2253,7 @@ def test_ap_wps_while_connected_no_autoconnect(dev, apdev):
hapd.request("WPS_PBC")
dev[0].dump_monitor()
dev[0].request("WPS_PBC " + apdev[0]['bssid'])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
status = dev[0].get_status()
if status['bssid'] != apdev[0]['bssid']:
raise Exception("Unexpected BSSID")
@ -2353,9 +2281,7 @@ def test_ap_wps_from_event(dev, apdev):
if vals[5] != '4':
raise Exception("Unexpected Device Password Id: " + vals[5])
hapd.request("WPS_PBC")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
def test_ap_wps_ap_scan_2(dev, apdev):
"""AP_SCAN 2 for WPS"""
@ -2377,13 +2303,9 @@ def test_ap_wps_ap_scan_2(dev, apdev):
ev = wpas.wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("WPS-SUCCESS event timed out")
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
wpas.wait_connected(timeout=30)
wpas.request("DISCONNECT")
wpas.request("BSS_FLUSH 0")
wpas.dump_monitor()
wpas.request("REASSOCIATE")
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
wpas.wait_connected(timeout=30)

View file

@ -126,12 +126,8 @@ def test_cfg80211_wep_key_idx_change(dev, apdev):
attrs += build_nl80211_attr_mac('MAC', apdev[0]['bssid'])
attrs += build_nl80211_attr_flag('LOCAL_STATE_CHANGE')
nl80211_command(dev[0], 'DEAUTHENTICATE', attrs)
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Local-deauth timed out")
dev[0].wait_disconnected(timeout=5, error="Local-deauth timed out")
# the previous command results in deauth event followed by auto-reconnect
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Reassociation with the AP timed out")
dev[0].wait_connected(timeout=10, error="Reassociation timed out")
hwsim_utils.test_connectivity(dev[0], hapd)

View file

@ -56,18 +56,14 @@ def test_erp(dev, apdev):
erp="1", scan_freq="2412")
for i in range(3):
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
dev[0].wait_disconnected(timeout=15)
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
if ev is None:
raise Exception("EAP success timed out")
if "EAP re-authentication completed successfully" not in ev:
raise Exception("Did not use ERP")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection timed out")
dev[0].wait_connected(timeout=15, error="Reconnection timed out")
def test_erp_server_no_match(dev, apdev):
"""ERP enabled on server and peer, but server has no key match"""
@ -87,9 +83,7 @@ def test_erp_server_no_match(dev, apdev):
password_hex="0123456789abcdef0123456789abcdef",
erp="1", scan_freq="2412")
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
dev[0].wait_disconnected(timeout=15)
hapd.request("ERP_FLUSH")
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS",
@ -105,9 +99,7 @@ def test_erp_server_no_match(dev, apdev):
raise Exception("EAP success timed out")
if "EAP re-authentication completed successfully" in ev:
raise Exception("Unexpected use of ERP")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection timed out")
dev[0].wait_connected(timeout=15, error="Reconnection timed out")
def start_erp_as(apdev):
params = { "ssid": "as", "beacon_int": "2000",
@ -147,18 +139,14 @@ def test_erp_radius(dev, apdev):
erp="1", scan_freq="2412")
for i in range(3):
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
dev[0].wait_disconnected(timeout=15)
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
if ev is None:
raise Exception("EAP success timed out")
if "EAP re-authentication completed successfully" not in ev:
raise Exception("Did not use ERP")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection timed out")
dev[0].wait_connected(timeout=15, error="Reconnection timed out")
def erp_test(dev, hapd, **kwargs):
hapd.dump_monitor()
@ -167,9 +155,7 @@ def erp_test(dev, hapd, **kwargs):
id = dev.connect("test-wpa2-eap", key_mgmt="WPA-EAP", erp="1",
scan_freq="2412", **kwargs)
dev.request("DISCONNECT")
ev = dev.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Disconnection timed out")
dev.wait_disconnected(timeout=15)
hapd.dump_monitor()
dev.request("RECONNECT")
ev = dev.wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
@ -177,9 +163,7 @@ def erp_test(dev, hapd, **kwargs):
raise Exception("EAP success timed out")
if "EAP re-authentication completed successfully" not in ev:
raise Exception("Did not use ERP")
ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection timed out")
dev.wait_connected(timeout=15, error="Reconnection timed out")
ev = hapd.wait_event([ "AP-STA-CONNECTED" ], timeout=5)
if ev is None:
raise Exception("No connection event received from hostapd")

View file

@ -207,9 +207,7 @@ def test_gas_concurrent_connect(dev, apdev):
get_gas_response(dev[0], bssid, ev)
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection timed out")
dev[0].wait_disconnected(timeout=5)
logger.debug("Wait six seconds for expiration of connect-without-scan")
time.sleep(6)
@ -230,9 +228,7 @@ def test_gas_concurrent_connect(dev, apdev):
if ev is None:
raise Exception("No new scan results reported")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Operation timed out")
ev = dev[0].wait_connected(timeout=20, error="Operation tiemd out")
if "CTRL-EVENT-CONNECTED" not in ev:
raise Exception("Unexpected operation order")

View file

@ -41,21 +41,13 @@ def test_hapd_ctrl_p2p_manager(dev, apdev):
addr = dev[0].p2p_dev_addr()
if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " p2p=2"):
raise Exception("DEAUTHENTICATE command failed")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Re-connection timed out")
dev[0].wait_disconnected(timeout=5)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
if "OK" not in hapd.request("DISASSOCIATE " + addr + " p2p=2"):
raise Exception("DISASSOCIATE command failed")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Re-connection timed out")
dev[0].wait_disconnected(timeout=5)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
def test_hapd_ctrl_sta(dev, apdev):
"""hostapd and STA ctrl_iface commands"""
@ -93,24 +85,16 @@ def test_hapd_ctrl_disconnect(dev, apdev):
if "OK" not in hapd.request("DEAUTHENTICATE ff:ff:ff:ff:ff:ff"):
raise Exception("Unexpected DEAUTHENTICATE failure")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Re-connection timed out")
dev[0].wait_disconnected(timeout=5)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
if "FAIL" not in hapd.request("DISASSOCIATE 00:11:22:33:44"):
raise Exception("Unexpected DISASSOCIATE success")
if "OK" not in hapd.request("DISASSOCIATE ff:ff:ff:ff:ff:ff"):
raise Exception("Unexpected DISASSOCIATE failure")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Re-connection timed out")
dev[0].wait_disconnected(timeout=5)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
def test_hapd_ctrl_chan_switch(dev, apdev):
"""hostapd and CHAN_SWITCH ctrl_iface command"""
@ -237,9 +221,7 @@ def test_hapd_ctrl_set_deny_mac_file(dev, apdev):
dev[1].connect(ssid, key_mgmt="NONE", scan_freq="2412")
if "OK" not in hapd.request("SET deny_mac_file hostapd.macaddr"):
raise Exception("Unexpected SET failure")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], 15)
if ev is None:
raise Exception("Disconnection timeout")
dev[0].wait_disconnected(timeout=15)
ev = dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"], 1)
if ev is not None:
raise Exception("Unexpected disconnection")
@ -254,9 +236,7 @@ def test_hapd_ctrl_set_accept_mac_file(dev, apdev):
hapd.request("SET macaddr_acl 1")
if "OK" not in hapd.request("SET accept_mac_file hostapd.macaddr"):
raise Exception("Unexpected SET failure")
ev = dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"], 15)
if ev is None:
raise Exception("Disconnection timeout")
dev[1].wait_disconnected(timeout=15)
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], 1)
if ev is not None:
raise Exception("Unexpected disconnection")

View file

@ -17,9 +17,8 @@ def connect_ibss_cmd(dev, id):
def wait_ibss_connection(dev):
logger.info(dev.ifname + " waiting for IBSS start/join to complete")
ev = dev.wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Connection to the IBSS timed out")
ev = dev.wait_connected(timeout=20,
error="Connection to the IBSS timed out")
exp = r'<.>(CTRL-EVENT-CONNECTED) - Connection to ([0-9a-f:]*) completed.*'
s = re.split(exp, ev)
if len(s) < 3:

View file

@ -563,9 +563,7 @@ def test_nfc_p2p_go_legacy_config_token(dev):
res = dev[1].request("WPS_NFC_TAG_READ " + conf)
if "FAIL" in res:
raise Exception("Failed to provide NFC tag contents to wpa_supplicant")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Joining the group timed out")
dev[1].wait_connected(timeout=15, error="Joining the group timed out")
hwsim_utils.test_connectivity_p2p(dev[0], dev[1])
dev[1].request("DISCONNECT")
dev[0].remove_group()
@ -588,9 +586,7 @@ def test_nfc_p2p_go_legacy_handover(dev):
res = dev[1].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to wpa_supplicant (legacy STA)")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Joining the group timed out")
dev[1].wait_connected(timeout=15, error="Joining the group timed out")
hwsim_utils.test_connectivity_p2p(dev[0], dev[1])
dev[1].request("DISCONNECT")
dev[0].remove_group()

View file

@ -53,9 +53,7 @@ def test_nfc_wps_password_token_sta(dev, apdev):
res = dev[0].request("WPS_NFC")
if "FAIL" in res:
raise Exception("Failed to start Enrollee using NFC password token")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid)
def test_nfc_wps_config_token(dev, apdev):
@ -72,9 +70,7 @@ def test_nfc_wps_config_token(dev, apdev):
res = dev[0].request("WPS_NFC_TAG_READ " + conf)
if "FAIL" in res:
raise Exception("Failed to provide NFC tag contents to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid)
def test_nfc_wps_config_token_init(dev, apdev):
@ -91,9 +87,7 @@ def test_nfc_wps_config_token_init(dev, apdev):
res = dev[0].request("WPS_NFC_TAG_READ " + conf)
if "FAIL" in res:
raise Exception("Failed to provide NFC tag contents to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid, mixed=True)
def test_nfc_wps_password_token_sta_init(dev, apdev):
@ -113,9 +107,7 @@ def test_nfc_wps_password_token_sta_init(dev, apdev):
res = dev[0].request("WPS_NFC")
if "FAIL" in res:
raise Exception("Failed to start Enrollee using NFC password token")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid, mixed=True)
def test_nfc_wps_password_token_ap(dev, apdev):
@ -140,9 +132,7 @@ def test_nfc_wps_password_token_ap(dev, apdev):
res = dev[0].request("WPS_REG " + apdev[0]['bssid'] + " nfc-pw " + new_ssid.encode("hex") + " WPA2PSK CCMP " + new_passphrase.encode("hex"))
if "FAIL" in res:
raise Exception("Failed to start Registrar using NFC password token")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
check_wpa2_connection(dev[0], apdev[0], hapd, new_ssid, mixed=True)
if "FAIL" in hapd.request("WPS_NFC_TOKEN disable"):
raise Exception("Failed to disable AP password token")
@ -170,9 +160,7 @@ def test_nfc_wps_handover_init(dev, apdev):
res = dev[0].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid, mixed=True)
def test_nfc_wps_handover_errors(dev, apdev):
@ -225,9 +213,7 @@ def test_nfc_wps_handover(dev, apdev):
res = dev[0].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid)
def test_nfc_wps_handover_5ghz(dev, apdev):
@ -253,9 +239,7 @@ def test_nfc_wps_handover_5ghz(dev, apdev):
res = dev[0].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid)
finally:
dev[0].request("DISCONNECT")
@ -287,9 +271,7 @@ def test_nfc_wps_handover_chan14(dev, apdev):
res = dev[0].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid)
finally:
dev[0].request("DISCONNECT")
@ -325,9 +307,7 @@ def test_nfc_wps_handover_with_pw_token_set(dev, apdev):
res = dev[0].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to to wpa_supplicant")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
check_wpa2_connection(dev[0], apdev[0], hapd, ssid)
def test_nfc_wps_handover_pk_hash_mismatch_sta(dev, apdev):
@ -443,9 +423,7 @@ def _test_nfc_wps_er_pw_token(dev, apdev):
ev = dev[0].wait_event(["WPS-SUCCESS"], timeout=15)
if ev is None:
raise Exception("WPS ER did not report success")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=15)
check_wpa2_connection(dev[1], apdev[0], hapd, ssid)
def test_nfc_wps_er_config_token(dev, apdev):
@ -470,9 +448,7 @@ def _test_nfc_wps_er_config_token(dev, apdev):
res = dev[1].request("WPS_NFC_TAG_READ " + conf)
if "FAIL" in res:
raise Exception("Failed to provide NFC tag contents to wpa_supplicant")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=15)
check_wpa2_connection(dev[1], apdev[0], hapd, ssid)
def test_nfc_wps_er_handover(dev, apdev):
@ -500,9 +476,7 @@ def _test_nfc_wps_er_handover(dev, apdev):
res = dev[1].request("NFC_REPORT_HANDOVER INIT WPS " + req + " " + sel)
if "FAIL" in res:
raise Exception("Failed to report NFC connection handover to to wpa_supplicant")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=15)
check_wpa2_connection(dev[1], apdev[0], hapd, ssid)
def test_nfc_wps_er_handover_pk_hash_mismatch_sta(dev, apdev):

View file

@ -313,9 +313,7 @@ def test_autogo_legacy(dev):
dev[2].request("P2P_SET disabled 1")
dev[2].dump_monitor()
dev[2].request("WPS_PIN any " + pin)
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the GO timed out")
dev[2].wait_connected(timeout=30)
status = dev[2].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")
@ -412,9 +410,7 @@ def test_autogo_passphrase_len(dev):
dev[2].request("P2P_SET disabled 1")
dev[2].dump_monitor()
dev[2].request("WPS_PIN any " + pin)
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the GO timed out")
dev[2].wait_connected(timeout=30)
status = dev[2].get_status()
if status['wpa_state'] != 'COMPLETED':
raise Exception("Not fully connected")

View file

@ -130,9 +130,7 @@ def test_discovery_group_client(dev):
# make group client non-responsive on operating channel
dev[1].dump_monitor()
dev[1].group_request("DISCONNECT")
ev = dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout on waiting disconnection")
dev[1].wait_disconnected(timeout=10)
dev[2].request("P2P_CONNECT {} {} display".format(dev[1].p2p_dev_addr(),
pin))
ev = dev[1].wait_event(["P2P-GO-NEG-REQUEST"], timeout=2)

View file

@ -359,9 +359,7 @@ def test_grpform_per_sta_psk_wps(dev):
dev[0].p2p_go_authorize_client_pbc()
dev[2].request("WPS_PBC")
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the GO timed out")
dev[2].wait_connected(timeout=30)
hwsim_utils.test_connectivity_p2p_sta(dev[1], dev[2])

View file

@ -61,9 +61,7 @@ def test_p2p_set_discoverability(dev):
raise Exception("P2P_SET discoverability 1 failed")
dev[1].dump_monitor()
dev[1].group_request("REASSOCIATE")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Reassociation with the GO timed out")
dev[1].wait_connected(timeout=20)
dev[2].request("P2P_FLUSH")
if not dev[2].discover_peer(addr1, timeout=10):

View file

@ -46,9 +46,7 @@ def test_pmksa_cache_on_roam_back(dev, apdev):
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
dev[0].wait_connected(timeout=10, error="Roaming timed out")
pmksa2 = dev[0].get_pmksa(bssid2)
if pmksa2 is None:
raise Exception("No PMKSA cache entry found")
@ -76,12 +74,8 @@ def test_pmksa_cache_on_roam_back(dev, apdev):
raise Exception("PMKSA_FLUSH failed")
if dev[0].get_pmksa(bssid) is not None or dev[0].get_pmksa(bssid2) is not None:
raise Exception("PMKSA_FLUSH did not remove PMKSA entries")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection timed out")
dev[0].wait_disconnected(timeout=5)
dev[0].wait_connected(timeout=15, error="Reconnection timed out")
def test_pmksa_cache_opportunistic_only_on_sta(dev, apdev):
"""Opportunistic PMKSA caching enabled only on station"""
@ -108,9 +102,7 @@ def test_pmksa_cache_opportunistic_only_on_sta(dev, apdev):
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
dev[0].wait_connected(timeout=10, error="Roaming timed out")
pmksa2 = dev[0].get_pmksa(bssid2)
if pmksa2 is None:
raise Exception("No PMKSA cache entry found")
@ -410,9 +402,7 @@ def test_pmksa_cache_disabled(dev, apdev):
ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
dev[0].wait_connected(timeout=10, error="Roaming timed out")
dev[0].dump_monitor()
logger.info("Roam back to AP1")
@ -447,16 +437,10 @@ def test_pmksa_cache_ap_expiration(dev, apdev):
raise Exception("Roaming with the AP timed out")
if "CTRL-EVENT-CONNECTED" in ev:
raise Exception("EAP exchange missing")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Reassociation with the AP timed out")
dev[0].wait_connected(timeout=20, error="Reconnect timed out")
dev[0].dump_monitor()
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=20)
if ev is None:
raise Exception("Disconnection event timed out")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Reassociation with the AP timed out")
dev[0].wait_disconnected(timeout=20)
dev[0].wait_connected(timeout=20, error="Reassociation timed out")
def test_pmksa_cache_multiple_sta(dev, apdev):
"""PMKSA cache with multiple stations"""
@ -494,9 +478,7 @@ def test_pmksa_cache_multiple_sta(dev, apdev):
ev = sta.wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = sta.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
sta.wait_connected(timeout=10, error="Roaming timed out")
logger.info("Roam back to AP1")
for sta in [ dev[1], wpas, dev[0], dev[2] ]:
@ -504,9 +486,7 @@ def test_pmksa_cache_multiple_sta(dev, apdev):
sta.scan(freq="2412")
sta.dump_monitor()
sta.request("ROAM " + bssid)
ev = sta.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
sta.wait_connected(timeout=10, error="Roaming timed out")
sta.dump_monitor()
time.sleep(4)
@ -517,9 +497,7 @@ def test_pmksa_cache_multiple_sta(dev, apdev):
sta.scan(freq="2412")
sta.dump_monitor()
sta.request("ROAM " + bssid2)
ev = sta.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
sta.wait_connected(timeout=10, error="Roaming timed out")
sta.dump_monitor()
def test_pmksa_cache_opportunistic_multiple_sta(dev, apdev):

View file

@ -164,9 +164,7 @@ def test_radius_acct_pmksa_caching(dev, apdev):
password_hex="0123456789abcdef0123456789abcdef")
for d in [ dev[0], dev[1] ]:
d.request("REASSOCIATE")
ev = d.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reassociation timed out")
d.wait_connected(timeout=15, error="Reassociation timed out")
count = 0
while True:
@ -423,12 +421,8 @@ def test_radius_das_disconnect(dev, apdev):
if reply.code != pyrad.packet.DisconnectACK:
raise Exception("Unexpected response code")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for disconnection")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for re-connection")
dev[0].wait_disconnected(timeout=10)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching User-Name")
req = radius_das.DisconnectPacket(dict=dict, secret="secret",
@ -442,12 +436,8 @@ def test_radius_das_disconnect(dev, apdev):
if reply.code != pyrad.packet.DisconnectACK:
raise Exception("Unexpected response code")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for disconnection")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for re-connection")
dev[0].wait_disconnected(timeout=10)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching Calling-Station-Id")
req = radius_das.DisconnectPacket(dict=dict, secret="secret",
@ -461,17 +451,13 @@ def test_radius_das_disconnect(dev, apdev):
if reply.code != pyrad.packet.DisconnectACK:
raise Exception("Unexpected response code")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for disconnection")
dev[0].wait_disconnected(timeout=10)
ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED", "CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for re-connection")
if "CTRL-EVENT-EAP-STARTED" not in ev:
raise Exception("Unexpected skipping of EAP authentication in reconnection")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for re-connection to complete")
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching Calling-Station-Id and non-matching CUI")
req = radius_das.DisconnectPacket(dict=dict, secret="secret",
@ -485,12 +471,8 @@ def test_radius_das_disconnect(dev, apdev):
if reply.code != pyrad.packet.DisconnectACK:
raise Exception("Unexpected response code")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for disconnection")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for re-connection")
dev[0].wait_disconnected(timeout=10)
dev[0].wait_connected(timeout=10, error="Re-connection timed out")
logger.info("Disconnect-Request with matching CUI")
dev[1].connect("radius-das", key_mgmt="WPA-EAP",
@ -507,12 +489,8 @@ def test_radius_das_disconnect(dev, apdev):
if reply.code != pyrad.packet.DisconnectACK:
raise Exception("Unexpected response code")
ev = dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for disconnection")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for re-connection")
dev[1].wait_disconnected(timeout=10)
dev[1].wait_connected(timeout=10, error="Re-connection timed out")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
if ev is not None:
@ -627,9 +605,7 @@ def test_radius_failover(dev, apdev):
'192.168.213.17'])
dev[0].request("SET EAPOL::authPeriod 5")
connect(dev[0], "radius-failover", wait_connect=False)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=60)
if ev is None:
raise Exception("Connection with the AP timed out")
dev[0].wait_connected(timeout=60)
finally:
dev[0].request("SET EAPOL::authPeriod 30")
subprocess.call(['sudo', 'ip', 'ro', 'del', '192.168.213.17'])

View file

@ -38,9 +38,8 @@ def test_rfkill_open(dev, apdev):
try:
logger.info("rfkill block")
subprocess.call(['sudo', 'rfkill', 'block', id])
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Missing disconnection event on rfkill block")
dev[0].wait_disconnected(timeout=10,
error="Missing disconnection event on rfkill block")
if "FAIL" not in dev[0].request("REASSOCIATE"):
raise Exception("REASSOCIATE accepted while disabled")
@ -53,9 +52,8 @@ def test_rfkill_open(dev, apdev):
logger.info("rfkill unblock")
subprocess.call(['sudo', 'rfkill', 'unblock', id])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Missing connection event on rfkill unblock")
dev[0].wait_connected(timeout=10,
error="Missing connection event on rfkill unblock")
hwsim_utils.test_connectivity(dev[0], hapd)
finally:
subprocess.call(['sudo', 'rfkill', 'unblock', id])
@ -74,15 +72,13 @@ def test_rfkill_wpa2_psk(dev, apdev):
try:
logger.info("rfkill block")
subprocess.call(['sudo', 'rfkill', 'block', id])
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Missing disconnection event on rfkill block")
dev[0].wait_disconnected(timeout=10,
error="Missing disconnection event on rfkill block")
logger.info("rfkill unblock")
subprocess.call(['sudo', 'rfkill', 'unblock', id])
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Missing connection event on rfkill unblock")
dev[0].wait_connected(timeout=10,
error="Missing connection event on rfkill unblock")
hwsim_utils.test_connectivity(dev[0], hapd)
finally:
subprocess.call(['sudo', 'rfkill', 'unblock', id])
@ -155,9 +151,7 @@ def test_rfkill_hostapd(dev, apdev):
ev = hapd.wait_event(["INTERFACE-DISABLED"], timeout=5)
if ev is None:
raise Exception("INTERFACE-DISABLED event not seen")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Missing disconnection event")
dev[0].wait_disconnected(timeout=10)
dev[0].request("DISCONNECT")
hapd.disable()

View file

@ -45,9 +45,7 @@ def test_sae_pmksa_caching(dev, apdev):
scan_freq="2412")
dev[0].request("DISCONNECT")
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnect timed out")
dev[0].wait_connected(timeout=15, error="Reconnect timed out")
if dev[0].get_status_field('sae_group') is not None:
raise Exception("SAE group claimed to have been used")
@ -64,9 +62,7 @@ def test_sae_pmksa_caching_disabled(dev, apdev):
scan_freq="2412")
dev[0].request("DISCONNECT")
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnect timed out")
dev[0].wait_connected(timeout=15, error="Reconnect timed out")
if dev[0].get_status_field('sae_group') != '19':
raise Exception("Expected default SAE group not used")
@ -124,9 +120,7 @@ def test_sae_anti_clogging(dev, apdev):
for i in range(0, 2):
dev[i].select_network(id[i])
for i in range(0, 2):
ev = dev[i].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Association with the AP timed out")
dev[i].wait_connected(timeout=10)
def test_sae_forced_anti_clogging(dev, apdev):
"""SAE anti clogging (forced)"""

View file

@ -341,9 +341,7 @@ def test_scan_for_auth(dev, apdev):
# cfg80211 BSS entry missing.
dev[0].request("RADIO_WORK done " + id)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=15)
def test_scan_hidden(dev, apdev):
"""Control interface behavior on scan parameters"""
@ -451,12 +449,8 @@ def test_scan_and_bss_entry_removed(dev, apdev):
dev[0].request("RADIO_WORK done " + id)
wpas.request("RADIO_WORK done " + w_id)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("No connection (sme-connect)")
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("No connection (connect)")
dev[0].wait_connected(timeout=15, error="No connection (sme-connect)")
wpas.wait_connected(timeout=15, error="No connection (connect)")
def test_scan_reqs_with_non_scan_radio_work(dev, apdev):
"""SCAN commands while non-scan radio_work is in progress"""
@ -594,6 +588,4 @@ def _test_scan_hidden_many(dev, apdev):
dev[0].set_network(id, "scan_ssid", "1")
dev[0].request("REASSOCIATE")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=30)

View file

@ -42,13 +42,9 @@ def test_sta_ap_scan_0(dev, apdev):
wpas.request("SCAN")
time.sleep(0.5)
subprocess.call(['sudo', 'iw', wpas.ifname, 'connect', 'test', '2412'])
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection not reported")
wpas.wait_connected(timeout=10)
wpas.request("SCAN")
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=5)
if ev is None:
raise Exception("Connection not reported")
wpas.wait_connected(timeout=5)
def test_sta_ap_scan_2(dev, apdev):
"""Dynamically added wpa_supplicant interface with AP_SCAN 2 connection"""
@ -73,14 +69,10 @@ def test_sta_ap_scan_2(dev, apdev):
'freq', '2412'])
time.sleep(1)
subprocess.call(['sudo', 'iw', wpas.ifname, 'connect', 'test', '2412'])
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection not reported")
wpas.wait_connected(timeout=10)
wpas.request("SET disallow_aps bssid " + bssid)
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection not reported")
wpas.wait_disconnected(timeout=10)
subprocess.call(['sudo', 'iw', wpas.ifname, 'connect', 'test', '2412'])
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=1)
@ -130,15 +122,11 @@ def test_sta_dynamic_down_up(dev, apdev):
wpas.connect("sta-dynamic", psk="12345678", scan_freq="2412")
hwsim_utils.test_connectivity(wpas, hapd)
subprocess.call(['sudo', 'ifconfig', wpas.ifname, 'down'])
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection not reported")
wpas.wait_disconnected(timeout=10)
if wpas.get_status_field("wpa_state") != "INTERFACE_DISABLED":
raise Exception("Unexpected wpa_state")
subprocess.call(['sudo', 'ifconfig', wpas.ifname, 'up'])
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection not reported")
wpas.wait_connected(timeout=15, error="Reconnection not reported")
hwsim_utils.test_connectivity(wpas, hapd)
def test_sta_dynamic_ext_mac_addr_change(dev, apdev):
@ -152,9 +140,7 @@ def test_sta_dynamic_ext_mac_addr_change(dev, apdev):
wpas.connect("sta-dynamic", psk="12345678", scan_freq="2412")
hwsim_utils.test_connectivity(wpas, hapd)
subprocess.call(['sudo', 'ifconfig', wpas.ifname, 'down'])
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection not reported")
wpas.wait_disconnected(timeout=10)
if wpas.get_status_field("wpa_state") != "INTERFACE_DISABLED":
raise Exception("Unexpected wpa_state")
prev_addr = wpas.p2p_interface_addr()
@ -163,9 +149,7 @@ def test_sta_dynamic_ext_mac_addr_change(dev, apdev):
subprocess.call(['sudo', 'ip', 'link', 'set', 'dev', wpas.ifname,
'address', new_addr])
subprocess.call(['sudo', 'ifconfig', wpas.ifname, 'up'])
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection not reported")
wpas.wait_connected(timeout=15, error="Reconnection not reported")
if wpas.get_driver_status_field('addr') != new_addr:
raise Exception("Address change not reported")
hwsim_utils.test_connectivity(wpas, hapd)

View file

@ -32,9 +32,7 @@ def test_suite_b(dev, apdev):
raise Exception("Unexpected BSS flags: " + bss['flags'])
dev[0].request("DISCONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=20)
if ev is None:
raise Exception("Disconnection event timed out")
dev[0].wait_disconnected(timeout=20)
dev[0].dump_monitor()
dev[0].request("RECONNECT")
ev = dev[0].wait_event(["CTRL-EVENT-EAP-STARTED",

View file

@ -22,9 +22,7 @@ def test_tnc_peap_soh(dev, apdev):
phase1="peapver=0 tnc=soh cryptobinding=0",
phase2="auth=MSCHAPV2",
wait_connect=False)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[0].wait_connected(timeout=10)
dev[1].connect("test-wpa2-eap", key_mgmt="WPA-EAP",
eap="PEAP", identity="user", password="password",
@ -32,9 +30,7 @@ def test_tnc_peap_soh(dev, apdev):
phase1="peapver=0 tnc=soh1 cryptobinding=1",
phase2="auth=MSCHAPV2",
wait_connect=False)
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[1].wait_connected(timeout=10)
dev[2].connect("test-wpa2-eap", key_mgmt="WPA-EAP",
eap="PEAP", identity="user", password="password",
@ -42,9 +38,7 @@ def test_tnc_peap_soh(dev, apdev):
phase1="peapver=0 tnc=soh2 cryptobinding=2",
phase2="auth=MSCHAPV2",
wait_connect=False)
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[2].wait_connected(timeout=10)
def test_tnc_ttls(dev, apdev):
"""TNC TTLS"""
@ -62,9 +56,7 @@ def test_tnc_ttls(dev, apdev):
phase2="auth=MSCHAPV2",
ca_cert="auth_serv/ca.pem",
wait_connect=False)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[0].wait_connected(timeout=10)
def test_tnc_fast(dev, apdev):
"""TNC FAST"""
@ -88,6 +80,4 @@ def test_tnc_fast(dev, apdev):
pac_file="blob://fast_pac_auth_tnc",
ca_cert="auth_serv/ca.pem",
wait_connect=False)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Connection timed out")
dev[0].wait_connected(timeout=10)

View file

@ -78,7 +78,5 @@ def test_wep_shared_key_auth_multi_key(dev, apdev):
dev[2].set_network(id, "wep_tx_keyidx", "1")
dev[2].request("REASSOCIATE")
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Reassociation with the AP timed out")
dev[2].wait_connected(timeout=10, error="Reassociation timed out")
hwsim_utils.test_connectivity(dev[2], hapd)

View file

@ -55,9 +55,7 @@ def test_wext_wpa2_psk(dev, apdev):
wpas.dump_monitor()
hapd.request("DEAUTHENTICATE " + wpas.p2p_interface_addr())
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=15)
if ev is None:
raise Exception("Timeout on disconnection event")
wpas.wait_disconnected(timeout=15)
def test_wext_wpa_psk(dev, apdev):
"""WEXT driver interface with WPA-PSK"""
@ -82,9 +80,8 @@ def test_wext_wpa_psk(dev, apdev):
with open(testfile, "w") as f:
f.write("ff:ff:ff:ff:ff:ff")
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("No disconnection after two Michael MIC failures")
ev = wpas.wait_disconnected(timeout=10,
error="No disconnection after two Michael MIC failures")
if "reason=14 locally_generated=1" not in ev:
raise Exception("Unexpected disconnection reason: " + ev)
@ -125,9 +122,7 @@ def test_wext_pmksa_cache(dev, apdev):
ev = wpas.wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=10)
if ev is None:
raise Exception("EAP success timed out")
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
wpas.wait_connected(timeout=10, error="Roaming timed out")
pmksa2 = wpas.get_pmksa(bssid2)
if pmksa2 is None:
raise Exception("No PMKSA cache entry found")
@ -155,12 +150,8 @@ def test_wext_pmksa_cache(dev, apdev):
raise Exception("PMKSA_FLUSH failed")
if wpas.get_pmksa(bssid) is not None or wpas.get_pmksa(bssid2) is not None:
raise Exception("PMKSA_FLUSH did not remove PMKSA entries")
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reconnection timed out")
wpas.wait_disconnected(timeout=5)
wpas.wait_connected(timeout=15, error="Reconnection timed out")
def test_wext_wep_open_auth(dev, apdev):
"""WEP Open System authentication"""
@ -191,9 +182,7 @@ def test_wext_wep_shared_key_auth(dev, apdev):
wep_key0='"hello12345678"', scan_freq="2412")
hwsim_utils.test_connectivity(wpas, hapd)
wpas.request("REMOVE_NETWORK all")
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection event timed out")
wpas.wait_disconnected(timeout=5)
wpas.connect("wep-shared-key", key_mgmt="NONE", auth_alg="OPEN SHARED",
wep_key0='"hello12345678"', scan_freq="2412")
@ -215,9 +204,7 @@ def test_wext_pmf(dev, apdev):
addr = wpas.p2p_interface_addr()
hapd.request("DEAUTHENTICATE " + addr)
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("No disconnection")
wpas.wait_disconnected(timeout=5)
def test_wext_scan_hidden(dev, apdev):
"""WEXT with hidden SSID"""
@ -264,15 +251,13 @@ def test_wext_rfkill(dev, apdev):
try:
logger.info("rfkill block")
subprocess.call(['rfkill', 'block', id])
ev = wpas.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Missing disconnection event on rfkill block")
wpas.wait_disconnected(timeout=10,
error="Missing disconnection event on rfkill block")
logger.info("rfkill unblock")
subprocess.call(['rfkill', 'unblock', id])
ev = wpas.wait_event(["CTRL-EVENT-CONNECTED"], timeout=20)
if ev is None:
raise Exception("Missing connection event on rfkill unblock")
wpas.wait_connected(timeout=20,
error="Missing connection event on rfkill unblock")
hwsim_utils.test_connectivity(wpas, hapd)
finally:
subprocess.call(['rfkill', 'unblock', id])

View file

@ -466,9 +466,7 @@ def test_wnm_bss_tm(dev, apdev):
raise Exception("BSS transition request was not accepted: " + ev)
if "target_bssid=" + apdev[1]['bssid'] not in ev:
raise Exception("Unexpected target BSS: " + ev)
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("No reassociation seen");
dev[0].wait_connected(timeout=15, error="No reassociation seen")
if apdev[1]['bssid'] not in ev:
raise Exception("Unexpected reassociation target: " + ev)
ev = dev[0].wait_event(["CTRL-EVENT-SCAN-STARTED"], timeout=0.1)

View file

@ -67,18 +67,10 @@ def test_wpas_ap_open(dev):
dev[2].dump_monitor()
dev[0].request("DEAUTHENTICATE " + addr1)
dev[0].request("DISASSOCIATE " + addr2)
ev = dev[1].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Disconnection timed out")
ev = dev[2].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Disconnection timed out")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Reconnection timed out")
ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED"])
if ev is None:
raise Exception("Reconnection timed out")
dev[1].wait_disconnected(timeout=10)
dev[2].wait_disconnected(timeout=10)
dev[1].wait_connected(timeout=10, error="Reconnection timed out")
dev[2].wait_connected(timeout=10, error="Reconnection timed out")
dev[1].request("DISCONNECT")
dev[2].request("DISCONNECT")
@ -167,9 +159,7 @@ def test_wpas_ap_wps(dev):
if ev is None:
raise Exception("PBC mode start timeout")
dev[1].request("WPS_PBC")
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("WPS PBC operation timed out")
dev[1].wait_connected(timeout=30, error="WPS PBC operation timed out")
hwsim_utils.test_connectivity(dev[0], dev[1])
logger.info("Test AP PIN to learn configuration")
@ -194,17 +184,13 @@ def test_wpas_ap_wps(dev):
pin = dev[1].wps_read_pin()
dev[0].request("WPS_PIN any " + pin)
dev[1].request("WPS_PIN any " + pin)
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=30)
dev[1].request("REMOVE_NETWORK all")
dev[1].dump_monitor()
dev[0].request("WPS_PIN any " + pin + " 100")
dev[1].request("WPS_PIN any " + pin)
ev = dev[1].wait_event(["CTRL-EVENT-CONNECTED"], timeout=30)
if ev is None:
raise Exception("Association with the AP timed out")
dev[1].wait_connected(timeout=30)
dev[1].request("REMOVE_NETWORK all")
dev[1].dump_monitor()
@ -221,9 +207,7 @@ def test_wpas_ap_wps(dev):
raise Exception("WPS operation timed out")
if "WPS-SUCCESS" in ev:
raise Exception("WPS operation succeeded unexpectedly")
ev = dev[2].wait_event(["CTRL-EVENT-DISCONNECTED"])
if ev is None:
raise Exception("Timeout while waiting for disconnection")
dev[2].wait_disconnected(timeout=10)
dev[2].request("WPS_CANCEL")
dev[2].request("REMOVE_NETWORK all")
ev = dev[0].wait_event(["WPS-AP-SETUP-LOCKED"])

View file

@ -674,18 +674,14 @@ def test_wpas_ctrl_disallow_aps(dev, apdev):
raise Exception("Failed to set disallow_aps")
if "OK" not in dev[0].request("SET disallow_aps bssid " + apdev[0]['bssid']):
raise Exception("Failed to set disallow_aps")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Reassociation timed out")
ev = dev[0].wait_connected(timeout=15, error="Reassociation timed out")
if apdev[1]['bssid'] not in ev:
raise Exception("Unexpected BSSID")
dev[0].dump_monitor()
if "OK" not in dev[0].request("SET disallow_aps ssid " + "test".encode("hex")):
raise Exception("Failed to set disallow_aps")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
if ev is None:
raise Exception("Disconnection not seen")
dev[0].wait_disconnected(timeout=5, error="Disconnection not seen")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=1)
if ev is not None:
raise Exception("Unexpected reassociation")
@ -945,26 +941,18 @@ def test_wpas_ctrl_enable_disable_network(dev, apdev):
raise Exception("Failed to disable networks")
if "OK" not in dev[0].request("ENABLE_NETWORK " + str(id)):
raise Exception("Failed to enable network")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=10)
if "OK" not in dev[0].request("DISABLE_NETWORK " + str(id)):
raise Exception("Failed to disable network")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection with the AP timed out")
dev[0].wait_disconnected(timeout=10)
time.sleep(0.1)
if "OK" not in dev[0].request("ENABLE_NETWORK all"):
raise Exception("Failed to enable network")
ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Association with the AP timed out")
dev[0].wait_connected(timeout=10)
if "OK" not in dev[0].request("DISABLE_NETWORK all"):
raise Exception("Failed to disable network")
ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Disconnection with the AP timed out")
dev[0].wait_disconnected(timeout=10)
def test_wpas_ctrl_country(dev, apdev):
"""wpa_supplicant SET/GET country code"""

View file

@ -269,9 +269,7 @@ class WpaSupplicant:
def connect_network(self, id, timeout=10):
self.dump_monitor()
self.select_network(id)
ev = self.wait_event(["CTRL-EVENT-CONNECTED"], timeout=timeout)
if ev is None:
raise Exception("Association with the AP timed out")
self.wait_connected(timeout=timeout)
self.dump_monitor()
def get_status(self, extra=None):
@ -861,9 +859,7 @@ class WpaSupplicant:
raise Exception("Unexpected connection")
self.dump_monitor()
return
ev = self.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
self.wait_connected(timeout=10, error="Roaming with the AP timed out")
self.dump_monitor()
def roam_over_ds(self, bssid, fail_test=False):
@ -876,9 +872,7 @@ class WpaSupplicant:
raise Exception("Unexpected connection")
self.dump_monitor()
return
ev = self.wait_event(["CTRL-EVENT-CONNECTED"], timeout=10)
if ev is None:
raise Exception("Roaming with the AP timed out")
self.wait_connected(timeout=10, error="Roaming with the AP timed out")
self.dump_monitor()
def wps_reg(self, bssid, pin, new_ssid=None, key_mgmt=None, cipher=None,
@ -901,9 +895,7 @@ class WpaSupplicant:
ev = self.wait_event(["WPS-FAIL"], timeout=15)
if ev is None:
raise Exception("WPS timed out")
ev = self.wait_event(["CTRL-EVENT-CONNECTED"], timeout=15)
if ev is None:
raise Exception("Association with the AP timed out")
self.wait_connected(timeout=15)
def relog(self):
self.global_request("RELOG")
@ -998,3 +990,15 @@ class WpaSupplicant:
msg['payload'] = frame[24:]
return msg
def wait_connected(self, timeout=10, error="Connection timed out"):
ev = self.wait_event(["CTRL-EVENT-CONNECTED"], timeout=timeout)
if ev is None:
raise Exception(error)
return ev
def wait_disconnected(self, timeout=10, error="Disconnection timed out"):
ev = self.wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=timeout)
if ev is None:
raise Exception(error)
return ev