blob: e8ab4caf00c2cfcad91be1f6f0e610c09ef2b3b8 [file] [log] [blame]
#!/usr/bin/python
"""Tests for connection_manager.py."""
import logging
import os
import shutil
import tempfile
import time
# This has to be called before another module calls it with a higher log level.
# pylint: disable=g-import-not-at-top
logging.basicConfig(level=logging.DEBUG)
# This is in site-packages on the device, but not when running tests, and so
# raises lint errors.
# pylint: disable=g-bad-import-order
import wpactrl
import experiment_testutils
import interface
from wvtest import wvtest
# pylint: disable=line-too-long
_IP_ADDR_SHOW_TPL = """4: {name}: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP qlen 1000
inet {ip}/21 brd 100.100.55.255 scope global {name}
valid_lft forever preferred_lft forever
"""
class FakeInterfaceMixin(object):
"""Replace Interface methods which interact with the system."""
def __init__(self, *args, **kwargs):
super(FakeInterfaceMixin, self).__init__(*args, **kwargs)
self.set_connection_check_result('succeed')
self.routing_table = {}
self.ip_testonly = None
def set_connection_check_result(self, result):
if result in ['succeed', 'fail', 'restricted']:
# pylint: disable=invalid-name
self.CONNECTION_CHECK = './test/' + result
else:
raise ValueError('Invalid fake connection_check script.')
def _really_ip_route(self, *args):
if not args:
return '\n'.join(self.routing_table.values() +
['1.2.3.4/24 dev fake0 proto kernel scope link',
# Non-subnet route, e.g. to NFS host.
'1.2.3.1 dev %s proto kernel scope link' % self.name,
'default via 1.2.3.4 dev fake0',
'random junk'])
metric = None
if 'metric' in args:
metric = args[args.index('metric') + 1]
if args[0] in ('add', 'del'):
route = args[1]
key = (self.name, route, metric)
if args[0] == 'add' and key not in self.routing_table:
logging.debug('Adding route for %r', key)
self.routing_table[key] = ' '.join(args[1:])
elif args[0] == 'del':
if key in self.routing_table:
logging.debug('Deleting route for %r', key)
del self.routing_table[key]
elif key[2] is None:
# pylint: disable=g-builtin-op
for k in self.routing_table.keys():
if k[:-1] == key[:-1]:
logging.debug('Deleting route for %r (generalized from %s)', k, key)
del self.routing_table[k]
break
def _ip_addr_show(self):
if self.ip_testonly:
return _IP_ADDR_SHOW_TPL.format(name=self.name, ip=self.ip_testonly)
return ''
def current_routes_normal_testonly(self):
result = self.current_routes()
return {k: v for k, v in result.iteritems() if int(v.get('metric', 0)) < 50}
class Bridge(FakeInterfaceMixin, interface.Bridge):
pass
class FakeWPACtrl(object):
"""Fake wpactrl.WPACtrl."""
# pylint: disable=unused-argument
def __init__(self, socket):
self._socket = socket
self.events = []
self.attached = False
self.connected = False
self.ssid_testonly = None
self.secure_testonly = False
self.request_status_fails = False
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):
if not os.path.exists(self._socket):
raise wpactrl.error('wpactrl_attach failed')
self.attached = True
def detach(self):
self.attached = False
self.ssid_testonly = None
self.secure_testonly = False
self.connected = False
self.check_socket_exists('wpactrl_detach failed')
def request(self, request_type):
if request_type == 'STATUS':
if self.request_status_fails:
raise wpactrl.error('test error')
if self.connected:
return ('foo\nwpa_state=COMPLETED\nssid=%s\nkey_mgmt=%s\nbar' %
(self.ssid_testonly,
'WPA2-PSK' if self.secure_testonly else 'NONE'))
else:
return 'wpa_state=SCANNING\naddress=12:34:56:78:90:ab'
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 add_event(self, event):
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):
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."""
CONNECTED_EVENT = '<2>CTRL-EVENT-CONNECTED'
DISCONNECTED_EVENT = '<2>CTRL-EVENT-DISCONNECTED'
TERMINATING_EVENT = '<2>CTRL-EVENT-TERMINATING'
WPACtrl = FakeWPACtrl
def __init__(self, *args, **kwargs):
super(Wifi, self).__init__(*args, **kwargs)
self._initial_ssid_testonly = None
self._secure_testonly = False
def attach_wpa_control(self, path):
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)
if self._initial_ssid_testonly:
result.connected = True
result.ssid_testonly = self._initial_ssid_testonly
result.secure_testonly = self._secure_testonly
return result
def add_connected_event(self):
if self.attached():
self._wpa_control.add_connected_event()
def add_disconnected_event(self):
self._initial_ssid_testonly = None
self._secure_testonly = False
if self.attached():
self._wpa_control.add_disconnected_event()
def add_terminating_event(self):
self._initial_ssid_testonly = None
self._secure_testonly = False
if self.attached():
self._wpa_control.add_terminating_event()
def detach_wpa_control(self):
self._initial_ssid_testonly = None
self._secure_testonly = False
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.ssid_testonly = None
self.secure_testonly = False
self.request_status_fails = False
def _qcsapi(self, *command):
return self.fake_qcsapi.get(command[0], None)
def add_connected_event(self):
self.fake_qcsapi['get_mode'] = 'Station'
self.fake_qcsapi['get_ssid'] = self.ssid_testonly
security = 'PSKAuthentication' if self.secure_testonly else 'NONE'
self.fake_qcsapi['ssid_get_authentication_mode'] = security
def add_disconnected_event(self):
self.ssid_testonly = None
self.secure_testonly = False
self.fake_qcsapi['get_ssid'] = None
self.fake_qcsapi['ssid_get_authentication_mode'] = 'NONE'
def add_terminating_event(self):
self.ssid_testonly = None
self.secure_testonly = False
self.fake_qcsapi['get_ssid'] = None
self.fake_qcsapi['get_mode'] = 'AP'
self.fake_qcsapi['ssid_get_authentication_mode'] = 'NONE'
def detach(self):
self.add_terminating_event()
super(FrenzyWPACtrl, self).detach()
def request(self, request_type):
if request_type == 'STATUS' and self.request_status_fails:
raise wpactrl.error('test error')
return super(FrenzyWPACtrl, self).request(request_type)
class FrenzyWifi(FakeInterfaceMixin, interface.FrenzyWifi):
WPACtrl = FrenzyWPACtrl
def __init__(self, *args, **kwargs):
super(FrenzyWifi, self).__init__(*args, **kwargs)
self._initial_ssid_testonly = None
self._secure_testonly = False
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.ssid_testonly = self._initial_ssid_testonly
self._wpa_control.secure_testonly = self._secure_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)
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.secure_testonly = self._secure_testonly
result.add_connected_event()
return result
def add_connected_event(self):
if self.attached():
self._wpa_control.add_connected_event()
def add_disconnected_event(self):
self._initial_ssid_testonly = None
self._secure_testonly = False
if self.attached():
self._wpa_control.add_disconnected_event()
def add_terminating_event(self):
self._initial_ssid_testonly = None
self._secure_testonly = False
if self.attached():
self._wpa_control.add_terminating_event()
def detach_wpa_control(self):
self._initial_ssid_testonly = None
self._secure_testonly = False
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():
"""Test Interface and Bridge."""
tmp_dir = tempfile.mkdtemp()
try:
autoprov_filepath = os.path.join(tmp_dir, 'autoprov')
b = Bridge('br0', '10', acs_autoprovisioning_filepath=autoprov_filepath)
b.set_connection_check_result('succeed')
wvtest.WVFAIL(b.acs())
wvtest.WVFAIL(b.internet())
wvtest.WVFAIL(b.current_routes())
wvtest.WVFAIL(b.current_routes_normal_testonly())
wvtest.WVFAIL(os.path.exists(autoprov_filepath))
b.add_moca_station(0)
wvtest.WVFAIL(os.path.exists(autoprov_filepath))
b.set_gateway_ip('192.168.1.1')
b.set_subnet('192.168.1.0/24')
wvtest.WVFAIL(os.path.exists(autoprov_filepath))
# Everything should fail because the interface is not initialized.
wvtest.WVFAIL(b.acs())
wvtest.WVFAIL(b.internet())
wvtest.WVFAIL(b.current_routes_normal_testonly())
wvtest.WVFAIL(os.path.exists(autoprov_filepath))
b.initialize()
wvtest.WVPASS(b.acs())
wvtest.WVPASS(b.internet())
current_routes = b.current_routes()
wvtest.WVPASSEQ(len(current_routes), 3)
wvtest.WVPASS('default' in current_routes)
wvtest.WVPASS('subnet' in current_routes)
wvtest.WVPASS('multicast' in current_routes)
wvtest.WVPASS(os.path.exists(autoprov_filepath))
b.add_moca_station(1)
wvtest.WVPASS(b.acs())
wvtest.WVPASS(b.internet())
wvtest.WVPASSEQ(len(b.current_routes()), 3)
wvtest.WVPASS(os.path.exists(autoprov_filepath))
b.remove_moca_station(0)
b.remove_moca_station(1)
wvtest.WVFAIL(b.acs())
wvtest.WVFAIL(b.internet())
# We have no links, so should have no routes.
wvtest.WVFAIL(b.current_routes())
wvtest.WVFAIL(os.path.exists(autoprov_filepath))
b.add_moca_station(2)
wvtest.WVPASS(b.acs())
wvtest.WVPASS(b.internet())
wvtest.WVPASSEQ(len(b.current_routes()), 3)
wvtest.WVPASS(os.path.exists(autoprov_filepath))
b.set_connection_check_result('fail')
b.update_routes()
wvtest.WVFAIL(b.acs())
wvtest.WVFAIL(b.internet())
# We have links but the connection check failed, so we should only have a
# low priority route, i.e. metric at least 50.
wvtest.WVPASSEQ(len(b.current_routes()), 3)
wvtest.WVFAIL(b.current_routes_normal_testonly())
wvtest.WVFAIL(os.path.exists(autoprov_filepath))
b.set_connection_check_result('restricted')
b.update_routes()
wvtest.WVPASS(b.acs())
wvtest.WVFAIL(b.internet())
wvtest.WVPASSEQ(len(b.current_routes()), 3)
wvtest.WVPASS(os.path.exists(autoprov_filepath))
wvtest.WVFAIL(b.get_ip_address())
b.ip_testonly = '192.168.1.100'
wvtest.WVPASSEQ(b.get_ip_address(), '192.168.1.100')
finally:
shutil.rmtree(tmp_dir)
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)
# pylint: disable=protected-access
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()
wvtest.WVPASS(w.wpa_supplicant)
w.set_gateway_ip('192.168.1.1')
# wpa_supplicant disconnects.
wpa_control.add_disconnected_event()
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()
# pylint: disable=protected-access
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')
w.initialize()
wvtest.WVPASSEQ(w.initial_ssid, None)
wvtest.WVPASSNE(w.wpa_status(), {})
w._wpa_control.request_status_fails = True
wvtest.WVPASSEQ(w.wpa_status(), {})
# The wpa_supplicant process disconnects and terminates.
wpa_control.add_disconnected_event()
wpa_control.add_terminating_event()
w.handle_wpa_events()
wvtest.WVFAIL(w.wpa_supplicant)
@wvtest.wvtest
def wifi_test():
"""Test Wifi."""
w = Wifi('wcli0', '21')
w.set_connection_check_result('succeed')
w.initialize()
try:
wpa_path = tempfile.mkdtemp()
generic_wifi_test(w, wpa_path)
finally:
shutil.rmtree(wpa_path)
@wvtest.wvtest
def frenzy_wifi_test():
"""Test FrenzyWifi."""
w = FrenzyWifi('wlan0', '20')
w.set_connection_check_result('succeed')
w.initialize()
try:
wpa_path = tempfile.mkdtemp()
FrenzyWifi.WPACtrl.WIFIINFO_PATH = wifiinfo_path = tempfile.mkdtemp()
generic_wifi_test(w, wpa_path)
finally:
shutil.rmtree(wpa_path)
shutil.rmtree(wifiinfo_path)
@wvtest.wvtest
def simulate_wireless_test():
"""Test the WifiSimulateWireless experiment."""
unused_raii = experiment_testutils.MakeExperimentDirs()
tmp_dir = tempfile.mkdtemp()
interface.CWMP_PATH = tempfile.mkdtemp()
interface.MAX_ACS_FAILURE_S = 1
contact = os.path.join(interface.CWMP_PATH, 'acscontact')
connected = os.path.join(interface.CWMP_PATH, 'acsconnected')
try:
autoprov_filepath = os.path.join(tmp_dir, 'autoprov')
b = Bridge('br0', '10', acs_autoprovisioning_filepath=autoprov_filepath)
b.add_moca_station(0)
b.set_gateway_ip('192.168.1.1')
b.set_connection_check_result('succeed')
b.initialize()
# Initially, the connection check passes.
wvtest.WVPASS(b.internet())
# Enable the experiment.
experiment_testutils.enable('WifiSimulateWireless')
# Calling update_routes overwrites the connection status cache, which we
# need in order to see the effects we are looking for immediately
# (ConnectionManager calls this every few seconds).
b.update_routes()
wvtest.WVFAIL(b.internet())
# Create an ACS connection attempt.
open(contact, 'w')
b.update_routes()
wvtest.WVFAIL(b.internet())
# Record success.
open(connected, 'w')
b.update_routes()
wvtest.WVFAIL(b.internet())
# Disable the experiment and the connection check should pass again.
experiment_testutils.disable('WifiSimulateWireless')
b.update_routes()
wvtest.WVPASS(b.internet())
# Reenable the experiment and the connection check should fail again.
experiment_testutils.enable('WifiSimulateWireless')
b.update_routes()
wvtest.WVFAIL(b.internet())
# Wait until we've failed for long enough for the experiment to "expire",
# then log another attempt without success. Make sure the connection check
# passes.
time.sleep(interface.MAX_ACS_FAILURE_S)
open(contact, 'w')
b.update_routes()
wvtest.WVPASS(b.internet())
finally:
shutil.rmtree(tmp_dir)
shutil.rmtree(interface.CWMP_PATH)
if __name__ == '__main__':
wvtest.wvtest_main()