Crash_Override/OVMS3-idf
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge branch 'bugfix/prov_retry_failures_v4.0' into 'release/v4.0'

Browse source 
ci: Add retries for all provisioning methods, some refactors (v4.0)

See merge request espressif/esp-idf!9389
This commit is contained in:
Angus Gratton 2020-07-23 13:17:21 +08:00
parent e5526e2953 d7f251aef3
commit 8cfc9f1609
11 changed files with 103 additions and 104 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
examples/provisioning/ble_prov/ble_prov_test.py
View file

@ -17,7 +17,6 @@
from __future__ import print_function
import re
import os
import time
import ttfw_idf
import esp_prov
@ -82,19 +81,7 @@ def test_examples_provisioning_ble(env, extra_data):
if not esp_prov.apply_wifi_config(transport, security):
raise RuntimeError("Failed to send apply config")
success = False
while True:
time.sleep(5)
print("Wi-Fi connection state")
ret = esp_prov.get_wifi_config(transport, security)
if (ret == 1):
continue
elif (ret == 0):
print("Provisioning was successful")
success = True
break
if not success:
if not esp_prov.wait_wifi_connected(transport, security):
raise RuntimeError("Provisioning failed")

2
examples/provisioning/ble_prov/main/app_prov.c
View file

@ -248,8 +248,8 @@ static void app_prov_event_handler(void* handler_arg, esp_event_base_t event_bas
/* If none of the expected reasons,
* retry connecting to host SSID */
g_prov->wifi_state = WIFI_PROV_STA_CONNECTING;
esp_wifi_connect();
}
esp_wifi_connect();
}
}

2
examples/provisioning/console_prov/main/app_prov.c
View file

@ -186,8 +186,8 @@ static void app_prov_event_handler(void* handler_arg, esp_event_base_t event_bas
/* If none of the expected reasons,
* retry connecting to host SSID */
g_prov->wifi_state = WIFI_PROV_STA_CONNECTING;
esp_wifi_connect();
}
esp_wifi_connect();
}
}

2
examples/provisioning/custom_config/main/app_prov.c
View file

@ -211,8 +211,8 @@ static void app_prov_event_handler(void* handler_arg, esp_event_base_t event_bas
/* If none of the expected reasons,
* retry connecting to host SSID */
g_prov->wifi_state = WIFI_PROV_STA_CONNECTING;
esp_wifi_connect();
}
esp_wifi_connect();
}
}

20
examples/provisioning/manager/wifi_prov_mgr_test.py
View file

@ -17,7 +17,6 @@
from __future__ import print_function
import re
import os
import time
import ttfw_idf
import esp_prov
@ -86,24 +85,7 @@ def test_examples_wifi_prov_mgr(env, extra_data):
if not esp_prov.apply_wifi_config(transport, security):
raise RuntimeError("Failed to send apply config")
success = False
retry = 0
while True:
time.sleep(5)
print("Wi-Fi connection state")
ret = esp_prov.get_wifi_config(transport, security)
if (ret == 1):
continue
elif (ret == 0):
print("Provisioning was successful")
success = True
elif (ret == 3 and retry < 3):
retry = retry + 1
print("Connection failed.. retry again...: ", ret)
continue
break
if not success:
if not esp_prov.wait_wifi_connected(transport, security):
raise RuntimeError("Provisioning failed")
# Check if BTDM memory is released after provisioning finishes

2
examples/provisioning/softap_prov/main/app_prov.c
View file

@ -197,8 +197,8 @@ static void app_prov_event_handler(void* handler_arg, esp_event_base_t event_bas
/* If none of the expected reasons,
* retry connecting to host SSID */
g_prov->wifi_state = WIFI_PROV_STA_CONNECTING;
esp_wifi_connect();
}
esp_wifi_connect();
}
}

92
examples/provisioning/softap_prov/softap_prov_test.py
View file

