99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
# Copyright (c) 2012 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.
|
|
|
|
"""Encapsulates interactions with the cellular testbed."""
|
|
|
|
import contextlib, logging, urllib2
|
|
|
|
# We'd really prefer not to depend on autotest proper here
|
|
from autotest_lib.client.bin import utils
|
|
|
|
from autotest_lib.client.cros import backchannel, flimflam_test_path
|
|
from autotest_lib.client.cros.cellular import cellular, cell_tools
|
|
from autotest_lib.client.cros.cellular import emulator_config
|
|
|
|
import flimflam
|
|
|
|
|
|
TIMEOUT = 60
|
|
|
|
class Error(Exception):
|
|
pass
|
|
|
|
|
|
class Environment(object):
|
|
"""Dispatch class: reads config and returns appropriate concrete type."""
|
|
def __new__(cls, config, *args, **kwargs):
|
|
return EmulatedEnvironment(config, *args, **kwargs)
|
|
|
|
|
|
class EmulatedEnvironment(object):
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.flim = None
|
|
self.emulator = None
|
|
|
|
def __enter__(self):
|
|
self.flim = flimflam.FlimFlam()
|
|
return self
|
|
|
|
def __exit__(self, exception, value, traceback):
|
|
if self.emulator:
|
|
self.emulator.Close()
|
|
return False
|
|
|
|
def StartDefault(self, technology):
|
|
(self.emulator, self.verifier) = emulator_config.StartDefault(
|
|
self.config, technology)
|
|
|
|
def CheckHttpConnectivity(self):
|
|
"""Check that the device can fetch HTTP pages."""
|
|
http_config = self.config.cell['http_connectivity']
|
|
response = urllib2.urlopen(http_config['url'], timeout=TIMEOUT).read()
|
|
|
|
if ('url_required_contents' in http_config and
|
|
http_config['url_required_contents'] not in response):
|
|
logging.error('Could not find %s in \n\t%s\n',
|
|
http_config['url_required_contents'], response)
|
|
raise Error('Content downloaded, but it was incorrect')
|
|
|
|
def CheckedConnectToCellular(self, timeout=TIMEOUT):
|
|
"""Connect to cellular, check if we are connected, return a service"""
|
|
(service, _) = cell_tools.ConnectToCellular(self.flim, timeout=timeout)
|
|
self.verifier.AssertDataStatusIn([
|
|
cellular.UeGenericDataStatus.CONNECTED])
|
|
return service
|
|
|
|
def CheckedDisconnectFromCellular(self, service):
|
|
"""Disconnect from cellular and check that we're disconnected."""
|
|
|
|
self.flim.DisconnectService(service)
|
|
|
|
def _ModemIsFullyDisconnected():
|
|
return self.verifier.IsDataStatusIn([
|
|
cellular.UeGenericDataStatus.REGISTERED,
|
|
cellular.UeGenericDataStatus.NONE,])
|
|
|
|
utils.poll_for_condition(
|
|
_ModemIsFullyDisconnected,
|
|
timeout=20,
|
|
exception=Error('modem not disconnected from base station'))
|
|
|
|
|
|
class DefaultCellularTestContext(object):
|
|
"""Wraps useful contexts for a cellular test in a single context."""
|
|
def __init__(self, config):
|
|
self._nested = contextlib.nested(
|
|
backchannel.Backchannel(),
|
|
cell_tools.OtherDeviceShutdownContext('cellular'),
|
|
Environment(config))
|
|
|
|
def __enter__(self):
|
|
(self.backchannel,
|
|
self.other_device_shutdown_context,
|
|
self.env) = self._nested.__enter__()
|
|
return self
|
|
|
|
def __exit__(self, exception, value, traceback):
|
|
return self._nested.__exit__(exception, value, traceback)
|