| # 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 test |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib import error |
| |
| import dbus |
| import dbus.types |
| import logging |
| import time |
| |
| from autotest_lib.client.cros import backchannel |
| from autotest_lib.client.cros.cellular import mm |
| from autotest_lib.client.cros.cellular.pseudomodem import modem_3gpp |
| from autotest_lib.client.cros.cellular.pseudomodem import mm1, pseudomodem, sim |
| |
| # pylint: disable=W0611 |
| from autotest_lib.client.cros import flimflam_test_path |
| # pylint: enable=W0611 |
| import cellular_proxy |
| |
| I_ACTIVATION_TEST = 'Interface.LTEActivationTest' |
| |
| LONG_TIMEOUT = 20 |
| SHORT_TIMEOUT = 10 |
| |
| class ActivationTest(object): |
| """ |
| Super class that implements setup code that is common to the individual |
| tests. |
| |
| """ |
| class TestModem(modem_3gpp.Modem3gpp): |
| """ |
| Base class for the custom 3GPP fake modems that are defined in this |
| test. |
| |
| """ |
| def _InitializeProperties(self): |
| props = modem_3gpp.Modem3gpp._InitializeProperties(self) |
| modem_props = props[mm1.I_MODEM] |
| modem_props['OwnNumbers'] = ['0000000000'] |
| modem_props['AccessTechnologies'] = dbus.types.UInt32( |
| mm1.MM_MODEM_ACCESS_TECHNOLOGY_LTE) |
| modem_props['ModemCapabilities'] = dbus.types.UInt32( |
| mm1.MM_MODEM_CAPABILITY_LTE) |
| modem_props['CurrentCapabilities'] = dbus.types.UInt32( |
| mm1.MM_MODEM_CAPABILITY_LTE) |
| |
| # For the purposes of this test, introduce a property to help |
| # verify that a reset has taken place. Expose this under a test |
| # specific interface. |
| if hasattr(self, '_properties'): |
| reset_called = \ |
| self._properties[I_ACTIVATION_TEST]['ResetCalled'] |
| else: |
| reset_called = False |
| props[I_ACTIVATION_TEST] = { |
| 'ResetCalled' : dbus.types.Boolean(reset_called) |
| } |
| return props |
| |
| def RegisterWithNetwork( |
| self, operator_id='', return_cb=None, raise_cb=None): |
| # Make this do nothing, so that we don't automatically |
| # register to a network after enable. |
| return |
| |
| def Reset(self): |
| self.Set( |
| I_ACTIVATION_TEST, 'ResetCalled', dbus.types.Boolean(True)) |
| modem_3gpp.Modem3gpp.Reset(self) |
| |
| def __init__(self, pmm_context, test): |
| self.pmm_context = pmm_context |
| self.test = test |
| self.test_modem = self.SetupTestModem() |
| |
| def Cleanup(self): |
| """ |
| Makes the modem look like it has been activated to satisfy the test |
| end condition. |
| |
| """ |
| # Set the MDN to a non-zero value, so that shill removes the ICCID from |
| # activating_iccid_store.profile. This way, individual test runs won't |
| # interfere with each other. |
| modem = self.test.GetModem() |
| modem.PropertiesInterface().Set(mm1.I_MODEM, |
| 'OwnNumbers', |
| ['1111111111']) |
| time.sleep(5) |
| if self.test.FindCellularService(False): |
| self.test.CheckServiceActivationState('activated') |
| |
| def Run(self): |
| """ |
| Configures the pseudomodem to run with the test modem, runs the test |
| and cleans up. |
| |
| """ |
| if not self.test_modem: |
| raise test.TestFail('Uninitialized test modem') |
| self.pmm_context.SetModem(self.test_modem) |
| time.sleep(5) |
| self.RunTest() |
| self.Cleanup() |
| |
| def SetupTestModem(self): |
| """ |
| Returns the modem.Modem3gpp implementation that will be used by the |
| test. Should be implemented by the subclass. |
| |
| @return An instance of ActivationTest.TestModem. |
| |
| """ |
| raise NotImplementedError() |
| |
| def RunTest(self): |
| """ |
| Runs the body of the test. Should be implemented by the subclass. |
| |
| """ |
| raise NotImplementedError() |
| |
| class TimeoutResetTest(ActivationTest): |
| """ |
| This test verifies that the modem resets after a timeout following online |
| payment. |
| |
| """ |
| def SetupTestModem(self): |
| return ActivationTest.TestModem() |
| |
| def RunTest(self): |
| # Service should appear as 'not-activated'. |
| self.test.CheckServiceActivationState('not-activated') |
| self.test.CheckResetCalled(False) |
| |
| # Call 'CompleteActivation' on the device. The service should become |
| # 'activating' and the modem should reset after 20 seconds. |
| service = self.test.FindCellularService() |
| service.CompleteCellularActivation() |
| self.test.CheckServiceActivationState('activating') |
| |
| # Wait until the register timeout. |
| time.sleep(LONG_TIMEOUT) |
| self.test.CheckResetCalled(True) |
| |
| # At this point, a service should never get created. |
| if self.test.FindCellularService(False): |
| raise error.TestError('There should be no cellular service.') |
| |
| class TimeoutActivatedTest(ActivationTest): |
| """ |
| This test verifies that the service eventually becomes 'activated' in the |
| case of a post-payment registration timeout but the modem finally registers |
| to a network after a reset. |
| |
| """ |
| def SetupTestModem(self): |
| class Modem(ActivationTest.TestModem): |
| """ |
| Fake modem that only becomes registered if it has been reset at |
| least once. |
| |
| """ |
| def RegisterWithNetwork( |
| self, operator_id='', return_cb=None, raise_cb=None): |
| if self.Get(I_ACTIVATION_TEST, 'ResetCalled'): |
| modem_3gpp.Modem3gpp.RegisterWithNetwork( |
| self, operator_id, return_cb, raise_cb) |
| return Modem() |
| |
| def RunTest(self): |
| # Service should appear as 'not-activated'. |
| self.test.CheckServiceActivationState('not-activated') |
| self.test.CheckResetCalled(False) |
| |
| # Call 'CompleteActivation' on the device. The service should become |
| # 'activating' and the modem should reset after 20 seconds. |
| service = self.test.FindCellularService() |
| service.CompleteCellularActivation() |
| self.test.CheckServiceActivationState('activating') |
| |
| # Wait until the register timeout. |
| time.sleep(LONG_TIMEOUT) |
| self.test.CheckResetCalled(True) |
| |
| # The service should register and be marked as 'activated'. |
| self.test.CheckServiceActivationState('activated') |
| |
| class ResetAfterRegisterTest(ActivationTest): |
| """ |
| This test verifies that shill resets the modem if the modem registers |
| with a network within the timeout interval. |
| |
| """ |
| def SetupTestModem(self): |
| class Modem(ActivationTest.TestModem): |
| """ |
| Fake modem that becomes registered once registration has been |
| triggered at least twice. |
| |
| """ |
| def __init__(self): |
| ActivationTest.TestModem.__init__(self) |
| self.register_count = 0 |
| |
| def RegisterWithNetwork( |
| self, operator_id='', return_cb=None, raise_cb=None): |
| # Make the initial registration due triggered by Enable do |
| # nothing. We expect exactly two Enable commands: |
| # 1. Triggered by shill to enable the modem, |
| # 2. Triggered by ResetCellularDevice in |
| # ResetAfterRegisterTest.RunTest. |
| self.register_count += 1 |
| if self.register_count > 1: |
| modem_3gpp.Modem3gpp.RegisterWithNetwork( |
| self, operator_id, return_cb, raise_cb) |
| return Modem() |
| |
| def RunTest(self): |
| # Service should appear as 'not-activated'. |
| self.test.CheckServiceActivationState('not-activated') |
| |
| # Call 'CompleteActivation' on the device. The service should become |
| # 'activating' and the modem should reset after 20 seconds. |
| service = self.test.FindCellularService() |
| service.CompleteCellularActivation() |
| self.test.CheckServiceActivationState('activating') |
| |
| time.sleep(5) |
| self.test.CheckResetCalled(False) |
| |
| # The service should register and trigger shill to reset the modem. |
| self.test.EnsureModemStateReached( |
| mm1.MM_MODEM_STATE_ENABLED, SHORT_TIMEOUT) |
| time.sleep(5) |
| mccmnc = self.test.sim.Get(mm1.I_SIM, 'OperatorIdentifier') |
| self.test.GetModem().GsmModem().Register(mccmnc) |
| |
| time.sleep(5) |
| self.test.CheckResetCalled(True) |
| |
| # The new service should get marked as 'activated'. |
| self.test.CheckServiceActivationState('activated') |
| |
| class ActivatedDueToMdnTest(ActivationTest): |
| """ |
| This test verifies that a valid MDN should cause the service to get marked |
| as 'activated'. |
| |
| """ |
| def SetupTestModem(self): |
| return ActivationTest.TestModem() |
| |
| def RunTest(self): |
| # Service should appear as 'not-activated'. |
| self.test.CheckServiceActivationState('not-activated') |
| |
| # Update the MDN. The service should get marked as activated. |
| modem = self.test.GetModem() |
| modem.PropertiesInterface().Set(mm1.I_MODEM, |
| 'OwnNumbers', |
| ['1111111111']) |
| time.sleep(5) |
| self.test.CheckServiceActivationState('activated') |
| |
| class network_LTEActivate(test.test): |
| """ |
| After an online payment to activate a network, shill keeps track of service |
| activation by monitoring changes to network registration and MDN updates |
| combined with a modem reset. The test checks that the |
| Cellular.ActivationState property of the service has the correct value |
| associated with it by simulating possible scenarios using the pseudo modem |
| manager. |
| |
| """ |
| version = 1 |
| |
| def FindModem(self): |
| """ |
| Tries to find a modem object exposed by the current modem manager and |
| returns a proxy to it. |
| |
| @return A modem proxy, or None if not found. |
| |
| """ |
| try: |
| manager, modem_path = mm.PickOneModem('') |
| return manager.GetModem(modem_path) |
| except ValueError as e: |
| # TODO(armansito): PickOneModem, for some predictably beautiful |
| # reason, raises a ValueError instead of something more specific. |
| # We should change that. |
| logging.info('Error while getting modem: ' + repr(e)) |
| return None |
| |
| def GetModem(self): |
| """ |
| Returns a modem proxy. This method will block for a LONG_TIMEOUT amount |
| of time and retry to obtain a modem proxy until the timeout expires. |
| |
| """ |
| utils.poll_for_condition( |
| lambda: self.FindModem() is not None, |
| exception=error.TestFail('Modem not found.'), |
| timeout=LONG_TIMEOUT) |
| return self.FindModem() |
| |
| def GetModemState(self): |
| """Returns the current ModemManager modem state.""" |
| modem = self.GetModem() |
| props = modem.GetAll(mm1.I_MODEM) |
| return props['State'] |
| |
| def GetResetCalled(self, modem): |
| """ |
| Returns the current value of the "ResetCalled" property of the current |
| modem. |
| |
| @param modem: Modem proxy to send the query to. |
| |
| """ |
| return modem.GetAll(I_ACTIVATION_TEST)['ResetCalled'] |
| |
| def _CheckResetCalledHelper(self, expected_value): |
| modem = self.GetModem() |
| try: |
| return self.GetResetCalled(modem) == expected_value |
| except dbus.exceptions.DBusException as e: |
| name = e.get_dbus_name() |
| unknown_method_str = 'org.feedesktop.DBus.Error.UnknownMethod' |
| unknown_object_str = 'org.feedesktop.DBus.Error.UnknowObject' |
| if name == unknown_method_str or name == unknown_object_str: |
| return False |
| raise e |
| |
| def CheckResetCalled(self, expected_value): |
| """ |
| Checks that the ResetCalled property on the modem matches the expect |
| value. |
| |
| @param expected_value: The expected value of ResetCalled. |
| |
| """ |
| utils.poll_for_condition( |
| lambda: self._CheckResetCalledHelper(expected_value), |
| exception=error.TestFail("\"ResetCalled\" did not match: " + |
| str(expected_value)), |
| timeout=LONG_TIMEOUT) |
| |
| def EnsureModemStateReached(self, expected_state, timeout): |
| """ |
| Asserts that the underlying modem state becomes |expected_state| within |
| |timeout|. |
| |
| @param expected_state: The expected modem state. |
| @param timeout: Timeout in which the condition should be met. |
| |
| """ |
| utils.poll_for_condition( |
| lambda: self.GetModemState() == expected_state, |
| exception=error.TestFail( |
| 'Modem failed to reach state ' + |
| mm1.ModemStateToString(expected_state)), |
| timeout=timeout) |
| |
| def CheckServiceActivationState(self, expected_state): |
| """ |
| Asserts that the service activation state matches |expected_state| |
| within SHORT_TIMEOUT. |
| |
| @param expected_state: The expected service activation state. |
| |
| """ |
| logging.info('Checking for service activation state: %s', |
| expected_state) |
| service = self.FindCellularService() |
| success, state, duration = self.shill.wait_for_property_in( |
| service, |
| 'Cellular.ActivationState', |
| [expected_state], |
| SHORT_TIMEOUT) |
| if not success and state != expected_state: |
| raise error.TestError( |
| 'Service activation state should be \'%s\', but it is \'%s\'.' |
| % (expected_state, state)) |
| |
| def FindCellularService(self, check_not_none=True): |
| """ |
| Returns the current cellular service. |
| |
| @param check_not_none: If True, an error will be raised if no service |
| was found. |
| |
| """ |
| if check_not_none: |
| utils.poll_for_condition( |
| lambda: (self.shill.find_cellular_service_object() is |
| not None), |
| exception=error.TestError( |
| 'Could not find cellular service within timeout.'), |
| timeout=LONG_TIMEOUT); |
| |
| service = self.shill.find_cellular_service_object() |
| |
| # Check once more, to make sure it's valid. |
| if check_not_none and not service: |
| raise error.TestError('Could not find cellular service.') |
| return service |
| |
| def FindCellularDevice(self): |
| """Returns the current cellular device.""" |
| device = self.shill.find_cellular_device_object() |
| if not device: |
| raise error.TestError('Could not find cellular device.') |
| return device |
| |
| def ResetCellularDevice(self): |
| """ |
| Resets all modems, guaranteeing that the operation succeeds and doesn't |
| fail due to race conditions in pseudomodem start-up and test execution. |
| |
| """ |
| self.EnsureModemStateReached( |
| mm1.MM_MODEM_STATE_ENABLED, SHORT_TIMEOUT) |
| self.shill.reset_modem(self.FindCellularDevice()) |
| self.EnsureModemStateReached( |
| mm1.MM_MODEM_STATE_ENABLED, SHORT_TIMEOUT) |
| |
| def run_once(self): |
| with backchannel.Backchannel(): |
| self.sim = sim.SIM(sim.SIM.Carrier('test'), |
| mm1.MM_MODEM_ACCESS_TECHNOLOGY_LTE) |
| with pseudomodem.TestModemManagerContext(True, |
| sim=self.sim) as tmmc: |
| self.shill = cellular_proxy.CellularProxy.get_proxy() |
| self.shill.set_logging_for_cellular_test() |
| |
| tests = [ |
| TimeoutResetTest(tmmc.GetPseudoModemManager(), self), |
| TimeoutActivatedTest(tmmc.GetPseudoModemManager(), self), |
| ResetAfterRegisterTest(tmmc.GetPseudoModemManager(), self), |
| ActivatedDueToMdnTest(tmmc.GetPseudoModemManager(), self) |
| ] |
| |
| for test in tests: |
| test.Run() |
