# Copyright (c) 2014 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.
import dbus
import logging
import os
import time
from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.cellular import mm1_constants
from autotest_lib.client.cros.cellular.pseudomodem import pm_constants
from autotest_lib.client.cros.cellular.pseudomodem import pseudomodem_context
from autotest_lib.client.cros.networking import cellular_proxy
# Used for software message propagation latencies.
SHORT_TIMEOUT_SECONDS = 2
STATE_MACHINE_SCAN = 'ScanMachine'
TEST_MODEMS_MODULE_PATH = os.path.join(os.path.dirname(__file__), 'files',
'modems.py')
class cellular_ScanningProperty(test.test):
"""
Test that the |Scanning| Property of the shill cellular device object is
updated correctly in the following two scenarios:
(1) When a user requests a network scan using the |ProposeScan| method of
the cellular device.
(2) During the initial modem enable-register-connect sequence.
"""
version = 1
def _find_mm_modem(self):
"""
Find the modemmanager modem object.
Assumption: There is only one modem in the system.
@raises: TestError unless exactly one modem is found.
"""
object_manager = dbus.Interface(
self._bus.get_object(mm1_constants.I_MODEM_MANAGER,
mm1_constants.MM1),
mm1_constants.I_OBJECT_MANAGER)
try:
modems = object_manager.GetManagedObjects()
except dbus.exceptions.DBusException as e:
raise error.TestFail('Failed to list the available modems. '
'DBus error: |%s|', repr(e))
if len(modems) != 1:
raise error.TestFail('Expected one modem object, found %d' %
len(modems))
modem_path = modems.keys()[0]
modem_object = self._bus.get_object(mm1_constants.I_MODEM_MANAGER,
modem_path)
# Check that this object is valid
try:
modem_object.GetAll(mm1_constants.I_MODEM,
dbus_interface=mm1_constants.I_PROPERTIES)
except dbus.exceptions.DBusException as e:
raise error.TestFail('Failed to obtain dbus object for the modem '
'DBus error: |%s|', repr(e))
return dbus.Interface(modem_object, mm1_constants.I_MODEM)
def _check_mm_state(self, modem, states):
"""
Verify that the modemmanager state is |state|.
@param modem: A DBus object for the modemmanager modem.
@param states: The expected state of the modem. This is either a single
state, or a list of states.
@raises: TestError if the state differs.
"""
if not isinstance(states, list):
states = [states]
properties = modem.GetAll(mm1_constants.I_MODEM,
dbus_interface=mm1_constants.I_PROPERTIES)
actual_state = properties[mm1_constants.MM_MODEM_PROPERTY_NAME_STATE]
if actual_state not in states:
state_names = [mm1_constants.ModemStateToString(x) for x in states]
raise error.TestFail(
'Expected modemmanager modem state to be one of %s but '
'found %s' %
(state_names,
mm1_constants.ModemStateToString(actual_state)))
def _check_shill_property_update(self, cellular_device, property_name,
old_state, new_state):
"""
Check the value of property of shill.
@param cellular_device: The DBus proxy object for the cellular device.
@param property_name: Name of the property to check.
@param old_state: old value of property.
@param new_state: new expected value of property.
@raises: TestError if the property fails to enter the given state.
"""
# If we don't expect a change in the value, there is a race between this
# check and a possible (erronous) update of the value. Allow some time
# for the property to be updated before checking.
if old_state == new_state:
time.sleep(SHORT_TIMEOUT_SECONDS)
polling_timeout = 0
else:
polling_timeout = SHORT_TIMEOUT_SECONDS
success, _, _ = self._cellular_proxy.wait_for_property_in(
cellular_device,
property_name,
(new_state,),
timeout_seconds=polling_timeout)
if not success:
raise error.TestFail('Shill failed to set |%s| to %s.' %
(property_name, str(new_state)))
def _itesting_machine(self, machine_name, timeout=SHORT_TIMEOUT_SECONDS):
"""
Get the testing interface of the given interactive state machine.
@param machine_name: The name of the interactive state machine.
@return dbus.Interface for the testing interface of
InteractiveScanningMachine, if found. None otherwise.
@raises utils.TimeoutError if a valid dbus object can't be found.
"""
def _get_machine():
machine = self._bus.get_object(
mm1_constants.I_MODEM_MANAGER,
'/'.join([pm_constants.TESTING_PATH, machine_name]))
if machine:
i_machine = dbus.Interface(machine, pm_constants.I_TESTING_ISM)
# Only way to know if this DBus object is valid is to call a
# method on it.
try:
i_machine.IsWaiting() # Ignore result.
return i_machine
except dbus.exceptions.DBusException as e:
logging.debug(e)
return None
utils.poll_for_condition(_get_machine, timeout=timeout)
return _get_machine()
def test_user_initiated_cellular_scan(self):
"""
Test that the |ProposeScan| DBus method exported by shill cellular
object correctly updates the cellular object |Scanning| property while
the scan is in progress.
"""
with pseudomodem_context.PseudoModemManagerContext(
True,
{'test-module' : TEST_MODEMS_MODULE_PATH,
'test-modem-class' : 'AsyncScanModem'}):
self._cellular_proxy = cellular_proxy.CellularProxy.get_proxy()
self._bus = dbus.SystemBus()
self._cellular_proxy.set_logging_for_cellular_test()
logging.info('Sanity check initial values')
utils.poll_for_condition(
self._cellular_proxy.find_cellular_device_object,
exception=error.TestFail(
'Bad initial state: Failed to obtain a cellular '
'device in pseudomodem context.'),
timeout=SHORT_TIMEOUT_SECONDS)
device = self._cellular_proxy.find_cellular_device_object()
try:
self._itesting_machine(STATE_MACHINE_SCAN, 0)
raise error.TestFail('Bad initial state: scan machine created '
'by pseudomodem before scan is proposed.')
except utils.TimeoutError:
pass
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
False,
False)
logging.info('Test actions and checks')
device.ProposeScan()
try:
itesting_scan_machine = self._itesting_machine(
STATE_MACHINE_SCAN)
except utils.TimeoutError:
raise error.TestFail('Pseudomodem failed to launch %s' %
STATE_MACHINE_SCAN)
utils.poll_for_condition(
itesting_scan_machine.IsWaiting,
exception=error.TestFail('Scan machine failed to enter '
'scan state'),
timeout=SHORT_TIMEOUT_SECONDS)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
False,
True)
itesting_scan_machine.Advance()
utils.poll_for_condition(
lambda: not itesting_scan_machine.IsWaiting(),
exception=error.TestFail('Scan machine failed to exit '
'scan state'),
timeout=SHORT_TIMEOUT_SECONDS)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
True,
False)
def test_activated_service_states(self):
"""
Test that shill |Scanning| property is updated correctly when an
activated 3GPP service connects.
"""
with pseudomodem_context.PseudoModemManagerContext(
True,
{'test-module' : TEST_MODEMS_MODULE_PATH,
'test-state-machine-factory-class' :
'InteractiveStateMachineFactory'}):
self._cellular_proxy = cellular_proxy.CellularProxy.get_proxy()
self._bus = dbus.SystemBus()
self._cellular_proxy.set_logging_for_cellular_test()
logging.info('Sanity check initial values')
enable_machine = self._itesting_machine(
pm_constants.STATE_MACHINE_ENABLE)
utils.poll_for_condition(
enable_machine.IsWaiting,
exception=error.TestFail(
'Bad initial state: Pseudomodem did not launch '
'Enable machine'),
timeout=SHORT_TIMEOUT_SECONDS)
utils.poll_for_condition(
self._cellular_proxy.find_cellular_device_object,
exception=error.TestFail(
'Bad initial state: Failed to obtain a cellular '
'device in pseudomodem context.'),
timeout=SHORT_TIMEOUT_SECONDS)
device = self._cellular_proxy.find_cellular_device_object()
mm_modem = self._find_mm_modem()
logging.info('Test Connect sequence')
self._check_mm_state(mm_modem,
mm1_constants.MM_MODEM_STATE_DISABLED)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_POWERED,
False,
False)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
False,
False)
logging.info('Expectation met: |Scanning| is False in MM state '
'Disabled')
enable_machine.Advance()
# MM state: Enabling
utils.poll_for_condition(
enable_machine.IsWaiting,
exception=error.TestFail('EnableMachine failed to wait in '
'Enabling state'),
timeout=SHORT_TIMEOUT_SECONDS)
self._check_mm_state(mm_modem,
mm1_constants.MM_MODEM_STATE_ENABLING)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
False,
True)
logging.info('Expectation met: |Scanning| is True in MM state '
'Enabling')
enable_machine.Advance()
# MM state: Enabled
utils.poll_for_condition(
enable_machine.IsWaiting,
exception=error.TestFail('EnableMachine failed to wait in '
'Enabled state'),
timeout=SHORT_TIMEOUT_SECONDS)
# Finish the enable call.
enable_machine.Advance()
self._check_mm_state(mm_modem, mm1_constants.MM_MODEM_STATE_ENABLED)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_POWERED,
False,
True)
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
True,
True)
register_machine = self._itesting_machine(
pm_constants.STATE_MACHINE_REGISTER)
utils.poll_for_condition(
register_machine.IsWaiting,
exception=error.TestFail('SearchingMachine failed to wait '
'in Enabled state'),
timeout=SHORT_TIMEOUT_SECONDS)
logging.info('Expectation met: |Scanning| is True in MM state '
'Enabled')
register_machine.Advance()
# MM state: Searching
utils.poll_for_condition(
register_machine.IsWaiting,
exception=error.TestFail('SearchingMachine failed to wait '
'in Searching state'),
timeout=SHORT_TIMEOUT_SECONDS)
self._check_mm_state(mm_modem,
mm1_constants.MM_MODEM_STATE_SEARCHING)
enable_machine.Advance()
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
True,
True)
logging.info('Expectation met: |Scanning| is True in MM state '
'Searching')
register_machine.Advance()
# MM state: >= Registered
utils.poll_for_condition(
self._cellular_proxy.find_cellular_service_object,
error.TestFail('Failed to create Cellular Service for a '
'registered modem'),
timeout=SHORT_TIMEOUT_SECONDS)
self._check_mm_state(mm_modem,
[mm1_constants.MM_MODEM_STATE_REGISTERED,
mm1_constants.MM_MODEM_STATE_CONNECTING,
mm1_constants.MM_MODEM_STATE_CONNECTED])
self._check_shill_property_update(
device,
self._cellular_proxy.DEVICE_PROPERTY_SCANNING,
True,
False)
logging.info('Expectation met: |Scanning| is False in MM state '
'Registered')
def run_once(self):
""" Autotest entry function """
self.test_user_initiated_cellular_scan()
self.test_activated_service_states()