conman: Don't use wpa_supplicant control interface.
The control interface is unreliable, both for requests and events.
Stop using it and go with the slightly more expensive but much simpler
approach of always talking to wpa_cli, which so far seems much more
reliable.
While writing and debugging this, it was useful to improve the
printing of mocked access points.
Change-Id: I05205abda499646aa35cd63bbc5f68fd331b34c1
diff --git a/conman/connection_manager.py b/conman/connection_manager.py
index b4c3887..e484e67 100755
--- a/conman/connection_manager.py
+++ b/conman/connection_manager.py
@@ -74,7 +74,7 @@
WIFI_SETCLIENT = ['wifi', 'setclient', '--persist']
WIFI_STOPCLIENT = ['wifi', 'stopclient', '--persist']
- def __init__(self, band, wifi, command_lines, wpa_control_interface):
+ def __init__(self, band, wifi, command_lines):
self.band = band
self.wifi = wifi
self.command = command_lines.splitlines()
@@ -83,7 +83,6 @@
self.passphrase = None
self.interface_suffix = None
self.access_point = None
- self._wpa_control_interface = wpa_control_interface
binwifi_option_attrs = {
'-s': 'ssid',
@@ -106,16 +105,12 @@
if self.ssid is None:
raise ValueError('Command file does not specify SSID')
- if self.wifi.initial_ssid == self.ssid:
+ if self.client_up:
logging.info('Connected to WLAN at startup')
@property
def client_up(self):
- wpa_status = self.wifi.wpa_status()
- return (wpa_status.get('wpa_state') == 'COMPLETED'
- # NONE indicates we're on a provisioning network; anything else
- # suggests we're already on the WLAN.
- and wpa_status.get('key_mgmt') != 'NONE')
+ return self.ssid and self.ssid == self.wifi.current_secure_ssid()
def start_access_point(self):
"""Start an access point."""
@@ -163,7 +158,8 @@
return
if self._actually_start_client():
- self._post_start_client()
+ self.wifi.status.connected_to_wlan = True
+ logging.info('Started wifi client on %s GHz', self.band)
def _actually_start_client(self):
"""Actually run wifi setclient.
@@ -187,26 +183,18 @@
return True
- def _post_start_client(self):
- self.wifi.handle_wpa_events()
- self.wifi.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:
logging.debug('Wifi client already stopped on %s GHz', self.band)
return
- self.wifi.detach_wpa_control()
-
try:
subprocess.check_output(self.WIFI_STOPCLIENT + ['-b', self.band],
stderr=subprocess.STDOUT)
# TODO(rofrankel): Make this work for dual-radio devices.
self.wifi.status.connected_to_wlan = False
logging.info('Stopped wifi client on %s GHz', self.band)
- self.wifi.handle_wpa_events()
+ self.wifi.update()
except subprocess.CalledProcessError as e:
logging.error('Failed to stop wifi client: %s', e.output)
@@ -242,7 +230,6 @@
tmp_dir='/tmp/conman',
config_dir='/config/conman',
moca_tmp_dir='/tmp/cwmp/monitoring/moca2',
- wpa_control_interface='/var/run/wpa_supplicant',
run_duration_s=1, interface_update_period=5,
wifi_scan_period_s=120, wlan_retry_s=120, associate_wait_s=15,
dhcp_wait_s=10, acs_connection_check_wait_s=1,
@@ -254,7 +241,6 @@
self._interface_status_dir = os.path.join(tmp_dir, 'interfaces')
self._status_dir = os.path.join(tmp_dir, 'status')
self._moca_tmp_dir = moca_tmp_dir
- self._wpa_control_interface = wpa_control_interface
self._run_duration_s = run_duration_s
self._interface_update_period = interface_update_period
self._wifi_scan_period_s = wifi_scan_period_s
@@ -310,17 +296,13 @@
self.ifplugd_action('eth0', ethernet_up)
self.bridge.ethernet = ethernet_up
- # Do the same for wifi interfaces , but rather than explicitly setting that
- # the wpa_supplicant link is up, attempt to attach to the wpa_supplicant
- # control interface.
+ # Do the same for wifi interfaces.
for wifi in self.wifi:
wifi_up = self.is_interface_up(wifi.name)
+ wifi.wpa_supplicant = wifi_up
if not os.path.exists(
os.path.join(self._interface_status_dir, wifi.name)):
self.ifplugd_action(wifi.name, wifi_up)
- if wifi_up:
- wifi.status.attached_to_wpa_supplicant = wifi.attach_wpa_control(
- self._wpa_control_interface)
for path, prefix in ((self._tmp_dir, self.GATEWAY_FILE_PREFIX),
(self._tmp_dir, self.SUBNET_FILE_PREFIX),
@@ -438,8 +420,6 @@
while True:
self.run_once()
finally:
- for wifi in self.wifi:
- wifi.detach_wpa_control()
self.notifier.stop()
def run_once(self):
@@ -510,12 +490,7 @@
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)
- wifi.status.attached_to_wpa_supplicant = wifi.attach_wpa_control(
- self._wpa_control_interface)
- wifi.handle_wpa_events()
+ wifi.update()
if continue_wifi:
logging.debug('Running AP on %s, nothing else to do.', wifi.name)
@@ -735,8 +710,7 @@
wifi = self.wifi_for_band(band)
if wifi:
self._update_wlan_configuration(
- self.WLANConfiguration(band, wifi, contents,
- self._wpa_control_interface))
+ self.WLANConfiguration(band, wifi, contents))
elif filename.startswith(self.ACCESS_POINT_FILE_PREFIX):
match = re.match(self.ACCESS_POINT_FILE_REGEXP, filename)
if match:
@@ -844,8 +818,7 @@
self.start_provisioning(wifi)
connected = self._try_bssid(wifi, bss_info)
if connected:
- wifi.attach_wpa_control(self._wpa_control_interface)
- wifi.handle_wpa_events()
+ wifi.update()
wifi.status.connected_to_open = True
now = _gettime()
wifi.complain_about_acs_at = now + 5
diff --git a/conman/connection_manager_test.py b/conman/connection_manager_test.py
index 22cf89f..1e447d0 100755
--- a/conman/connection_manager_test.py
+++ b/conman/connection_manager_test.py
@@ -106,7 +106,7 @@
'--bridge=br0', '-s', 'my ssid=1', '--interface-suffix', '_suffix',
])
config = connection_manager.WLANConfiguration(
- '5', interface_test.Wifi('wcli0', 20), cmd, None)
+ '5', interface_test.Wifi('wcli0', 20), cmd)
wvtest.WVPASSEQ('my ssid=1', config.ssid)
wvtest.WVPASSEQ('abcdWIFI_PSK=qwer', config.passphrase)
@@ -157,10 +157,6 @@
subprocess.mock('wifi', 'remote_ap', band=band, ssid=ssid, psk=psk,
bssid='00:00:00:00:00:00')
- # Also create the wpa_supplicant socket to which to attach.
- open(os.path.join(kwargs['wpa_control_interface'], interface_name),
- 'w')
-
super(ConnectionManager, self).__init__(*args, **kwargs)
# Just looking for last_wifi_scan_time to change doesn't work because the
@@ -261,7 +257,6 @@
moca_tmp_dir = tempfile.mkdtemp()
wpa_control_interface = tempfile.mkdtemp()
subprocess.mock('wifi', 'wpa_path', wpa_control_interface)
- FrenzyWifi.WPACtrl.WIFIINFO_PATH = tempfile.mkdtemp()
connection_manager.CWMP_PATH = tempfile.mkdtemp()
subprocess.set_conman_paths(tmp_dir, config_dir,
connection_manager.CWMP_PATH)
@@ -276,7 +271,6 @@
c = ConnectionManager(tmp_dir=tmp_dir,
config_dir=config_dir,
moca_tmp_dir=moca_tmp_dir,
- wpa_control_interface=wpa_control_interface,
run_duration_s=run_duration_s,
interface_update_period=interface_update_period,
wlan_retry_s=0,
@@ -301,7 +295,6 @@
shutil.rmtree(config_dir)
shutil.rmtree(moca_tmp_dir)
shutil.rmtree(wpa_control_interface)
- shutil.rmtree(FrenzyWifi.WPACtrl.WIFIINFO_PATH)
shutil.rmtree(connection_manager.CWMP_PATH)
actual_test.func_name = f.func_name
@@ -478,6 +471,7 @@
ssid = 'wlan2'
psk = 'password2'
subprocess.mock('cwmp', band, ssid=ssid, psk=psk)
+ # Overwrites previous one due to same BSSID.
subprocess.mock('wifi', 'remote_ap',
bssid='11:22:33:44:55:66',
ssid=ssid, psk=psk, band=band, security='WPA2')
@@ -487,10 +481,6 @@
wvtest.WVPASS(c._connected_to_open(c.wifi_for_band(band)))
wvtest.WVPASSEQ(c.wifi_for_band(band).last_attempted_bss_info.ssid, 's2')
- # Overwrites previous one due to same BSSID.
- subprocess.mock('wifi', 'remote_ap',
- bssid='11:22:33:44:55:66',
- ssid=ssid, psk=psk, band=band, security='WPA2')
# Run once for cwmp wakeup to get called, then once more for the new config to
# be received.
c.run_once()
diff --git a/conman/interface.py b/conman/interface.py
index b741555..68aa35b 100755
--- a/conman/interface.py
+++ b/conman/interface.py
@@ -12,7 +12,6 @@
logging.basicConfig(level=logging.DEBUG)
import experiment
-import wpactrl
METRIC_5GHZ = 20
METRIC_24GHZ_5GHZ = 21
@@ -428,57 +427,19 @@
class Wifi(Interface):
"""Represents a wireless interface."""
- WPA_EVENT_RE = re.compile(r'<\d+>CTRL-EVENT-(?P<event>[A-Z\-]+).*')
- # pylint: disable=invalid-name
- WPACtrl = wpactrl.WPACtrl
-
def __init__(self, *args, **kwargs):
self.bands = kwargs.pop('bands', [])
super(Wifi, self).__init__(*args, **kwargs)
- self._wpa_control = None
- self.initial_ssid = None
@property
def wpa_supplicant(self):
+ self.update()
return 'wpa_supplicant' in self.links
@wpa_supplicant.setter
def wpa_supplicant(self, is_up):
self._set_link_status('wpa_supplicant', is_up)
- def attached(self):
- return self._wpa_control and self._wpa_control.attached
-
- def attach_wpa_control(self, path):
- """Attach to the wpa_supplicant control interface.
-
- Args:
- path: The path containing the wpa_supplicant control interface socket.
-
- Returns:
- Whether attaching was successful.
- """
- if self.attached():
- return True
-
- socket = os.path.join(path, self.name)
- logging.debug('%s socket is %s', self.name, socket)
- try:
- self._wpa_control = self.get_wpa_control(socket)
- self._wpa_control.attach()
- logging.debug('%s successfully attached', self.name)
- except (wpactrl.error, OSError) as e:
- logging.error('Error attaching to wpa_supplicant: %s', e)
- return False
-
- status = self.wpa_status()
- logging.debug('%s status after attaching is %s', self.name, status)
- self.wpa_supplicant = status.get('wpa_state') == 'COMPLETED'
- if not self._initialized:
- self.initial_ssid = status.get('ssid')
-
- return True
-
def wpa_status(self):
"""Parse the STATUS response from the wpa_supplicant control interface.
@@ -488,82 +449,41 @@
"""
status = {}
- if self.attached():
- lines = []
- try:
- lines = self._wpa_control.request('STATUS').splitlines()
- except (wpactrl.error, OSError) as e:
- logging.error('wpa_control STATUS request failed %s args %s',
- e.message, e.args)
- lines = self.wpa_cli_status().splitlines()
- for line in lines:
- if '=' not in line:
- continue
- k, v = line.strip().split('=', 1)
- status[k] = v
+ try:
+ lines = subprocess.check_output(['wpa_cli', '-i', self.name,
+ 'status']).splitlines()
+ except subprocess.CalledProcessError:
+ logging.error('wpa_cli status request failed')
+ return {}
+
+ for line in lines:
+ if '=' not in line:
+ continue
+ k, v = line.strip().split('=', 1)
+ status[k] = v
logging.debug('wpa_status is %r', status)
return status
- def get_wpa_control(self, socket):
- return self.WPACtrl(socket)
-
- def detach_wpa_control(self):
- if self.attached():
- try:
- self._wpa_control.detach()
- except (wpactrl.error, OSError):
- logging.error('Failed to detach from wpa_supplicant interface. This '
- 'may mean something else killed wpa_supplicant.')
- self._wpa_control = None
-
- self.wpa_supplicant = False
-
- def handle_wpa_events(self):
- if not self.attached():
- self.wpa_supplicant = False
- return
-
- # b/31261343: Make sure we didn't miss wpa_supplicant being up.
+ def update(self):
self.wpa_supplicant = self.wpa_status().get('wpa_state', '') == 'COMPLETED'
- while self._wpa_control.pending():
- match = self.WPA_EVENT_RE.match(self._wpa_control.recv())
- if match:
- event = match.group('event')
- logging.debug('%s got wpa_supplicant event %s', self.name, event)
- if event == 'CONNECTED':
- self.wpa_supplicant = True
- elif event in ('DISCONNECTED', 'TERMINATING', 'ASSOC-REJECT',
- 'SSID-TEMP-DISABLED', 'AUTH-REJECT'):
- self.wpa_supplicant = False
- if event == 'TERMINATING':
- self.detach_wpa_control()
- break
-
- self.update_routes()
-
- def initialize(self):
- """Unset self.initial_ssid, which is only relevant during initialization."""
- self.initial_ssid = None
- super(Wifi, self).initialize()
-
def connected_to_open(self):
status = self.wpa_status()
return (status.get('wpa_state', None) == 'COMPLETED' and
status.get('key_mgmt', None) == 'NONE')
- # TODO(rofrankel): Remove this if and when the wpactrl failures are fixed.
- def wpa_cli_status(self):
- """Fallback for wpa_supplicant control interface status requests."""
- try:
- return subprocess.check_output(['wpa_cli', '-i', self.name, 'status'])
- except subprocess.CalledProcessError:
- logging.error('wpa_cli status request failed')
- return ''
+ def current_secure_ssid(self):
+ """Returns SSID if connected to a secure network, False otherwise."""
+ status = self.wpa_status()
+ return (status.get('wpa_state', None) == 'COMPLETED' and
+ # NONE indicates we're on a provisioning network; anything else
+ # suggests we're already on the WLAN.
+ status.get('key_mgmt', None) != 'NONE' and
+ status.get('ssid'))
-class FrenzyWPACtrl(object):
+class FrenzyWifi(Wifi):
"""A WPACtrl for Frenzy devices.
Implements the same functions used on the normal WPACtrl, using a combination
@@ -571,19 +491,6 @@
diffing saved state with current system state.
"""
- WIFIINFO_PATH = '/tmp/wifi/wifiinfo'
-
- def __init__(self, socket):
- self.ctrl_iface_path, self._interface = os.path.split(socket)
-
- # State from QCSAPI and wifi_files.
- self._client_mode = False
- self._ssid = None
- self._status = None
- self._security = None
-
- self._events = []
-
def _qcsapi(self, *command):
try:
return subprocess.check_output(['qcsapi'] + list(command)).strip()
@@ -591,80 +498,27 @@
logging.error('QCSAPI call failed: %s: %s', e, e.output)
raise
- def attach(self):
- self._update()
-
- @property
- def attached(self):
- return self._client_mode
-
- def detach(self):
- self._events = []
- raise wpactrl.error('Real WPACtrl always raises this when detaching.')
-
- def pending(self):
- self._update()
- return bool(self._events)
-
- def _update(self):
+ def wpa_status(self):
"""Generate and cache events, update state."""
try:
client_mode = self._qcsapi('get_mode', 'wifi0') == 'Station'
ssid = self._qcsapi('get_ssid', 'wifi0')
- status = self._qcsapi('get_status', 'wifi0')
security = (self._qcsapi('ssid_get_authentication_mode', 'wifi0', ssid)
if ssid else None)
except subprocess.CalledProcessError:
- # If QCSAPI failed, skip update.
- return
+ # If QCSAPI failed, don't crash.
+ return {}
- # If we have an SSID and are in client mode, and at least one of those is
- # new, then we have just connected.
- if client_mode and ssid and (not self._client_mode or ssid != self._ssid):
- self._events.append('<2>CTRL-EVENT-CONNECTED')
+ up = bool(client_mode and ssid)
+ self.wpa_supplicant = up
- # If we are in client mode but lost SSID, we disconnected.
- if client_mode and self._ssid and not ssid:
- self._events.append('<2>CTRL-EVENT-DISCONNECTED')
-
- # If there is an auth/assoc failure, then status (above) is 'Error'. We
- # really want the converse of this implication (i.e. that 'Error' implies an
- # auth/assoc failure), but due to limited documentation this will have to
- # do. It should be good enough: if something else causes get_status to
- # return 'Error', we are probably not connected, and we don't do anything
- # special with auth/assoc failures specifically.
- if client_mode and status == 'Error' and self._status != 'Error':
- self._events.append('<2>CTRL-EVENT-SSID-TEMP-DISABLED')
-
- # If we left client mode, wpa_supplicant has terminated.
- if self._client_mode and not client_mode:
- self._events.append('<2>CTRL-EVENT-TERMINATING')
-
- self._client_mode = client_mode
- self._ssid = ssid
- self._status = status
- self._security = security
-
- def recv(self):
- return self._events.pop(0)
-
- def request(self, request_type):
- """Partial implementation of WPACtrl.request."""
-
- if request_type != 'STATUS':
- return ''
-
- self._update()
-
- if not self._client_mode or not self._ssid:
- return ''
-
- return ('wpa_state=COMPLETED\nssid=%s\nkey_mgmt=%s' %
- (self._ssid, self._security or 'NONE'))
-
-
-class FrenzyWifi(Wifi):
- """Represents a Frenzy wireless interface."""
-
- # pylint: disable=invalid-name
- WPACtrl = FrenzyWPACtrl
+ if up:
+ return {
+ 'wpa_state': 'COMPLETED',
+ 'ssid': ssid,
+ 'key_mgmt': security or 'NONE',
+ }
+ else:
+ return {
+ 'wpa_state': 'SCANNING',
+ }
diff --git a/conman/interface_test.py b/conman/interface_test.py
index 14fb795..f080285 100755
--- a/conman/interface_test.py
+++ b/conman/interface_test.py
@@ -176,7 +176,6 @@
def generic_wifi_test(w, wpa_path):
# Not currently connected.
subprocess.wifi.WPA_PATH = wpa_path
- w.attach_wpa_control(wpa_path)
wvtest.WVFAIL(w.wpa_supplicant)
# wpa_supplicant connects.
@@ -186,38 +185,15 @@
bssid='00:00:00:00:00:00', connection_check_result='succeed')
subprocess.check_call(['wifi', 'setclient', '--ssid', ssid, '--band', '5'],
env={'WIFI_CLIENT_PSK': psk})
- wvtest.WVFAIL(w.wpa_supplicant)
- w.attach_wpa_control(wpa_path)
- w.handle_wpa_events()
wvtest.WVPASS(w.wpa_supplicant)
w.set_gateway_ip('192.168.1.1')
# wpa_supplicant disconnects.
subprocess.mock('wifi', 'disconnected_event', '5')
- w.handle_wpa_events()
wvtest.WVFAIL(w.wpa_supplicant)
- # Now, start over so we can test what happens when wpa_supplicant is already
- # connected when we attach.
- w.detach_wpa_control()
- w._initialized = False
- subprocess.check_call(['wifi', 'setclient', '--ssid', ssid, '--band', '5'],
- env={'WIFI_CLIENT_PSK': psk})
- w.attach_wpa_control(wpa_path)
-
- # wpa_supplicant was already connected when we attached.
- wvtest.WVPASS(w.wpa_supplicant)
- wvtest.WVPASSEQ(w.initial_ssid, ssid)
- w.initialize()
- wvtest.WVPASSEQ(w.initial_ssid, None)
-
- wvtest.WVPASSNE(w.wpa_status(), {})
- w._wpa_control.request_status_fails = True
- wvtest.WVPASSNE(w.wpa_status(), {})
-
# The wpa_supplicant process disconnects and terminates.
subprocess.check_call(['wifi', 'stopclient', '--band', '5'])
- w.handle_wpa_events()
wvtest.WVFAIL(w.wpa_supplicant)
@@ -254,14 +230,11 @@
subprocess.mock('wifi', 'interfaces',
subprocess.wifi.MockInterface(phynum='0', bands=['5'],
driver='frenzy'))
- FrenzyWifi.WPACtrl.WIFIINFO_PATH = wifiinfo_path = tempfile.mkdtemp()
-
generic_wifi_test(w, wpa_path)
finally:
shutil.rmtree(wpa_path)
shutil.rmtree(conman_path)
- shutil.rmtree(wifiinfo_path)
@wvtest.wvtest
@@ -329,62 +302,5 @@
shutil.rmtree(interface.CWMP_PATH)
-@wvtest.wvtest
-def b31261343_test():
- """Test Wifi."""
- w = Wifi('wcli0', '21')
- w.initialize()
-
- try:
- wpa_path = tempfile.mkdtemp()
- conman_path = tempfile.mkdtemp()
- subprocess.set_conman_paths(conman_path, None)
- subprocess.mock('wifi', 'interfaces',
- subprocess.wifi.MockInterface(phynum='0', bands=['5'],
- driver='cfg80211'))
- subprocess.wifi.WPA_PATH = wpa_path
-
- w.attach_wpa_control(wpa_path)
- wvtest.WVFAIL(w.wpa_supplicant)
-
- # Set up.
- ssid = 'my=ssid'
- psk = 'passphrase'
- subprocess.mock('wifi', 'remote_ap', ssid=ssid, psk=psk, band='5',
- bssid='00:00:00:00:00:00', connection_check_result='succeed')
- subprocess.check_call(['wifi', 'setclient', '--ssid', ssid, '--band', '5'],
- env={'WIFI_CLIENT_PSK': psk})
-
- w.set_gateway_ip('192.168.1.1')
- w.set_subnet('192.168.1.0/24')
- wvtest.WVFAIL(w.wpa_supplicant)
- w.attach_wpa_control(wpa_path)
- w.handle_wpa_events()
-
- def check_working():
- w.update_routes(True)
- wvtest.WVPASS(w.wpa_supplicant)
- wvtest.WVPASS('default' in w.current_routes())
-
- def check_broken():
- w.update_routes(True)
- wvtest.WVFAIL(w.wpa_supplicant)
- wvtest.WVFAIL('default' in w.current_routes())
-
- check_working()
-
- # This is the buggy state.
- w.wpa_supplicant = False
- check_broken()
-
- # Should fix itself when we next run handle_wpa_events.
- w.handle_wpa_events()
- check_working()
-
- finally:
- shutil.rmtree(wpa_path)
- shutil.rmtree(conman_path)
-
-
if __name__ == '__main__':
wvtest.wvtest_main()
diff --git a/conman/status.py b/conman/status.py
index c5d4187..b5b1bae 100644
--- a/conman/status.py
+++ b/conman/status.py
@@ -30,7 +30,6 @@
COULD_REACH_ACS = 'COULD_REACH_ACS'
CAN_REACH_INTERNET = 'CAN_REACH_INTERNET'
PROVISIONING_FAILED = 'PROVISIONING_FAILED'
- ATTACHED_TO_WPA_SUPPLICANT = 'ATTACHED_TO_WPA_SUPPLICANT'
WAITING_FOR_PROVISIONING = 'WAITING_FOR_PROVISIONING'
WAITING_FOR_DHCP = 'WAITING_FOR_DHCP'
@@ -82,10 +81,6 @@
(P.HAVE_CONFIG,),
(),
),
- P.ATTACHED_TO_WPA_SUPPLICANT: (
- (),
- (),
- ),
P.WAITING_FOR_PROVISIONING: (
(P.CONNECTED_TO_OPEN,),
(),
diff --git a/conman/test/fake_python/subprocess/wifi.py b/conman/test/fake_python/subprocess/wifi.py
index 13d1be3..900b908 100644
--- a/conman/test/fake_python/subprocess/wifi.py
+++ b/conman/test/fake_python/subprocess/wifi.py
@@ -67,8 +67,9 @@
class AccessPoint(object):
def __init__(self, **kwargs):
- for attr in ('ssid', 'psk', 'band', 'bssid', 'security', 'rssi',
- 'vendor_ies', 'connection_check_result', 'hidden'):
+ self._attrs = ('ssid', 'psk', 'band', 'bssid', 'security', 'rssi',
+ 'vendor_ies', 'connection_check_result', 'hidden')
+ for attr in self._attrs:
setattr(self, attr, kwargs.get(attr, None))
def scan_str(self):
@@ -86,6 +87,13 @@
rssi='%.2f dBm' % (self.rssi or 0),
security=security_strs.get(self.security, ''))
+ def __str__(self):
+ return 'AccessPoint<%s>' % ' '.join('%s=%s' % (attr, getattr(self, attr))
+ for attr in self._attrs)
+
+ def __repr__(self):
+ return str(self)
+
def call(*args, **kwargs):
wifi_commands = {
@@ -140,14 +148,14 @@
ap = REMOTE_ACCESS_POINTS[band].get(bssid, None)
if not ap or ap.ssid != ssid:
_setclient_error_not_found(interface_name, ssid, interface.driver)
- return 1, ('AP with band %r and BSSID %r and ssid %s not found'
- % (band, bssid, ssid))
+ return 1, ('AP with band %r and BSSID %r and ssid %s not found: %s'
+ % (band, bssid, ssid, REMOTE_ACCESS_POINTS))
elif ssid:
candidates = [ap for ap in REMOTE_ACCESS_POINTS[band].itervalues()
if ap.ssid == ssid]
if not candidates:
_setclient_error_not_found(interface_name, ssid, interface.driver)
- return 1, 'AP with SSID %r not found' % ssid
+ return 1, 'AP with SSID %r not found: %s' % (ssid, REMOTE_ACCESS_POINTS)
ap = random.choice(candidates)
else:
raise ValueError('Did not specify BSSID or SSID in %r' % args)
diff --git a/conman/test/fake_python/wpactrl.py b/conman/test/fake_python/wpactrl.py
deleted file mode 100644
index 3d8e300..0000000
--- a/conman/test/fake_python/wpactrl.py
+++ /dev/null
@@ -1,73 +0,0 @@
-#!/usr/bin/python
-
-"""Fake WPACtrl implementation."""
-
-import os
-
-import subprocess
-import subprocess.wifi
-
-
-CONNECTED_EVENT = '<2>CTRL-EVENT-CONNECTED'
-DISCONNECTED_EVENT = '<2>CTRL-EVENT-DISCONNECTED'
-TERMINATING_EVENT = '<2>CTRL-EVENT-TERMINATING'
-
-
-# pylint: disable=invalid-name
-class error(Exception):
- pass
-
-
-class WPACtrl(object):
- """Fake wpactrl.WPACtrl."""
-
- # pylint: disable=unused-argument
- def __init__(self, wpa_socket):
- self._socket = wpa_socket
- self.interface_name = os.path.split(self._socket)[-1]
- self.attached = False
- self.connected = False
- self.request_status_fails = False
- self._clear_events()
-
- def pending(self):
- return bool(subprocess.wifi.INTERFACE_EVENTS[self.interface_name])
-
- def recv(self):
- return subprocess.wifi.INTERFACE_EVENTS[self.interface_name].pop(0)
-
- def attach(self):
- if not os.path.exists(self._socket):
- raise error('wpactrl_attach failed')
- self.attached = True
-
- def detach(self):
- self.attached = False
- self.connected = False
- self.check_socket_exists('wpactrl_detach failed')
- self._clear_events()
-
- def request(self, request_type):
- if request_type == 'STATUS':
- if self.request_status_fails:
- raise error('test error')
- try:
- return subprocess.check_output(['wpa_cli', '-i', self.interface_name,
- 'status'])
- except subprocess.CalledProcessError as e:
- raise error(e.output)
- else:
- raise ValueError('Invalid request_type %s' % request_type)
-
- @property
- def ctrl_iface_path(self):
- return os.path.split(self._socket)[0]
-
- # Below methods are not part of WPACtrl.
-
- def check_socket_exists(self, msg='Socket does not exist'):
- if not os.path.exists(self._socket):
- raise error(msg)
-
- def _clear_events(self):
- subprocess.wifi.INTERFACE_EVENTS[self.interface_name] = []
