| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| from autotest_lib.client.bin import site_utils, test |
| from autotest_lib.client.common_lib import error |
| |
| import dbus |
| import dbus.types |
| import gobject |
| import time |
| |
| from autotest_lib.client.cros import backchannel, network, flimflam_test_path |
| from autotest_lib.client.cros.cellular import cell_tools |
| from autotest_lib.client.cros.cellular import mm |
| from autotest_lib.client.cros.cellular.pseudomodem import mm1 |
| from autotest_lib.client.cros.cellular.pseudomodem import modem_cdma |
| from autotest_lib.client.cros.cellular.pseudomodem import pseudomodem |
| import flimflam |
| |
| I_ACTIVATION_TEST = 'Interface.CDMAActivationTest' |
| ACTIVATION_STATE_TIMEOUT = 10 |
| MODEM_STATE_TIMEOUT = 10 |
| |
| class ActivationTest(object): |
| """ |
| Super class that implements setup code that is common to the individual |
| tests. |
| |
| """ |
| def __init__(self, pmm_context, test): |
| self.pmm_context = pmm_context |
| self.test = test |
| self.test_modem = self._setup_test_modem() |
| self.modem_properties_interface = None |
| |
| def run(self): |
| """ |
| Restarts the pseudomodem with the modem object to be used for this |
| test and runs the test. |
| |
| """ |
| if not self.test_modem: |
| raise test.TestFail('Uninitialized test modem') |
| self.pmm_context.SetModem(self.test_modem) |
| self._run_test() |
| |
| def _get_modem_properties_interface(self): |
| if not self.modem_properties_interface: |
| self.modem_properties_interface = \ |
| self.test.modem().PropertiesInterface() |
| return self.modem_properties_interface |
| |
| def _set_modem_activation_state(self, state): |
| interface = self._get_modem_properties_interface() |
| interface.Set( |
| mm1.I_MODEM_CDMA, 'ActivationState', dbus.types.UInt32(state)) |
| |
| def _get_modem_activation_state(self): |
| interface = self._get_modem_properties_interface() |
| return interface.Get(mm1.I_MODEM_CDMA, 'ActivationState') |
| |
| def _setup_test_modem(self): |
| raise NotImplementedError() |
| |
| def _run_test(self): |
| raise NotImplementedError() |
| |
| class ActivationStateTest(ActivationTest): |
| """ |
| This test verifies that the service "ActivationState" property matches the |
| cdma activation state exposed by ModemManager. |
| |
| """ |
| def _setup_test_modem(self): |
| return modem_cdma.ModemCdma() |
| |
| def _run_test(self): |
| network.ResetAllModems(self.test.flim) |
| |
| # The modem state should be REGISTERED. |
| self.test.check_modem_state(mm1.MM_MODEM_STATE_REGISTERED) |
| |
| # Service should appear as 'activated'. |
| self.test.check_service_activation_state('activated') |
| |
| # Service activation state should change to 'not-activated'. |
| self._set_modem_activation_state( |
| mm1.MM_MODEM_CDMA_ACTIVATION_STATE_NOT_ACTIVATED) |
| self.test.check_service_activation_state('not-activated') |
| |
| # Service activation state should change to 'activating'. |
| self._set_modem_activation_state( |
| mm1.MM_MODEM_CDMA_ACTIVATION_STATE_ACTIVATING) |
| self.test.check_service_activation_state('activating') |
| |
| # Service activation state should change to 'partially-activated'. |
| self._set_modem_activation_state( |
| mm1.MM_MODEM_CDMA_ACTIVATION_STATE_PARTIALLY_ACTIVATED) |
| self.test.check_service_activation_state('partially-activated') |
| |
| # Service activation state should change to 'activated'. |
| self._set_modem_activation_state( |
| mm1.MM_MODEM_CDMA_ACTIVATION_STATE_ACTIVATED) |
| self.test.check_service_activation_state('activated') |
| |
| class ActivationSuccessTest(ActivationTest): |
| """ |
| This test verifies that the service finally bacomes "activated" when the |
| service is told to initiate OTASP activation. |
| |
| """ |
| def _setup_test_modem(self): |
| return modem_cdma.ModemCdma( |
| modem_cdma.ModemCdma.CdmaNetwork(activated=False)) |
| |
| def _run_test(self): |
| network.ResetAllModems(self.test.flim) |
| |
| # The modem state should be REGISTERED. |
| self.test.check_modem_state(mm1.MM_MODEM_STATE_REGISTERED) |
| |
| # Service should appear as 'not-activated'. |
| self.test.check_service_activation_state('not-activated') |
| |
| # Call 'CompleteActivation' on the service. The service should become |
| # 'activating'. |
| service = self.test.find_cellular_service() |
| service.CompleteCellularActivation() |
| self.test.check_service_activation_state('activating') |
| |
| # The modem should reset in 5 seconds. Wait 5 more seconds to make sure |
| # a new service gets created. |
| time.sleep(10) |
| self.test.check_service_activation_state('activated') |
| |
| class ActivationFailureRetryTest(ActivationTest): |
| """ |
| This test verifies that if "ActivateAutomatic" fails, a retry will be |
| scheduled. |
| |
| """ |
| NUM_ACTIVATE_RETRIES = 5 |
| def _setup_test_modem(self): |
| class TestModem(modem_cdma.ModemCdma): |
| """ |
| TestModem to test that shill retries OTASP activation until it |
| succeeds. |
| |
| """ |
| def __init__(self): |
| self.activate_count = 0 |
| modem_cdma.ModemCdma.__init__(self, |
| modem_cdma.ModemCdma.CdmaNetwork(activated=False)) |
| |
| def _InitializeProperties(self): |
| props = modem_cdma.ModemCdma._InitializeProperties(self) |
| |
| # For the purposes of this test, introduce a property that |
| # stores how many times "Activate" has been called on this |
| # modem. |
| props[I_ACTIVATION_TEST] = { |
| 'ActivateCount' : dbus.types.UInt32(self.activate_count) |
| } |
| return props |
| |
| def _IncrementActivateCount(self): |
| self.activate_count += 1 |
| self.Set(I_ACTIVATION_TEST, |
| 'ActivateCount', |
| self.activate_count) |
| |
| def Activate(self, carrier, return_cb, raise_cb): |
| """ |
| Activation will only succeed on the NUM_ACTIVATE_RETRIESth try. |
| |
| """ |
| self._IncrementActivateCount() |
| if self.activate_count == \ |
| ActivationFailureRetryTest.NUM_ACTIVATE_RETRIES: |
| modem_cdma.ModemCdma.Activate( |
| self, carrier, return_cb, raise_cb) |
| else: |
| def _raise_activation_error(): |
| raise_cb(mm1.MMCdmaActivationError( |
| mm1.MMCdmaActivationError.START_FAILED)) |
| gobject.idle_add(_raise_activation_error) |
| return TestModem() |
| |
| def _run_test(self): |
| network.ResetAllModems(self.test.flim) |
| |
| # The modem state should be REGISTERED. |
| self.test.check_modem_state(mm1.MM_MODEM_STATE_REGISTERED) |
| |
| # Service should appear as 'not-activated'. |
| self.test.check_service_activation_state('not-activated') |
| |
| # Call 'CompleteActivation' on the service. The service should remain |
| # 'not-activated'. |
| service = self.test.find_cellular_service() |
| service.CompleteCellularActivation() |
| |
| modem_props = self._get_modem_properties_interface() |
| while modem_props.Get(I_ACTIVATION_TEST, 'ActivateCount') < \ |
| self.NUM_ACTIVATE_RETRIES: |
| self.test.check_service_activation_state('not-activated') |
| |
| # Activation should succeed after the latest retry. |
| self.test.check_service_activation_state('activating') |
| |
| # The modem should reset in 5 seconds. Wait 5 more seconds to make sure |
| # a new service gets created. |
| time.sleep(10) |
| self.test.check_service_activation_state('activated') |
| |
| class network_CDMAActivate(test.test): |
| """ |
| Tests various scenarios that may arise during the post-payment CDMA |
| activation process when shill accesses the modem via ModemManager. |
| |
| """ |
| version = 1 |
| |
| def modem(self): |
| """ |
| Returns a D-Bus proxy for the current modem object exposed by the |
| modem manager. |
| |
| @return A dbus.service.Object instance. |
| |
| """ |
| manager, modem_path = mm.PickOneModem('') |
| return manager.GetModem(modem_path) |
| |
| def modem_state(self): |
| """ |
| Gets the current modem state from the current modem manager. |
| |
| @return The modem state, as a uint32 value. |
| |
| """ |
| modem = self.modem() |
| props = modem.GetAll(mm1.I_MODEM) |
| return props['State'] |
| |
| def check_modem_state(self, expected_state, timeout=MODEM_STATE_TIMEOUT): |
| """ |
| Polls until the modem has the expected state within |timeout| seconds. |
| |
| @param expected_state: The modem state the modem is expected to be in. |
| @param timeout: The timeout interval for polling. |
| |
| @raises error.TestFail, if the modem doesn't transition to |
| |expected_state| within |timeout|. |
| |
| """ |
| site_utils.poll_for_condition( |
| lambda: self.modem_state() == expected_state, |
| exception=error.TestFail('Timed out waiting for modem state ' + |
| str(expected_state)), |
| timeout=timeout); |
| |
| def find_cellular_service(self): |
| """ |
| Searches for a cellular service and returns it. |
| |
| @return A dbus.service.Object instance. |
| @raises error.TestFail, if no cellular service is found. |
| |
| """ |
| service = self.flim.FindCellularService() |
| if not service: |
| raise error.TestFail('Could not find cellular service.') |
| return service |
| |
| def check_service_activation_state(self, expected_state): |
| """ |
| Waits until the current cellular service has the expected activation |
| state within ACTIVATION_STATE_TIMEOUT seconds. |
| |
| @param expected_state: The activation state the service is expected to |
| be in. |
| @raises error.TestFail, if no cellular service is found or the service |
| activation state doesn't match |expected_state| within timeout. |
| |
| """ |
| service = self.find_cellular_service() |
| state = self.flim.WaitForServiceState( |
| service=service, |
| expected_states=[expected_state], |
| timeout=ACTIVATION_STATE_TIMEOUT, |
| property_name='Cellular.ActivationState')[0] |
| if state != expected_state: |
| raise error.TestFail( |
| 'Service activation state should be \'%s\', but it is ' |
| '\'%s\'.' % (expected_state, state)) |
| |
| def run_once(self): |
| with backchannel.Backchannel(): |
| with pseudomodem.TestModemManagerContext(True, 'CDMA') as tmmc: |
| self.flim = flimflam.FlimFlam() |
| self.device_manager = flimflam.DeviceManager(self.flim) |
| self.flim.SetDebugTags( |
| 'dbus+service+device+modem+cellular+portal+network+' |
| 'manager+dhcp') |
| |
| tests = [ |
| ActivationStateTest(tmmc.GetPseudoModemManager(), self), |
| ActivationSuccessTest(tmmc.GetPseudoModemManager(), self), |
| ActivationFailureRetryTest( |
| tmmc.GetPseudoModemManager(), self) |
| ] |
| |
| with cell_tools.OtherDeviceShutdownContext('cellular'): |
| for test in tests: |
| test.run() |
