conman: Wait for provisioning properly.
Adds a Ratchet class that allows us to track, and optionally time out,
each step of the process (DHCP lease, ACS session started, ACS session
completed), and call 'cwmp wakeup' after receiving a DHCP lease to
trigger an ACS session. This subsumes the existing wait-for-DHCP
code.
The Ratchet class may seem overengineered at first glance, but I
started writing the alternative and it was leading to spaghetti code
that was hard to test (and would be hard to maintain). Now the serial
timeout logic is separate and can be tested cleanly.
Other minor fixes/refactors:
- Clear gateway_ip and subnet before all 'wifi setclient' calls,
in particular WLANConfiguration calls (this was already done for
provisioning calls).
- Create a ConnectionManager.interfaces() method to replace multiple
references to "[self.bridge] + self.wifi".
- Remove some spurious variables in connection_manager_test.
- Each interface now has its own Status, and there is a composite
Status that takes the disjunction of each interface's status. In
practice, this should not cause problems (and was necessary to
resolve some), but it does mean that the contents of
/tmp/conman/status may not provide the same coherent summary as
originally intended. Any future /bin/ledmonitor changes will have to
take this into account (or the composite status will have to be made
smarter). There may be some edge cases for two-radio devices which
will need to be resolved before we launch any which run conman (e.g.
as written, ledmonitor will flash fast blue if CONNECTED_TO_OPEN is
present, even if CONNECTED_TO_WLAN is also present, which could
happen with a two-radio device).
Change-Id: I8e44babba431f70dcc7a3320c33415b292c05271
diff --git a/conman/connection_manager.py b/conman/connection_manager.py
index d530639..5e6edd0 100755
--- a/conman/connection_manager.py
+++ b/conman/connection_manager.py
@@ -23,11 +23,13 @@
import experiment
import interface
import iw
+import ratchet
import status
HOSTNAME = socket.gethostname()
TMP_HOSTS = '/tmp/hosts'
+CWMP_PATH = '/tmp/cwmp'
experiment.register('WifiNo2GClient')
@@ -63,7 +65,7 @@
WIFI_SETCLIENT = ['wifi', 'setclient', '--persist']
WIFI_STOPCLIENT = ['wifi', 'stopclient', '--persist']
- def __init__(self, band, wifi, command_lines, _status, wpa_control_interface):
+ def __init__(self, band, wifi, command_lines, wpa_control_interface):
self.band = band
self.wifi = wifi
self.command = command_lines.splitlines()
@@ -72,7 +74,6 @@
self.passphrase = None
self.interface_suffix = None
self.access_point = None
- self._status = _status
self._wpa_control_interface = wpa_control_interface
binwifi_option_attrs = {
@@ -161,22 +162,24 @@
Returns:
Whether the command succeeded.
"""
+ self.wifi.set_gateway_ip(None)
+ self.wifi.set_subnet(None)
command = self.WIFI_SETCLIENT + ['--ssid', self.ssid, '--band', self.band]
env = dict(os.environ)
if self.passphrase:
env['WIFI_CLIENT_PSK'] = self.passphrase
try:
- self._status.trying_wlan = True
+ self.wifi.status.trying_wlan = True
subprocess.check_output(command, stderr=subprocess.STDOUT, env=env)
except subprocess.CalledProcessError as e:
logging.error('Failed to start wifi client: %s', e.output)
- self._status.wlan_failed = True
+ self.wifi.status.wlan_failed = True
return False
return True
def _post_start_client(self):
- self._status.connected_to_wlan = True
+ 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)
@@ -191,7 +194,7 @@
subprocess.check_output(self.WIFI_STOPCLIENT + ['-b', self.band],
stderr=subprocess.STDOUT)
# TODO(rofrankel): Make this work for dual-radio devices.
- self._status.connected_to_wlan = False
+ self.wifi.status.connected_to_wlan = False
logging.info('Stopped wifi client on %s GHz', self.band)
except subprocess.CalledProcessError as e:
logging.error('Failed to stop wifi client: %s', e.output)
@@ -221,6 +224,7 @@
IFPLUGD_ACTION = ['/etc/ifplugd/ifplugd.action']
BINWIFI = ['wifi']
UPLOAD_LOGS_AND_WAIT = ['timeout', '60', 'upload-logs-and-wait']
+ CWMP_WAKEUP = ['cwmp', 'wakeup']
def __init__(self,
bridge_interface='br0',
@@ -229,8 +233,9 @@
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=15, acs_update_wait_s=10,
- dhcp_wait_s=10, bssid_cycle_length_s=30):
+ wifi_scan_period_s=120, wlan_retry_s=15, associate_wait_s=15,
+ dhcp_wait_s=10, acs_start_wait_s=20, acs_finish_wait_s=120,
+ bssid_cycle_length_s=30):
self._tmp_dir = tmp_dir
self._config_dir = config_dir
@@ -242,8 +247,10 @@
self._interface_update_period = interface_update_period
self._wifi_scan_period_s = wifi_scan_period_s
self._wlan_retry_s = wlan_retry_s
- self._acs_update_wait_s = acs_update_wait_s
+ self._associate_wait_s = associate_wait_s
self._dhcp_wait_s = dhcp_wait_s
+ self._acs_start_wait_s = acs_start_wait_s
+ self._acs_finish_wait_s = acs_finish_wait_s
self._bssid_cycle_length_s = bssid_cycle_length_s
self._wlan_configuration = {}
self._try_to_upload_logs = False
@@ -263,7 +270,13 @@
self.create_wifi_interfaces()
- self._status = status.Status(self._status_dir)
+ for ifc in self.interfaces():
+ status_dir = os.path.join(self._status_dir, ifc.name)
+ if not os.path.exists(status_dir):
+ os.makedirs(status_dir)
+ ifc.status = status.Status(status_dir)
+ self._status = status.CompositeStatus(self._status_dir,
+ [i.status for i in self.interfaces()])
wm = pyinotify.WatchManager()
wm.add_watch(self._config_dir,
@@ -293,7 +306,7 @@
wifi_up = self.is_interface_up(wifi.name)
self.ifplugd_action(wifi.name, wifi_up)
if wifi_up:
- self._status.attached_to_wpa_supplicant = wifi.attach_wpa_control(
+ 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),
@@ -324,13 +337,32 @@
# Now that we've read any existing state, it's okay to let interfaces touch
# the routing table.
- for ifc in [self.bridge] + self.wifi:
+ for ifc in self.interfaces():
ifc.initialize()
logging.info('%s initialized', ifc.name)
self._interface_update_counter = 0
self._try_wlan_after = {'5': 0, '2.4': 0}
+ for wifi in self.wifi:
+ ratchet_name = '%s provisioning' % wifi.name
+ wifi.provisioning_ratchet = ratchet.Ratchet(ratchet_name, [
+ ratchet.Condition('trying_open', wifi.connected_to_open,
+ self._associate_wait_s,
+ callback=wifi.expire_connection_status_cache),
+ ratchet.Condition('waiting_for_dhcp', wifi.gateway, self._dhcp_wait_s,
+ callback=self.cwmp_wakeup),
+ ratchet.FileTouchedCondition('waiting_for_cwmp_wakeup',
+ os.path.join(CWMP_PATH, 'acscontact'),
+ self._acs_start_wait_s),
+ ratchet.FileTouchedCondition('waiting_for_acs_session',
+ os.path.join(CWMP_PATH, 'acsconnected'),
+ self._acs_finish_wait_s),
+ ], wifi.status)
+
+ def interfaces(self):
+ return [self.bridge] + self.wifi
+
def create_wifi_interfaces(self):
"""Create Wifi interfaces."""
@@ -444,7 +476,7 @@
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(
+ wifi.status.attached_to_wpa_supplicant = wifi.attach_wpa_control(
self._wpa_control_interface)
wifi.handle_wpa_events()
@@ -455,13 +487,13 @@
# If this interface is connected to the user's WLAN, there is nothing else
# to do.
if self._connected_to_wlan(wifi):
- self._status.connected_to_wlan = True
+ wifi.status.connected_to_wlan = True
logging.debug('Connected to WLAN on %s, nothing else to do.', wifi.name)
break
# This interface is not connected to the WLAN, so scan for potential
# routes to the ACS for provisioning.
- if (not self.acs() and
+ if ((not self.acs() or self.provisioning_failed(wifi)) and
not getattr(wifi, 'last_successful_bss_info', None) and
time.time() > wifi.last_wifi_scan_time + self._wifi_scan_period_s):
logging.debug('Performing scan on %s.', wifi.name)
@@ -479,12 +511,12 @@
wlan_configuration.start_client()
if self._connected_to_wlan(wifi):
logging.info('Joined WLAN on %s.', wifi.name)
- self._status.connected_to_wlan = True
+ wifi.status.connected_to_wlan = True
self._try_wlan_after[band] = 0
break
else:
logging.error('Failed to connect to WLAN on %s.', wifi.name)
- self._status.connected_to_wlan = False
+ wifi.status.connected_to_wlan = False
self._try_wlan_after[band] = time.time() + self._wlan_retry_s
else:
# If we are aren't on the WLAN, can ping the ACS, and haven't gotten a
@@ -494,7 +526,8 @@
# 2) cwmpd isn't writing a configuration, possibly because the device
# isn't registered to any accounts.
logging.debug('Unable to join WLAN on %s', wifi.name)
- self._status.connected_to_wlan = False
+ wifi.status.connected_to_wlan = False
+ provisioning_failed = self.provisioning_failed(wifi)
if self.acs():
logging.debug('Connected to ACS')
if self._try_to_upload_logs:
@@ -505,54 +538,52 @@
wifi.last_successful_bss_info = getattr(wifi,
'last_attempted_bss_info',
None)
+ if provisioning_failed:
+ wifi.last_successful_bss_info = None
+
now = time.time()
- if (self._wlan_configuration and
- hasattr(wifi, 'waiting_for_acs_since')):
- if now - wifi.waiting_for_acs_since > self._acs_update_wait_s:
- logging.info('ACS has not updated WLAN configuration; will retry '
- ' with old config.')
- for w in self.wifi:
- for b in w.bands:
- self._try_wlan_after[b] = now
- continue
+ if self._wlan_configuration:
+ logging.info('ACS has not updated WLAN configuration; will retry '
+ ' with old config.')
+ for w in self.wifi:
+ for b in w.bands:
+ self._try_wlan_after[b] = now
+ continue
# We don't want to want to log this spammily, so do exponential
# backoff.
elif (hasattr(wifi, 'complain_about_acs_at')
and now >= wifi.complain_about_acs_at):
- wait = wifi.complain_about_acs_at - wifi.waiting_for_acs_since
+ wait = wifi.complain_about_acs_at - self.provisioning_since(wifi)
logging.info('Can ping ACS, but no WLAN configuration for %ds.',
wait)
wifi.complain_about_acs_at += wait
- # If we didn't manage to join the WLAN and we don't have an ACS
- # connection, we should try to establish one.
- else:
- # If we are associated but waiting for a DHCP lease, try again later.
+ # If we didn't manage to join the WLAN, and we don't have an ACS
+ # connection or the ACS session failed, we should try another open AP.
+ if not self.acs() or provisioning_failed:
now = time.time()
- connected_to_open = (
- wifi.wpa_status().get('wpa_state', None) == 'COMPLETED' and
- wifi.wpa_status().get('key_mgmt', None) == 'NONE')
- wait_for_dhcp = (
- not wifi.gateway() and
- hasattr(wifi, 'waiting_for_dhcp_since') and
- now - wifi.waiting_for_dhcp_since < self._dhcp_wait_s)
- if connected_to_open and wait_for_dhcp:
- logging.debug('Waiting for DHCP lease after %ds.',
- now - wifi.waiting_for_acs_since)
+ if self._connected_to_open(wifi) and not provisioning_failed:
+ logging.debug('Waiting for provisioning for %ds.',
+ now - self.provisioning_since(wifi))
else:
- logging.debug('Not connected to ACS')
+ logging.debug('Not connected to ACS or provisioning failed')
self._try_next_bssid(wifi)
time.sleep(max(0, self._run_duration_s - (time.time() - start_time)))
def acs(self):
- result = self.bridge.acs() or any(wifi.acs() for wifi in self.wifi)
- self._status.can_reach_acs = result
+ result = False
+ for ifc in self.interfaces():
+ acs = ifc.acs()
+ ifc.status.can_reach_acs = acs
+ result |= acs
return result
def internet(self):
- result = self.bridge.internet() or any(wifi.internet()
- for wifi in self.wifi)
- self._status.can_reach_internet = result
+ result = False
+ for ifc in self.interfaces():
+ internet = ifc.internet()
+ ifc.status.can_reach_internet = internet
+ result |= internet
return result
def _update_interfaces_and_routes(self):
@@ -585,7 +616,7 @@
def _update_tmp_hosts(self):
"""Update the contents of /tmp/hosts."""
lowest_metric_interface = None
- for ifc in [self.bridge] + self.wifi:
+ for ifc in self.interfaces():
route = ifc.current_routes().get('default', None)
if route:
metric = route.get('metric', 0)
@@ -638,7 +669,7 @@
config.stop_access_point()
del self._wlan_configuration[band]
if not self._wlan_configuration:
- self._status.have_config = False
+ self.wifi_for_band(band).status.have_config = False
def _process_file(self, path, filename):
"""Process or ignore an updated file in a watched directory."""
@@ -672,7 +703,7 @@
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._wpa_control_interface))
elif filename.startswith(self.ACCESS_POINT_FILE_PREFIX):
match = re.match(self.ACCESS_POINT_FILE_REGEXP, filename)
@@ -719,7 +750,7 @@
self.ifplugd_action('moca0', has_moca)
def interface_by_name(self, interface_name):
- for ifc in [self.bridge] + self.wifi:
+ for ifc in self.interfaces():
if ifc.name == interface_name:
return ifc
@@ -774,20 +805,19 @@
if bss_info is not None:
logging.info('Attempting to connect to SSID %s (%s) for provisioning',
bss_info.ssid, bss_info.bssid)
- self._status.trying_open = True
- wifi.set_gateway_ip(None)
+ self.start_provisioning(wifi)
connected = self._try_bssid(wifi, bss_info)
if connected:
- self._status.connected_to_open = True
+ wifi.status.connected_to_open = True
now = time.time()
- wifi.waiting_for_acs_since = now
- wifi.waiting_for_dhcp_since = now
wifi.complain_about_acs_at = now + 5
logging.info('Attempting to provision via SSID %s', bss_info.ssid)
self._try_to_upload_logs = True
# If we can no longer connect to this, it's no longer successful.
- elif bss_info == last_successful_bss_info:
- wifi.last_successful_bss_info = None
+ else:
+ wifi.status.connected_to_open = False
+ if bss_info == last_successful_bss_info:
+ wifi.last_successful_bss_info = None
return connected
else:
# TODO(rofrankel): There are probably more cases in which this should be
@@ -795,7 +825,8 @@
# Relatedly, once we find ACS access on an open network we may want to
# save that SSID/BSSID and that first in future. If we do that then we
# can declare that provisioning has failed much more aggressively.
- self._status.provisioning_failed = True
+ logging.info('Ran out of BSSIDs to try on %s', wifi.name)
+ wifi.status.provisioning_failed = True
return False
@@ -812,6 +843,11 @@
in self._wlan_configuration.iteritems()
if band in wifi.bands))
+ def _connected_to_open(self, wifi):
+ result = wifi.connected_to_open()
+ wifi.status.connected_to_open = result
+ return result
+
def _update_wlan_configuration(self, wlan_configuration):
band = wlan_configuration.band
current = self._wlan_configuration.get(band, None)
@@ -825,7 +861,7 @@
if wlan_configuration.interface_suffix else '') + band)
wlan_configuration.access_point = os.path.exists(ap_file)
self._wlan_configuration[band] = wlan_configuration
- self._status.have_config = True
+ self.wifi_for_band(band).status.have_config = True
logging.info('Updated WLAN configuration for %s GHz', band)
self._update_access_point(wlan_configuration)
@@ -885,6 +921,31 @@
if subprocess.call(self.UPLOAD_LOGS_AND_WAIT) != 0:
logging.error('Failed to upload logs')
+ def cwmp_wakeup(self):
+ if subprocess.call(self.CWMP_WAKEUP) != 0:
+ logging.error('cwmp wakeup failed')
+
+ def start_provisioning(self, wifi):
+ wifi.set_gateway_ip(None)
+ wifi.set_subnet(None)
+ wifi.provisioning_ratchet.reset()
+
+ def provisioning_failed(self, wifi):
+ try:
+ wifi.provisioning_ratchet.check()
+ if wifi.provisioning_ratchet.done_after:
+ wifi.status.provisioning_completed = True
+ logging.info('%s successfully provisioned', wifi.name)
+ return False
+ except ratchet.TimeoutException:
+ wifi.status.provisioning_failed = True
+ logging.info('%s failed to provision: %s', wifi.name,
+ wifi.provisioning_ratchet.current_step().name)
+ return True
+
+ def provisioning_since(self, wifi):
+ return wifi.provisioning_ratchet.t0
+
def _wifi_show():
try:
diff --git a/conman/connection_manager_test.py b/conman/connection_manager_test.py
index 7cf1785..a64ab79 100755
--- a/conman/connection_manager_test.py
+++ b/conman/connection_manager_test.py
@@ -255,8 +255,7 @@
'WIFI_PSK=abcdWIFI_PSK=qwer', 'wifi', 'set', '-P', '-b', '5',
'--bridge=br0', '-s', 'my ssid=1', '--interface-suffix', '_suffix',
])
- config = WLANConfiguration('5', interface_test.Wifi('wcli0', 20), cmd, None,
- None)
+ config = WLANConfiguration('5', interface_test.Wifi('wcli0', 20), cmd, None)
wvtest.WVPASSEQ('my ssid=1', config.ssid)
wvtest.WVPASSEQ('abcdWIFI_PSK=qwer', config.passphrase)
@@ -291,6 +290,7 @@
IFPLUGD_ACTION = ['echo', 'ifplugd.action']
BINWIFI = ['echo', 'wifi']
UPLOAD_LOGS_AND_WAIT = ['echo', 'upload-logs-and-wait']
+ CWMP_WAKEUP = ['echo', 'cwmp', 'wakeup']
def __init__(self, *args, **kwargs):
self._binwifi_commands = []
@@ -326,6 +326,7 @@
# Will s3 fail to acquire a DHCP lease?
self.dhcp_failure_on_s3 = False
self.log_upload_count = 0
+ self.acs_session_fails = False
def create_wifi_interfaces(self):
super(ConnectionManager, self).create_wifi_interfaces()
@@ -461,6 +462,7 @@
gateway_file = os.path.join(self._tmp_dir,
self.GATEWAY_FILE_PREFIX + interface_name)
with open(gateway_file, 'w') as f:
+ logging.debug('Writing gateway file %s', gateway_file)
# This value doesn't matter to conman, so it's fine to hard code it here.
f.write('192.168.1.1')
@@ -468,6 +470,7 @@
subnet_file = os.path.join(self._tmp_dir,
self.SUBNET_FILE_PREFIX + interface_name)
with open(subnet_file, 'w') as f:
+ logging.debug('Writing subnet file %s', subnet_file)
# This value doesn't matter to conman, so it's fine to hard code it here.
f.write('192.168.1.0/24')
@@ -509,6 +512,19 @@
def has_status_files(self, files):
return not set(files) - set(os.listdir(self._status_dir))
+ def cwmp_wakeup(self):
+ super(ConnectionManager, self).cwmp_wakeup()
+ self.write_acscontact()
+ if self.acs():
+ self.write_acsconnected()
+
+ def write_acscontact(self):
+ open(os.path.join(connection_manager.CWMP_PATH, 'acscontact'), 'w')
+
+ def write_acsconnected(self):
+ if not self.acs_session_fails:
+ open(os.path.join(connection_manager.CWMP_PATH, 'acsconnected'), 'w')
+
def wlan_config_filename(path, band):
return os.path.join(path, 'command.%s' % band)
@@ -560,7 +576,10 @@
interface_update_period = 5
wifi_scan_period = 15
wifi_scan_period_s = run_duration_s * wifi_scan_period
+ associate_wait_s = 0
dhcp_wait_s = .5
+ acs_start_wait_s = 0
+ acs_finish_wait_s = 0
# pylint: disable=protected-access
old_wifi_show = connection_manager._wifi_show
@@ -579,6 +598,7 @@
moca_tmp_dir = tempfile.mkdtemp()
wpa_control_interface = tempfile.mkdtemp()
FrenzyWifi.WPACtrl.WIFIINFO_PATH = tempfile.mkdtemp()
+ connection_manager.CWMP_PATH = tempfile.mkdtemp()
for band, access_point in wlan_configs.iteritems():
write_wlan_config(config_dir, band, 'initial ssid', 'initial psk')
@@ -595,14 +615,13 @@
run_duration_s=run_duration_s,
interface_update_period=interface_update_period,
wifi_scan_period_s=wifi_scan_period_s,
+ associate_wait_s=associate_wait_s,
dhcp_wait_s=dhcp_wait_s,
+ acs_start_wait_s=acs_start_wait_s,
+ acs_finish_wait_s=acs_finish_wait_s,
bssid_cycle_length_s=1,
**cm_kwargs)
- c.test_interface_update_period = interface_update_period
- c.test_wifi_scan_period = wifi_scan_period
- c.test_dhcp_wait_s = dhcp_wait_s
-
f(c)
finally:
if os.path.exists(connection_manager.TMP_HOSTS):
@@ -612,6 +631,7 @@
shutil.rmtree(moca_tmp_dir)
shutil.rmtree(wpa_control_interface)
shutil.rmtree(FrenzyWifi.WPACtrl.WIFIINFO_PATH)
+ shutil.rmtree(connection_manager.CWMP_PATH)
# pylint: disable=protected-access
connection_manager._wifi_show = old_wifi_show
connection_manager._get_quantenna_interfaces = old_gqi
@@ -889,7 +909,7 @@
c.can_connect_to_s2 = True
# Give it time to try all BSSIDs. This means sleeping long enough that
# everything in the cycler is active, then doing n+1 loops (the n+1st loop is
- # when we decided that the SSID in the nth loop was successful).
+ # when we decide that the SSID in the nth loop was successful).
time.sleep(c._bssid_cycle_length_s)
for _ in range(len(c.wifi_for_band(band).cycler) + 1):
c.run_once()
@@ -931,13 +951,68 @@
last_bss_info = c.wifi_for_band(band).last_attempted_bss_info
wvtest.WVPASSEQ(last_bss_info.ssid, 's3')
wvtest.WVPASSEQ(last_bss_info.bssid, 'ff:ee:dd:cc:bb:aa')
+ wvtest.WVPASS(c.has_status_files([status.P.WAITING_FOR_DHCP,
+ status.P.WAITING_FOR_PROVISIONING]))
# Third iteration: sleep for dhcp_wait_s and check that we try another AP.
- time.sleep(c.test_dhcp_wait_s)
+ time.sleep(c._dhcp_wait_s)
c.run_once()
last_bss_info = c.wifi_for_band(band).last_attempted_bss_info
wvtest.WVPASSNE(last_bss_info.ssid, 's3')
wvtest.WVPASSNE(last_bss_info.bssid, 'ff:ee:dd:cc:bb:aa')
+ # Now repeat the above, but for an ACS session that takes a while. We don't
+ # necessarily want to leave if it fails (so we don't want the third check),
+ # but we do want to make sure we don't leave while we're still waiting for it.
+ #
+ # Unlike DHCP, which we can always simulate working immediately above, it is
+ # wrong to simulate ACS sessions working for connections without ACS access.
+ # This means we can either always wait for the ACS session timeout in every
+ # test above, making the tests much slower, or we can set that timeout to 0
+ # and then be a little gross here and change it. The latter is unfortunately
+ # the lesser evil, because slow tests are bad.
+ del c.wifi_for_band(band).cycler
+ c.dhcp_failure_on_s3 = False
+ c.acs_session_fails = True
+ # First iteration: check that we try s3.
+ c.run_until_scan(band)
+ last_bss_info = c.wifi_for_band(band).last_attempted_bss_info
+ wvtest.WVPASSEQ(last_bss_info.ssid, 's3')
+ wvtest.WVPASSEQ(last_bss_info.bssid, 'ff:ee:dd:cc:bb:aa')
+
+ # This is the gross part.
+ # pylint: disable=protected-access
+ c.wifi_for_band(band).provisioning_ratchet.steps[3].timeout = 0.5
+
+ # Second iteration: check that we don't leave while waiting.
+ c.run_once()
+ last_bss_info = c.wifi_for_band(band).last_attempted_bss_info
+ wvtest.WVPASSEQ(last_bss_info.ssid, 's3')
+ wvtest.WVPASSEQ(last_bss_info.bssid, 'ff:ee:dd:cc:bb:aa')
+ wvtest.WVPASS(c.has_status_files([status.P.WAITING_FOR_ACS_SESSION,
+ status.P.WAITING_FOR_PROVISIONING]))
+ time.sleep(0.5)
+ c.run_once()
+ wvtest.WVPASS(c.has_status_files([status.P.PROVISIONING_FAILED]))
+ c.wifi_for_band(band).provisioning_ratchet.steps[3].timeout = 0
+
+ # Finally, test successful provisioning.
+ del c.wifi_for_band(band).cycler
+ c.dhcp_failure_on_s3 = False
+ c.acs_session_fails = False
+ # First iteration: check that we try s3.
+ c.run_until_scan(band)
+ last_bss_info = c.wifi_for_band(band).last_attempted_bss_info
+ wvtest.WVPASSEQ(last_bss_info.ssid, 's3')
+ wvtest.WVPASSEQ(last_bss_info.bssid, 'ff:ee:dd:cc:bb:aa')
+
+ c.run_once()
+ last_bss_info = c.wifi_for_band(band).last_attempted_bss_info
+ wvtest.WVPASSEQ(last_bss_info.ssid, 's3')
+ wvtest.WVPASSEQ(last_bss_info.bssid, 'ff:ee:dd:cc:bb:aa')
+ wvtest.WVFAIL(c.has_status_files([status.P.WAITING_FOR_ACS_SESSION,
+ status.P.WAITING_FOR_PROVISIONING]))
+ wvtest.WVPASS(c.has_status_files([status.P.PROVISIONING_COMPLETED]))
+
@wvtest.wvtest
@connection_manager_test(WIFI_SHOW_OUTPUT_MARVELL8897)
diff --git a/conman/interface.py b/conman/interface.py
index 2d0fbbe..ac258e9 100755
--- a/conman/interface.py
+++ b/conman/interface.py
@@ -547,6 +547,10 @@
self.initial_ssid = None
super(Wifi, self).initialize()
+ def connected_to_open(self):
+ return (self.wpa_status().get('wpa_state', None) == 'COMPLETED' and
+ self.wpa_status().get('key_mgmt', None) == 'NONE')
+
class FrenzyWPACtrl(object):
"""A WPACtrl for Frenzy devices.
diff --git a/conman/interface_test.py b/conman/interface_test.py
index 79c58ae..dd5e037 100755
--- a/conman/interface_test.py
+++ b/conman/interface_test.py
@@ -248,8 +248,10 @@
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')
+ wpa_socket = os.path.join(path, self.name)
+ logging.debug('Starting fake wpa_supplicant for %s: %s',
+ self.name, wpa_socket)
+ open(wpa_socket, 'w')
def kill_wpa_supplicant_testonly(self, path):
logging.debug('Killing fake wpa_supplicant for %s', self.name)
diff --git a/conman/ratchet.py b/conman/ratchet.py
new file mode 100644
index 0000000..7873c39
--- /dev/null
+++ b/conman/ratchet.py
@@ -0,0 +1,181 @@
+#!/usr/bin/python -S
+
+"""Utility for ensuring a series of events occur or time out."""
+
+import logging
+import os
+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)
+
+
+class TimeoutException(Exception):
+ pass
+
+
+class Condition(object):
+ """Wrapper for a function that may time out."""
+
+ def __init__(self, name, evaluate, timeout, logger=None, callback=None):
+ self.name = name
+ if evaluate:
+ self.evaluate = evaluate
+ self.timeout = timeout
+ self.logger = logger or logging
+ self.callback = callback
+ self.reset()
+
+ def reset(self, t0=None, start_at=None):
+ """Reset the Condition to an initial state.
+
+ Takes two different timestamp values to account for uncertainty in when a
+ previous condition may have been met.
+
+ Args:
+ t0: The timestamp after which to evaluate the condition.
+ start_at: The timestamp from which to compute the timeout.
+ """
+ self.t0 = t0 or time.time()
+ self.start_at = start_at or time.time()
+ self.done_after = None
+ self.done_by = None
+ self.timed_out = False
+ self.not_done_before = self.t0
+
+ def check(self):
+ """Check whether the condition has completed or timed out."""
+ if self.timed_out:
+ raise TimeoutException()
+
+ if self.done_after:
+ return True
+
+ if self.evaluate():
+ self.mark_done()
+ return True
+
+ now = time.time()
+ if now > self.start_at + self.timeout:
+ self.timed_out = True
+ self.logger.info('%s timed out after %.2f seconds',
+ self.name, now - self.t0)
+ raise TimeoutException()
+
+ self.not_done_before = time.time()
+ return False
+
+ def mark_done(self):
+ # In general, we don't know when a condition finished, but we know it was
+ # *after* whenever it was most recently not done.
+ self.done_after = self.not_done_before
+ self.done_by = time.time()
+ self.logger.info('%s completed after %.2f seconds',
+ self.name, self.done_by - self.t0)
+
+ if self.callback:
+ self.callback()
+
+
+class FileExistsCondition(Condition):
+ """A condition that checks for the existence of a file."""
+
+ def __init__(self, name, filename, timeout):
+ self._filename = filename
+ super(FileExistsCondition, self).__init__(name, None, timeout)
+
+ def evaluate(self):
+ return os.path.exists(self._filename)
+
+ def mtime(self):
+ if os.path.exists(self._filename):
+ return os.stat(self._filename).st_mtime
+
+ return None
+
+ def mark_done(self):
+ super(FileExistsCondition, self).mark_done()
+ # We have to check this because the file could have been deleted while this
+ # was being called. But this condition should almost always be true.
+ mtime = self.mtime()
+ if mtime:
+ self.done_after = self.done_by = mtime
+
+
+class FileTouchedCondition(FileExistsCondition):
+ """A condition that checks that a file was touched after a certain time."""
+
+ def reset(self, t0=None, start_at=None):
+ mtime = self.mtime
+ if t0 and mtime and mtime < t0:
+ self.initial_mtime = self.mtime()
+ else:
+ self.initial_mtime = None
+ super(FileTouchedCondition, self).reset(t0, start_at)
+
+ def evaluate(self):
+ if not super(FileTouchedCondition, self).evaluate():
+ return False
+
+ if self.initial_mtime:
+ return self.mtime() > self.initial_mtime
+
+ return self.mtime() >= self.t0
+
+
+class Ratchet(Condition):
+ """A condition that comprises a series of subconditions."""
+
+ def __init__(self, name, steps, status):
+ self.name = name
+ self.steps = steps
+ for step in self.steps:
+ step.logger = logging.getLogger(self.name)
+ self._status = status
+ super(Ratchet, self).__init__(name, None, 0)
+
+ def reset(self):
+ self._current_step = 0
+ for step in self.steps:
+ step.reset()
+ self._set_step_status(step, False)
+ self._set_current_step_status(True)
+ super(Ratchet, self).reset()
+
+ # Override check rather than evaluate because we don't want the Ratchet to
+ # time out unless one of its steps does.
+ def check(self):
+ if not self.done_after:
+ while self.current_step().check():
+ if not self.advance():
+ self.mark_done()
+ break
+
+ return self.done_after
+
+ def current_step(self):
+ return self.steps[self._current_step]
+
+ def on_final_step(self):
+ return self._current_step == len(self.steps) - 1
+
+ def advance(self):
+ if self.on_final_step():
+ return False
+ else:
+ prev_step = self.current_step()
+ self._current_step += 1
+ self.current_step().start_at = prev_step.done_by
+ self._set_current_step_status(True)
+ return True
+
+ def mark_done(self):
+ super(Ratchet, self).mark_done()
+ self.done_after = self.steps[-1].done_after
+
+ def _set_step_status(self, step, value):
+ setattr(self._status, step.name, value)
+
+ def _set_current_step_status(self, value):
+ self._set_step_status(self.current_step(), value)
diff --git a/conman/ratchet_test.py b/conman/ratchet_test.py
new file mode 100755
index 0000000..10c10ab
--- /dev/null
+++ b/conman/ratchet_test.py
@@ -0,0 +1,128 @@
+#!/usr/bin/python
+
+"""Tests for ratchet.py."""
+
+import os
+import shutil
+import tempfile
+import time
+
+import ratchet
+import status
+from wvtest import wvtest
+
+
+@wvtest.wvtest
+def condition_test():
+ """Test basic Condition functionality."""
+ x = y = 0
+ callback_sink = []
+ cx = ratchet.Condition('cx', lambda: x != 0, 0)
+ cy = ratchet.Condition('cx', lambda: y != 0, 0.1,
+ callback=lambda: callback_sink.append([0]))
+ wvtest.WVEXCEPT(ratchet.TimeoutException, cx.check)
+ wvtest.WVFAIL(cy.check())
+ time.sleep(0.1)
+ wvtest.WVEXCEPT(ratchet.TimeoutException, cy.check)
+
+ x = 1
+ wvtest.WVEXCEPT(ratchet.TimeoutException, cx.check)
+ cx.reset()
+ wvtest.WVPASS(cx.check())
+
+ y = 1
+ cy.reset()
+ wvtest.WVPASS(cy.check())
+ wvtest.WVPASSEQ(len(callback_sink), 1)
+ # Callback shouldn't fire again.
+ wvtest.WVPASS(cy.check())
+ wvtest.WVPASSEQ(len(callback_sink), 1)
+ cy.reset()
+ wvtest.WVPASS(cy.check())
+ wvtest.WVPASSEQ(len(callback_sink), 2)
+
+
+@wvtest.wvtest
+def file_condition_test():
+ """Test File*Condition functionality."""
+ try:
+ _, filename = tempfile.mkstemp()
+ c_exists = ratchet.FileExistsCondition('c exists', filename, 0.1)
+ c_mtime = ratchet.FileTouchedCondition('c mtime', filename, 0.1)
+ wvtest.WVPASS(c_exists.check())
+ wvtest.WVFAIL(c_mtime.check())
+ # mtime precision is too low to notice that we're touching the file *after*
+ # capturing its initial mtime rather than at the same time, so take a short
+ # nap before touching it.
+ time.sleep(0.01)
+ open(filename, 'w')
+ wvtest.WVPASS(c_mtime.check())
+
+ # Test that old mtimes don't count.
+ time.sleep(0.01)
+ c_mtime.reset()
+ wvtest.WVFAIL(c_mtime.check())
+ time.sleep(0.1)
+ wvtest.WVEXCEPT(ratchet.TimeoutException, c_mtime.check)
+
+ # Test t0 and start_at.
+ os.unlink(filename)
+ now = time.time()
+ c_mtime.reset(t0=now, start_at=now + 0.2)
+ wvtest.WVFAIL(c_mtime.check())
+ time.sleep(0.15)
+ wvtest.WVFAIL(c_mtime.check())
+ open(filename, 'w')
+ wvtest.WVPASS(c_mtime.check())
+
+ finally:
+ os.unlink(filename)
+
+
+@wvtest.wvtest
+def ratchet_test():
+ """Test Ratchet functionality."""
+
+ class P(object):
+ X = 'X'
+ Y = 'Y'
+ Z = 'Z'
+ status.P = P
+ status.IMPLICATIONS = {}
+
+ status_export_path = tempfile.mkdtemp()
+ try:
+ x = y = z = 0
+ r = ratchet.Ratchet('test ratchet', [
+ ratchet.Condition('x', lambda: x, 0.1),
+ ratchet.Condition('y', lambda: y, 0.1),
+ ratchet.Condition('z', lambda: z, 0.1),
+ ], status.Status(status_export_path))
+ x = y = 1
+
+ # Test that timeouts are not just summed, but start whenever the previous
+ # step completed.
+ wvtest.WVPASSEQ(r._current_step, 0) # pylint: disable=protected-access
+ wvtest.WVPASS(os.path.isfile(os.path.join(status_export_path, 'X')))
+ wvtest.WVFAIL(os.path.isfile(os.path.join(status_export_path, 'Y')))
+ wvtest.WVFAIL(os.path.isfile(os.path.join(status_export_path, 'Z')))
+ time.sleep(0.05)
+ wvtest.WVFAIL(r.check())
+ wvtest.WVPASSEQ(r._current_step, 2) # pylint: disable=protected-access
+ wvtest.WVPASS(os.path.isfile(os.path.join(status_export_path, 'X')))
+ wvtest.WVPASS(os.path.isfile(os.path.join(status_export_path, 'Y')))
+ wvtest.WVPASS(os.path.isfile(os.path.join(status_export_path, 'Z')))
+ wvtest.WVFAIL(r.check())
+ wvtest.WVPASSEQ(r._current_step, 2) # pylint: disable=protected-access
+ time.sleep(0.1)
+ wvtest.WVEXCEPT(ratchet.TimeoutException, r.check)
+
+ x = y = z = 1
+ r.reset()
+ wvtest.WVPASS(r.check())
+ finally:
+ shutil.rmtree(status_export_path)
+
+
+if __name__ == '__main__':
+ wvtest.wvtest_main()
diff --git a/conman/status.py b/conman/status.py
index 7f75682..8f8d3a1 100644
--- a/conman/status.py
+++ b/conman/status.py
@@ -32,17 +32,25 @@
PROVISIONING_FAILED = 'PROVISIONING_FAILED'
ATTACHED_TO_WPA_SUPPLICANT = 'ATTACHED_TO_WPA_SUPPLICANT'
+ WAITING_FOR_PROVISIONING = 'WAITING_FOR_PROVISIONING'
+ WAITING_FOR_DHCP = 'WAITING_FOR_DHCP'
+ WAITING_FOR_CWMP_WAKEUP = 'WAITING_FOR_CWMP_WAKEUP'
+ WAITING_FOR_ACS_SESSION = 'WAITING_FOR_ACS_SESSION'
+ PROVISIONING_COMPLETED = 'PROVISIONING_COMPLETED'
+
# Format: { proposition: (implications, counter-implications), ... }
# If you want to add a new proposition to the Status class, just edit this dict.
IMPLICATIONS = {
P.TRYING_OPEN: (
(),
- (P.CONNECTED_TO_OPEN, P.TRYING_WLAN, P.CONNECTED_TO_WLAN)
+ (P.CONNECTED_TO_OPEN, P.TRYING_WLAN, P.CONNECTED_TO_WLAN,
+ P.WAITING_FOR_PROVISIONING)
),
P.TRYING_WLAN: (
(),
- (P.TRYING_OPEN, P.CONNECTED_TO_OPEN, P.CONNECTED_TO_WLAN)
+ (P.TRYING_OPEN, P.CONNECTED_TO_OPEN, P.CONNECTED_TO_WLAN,
+ P.WAITING_FOR_PROVISIONING)
),
P.WLAN_FAILED: (
(),
@@ -54,7 +62,8 @@
),
P.CONNECTED_TO_WLAN: (
(P.HAVE_WORKING_CONFIG,),
- (P.CONNECTED_TO_OPEN, P.TRYING_OPEN, P.TRYING_WLAN)
+ (P.CONNECTED_TO_OPEN, P.TRYING_OPEN, P.TRYING_WLAN,
+ P.WAITING_FOR_PROVISIONING)
),
P.CAN_REACH_ACS: (
(P.COULD_REACH_ACS,),
@@ -62,11 +71,11 @@
),
P.COULD_REACH_ACS: (
(),
- (P.PROVISIONING_FAILED,),
+ (),
),
P.PROVISIONING_FAILED: (
(),
- (P.COULD_REACH_ACS,),
+ (P.WAITING_FOR_PROVISIONING, P.PROVISIONING_COMPLETED,),
),
P.HAVE_WORKING_CONFIG: (
(P.HAVE_CONFIG,),
@@ -75,24 +84,60 @@
P.ATTACHED_TO_WPA_SUPPLICANT: (
(),
(),
- )
+ ),
+ P.WAITING_FOR_PROVISIONING: (
+ (P.CONNECTED_TO_OPEN,),
+ (),
+ ),
+ P.WAITING_FOR_DHCP: (
+ (P.WAITING_FOR_PROVISIONING,),
+ (P.WAITING_FOR_CWMP_WAKEUP, P.WAITING_FOR_ACS_SESSION),
+ ),
+ P.WAITING_FOR_CWMP_WAKEUP: (
+ (P.WAITING_FOR_PROVISIONING,),
+ (P.WAITING_FOR_DHCP, P.WAITING_FOR_ACS_SESSION),
+ ),
+ P.WAITING_FOR_ACS_SESSION: (
+ (P.WAITING_FOR_PROVISIONING,),
+ (P.WAITING_FOR_DHCP, P.WAITING_FOR_CWMP_WAKEUP),
+ ),
+ P.PROVISIONING_COMPLETED: (
+ (),
+ (P.WAITING_FOR_PROVISIONING, P.PROVISIONING_FAILED,),
+ ),
}
-class Proposition(object):
+class ExportedValue(object):
+
+ def __init__(self, name, export_path):
+ self._name = name
+ self._export_path = export_path
+
+ def export(self):
+ filepath = os.path.join(self._export_path, self._name)
+ if self.value():
+ if not os.path.exists(filepath):
+ open(filepath, 'w')
+ else:
+ if os.path.exists(filepath):
+ os.unlink(filepath)
+
+
+class Proposition(ExportedValue):
"""Represents a proposition.
May imply truth or falsity of other propositions.
"""
- def __init__(self, name, export_path):
- self._name = name
- self._export_path = export_path
+ def __init__(self, *args, **kwargs):
+ super(Proposition, self).__init__(*args, **kwargs)
self._value = None
self._implications = set()
self._counter_implications = set()
self._impliers = set()
self._counter_impliers = set()
+ self.parents = set()
def implies(self, implication):
self._counter_implications.discard(implication)
@@ -115,11 +160,20 @@
self._counter_impliers.add(counter_implier)
def set(self, value):
+ """Set this Proposition's value.
+
+ If the value changed, update any dependent values/files.
+
+ Args:
+ value: The new value.
+ """
if value == self._value:
return
self._value = value
self.export()
+ for parent in self.parents:
+ parent.export()
logging.debug('%s is now %s', self._name, self._value)
if value:
@@ -135,14 +189,20 @@
for implier in self._impliers:
implier.set(False)
- def export(self):
- filepath = os.path.join(self._export_path, self._name)
- if self._value:
- if not os.path.exists(filepath):
- open(filepath, 'w')
- else:
- if os.path.exists(filepath):
- os.unlink(filepath)
+ def value(self):
+ return self._value
+
+
+class Disjunction(ExportedValue):
+
+ def __init__(self, name, export_path, disjuncts):
+ super(Disjunction, self).__init__(name, export_path)
+ self.disjuncts = disjuncts
+ for disjunct in self.disjuncts:
+ disjunct.parents.add(self)
+
+ def value(self):
+ return any(d.value() for d in self.disjuncts)
class Status(object):
@@ -154,6 +214,9 @@
self._export_path = export_path
+ self._set_up_propositions()
+
+ def _set_up_propositions(self):
self._propositions = {
p: Proposition(p, self._export_path)
for p in dict(inspect.getmembers(P)) if not p.startswith('_')
@@ -166,7 +229,7 @@
self._propositions[p].implies_not(
self._propositions[counter_implication])
- def _proposition(self, p):
+ def proposition(self, p):
return self._propositions[p]
def __setattr__(self, attr, value):
@@ -187,3 +250,17 @@
return
super(Status, self).__setattr__(attr, value)
+
+
+class CompositeStatus(Status):
+
+ def __init__(self, export_path, children):
+ self._children = children
+ super(CompositeStatus, self).__init__(export_path)
+
+ def _set_up_propositions(self):
+ self._propositions = {
+ p: Disjunction(p, self._export_path,
+ [c.proposition(p) for c in self._children])
+ for p in dict(inspect.getmembers(P)) if not p.startswith('_')
+ }
diff --git a/conman/status_test.py b/conman/status_test.py
index 03223f8..befebbf 100755
--- a/conman/status_test.py
+++ b/conman/status_test.py
@@ -17,6 +17,10 @@
return os.path.exists(os.path.join(path, filename))
+def has_file(s, filename):
+ return file_in(s._export_path, filename)
+
+
@wvtest.wvtest
def test_proposition():
export_path = tempfile.mkdtemp()
@@ -71,51 +75,85 @@
@wvtest.wvtest
def test_status():
- export_path = tempfile.mkdtemp()
+ export_path_s = tempfile.mkdtemp()
+ export_path_t = tempfile.mkdtemp()
+ export_path_st = tempfile.mkdtemp()
try:
- s = status.Status(export_path)
+ s = status.Status(export_path_s)
+ t = status.Status(export_path_t)
+ st = status.CompositeStatus(export_path_st, [s, t])
# Sanity check that there are no contradictions.
for p, (want_true, want_false) in status.IMPLICATIONS.iteritems():
setattr(s, p.lower(), True)
- wvtest.WVPASS(file_in(export_path, p))
+ wvtest.WVPASS(has_file(s, p))
for wt in want_true:
- wvtest.WVPASS(file_in(export_path, wt))
+ wvtest.WVPASS(has_file(s, wt))
for wf in want_false:
- wvtest.WVFAIL(file_in(export_path, wf))
+ wvtest.WVFAIL(has_file(s, wf))
+
+ def check_exported(check_s, check_t, filename):
+ wvtest.WVPASSEQ(check_s, has_file(s, filename))
+ wvtest.WVPASSEQ(check_t, has_file(t, filename))
+ wvtest.WVPASSEQ(check_s or check_t, has_file(st, filename))
s.trying_wlan = True
- wvtest.WVPASS(file_in(export_path, status.P.TRYING_WLAN))
- wvtest.WVFAIL(file_in(export_path, status.P.CONNECTED_TO_WLAN))
+ t.trying_wlan = False
+ check_exported(True, False, status.P.TRYING_WLAN)
+ check_exported(False, False, status.P.CONNECTED_TO_WLAN)
s.connected_to_open = True
- wvtest.WVPASS(file_in(export_path, status.P.CONNECTED_TO_OPEN))
- wvtest.WVFAIL(file_in(export_path, status.P.CONNECTED_TO_WLAN))
+ check_exported(True, False, status.P.CONNECTED_TO_OPEN)
+ check_exported(False, False, status.P.CONNECTED_TO_WLAN)
s.connected_to_wlan = True
- wvtest.WVPASS(file_in(export_path, status.P.CONNECTED_TO_WLAN))
- wvtest.WVPASS(file_in(export_path, status.P.HAVE_WORKING_CONFIG))
- wvtest.WVFAIL(file_in(export_path, status.P.CONNECTED_TO_OPEN))
- wvtest.WVFAIL(file_in(export_path, status.P.TRYING_WLAN))
- wvtest.WVFAIL(file_in(export_path, status.P.TRYING_OPEN))
+ t.trying_wlan = True
+ check_exported(True, False, status.P.CONNECTED_TO_WLAN)
+ check_exported(True, False, status.P.HAVE_WORKING_CONFIG)
+ check_exported(False, False, status.P.CONNECTED_TO_OPEN)
+ check_exported(False, True, status.P.TRYING_WLAN)
+ check_exported(False, False, status.P.TRYING_OPEN)
s.can_reach_acs = True
s.can_reach_internet = True
- wvtest.WVPASS(file_in(export_path, status.P.CAN_REACH_ACS))
- wvtest.WVPASS(file_in(export_path, status.P.COULD_REACH_ACS))
- wvtest.WVPASS(file_in(export_path, status.P.CAN_REACH_INTERNET))
- wvtest.WVFAIL(file_in(export_path, status.P.PROVISIONING_FAILED))
+ check_exported(True, False, status.P.CAN_REACH_ACS)
+ check_exported(True, False, status.P.COULD_REACH_ACS)
+ check_exported(True, False, status.P.CAN_REACH_INTERNET)
+ check_exported(False, False, status.P.PROVISIONING_FAILED)
# These should not have changed
- wvtest.WVPASS(file_in(export_path, status.P.CONNECTED_TO_WLAN))
- wvtest.WVPASS(file_in(export_path, status.P.HAVE_WORKING_CONFIG))
- wvtest.WVFAIL(file_in(export_path, status.P.CONNECTED_TO_OPEN))
- wvtest.WVFAIL(file_in(export_path, status.P.TRYING_WLAN))
- wvtest.WVFAIL(file_in(export_path, status.P.TRYING_OPEN))
+ check_exported(True, False, status.P.CONNECTED_TO_WLAN)
+ check_exported(True, False, status.P.HAVE_WORKING_CONFIG)
+ check_exported(False, False, status.P.CONNECTED_TO_OPEN)
+ check_exported(False, True, status.P.TRYING_WLAN)
+ check_exported(False, False, status.P.TRYING_OPEN)
+
+ # Test provisioning statuses.
+ s.waiting_for_dhcp = True
+ check_exported(False, True, status.P.TRYING_WLAN)
+ check_exported(False, False, status.P.TRYING_OPEN)
+ check_exported(False, False, status.P.CONNECTED_TO_WLAN)
+ check_exported(True, False, status.P.CONNECTED_TO_OPEN)
+ check_exported(True, False, status.P.WAITING_FOR_PROVISIONING)
+ check_exported(True, False, status.P.WAITING_FOR_DHCP)
+ s.waiting_for_cwmp_wakeup = True
+ check_exported(False, False, status.P.WAITING_FOR_DHCP)
+ check_exported(True, False, status.P.WAITING_FOR_CWMP_WAKEUP)
+ s.waiting_for_acs_session = True
+ check_exported(False, False, status.P.WAITING_FOR_DHCP)
+ check_exported(False, False, status.P.WAITING_FOR_CWMP_WAKEUP)
+ check_exported(True, False, status.P.WAITING_FOR_ACS_SESSION)
+ s.provisioning_completed = True
+ check_exported(False, False, status.P.WAITING_FOR_PROVISIONING)
+ check_exported(False, False, status.P.WAITING_FOR_DHCP)
+ check_exported(False, False, status.P.WAITING_FOR_CWMP_WAKEUP)
+ check_exported(False, False, status.P.WAITING_FOR_CWMP_WAKEUP)
finally:
- shutil.rmtree(export_path)
+ shutil.rmtree(export_path_s)
+ shutil.rmtree(export_path_t)
+ shutil.rmtree(export_path_st)
if __name__ == '__main__':
