eaw/esp-idf
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ble-wifi-example-tests: Add fixes and cleanups to ble and wifi tests

Browse Source 
(cherry picked from commit 2d22374460)
This commit is contained in:
Shivani Tipnis
2021-06-01 12:42:28 +05:30
parent 71cf821659
commit be960bc3dc
10 changed files with 1088 additions and 1110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
tools/ci/python_packages/wifi_tools.py
View File

@@ -13,10 +13,11 @@
# limitations under the License.
#
import time
import dbus
import dbus.mainloop.glib
import netifaces
import time
def get_wiface_name():
@@ -43,25 +44,30 @@ class wpa_cli:
self.new_network = None
self.connected = False
self.reset_on_exit = reset_on_exit
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SystemBus()
service = dbus.Interface(bus.get_object("fi.w1.wpa_supplicant1", "/fi/w1/wpa_supplicant1"),
"fi.w1.wpa_supplicant1")
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')
try:
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SystemBus()
self.old_network = self._get_iface_property("CurrentNetwork")
print("Old network is %s" % self.old_network)
service = dbus.Interface(bus.get_object('fi.w1.wpa_supplicant1', '/fi/w1/wpa_supplicant1'),
'fi.w1.wpa_supplicant1')
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')
if self.old_network == '/':
self.old_network = None
else:
self.connected = True
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
except Exception as err:
raise Exception('Failure in wpa_cli init: {}'.format(err))
def _get_iface_property(self, name):
""" Read the property with 'name' from the wi-fi interface object
@@ -69,40 +75,44 @@ class wpa_cli:
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)
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()
self.connected = False
try:
if self.connected is True:
self.iface_ifc.Disconnect()
self.connected = False
if self.new_network is not None:
self.iface_ifc.RemoveNetwork(self.new_network)
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)))
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)
self.new_network = self.iface_ifc.AddNetwork({'ssid': ssid, 'psk': password})
self.iface_ifc.SelectNetwork(self.new_network)
time.sleep(10)
ip = None
retry = 10
while retry > 0:
time.sleep(5)
ip = None
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
retry -= 1
time.sleep(3)
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()
self.reset()
print('wpa_cli : Connection failed')
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
retry -= 1
self.reset()
raise RuntimeError('wpa_cli : Connection failed')
except Exception as err:
raise Exception('Failure in wpa_cli init: {}'.format(err))
def reset(self):
if self.iface_ifc is not None: