conman: Always check client status.
Currently, there is caching code (WLANConfiguration.client_up) that
was intended to avoid unnecessary checks. But this is hard to make
correct if e.g. something else kills wpa_supplicant, and in any case
was a premature optimization. For now, always make the expensive
check; if it becomes a problem we can revisit caching the result
properly.
This change revealed a few small inconsistencies in unit tests which
have also been addressed, including:
* The fake wpa_supplicant control socket was touched far more often
than necessary. In one case, it was actually created when instead
the tests should have tested the behavior that occurs when it does
not already exist (in interface_test.Wifi.attach_wpa_control).
* In the test overrides of start_client and stop_client, some things
were happening in an unrealistic order (e.g. fakes were being
started before calling the super method, rather than after).
* 'my ssid' was set by default in fake wifi interfaces, meaning that
tests were not verifying that the SSID was set appropriately.
Finally, some comments which were incorrectly specific to non-Frenzy
devices (e.g. by talking about wcli*, when Frenzy uses wlan* for its
client interfaces) have been updated.
Change-Id: I95d0afe939e67fd8fc39befc9b6085cd57e0fa87
diff --git a/conman/connection_manager.py b/conman/connection_manager.py
index 45a848b..e9a2796 100755
--- a/conman/connection_manager.py
+++ b/conman/connection_manager.py
@@ -57,17 +57,17 @@
WIFI_SETCLIENT = ['wifi', 'setclient', '--persist']
WIFI_STOPCLIENT = ['wifi', 'stopclient', '--persist']
- def __init__(self, band, wifi, command_lines, _status):
+ def __init__(self, band, wifi, command_lines, _status, wpa_control_interface):
self.band = band
self.wifi = wifi
self.command = command_lines.splitlines()
self.access_point_up = False
- self.client_up = False
self.ssid = None
self.passphrase = None
self.interface_suffix = None
self.access_point = None
self._status = _status
+ self._wpa_control_interface = wpa_control_interface
binwifi_option_attrs = {
'-s': 'ssid',
@@ -92,7 +92,12 @@
if self.wifi.initial_ssid == self.ssid:
logging.debug('Connected to WLAN at startup')
- self.client_up = True
+
+ @property
+ def client_up(self):
+ wpa_cli_status = self.wifi.wpa_cli_status()
+ return (wpa_cli_status.get('wpa_state') == 'COMPLETED'
+ and wpa_cli_status.get('ssid') == self.ssid)
def start_access_point(self):
"""Start an access point."""
@@ -130,12 +135,11 @@
def start_client(self):
"""Join the WLAN as a client."""
- if self.client_up:
+ up = self.client_up
+ if up:
logging.debug('Wifi client already started on %s GHz', self.band)
return
- self.wifi.detach_wpa_control()
-
command = self.WIFI_SETCLIENT + ['--ssid', self.ssid, '--band', self.band]
env = dict(os.environ)
if self.passphrase:
@@ -143,11 +147,13 @@
try:
self._status.trying_wlan = True
subprocess.check_output(command, stderr=subprocess.STDOUT, env=env)
- self.client_up = True
- self._status.connected_to_wlan = True
- logging.info('Started wifi client on %s GHz', self.band)
except subprocess.CalledProcessError as e:
logging.error('Failed to start wifi client: %s', e.output)
+ return
+
+ self._status.connected_to_wlan = True
+ logging.info('Started wifi client on %s GHz', self.band)
+ self.wifi.attach_wpa_control(self._wpa_control_interface)
def stop_client(self):
if not self.client_up:
@@ -159,7 +165,6 @@
try:
subprocess.check_output(self.WIFI_STOPCLIENT + ['-b', self.band],
stderr=subprocess.STDOUT)
- self.client_up = False
# TODO(rofrankel): Make this work for dual-radio devices.
self._status.connected_to_wlan = False
logging.debug('Stopped wifi client on %s GHz', self.band)
@@ -225,28 +230,7 @@
bridge_interface, '10',
acs_autoprovisioning_filepath=acs_autoprov_filepath)
- # If we have multiple wcli interfaces, 5 GHz-only < both < 2.4 GHz-only.
- def metric_for_bands(bands):
- if '5' in bands:
- if '2.4' in bands:
- return interface.METRIC_24GHZ_5GHZ
- return interface.METRIC_5GHZ
- return interface.METRIC_24GHZ
-
- def wifi_class(attrs):
- return self.FrenzyWifi if 'frenzy' in attrs else self.Wifi
-
- self.wifi = sorted([
- wifi_class(attrs)(interface_name,
- metric_for_bands(attrs['bands']),
- # Prioritize 5 GHz over 2.4.
- bands=sorted(attrs['bands'], reverse=True))
- for interface_name, attrs
- in get_client_interfaces().iteritems()
- ], key=lambda w: w.metric)
-
- for wifi in self.wifi:
- wifi.last_wifi_scan_time = -self._wifi_scan_period_s
+ self.create_wifi_interfaces()
self._status = status.Status(self._status_dir)
@@ -315,6 +299,32 @@
self._interface_update_counter = 0
self._try_wlan_after = {'5': 0, '2.4': 0}
+ def create_wifi_interfaces(self):
+ """Create Wifi interfaces."""
+
+ # If we have multiple client interfaces, 5 GHz-only < both < 2.4 GHz-only.
+ def metric_for_bands(bands):
+ if '5' in bands:
+ if '2.4' in bands:
+ return interface.METRIC_24GHZ_5GHZ
+ return interface.METRIC_5GHZ
+ return interface.METRIC_24GHZ
+
+ def wifi_class(attrs):
+ return self.FrenzyWifi if 'frenzy' in attrs else self.Wifi
+
+ self.wifi = sorted([
+ wifi_class(attrs)(interface_name,
+ metric_for_bands(attrs['bands']),
+ # Prioritize 5 GHz over 2.4.
+ bands=sorted(attrs['bands'], reverse=True))
+ for interface_name, attrs
+ in get_client_interfaces().iteritems()
+ ], key=lambda w: w.metric)
+
+ for wifi in self.wifi:
+ wifi.last_wifi_scan_time = -self._wifi_scan_period_s
+
def is_interface_up(self, interface_name):
"""Explicitly check whether an interface is up.
@@ -378,12 +388,6 @@
for wifi in self.wifi:
continue_wifi = False
- if not wifi.attached():
- logging.debug('Attempting to attach to wpa control interface for %s',
- wifi.name)
- self._status.attached_to_wpa_supplicant = wifi.attach_wpa_control(
- self._wpa_control_interface)
- wifi.handle_wpa_events()
# Only one wlan_configuration per interface will have access_point ==
# True. Try 5 GHz first, then 2.4 GHz. If both bands are supported by
@@ -405,6 +409,13 @@
if wlan_configuration.access_point_up:
continue_wifi = True
+ if not wifi.attached():
+ logging.debug('Attempting to attach to wpa control interface for %s',
+ wifi.name)
+ self._status.attached_to_wpa_supplicant = wifi.attach_wpa_control(
+ self._wpa_control_interface)
+ wifi.handle_wpa_events()
+
if continue_wifi:
logging.debug('Running AP on %s, nothing else to do.', wifi.name)
continue
@@ -562,7 +573,8 @@
wifi = self.wifi_for_band(band)
if wifi:
self._update_wlan_configuration(
- self.WLANConfiguration(band, wifi, contents, self._status))
+ self.WLANConfiguration(band, wifi, contents, self._status,
+ self._wpa_control_interface))
elif filename.startswith(self.ACCESS_POINT_FILE_PREFIX):
match = re.match(self.ACCESS_POINT_FILE_REGEXP, filename)
if match:
diff --git a/conman/connection_manager_test.py b/conman/connection_manager_test.py
index 82df636..7b202b7 100755
--- a/conman/connection_manager_test.py
+++ b/conman/connection_manager_test.py
@@ -182,16 +182,26 @@
WIFI_STOPCLIENT = ['echo', 'stopclient']
def start_client(self):
- if not self.client_up:
+ client_was_up = self.client_up
+ was_attached = self.wifi.attached()
+ # Do this before calling the super method so that the attach call at the end
+ # succeeds.
+ if not client_was_up and not was_attached:
+ self.wifi._initial_ssid_testonly = self.ssid
+ self.wifi.start_wpa_supplicant_testonly(self._wpa_control_interface)
+
+ super(WLANConfiguration, self).start_client()
+
+ if not client_was_up:
self.wifi.set_connection_check_result('succeed')
- if self.wifi.attached():
+ if was_attached:
+ self.wifi._wpa_control.ssid_testonly = self.ssid
self.wifi.add_connected_event()
- else:
- open(self._socket(), 'w')
- # Normally, wpa_supplicant would bring up wcli*, which would trigger
- # ifplugd, which would run ifplugd.action, which would do two things:
+ # Normally, wpa_supplicant would bring up the client interface, which
+ # would trigger ifplugd, which would run ifplugd.action, which would do
+ # two things:
#
# 1) Write an interface status file.
# 2) Call run-dhclient, which would call dhclient-script, which would
@@ -201,21 +211,18 @@
self.write_interface_status_file('1')
self.write_gateway_file()
- super(WLANConfiguration, self).start_client()
-
def stop_client(self):
- if self.client_up:
+ client_was_up = self.client_up
+
+ super(WLANConfiguration, self).stop_client()
+
+ if client_was_up:
self.wifi.add_terminating_event()
self.wifi.set_connection_check_result('fail')
# See comments in start_client.
self.write_interface_status_file('0')
- super(WLANConfiguration, self).stop_client()
-
- def _socket(self):
- return os.path.join(self._wpa_control_interface, self.wifi.name)
-
def write_gateway_file(self):
gateway_file = os.path.join(self.tmp_dir,
self.gateway_file_prefix + self.wifi.name)
@@ -234,8 +241,6 @@
def __init__(self, *args, **kwargs):
super(Wifi, self).__init__(*args, **kwargs)
- # Whether wpa_supplicant is connected to a network.
- self._initially_connected = True
self.wifi_scan_counter = 0
@@ -243,8 +248,6 @@
def __init__(self, *args, **kwargs):
super(FrenzyWifi, self).__init__(*args, **kwargs)
- # Whether wpa_supplicant is connected to a network.
- self._initially_connected = True
self.wifi_scan_counter = 0
@@ -268,11 +271,12 @@
self.interfaces_already_up = kwargs.pop('__test_interfaces_already_up',
['eth0'])
- wifi_interfaces_already_up = [ifc for ifc in self.interfaces_already_up
- if ifc.startswith('wcli')]
- for wifi in wifi_interfaces_already_up:
- # wcli1 is always 5 GHz. wcli0 always *includes* 2.4.
- band = '5' if wifi == 'wcli1' else '2.4'
+ self.wifi_interfaces_already_up = [ifc for ifc in self.interfaces_already_up
+ if ifc.startswith('w')]
+ for wifi in self.wifi_interfaces_already_up:
+ # wcli1 is always 5 GHz. wcli0 always *includes* 2.4. wlan* client
+ # interfaces are Frenzy interfaces and therefore 5 GHz-only.
+ band = '5' if wifi in ('wlan0', 'wlan1', 'wcli1') else '2.4'
# This will happen in the super function, but in order for
# write_wlan_config to work we have to do it now. This has to happen
# before the super function so that the files exist before the inotify
@@ -285,42 +289,40 @@
super(ConnectionManager, self).__init__(*args, **kwargs)
- for wifi in wifi_interfaces_already_up:
- # pylint: disable=protected-access
- self.interface_by_name(wifi)._initially_connected = True
-
self.interface_with_scan_results = None
self.scan_results_include_hidden = False
self.can_connect_to_s2 = True
+ def create_wifi_interfaces(self):
+ super(ConnectionManager, self).create_wifi_interfaces()
+ for wifi in self.wifi_interfaces_already_up:
+ # pylint: disable=protected-access
+ self.interface_by_name(wifi)._initial_ssid_testonly = 'my ssid'
+
@property
def IP_LINK(self):
return ['echo'] + ['%s LOWER_UP' % ifc
for ifc in self.interfaces_already_up]
- def _update_access_point(self, wlan_configuration):
- client_was_up = wlan_configuration.client_up
- super(ConnectionManager, self)._update_access_point(wlan_configuration)
- if wlan_configuration.access_point_up:
- if client_was_up:
- wifi = self.wifi_for_band(wlan_configuration.band)
- wifi.add_terminating_event()
-
def _try_next_bssid(self, wifi):
if hasattr(wifi, 'cycler'):
bss_info = wifi.cycler.peek()
if bss_info:
self.last_provisioning_attempt = bss_info
+ # pylint: disable=protected-access
+ if wifi._wpa_control:
+ wifi._wpa_control.ssid_testonly = bss_info.ssid
super(ConnectionManager, self)._try_next_bssid(wifi)
- socket = os.path.join(self._wpa_control_interface, wifi.name)
-
def connect(connection_check_result):
+ # pylint: disable=protected-access
if wifi.attached():
+ wifi._wpa_control._ssid_testonly = bss_info.ssid
wifi.add_connected_event()
else:
- open(socket, 'w')
+ wifi._initial_ssid_testonly = bss_info.ssid
+ wifi.start_wpa_supplicant_testonly(self._wpa_control_interface)
wifi.set_connection_check_result(connection_check_result)
self.ifplugd_action(wifi.name, True)
@@ -340,7 +342,6 @@
# pylint: disable=unused-argument,protected-access
def _find_bssids(self, band):
- # Only wcli0 finds anything.
scan_output = ''
if (self.interface_with_scan_results and
band in self.interface_by_name(self.interface_with_scan_results).bands):
@@ -656,6 +657,8 @@
wvtest.WVPASS(c.internet())
wvtest.WVFAIL(c.client_up(band))
wvtest.WVPASS(c.wifi_for_band(band).current_route())
+ # Disable scan results again.
+ c.interface_with_scan_results = None
# Now, create a WLAN configuration which should be connected to.
ssid = 'wlan'
@@ -667,17 +670,28 @@
wvtest.WVPASS(c.wifi_for_band(band).current_route())
wvtest.WVPASS(c.has_status_files([status.P.CONNECTED_TO_WLAN]))
- # Now, remove the WLAN configuration and make sure we are disconnected. Then
- # disable the previously used ACS connection via s2, add the user's WLAN to
- # the scan results, and scan again. This time, the first SSID tried should be
- # 's3', which is not present in the scan results but *is* advertised by the
- # secure AP running the user's WLAN.
- c.delete_wlan_config(band)
+ # Kill wpa_supplicant. conman should restart it.
+ wvtest.WVPASS(c.client_up(band))
+ wvtest.WVPASS(c._connected_to_wlan(c.wifi_for_band(band)))
+ c.wifi_for_band(band).kill_wpa_supplicant_testonly(c._wpa_control_interface)
+ wvtest.WVFAIL(c.client_up(band))
+ wvtest.WVFAIL(c._connected_to_wlan(c.wifi_for_band(band)))
c.run_once()
+ wvtest.WVPASS(c.has_status_files([status.P.CONNECTED_TO_WLAN]))
+ wvtest.WVPASS(c.client_up(band))
+ wvtest.WVPASS(c._connected_to_wlan(c.wifi_for_band(band)))
+
+ # Now, remove the WLAN configuration and make sure we are disconnected. Then
+ # disable the previously used ACS connection via s2, re-enable scan results,
+ # add the user's WLAN to the scan results, and scan again. This time, the
+ # first SSID tried should be 's3', which is now present in the scan results
+ # (with its SSID hidden, but included via vendor IE).
+ c.delete_wlan_config(band)
c.can_connect_to_s2 = False
+ c.interface_with_scan_results = c.wifi_for_band(band).name
c.scan_results_include_hidden = True
- wvtest.WVFAIL(c.has_status_files([status.P.CONNECTED_TO_WLAN]))
c.run_until_interface_update_and_scan(band)
+ wvtest.WVFAIL(c.has_status_files([status.P.CONNECTED_TO_WLAN]))
c.run_until_interface_update()
wvtest.WVPASS(c.has_status_files([status.P.CONNECTED_TO_OPEN]))
wvtest.WVPASSEQ(c.last_provisioning_attempt.ssid, 's3')
@@ -809,6 +823,8 @@
wvtest.WVPASS(c.access_point_up('2.4'))
wvtest.WVPASS(c.access_point_up('5'))
wvtest.WVPASS(c.bridge.current_route())
+ wvtest.WVFAIL(c.client_up('2.4'))
+ wvtest.WVFAIL(c.client_up('5'))
wvtest.WVFAIL(c.wifi_for_band('2.4').current_route())
wvtest.WVFAIL(c.wifi_for_band('5').current_route())
@@ -855,7 +871,7 @@
wvtest.WVFAIL(c.wifi_for_band('5').current_route())
# The next 2.4 GHz scan will have results.
- c.interface_with_scan_results = 'wcli0'
+ c.interface_with_scan_results = c.wifi_for_band('2.4').name
c.run_until_scan('2.4')
# Now run 3 cycles, so that s2 will have been tried.
for _ in range(3):
@@ -946,8 +962,8 @@
wvtest.WVFAIL(c.wifi_for_band('2.4').current_route())
wvtest.WVFAIL(c.wifi_for_band('5').current_route())
- # The wcli0 scan will have results that will lead to ACS access.
- c.interface_with_scan_results = 'wcli0'
+ # The 2.4 GHz scan will have results that will lead to ACS access.
+ c.interface_with_scan_results = c.wifi_for_band('2.4').name
c.run_until_scan('5')
for _ in range(3):
c.run_once()
diff --git a/conman/interface.py b/conman/interface.py
index 72818d2..d45f42e 100755
--- a/conman/interface.py
+++ b/conman/interface.py
@@ -355,17 +355,32 @@
logging.error('Error attaching to wpa_supplicant: %s', e)
return False
- for line in self._wpa_control.request('STATUS').splitlines():
- if '=' not in line:
- continue
- key, value = line.split('=', 1)
- if key == 'wpa_state':
- self.wpa_supplicant = value == 'COMPLETED'
- elif key == 'ssid' and not self._initialized:
- self.initial_ssid = value
+ status = self.wpa_cli_status()
+ self.wpa_supplicant = status.get('wpa_state') == 'COMPLETED'
+ if not self._initialized:
+ self.initial_ssid = status.get('ssid')
return True
+ def wpa_cli_status(self):
+ """Parse the STATUS response from the wpa_supplicant CLI.
+
+ Returns:
+ A dict containing the parsed results, where key and value are separated by
+ '=' on each line.
+ """
+ status = {}
+
+ if self._wpa_control:
+ lines = self._wpa_control.request('STATUS').splitlines()
+ for line in lines:
+ if '=' not in line:
+ continue
+ k, v = line.strip().split('=', 1)
+ status[k] = v
+
+ return status
+
def get_wpa_control(self, socket):
return self.WPACtrl(socket)
@@ -408,10 +423,10 @@
class FrenzyWPACtrl(object):
- """A fake WPACtrl for Frenzy devices.
+ """A WPACtrl for Frenzy devices.
Implements the same functions used on the normal WPACtrl, using a combination
- of the QCSAPI and wifi_files. Keeps state in order to generate fake events by
+ of the QCSAPI and wifi_files. Keeps state in order to generate events by
diffing saved state with current system state.
"""
@@ -500,10 +515,17 @@
return self._events.pop(0)
def request(self, request_type):
- if request_type == 'STATUS' and self._client_mode and self._ssid:
- return 'wpa_state=COMPLETED\nssid=%s\n' % self._ssid
+ """Partial implementation of WPACtrl.request."""
- return ''
+ if request_type != 'STATUS':
+ return ''
+
+ self._update()
+
+ if not self._client_mode or not self._ssid:
+ return ''
+
+ return 'wpa_state=COMPLETED\nssid=%s' % self._ssid
class FrenzyWifi(Wifi):
diff --git a/conman/interface_test.py b/conman/interface_test.py
index f6dafbd..8a376a6 100755
--- a/conman/interface_test.py
+++ b/conman/interface_test.py
@@ -74,11 +74,14 @@
self.events = []
self.attached = False
self.connected = False
+ self.ssid_testonly = None
def pending(self):
+ self.check_socket_exists('pending: socket does not exist')
return bool(self.events)
def recv(self):
+ self.check_socket_exists('recv: socket does not exist')
return self.events.pop(0)
def attach(self):
@@ -87,14 +90,15 @@
self.attached = True
def detach(self):
- if not os.path.exists(self._socket):
- raise wpactrl.error('wpactrl_detach failed')
self.attached = False
+ self.ssid_testonly = None
+ self.connected = False
+ self.check_socket_exists('wpactrl_detach failed')
def request(self, request_type):
if request_type == 'STATUS':
- return ('foo\nwpa_state=COMPLETED\nssid=my ssid\nbar' if self.connected
- else 'foo')
+ return ('foo\nwpa_state=COMPLETED\nssid=%s\nbar' % self.ssid_testonly
+ if self.connected else 'foo')
else:
raise ValueError('Invalid request_type %s' % request_type)
@@ -104,15 +108,21 @@
self.events.append(event)
def add_connected_event(self):
+ self.connected = True
self.add_event(Wifi.CONNECTED_EVENT)
def add_disconnected_event(self):
+ self.connected = False
self.add_event(Wifi.DISCONNECTED_EVENT)
def add_terminating_event(self):
- os.unlink(self._socket)
+ self.connected = False
self.add_event(Wifi.TERMINATING_EVENT)
+ def check_socket_exists(self, msg='Fake socket does not exist'):
+ if not os.path.exists(self._socket):
+ raise wpactrl.error(msg)
+
class Wifi(FakeInterfaceMixin, interface.Wifi):
"""Fake Wifi for testing."""
@@ -125,17 +135,18 @@
def __init__(self, *args, **kwargs):
super(Wifi, self).__init__(*args, **kwargs)
- self._initially_connected = False
+ self._initial_ssid_testonly = None
def attach_wpa_control(self, path):
- open(os.path.join(path, self.name), 'w')
- if self._initially_connected and self._wpa_control:
+ if self._initial_ssid_testonly and self._wpa_control:
self._wpa_control.connected = True
super(Wifi, self).attach_wpa_control(path)
def get_wpa_control(self, *args, **kwargs):
result = super(Wifi, self).get_wpa_control(*args, **kwargs)
- result.connected = self._initially_connected
+ if self._initial_ssid_testonly:
+ result.connected = True
+ result.ssid_testonly = self._initial_ssid_testonly
return result
def add_connected_event(self):
@@ -143,35 +154,58 @@
self._wpa_control.add_connected_event()
def add_disconnected_event(self):
+ self._initial_ssid_testonly = None
if self.attached():
self._wpa_control.add_disconnected_event()
def add_terminating_event(self):
+ self._initial_ssid_testonly = None
if self.attached():
self._wpa_control.add_terminating_event()
+ def detach_wpa_control(self):
+ self._initial_ssid_testonly = None
+ super(Wifi, self).detach_wpa_control()
+
+ def start_wpa_supplicant_testonly(self, path):
+ logging.debug('Starting fake wpa_supplicant for %s', self.name)
+ open(os.path.join(path, self.name), 'w')
+
+ def kill_wpa_supplicant_testonly(self, path):
+ logging.debug('Killing fake wpa_supplicant for %s', self.name)
+ if self.attached():
+ self.detach_wpa_control()
+ os.unlink(os.path.join(path, self.name))
+ else:
+ raise RuntimeError('Trying to kill wpa_supplicant while not attached')
+
class FrenzyWPACtrl(interface.FrenzyWPACtrl):
def __init__(self, *args, **kwargs):
super(FrenzyWPACtrl, self).__init__(*args, **kwargs)
- self.fake_qcsapi = {}
+ self.ssid_testonly = None
def _qcsapi(self, *command):
return self.fake_qcsapi.get(command[0], None)
def add_connected_event(self):
self.fake_qcsapi['get_mode'] = 'Station'
- json.dump({'SSID': 'my ssid'}, open(self._wifiinfo_filename(), 'w'))
- self._update()
+ json.dump({'SSID': self.ssid_testonly},
+ open(self._wifiinfo_filename(), 'w'))
def add_disconnected_event(self):
+ self.ssid_testonly = None
json.dump({'SSID': ''}, open(self._wifiinfo_filename(), 'w'))
- self._update()
def add_terminating_event(self):
+ self.ssid_testonly = None
+ json.dump({'SSID': ''}, open(self._wifiinfo_filename(), 'w'))
self.fake_qcsapi['get_mode'] = 'AP'
- self._update()
+
+ def detach(self):
+ self.add_terminating_event()
+ super(FrenzyWPACtrl, self).detach()
class FrenzyWifi(FakeInterfaceMixin, interface.FrenzyWifi):
@@ -179,17 +213,22 @@
def __init__(self, *args, **kwargs):
super(FrenzyWifi, self).__init__(*args, **kwargs)
- self._initially_connected = False
+ self._initial_ssid_testonly = None
+ self.fake_qcsapi = {}
def attach_wpa_control(self, *args, **kwargs):
super(FrenzyWifi, self).attach_wpa_control(*args, **kwargs)
if self._wpa_control:
- self._wpa_control.fake_qcsapi['get_mode'] = 'Station'
+ self._wpa_control.ssid_testonly = self._initial_ssid_testonly
+ if self._initial_ssid_testonly:
+ self._wpa_control.add_connected_event()
def get_wpa_control(self, *args, **kwargs):
result = super(FrenzyWifi, self).get_wpa_control(*args, **kwargs)
- if self._initially_connected:
+ result.fake_qcsapi = self.fake_qcsapi
+ if self._initial_ssid_testonly:
result.fake_qcsapi['get_mode'] = 'Station'
+ result.ssid_testonly = self._initial_ssid_testonly
result.add_connected_event()
return result
@@ -198,13 +237,32 @@
self._wpa_control.add_connected_event()
def add_disconnected_event(self):
+ self._initial_ssid_testonly = None
if self.attached():
self._wpa_control.add_disconnected_event()
def add_terminating_event(self):
+ self._initial_ssid_testonly = None
if self.attached():
self._wpa_control.add_terminating_event()
+ def detach_wpa_control(self):
+ self._initial_ssid_testonly = None
+ super(FrenzyWifi, self).detach_wpa_control()
+
+ def start_wpa_supplicant_testonly(self, unused_path):
+ logging.debug('Starting fake wpa_supplicant for %s', self.name)
+ self.fake_qcsapi['get_mode'] = 'Station'
+
+ def kill_wpa_supplicant_testonly(self, unused_path):
+ logging.debug('Killing fake wpa_supplicant for %s', self.name)
+ if self.attached():
+ # This happens to do what we need.
+ self.add_terminating_event()
+ self.detach_wpa_control()
+ else:
+ raise RuntimeError('Trying to kill wpa_supplicant while not attached')
+
@wvtest.wvtest
def bridge_test():
@@ -273,6 +331,7 @@
def generic_wifi_test(w, wpa_path):
# Not currently connected.
+ w.start_wpa_supplicant_testonly(wpa_path)
w.attach_wpa_control(wpa_path)
wvtest.WVFAIL(w.wpa_supplicant)
@@ -280,6 +339,7 @@
wpa_control = w._wpa_control
# wpa_supplicant connects.
+ wpa_control.ssid_testonly = 'my=ssid'
wpa_control.add_connected_event()
wvtest.WVFAIL(w.wpa_supplicant)
w.handle_wpa_events()
@@ -295,14 +355,14 @@
# connected when we attach.
w.detach_wpa_control()
# pylint: disable=protected-access
- w._initially_connected = True
+ w._initial_ssid_testonly = 'my=ssid'
w._initialized = False
w.attach_wpa_control(wpa_path)
wpa_control = w._wpa_control
# wpa_supplicant was already connected when we attached.
wvtest.WVPASS(w.wpa_supplicant)
- wvtest.WVPASSEQ(w.initial_ssid, 'my ssid')
+ wvtest.WVPASSEQ(w.initial_ssid, 'my=ssid')
w.initialize()
wvtest.WVPASSEQ(w.initial_ssid, None)