@ -17,7 +17,6 @@
from __future__ import print_function
import re
import os
import time
import ttfw_idf
import esp_prov
@ -52,64 +51,55 @@ def test_examples_provisioning_softap(env, extra_data):
print("SoftAP SSID : " + ssid)
print("SoftAP Password : " + password)
ctrl = wifi_tools.wpa_cli(iface, reset_on_exit=True)
print("Connecting to DUT SoftAP...")
ip = ctrl.connect(ssid, password)
got_ip = dut1.expect(re.compile(r"softAP assign IP to station,IP is: (\d+.\d+.\d+.\d+)"), timeout=30)[0]
if ip != got_ip:
raise RuntimeError("SoftAP connected to another host! " + ip + "!=" + got_ip)
print("Connected to DUT SoftAP")
try:
ctrl = wifi_tools.wpa_cli(iface, reset_on_exit=True)
print("Connecting to DUT SoftAP...")
ip = ctrl.connect(ssid, password)
got_ip = dut1.expect(re.compile(r"softAP assign IP to station,IP is: (\d+.\d+.\d+.\d+)"), timeout=60)[0]
if ip != got_ip:
raise RuntimeError("SoftAP connected to another host! " + ip + "!=" + got_ip)
print("Connected to DUT SoftAP")
print("Starting Provisioning")
verbose = False
protover = "V0.1"
secver = 1
pop = "abcd1234"
provmode = "softap"
ap_ssid = "myssid"
ap_password = "mypassword"
softap_endpoint = ip.split('.')[0] + "." + ip.split('.')[1] + "." + ip.split('.')[2] + ".1:80"
print("Starting Provisioning")
verbose = False
protover = "V0.1"
secver = 1
pop = "abcd1234"
provmode = "softap"
ap_ssid = "myssid"
ap_password = "mypassword"
softap_endpoint = ip.split('.')[0] + "." + ip.split('.')[1] + "." + ip.split('.')[2] + ".1:80"
print("Getting security")
security = esp_prov.get_security(secver, pop, verbose)
if security is None:
raise RuntimeError("Failed to get security")
print("Getting security")
security = esp_prov.get_security(secver, pop, verbose)
if security is None:
raise RuntimeError("Failed to get security")
print("Getting transport")
transport = esp_prov.get_transport(provmode, softap_endpoint)
if transport is None:
raise RuntimeError("Failed to get transport")
print("Getting transport")
transport = esp_prov.get_transport(provmode, softap_endpoint)
if transport is None:
raise RuntimeError("Failed to get transport")
print("Verifying protocol version")
if not esp_prov.version_match(transport, protover):
raise RuntimeError("Mismatch in protocol version")
print("Verifying protocol version")
if not esp_prov.version_match(transport, protover):
raise RuntimeError("Mismatch in protocol version")
print("Starting Session")
if not esp_prov.establish_session(transport, security):
raise RuntimeError("Failed to start session")
print("Starting Session")
if not esp_prov.establish_session(transport, security):
raise RuntimeError("Failed to start session")
print("Sending Wifi credential to DUT")
if not esp_prov.send_wifi_config(transport, security, ap_ssid, ap_password):
raise RuntimeError("Failed to send Wi-Fi config")
print("Sending Wifi credential to DUT")
if not esp_prov.send_wifi_config(transport, security, ap_ssid, ap_password):
raise RuntimeError("Failed to send Wi-Fi config")
print("Applying config")
if not esp_prov.apply_wifi_config(transport, security):
raise RuntimeError("Failed to send apply config")
print("Applying config")
if not esp_prov.apply_wifi_config(transport, security):
raise RuntimeError("Failed to send apply config")
success = False
while True:
time.sleep(5)
print("Wi-Fi connection state")
ret = esp_prov.get_wifi_config(transport, security)
if (ret == 1):
continue
elif (ret == 0):
print("Provisioning was successful")
success = True
break
if not success:
raise RuntimeError("Provisioning failed")
if not esp_prov.wait_wifi_connected(transport, security):
raise RuntimeError("Provisioning failed")
finally:
ctrl.reset()
if __name__ == '__main__':

23
tools/ci/python_packages/wifi_tools.py
View file

