1# Copyright (c) 2012 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5"""Encapsulates interactions with the cellular testbed.""" 6 7import contextlib, logging, urllib2 8 9# We'd really prefer not to depend on autotest proper here 10from autotest_lib.client.bin import utils 11 12from autotest_lib.client.cros import backchannel, flimflam_test_path 13from autotest_lib.client.cros.cellular import cellular, cell_tools 14from autotest_lib.client.cros.cellular import emulator_config 15 16import flimflam 17 18 19TIMEOUT = 60 20 21class Error(Exception): 22 pass 23 24 25class Environment(object): 26 """Dispatch class: reads config and returns appropriate concrete type.""" 27 def __new__(cls, config, *args, **kwargs): 28 return EmulatedEnvironment(config, *args, **kwargs) 29 30 31class EmulatedEnvironment(object): 32 def __init__(self, config): 33 self.config = config 34 self.flim = None 35 self.emulator = None 36 37 def __enter__(self): 38 self.flim = flimflam.FlimFlam() 39 return self 40 41 def __exit__(self, exception, value, traceback): 42 if self.emulator: 43 self.emulator.Close() 44 return False 45 46 def StartDefault(self, technology): 47 (self.emulator, self.verifier) = emulator_config.StartDefault( 48 self.config, technology) 49 50 def CheckHttpConnectivity(self): 51 """Check that the device can fetch HTTP pages.""" 52 http_config = self.config.cell['http_connectivity'] 53 response = urllib2.urlopen(http_config['url'], timeout=TIMEOUT).read() 54 55 if ('url_required_contents' in http_config and 56 http_config['url_required_contents'] not in response): 57 logging.error('Could not find %s in \n\t%s\n', 58 http_config['url_required_contents'], response) 59 raise Error('Content downloaded, but it was incorrect') 60 61 def CheckedConnectToCellular(self, timeout=TIMEOUT): 62 """Connect to cellular, check if we are connected, return a service""" 63 (service, _) = cell_tools.ConnectToCellular(self.flim, timeout=timeout) 64 self.verifier.AssertDataStatusIn([ 65 cellular.UeGenericDataStatus.CONNECTED]) 66 return service 67 68 def CheckedDisconnectFromCellular(self, service): 69 """Disconnect from cellular and check that we're disconnected.""" 70 71 self.flim.DisconnectService(service) 72 73 def _ModemIsFullyDisconnected(): 74 return self.verifier.IsDataStatusIn([ 75 cellular.UeGenericDataStatus.REGISTERED, 76 cellular.UeGenericDataStatus.NONE,]) 77 78 utils.poll_for_condition( 79 _ModemIsFullyDisconnected, 80 timeout=20, 81 exception=Error('modem not disconnected from base station')) 82 83 84class DefaultCellularTestContext(object): 85 """Wraps useful contexts for a cellular test in a single context.""" 86 def __init__(self, config): 87 self._nested = contextlib.nested( 88 backchannel.Backchannel(), 89 cell_tools.OtherDeviceShutdownContext('cellular'), 90 Environment(config)) 91 92 def __enter__(self): 93 (self.backchannel, 94 self.other_device_shutdown_context, 95 self.env) = self._nested.__enter__() 96 return self 97 98 def __exit__(self, exception, value, traceback): 99 return self._nested.__exit__(exception, value, traceback) 100