@ -51,16 +51,26 @@ class wpa_cli:
iface_path = service.GetInterface(self.iface_name)
self.iface_obj = bus.get_object("fi.w1.wpa_supplicant1", iface_path)
self.iface_ifc = dbus.Interface(self.iface_obj, "fi.w1.wpa_supplicant1.Interface")
self.iface_props = dbus.Interface(self.iface_obj, 'org.freedesktop.DBus.Properties')
if self.iface_ifc is None:
raise RuntimeError('supplicant : Failed to fetch interface')
self.old_network = self.iface_obj.Get("fi.w1.wpa_supplicant1.Interface", "CurrentNetwork",
dbus_interface='org.freedesktop.DBus.Properties')
self.old_network = self._get_iface_property("CurrentNetwork")
print("Old network is %s" % self.old_network)
if self.old_network == '/':
self.old_network = None
else:
self.connected = True
def _get_iface_property(self, name):
""" Read the property with 'name' from the wi-fi interface object
Note: The result is a dbus wrapped type, so should usually convert it to the corresponding native
Python type
"""
return self.iface_props.Get("fi.w1.wpa_supplicant1.Interface", name)
def connect(self, ssid, password):
if self.connected is True:
self.iface_ifc.Disconnect()
@ -69,6 +79,8 @@ class wpa_cli:
if self.new_network is not None:
self.iface_ifc.RemoveNetwork(self.new_network)
print("Pre-connect state is %s, IP is %s" % (self._get_iface_property("State"), get_wiface_IPv4(self.iface_name)))
self.new_network = self.iface_ifc.AddNetwork({"ssid": ssid, "psk": password})
self.iface_ifc.SelectNetwork(self.new_network)
@ -76,7 +88,14 @@ class wpa_cli:
retry = 10
while retry > 0:
time.sleep(5)
state = str(self._get_iface_property("State"))
print("wpa iface state %s (scanning %s)" % (state, bool(self._get_iface_property("Scanning"))))
if state in ["disconnected", "inactive"]:
self.iface_ifc.Reconnect()
ip = get_wiface_IPv4(self.iface_name)
print("wpa iface %s IP %s" % (self.iface_name, ip))
if ip is not None:
self.connected = True
return ip

38
tools/esp_prov/esp_prov.py
View file

@ -266,6 +266,32 @@ def get_wifi_config(tp, sec):
return None
def wait_wifi_connected(tp, sec):
"""
Wait for provisioning to report Wi-Fi is connected
Returns True if Wi-Fi connection succeeded, False if connection consistently failed
"""
TIME_PER_POLL = 5
retry = 3
while True:
time.sleep(TIME_PER_POLL)
print("\n==== Wi-Fi connection state ====")
ret = get_wifi_config(tp, sec)
if ret == "connecting":
continue
elif ret == "connected":
print("==== Provisioning was successful ====")
return True
elif retry > 0:
retry -= 1
print("Waiting to poll status again (status %s, %d tries left)..." % (ret, retry))
else:
print("---- Provisioning failed ----")
return False
def desc_format(*args):
desc = ''
for arg in args:
@ -449,14 +475,4 @@ if __name__ == '__main__':
exit(7)
print("==== Apply config sent successfully ====")
while True:
time.sleep(5)
print("\n==== Wi-Fi connection state ====")
ret = get_wifi_config(obj_transport, obj_security)
if (ret == 1):
continue
elif (ret == 0):
print("==== Provisioning was successful ====")
else:
print("---- Provisioning failed ----")
break
wait_wifi_connected(obj_transport, obj_security)

7
tools/esp_prov/prov/wifi_prov.py
View file

@ -45,19 +45,24 @@ def config_get_status_response(security_ctx, response_data):
cmd_resp1.ParseFromString(decrypted_message)
print_verbose(security_ctx, "Response type " + str(cmd_resp1.msg))
print_verbose(security_ctx, "Response status " + str(cmd_resp1.resp_get_status.status))
if cmd_resp1.resp_get_status.sta_state == 0:
print("++++ WiFi state: " + "connected ++++")
return "connected"
elif cmd_resp1.resp_get_status.sta_state == 1:
print("++++ WiFi state: " + "connecting... ++++")
return "connecting"
elif cmd_resp1.resp_get_status.sta_state == 2:
print("++++ WiFi state: " + "disconnected ++++")
return "disconnected"
elif cmd_resp1.resp_get_status.sta_state == 3:
print("++++ WiFi state: " + "connection failed ++++")
if cmd_resp1.resp_get_status.fail_reason == 0:
print("++++ Failure reason: " + "Incorrect Password ++++")
elif cmd_resp1.resp_get_status.fail_reason == 1:
print("++++ Failure reason: " + "Incorrect SSID ++++")
return cmd_resp1.resp_get_status.sta_state
return "failed"
return "unknown"
def config_set_config_request(security_ctx, ssid, passphrase):

4
tools/esp_prov/transport/transport_http.py
View file

@ -31,10 +31,10 @@ class Transport_HTTP(Transport):
raise RuntimeError("Unable to resolve hostname :" + hostname)
if certfile is None:
self.conn = http.client.HTTPConnection(hostname, timeout=30)
self.conn = http.client.HTTPConnection(hostname, timeout=45)
else:
ssl_ctx = ssl.create_default_context(cafile=certfile)
self.conn = http.client.HTTPSConnection(hostname, context=ssl_ctx, timeout=30)
self.conn = http.client.HTTPSConnection(hostname, context=ssl_ctx, timeout=45)
try:
print("Connecting to " + hostname)
self.conn.connect()