• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2012 The Chromium 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"""Provides stubs for os, sys and subprocess for testing
6
7This test allows one to test code that itself uses os, sys, and subprocess.
8"""
9
10import os
11import re
12import shlex
13import sys
14
15
16class Override(object):
17  def __init__(self, base_module, module_list):
18    stubs = {'adb_commands': AdbCommandsModuleStub,
19             'cloud_storage': CloudStorageModuleStub,
20             'open': OpenFunctionStub,
21             'os': OsModuleStub,
22             'perf_control': PerfControlModuleStub,
23             'raw_input': RawInputFunctionStub,
24             'subprocess': SubprocessModuleStub,
25             'sys': SysModuleStub,
26             'thermal_throttle': ThermalThrottleModuleStub,
27             'logging': LoggingStub
28    }
29    self.adb_commands = None
30    self.os = None
31    self.subprocess = None
32    self.sys = None
33
34    self._base_module = base_module
35    self._overrides = {}
36
37    for module_name in module_list:
38      self._overrides[module_name] = getattr(base_module, module_name, None)
39      setattr(self, module_name, stubs[module_name]())
40      setattr(base_module, module_name, getattr(self, module_name))
41
42    if self.os and self.sys:
43      self.os.path.sys = self.sys
44
45  def __del__(self):
46    assert not len(self._overrides)
47
48  def Restore(self):
49    for module_name, original_module in self._overrides.iteritems():
50      setattr(self._base_module, module_name, original_module)
51    self._overrides = {}
52
53
54class AndroidCommands(object):
55
56  def __init__(self):
57    self.can_access_protected_file_contents = False
58
59  def CanAccessProtectedFileContents(self):
60    return self.can_access_protected_file_contents
61
62
63class AdbDevice(object):
64
65  def __init__(self):
66    self.shell_command_handlers = {}
67    self.mock_content = []
68    self.system_properties = {}
69    if self.system_properties.get('ro.product.cpu.abi') == None:
70      self.system_properties['ro.product.cpu.abi'] = 'armeabi-v7a'
71    self.old_interface = AndroidCommands()
72
73  def RunShellCommand(self, args):
74    if isinstance(args, basestring):
75      args = shlex.split(args)
76    handler = self.shell_command_handlers[args[0]]
77    return handler(args)
78
79  def FileExists(self, _):
80    return False
81
82  def ReadFile(self, device_path, as_root=False):  # pylint: disable=W0613
83    return self.mock_content
84
85  def GetProp(self, property_name):
86    return self.system_properties[property_name]
87
88  def SetProp(self, property_name, property_value):
89    self.system_properties[property_name] = property_value
90
91
92class AdbCommandsModuleStub(object):
93
94  class AdbCommandsStub(object):
95
96    def __init__(self, module, device):
97      self._module = module
98      self._device = device
99      self.is_root_enabled = True
100      self._adb_device = module.adb_device
101
102    def IsRootEnabled(self):
103      return self.is_root_enabled
104
105    def RestartAdbdOnDevice(self):
106      pass
107
108    def IsUserBuild(self):
109      return False
110
111    def WaitForDevicePm(self):
112      pass
113
114    def device(self):
115      return self._adb_device
116
117  def __init__(self):
118    self.attached_devices = []
119    self.adb_device = AdbDevice()
120
121    def AdbCommandsStubConstructor(device=None):
122      return AdbCommandsModuleStub.AdbCommandsStub(self, device)
123    self.AdbCommands = AdbCommandsStubConstructor
124
125  @staticmethod
126  def IsAndroidSupported():
127    return True
128
129  def GetAttachedDevices(self):
130    return self.attached_devices
131
132  def SetupPrebuiltTools(self, _):
133    return True
134
135  def CleanupLeftoverProcesses(self):
136    pass
137
138
139class CloudStorageModuleStub(object):
140  PUBLIC_BUCKET = 'chromium-telemetry'
141  PARTNER_BUCKET = 'chrome-partner-telemetry'
142  INTERNAL_BUCKET = 'chrome-telemetry'
143  BUCKET_ALIASES = {
144    'public': PUBLIC_BUCKET,
145    'partner': PARTNER_BUCKET,
146    'internal': INTERNAL_BUCKET,
147  }
148
149  class CloudStorageError(Exception):
150    pass
151
152  def __init__(self):
153    self.remote_paths = []
154    self.local_file_hashes = {}
155    self.local_hash_files = {}
156
157  def List(self, _):
158    return self.remote_paths
159
160  def Insert(self, bucket, remote_path, local_path):
161    pass
162
163  def CalculateHash(self, file_path):
164    return self.local_file_hashes[file_path]
165
166  def ReadHash(self, hash_path):
167    return self.local_hash_files[hash_path]
168
169
170class LoggingStub(object):
171  def __init__(self):
172    self.warnings = []
173
174  def info(self, msg, *args):
175    pass
176
177  def warn(self, msg, *args):
178    self.warnings.append(msg % args)
179
180
181class OpenFunctionStub(object):
182  class FileStub(object):
183    def __init__(self, data):
184      self._data = data
185
186    def __enter__(self):
187      return self
188
189    def __exit__(self, *args):
190      pass
191
192    def read(self, size=None):
193      if size:
194        return self._data[:size]
195      else:
196        return self._data
197
198    def write(self, data):
199      self._data.write(data)
200
201    def close(self):
202      pass
203
204  def __init__(self):
205    self.files = {}
206
207  def __call__(self, name, *args, **kwargs):
208    return OpenFunctionStub.FileStub(self.files[name])
209
210
211class OsModuleStub(object):
212  class OsEnvironModuleStub(object):
213    def get(self, _):
214      return None
215
216  class OsPathModuleStub(object):
217    def __init__(self, sys_module):
218      self.sys = sys_module
219      self.files = []
220
221    def exists(self, path):
222      return path in self.files
223
224    def isfile(self, path):
225      return path in self.files
226
227    def join(self, *paths):
228      def IsAbsolutePath(path):
229        if self.sys.platform.startswith('win'):
230          return re.match('[a-zA-Z]:\\\\', path)
231        else:
232          return path.startswith('/')
233
234      # Per Python specification, if any component is an absolute path,
235      # discard previous components.
236      for index, path in reversed(list(enumerate(paths))):
237        if IsAbsolutePath(path):
238          paths = paths[index:]
239          break
240
241      if self.sys.platform.startswith('win'):
242        tmp = os.path.join(*paths)
243        return tmp.replace('/', '\\')
244      else:
245        tmp = os.path.join(*paths)
246        return tmp.replace('\\', '/')
247
248    @staticmethod
249    def expanduser(path):
250      return os.path.expanduser(path)
251
252    @staticmethod
253    def dirname(path):
254      return os.path.dirname(path)
255
256    @staticmethod
257    def splitext(path):
258      return os.path.splitext(path)
259
260  X_OK = os.X_OK
261
262  pathsep = os.pathsep
263
264  def __init__(self, sys_module=sys):
265    self.path = OsModuleStub.OsPathModuleStub(sys_module)
266    self.environ = OsModuleStub.OsEnvironModuleStub()
267    self.display = ':0'
268    self.local_app_data = None
269    self.sys_path = None
270    self.program_files = None
271    self.program_files_x86 = None
272    self.devnull = os.devnull
273
274  def access(self, path, _):
275    return path in self.path.files
276
277  def getenv(self, name, value=None):
278    if name == 'DISPLAY':
279      env = self.display
280    elif name == 'LOCALAPPDATA':
281      env = self.local_app_data
282    elif name == 'PATH':
283      env = self.sys_path
284    elif name == 'PROGRAMFILES':
285      env = self.program_files
286    elif name == 'PROGRAMFILES(X86)':
287      env = self.program_files_x86
288    else:
289      raise NotImplementedError('Unsupported getenv')
290    return env if env else value
291
292  def chdir(self, path):
293    pass
294
295
296class PerfControlModuleStub(object):
297  class PerfControlStub(object):
298    def __init__(self, adb):
299      pass
300
301  def __init__(self):
302    self.PerfControl = PerfControlModuleStub.PerfControlStub
303
304
305class RawInputFunctionStub(object):
306  def __init__(self):
307    self.input = ''
308
309  def __call__(self, name, *args, **kwargs):
310    return self.input
311
312
313class SubprocessModuleStub(object):
314  class PopenStub(object):
315    def __init__(self):
316      self.communicate_result = ('', '')
317
318    def __call__(self, args, **kwargs):
319      return self
320
321    def communicate(self):
322      return self.communicate_result
323
324  def __init__(self):
325    self.Popen = SubprocessModuleStub.PopenStub()
326    self.PIPE = None
327
328  def call(self, *args, **kwargs):
329    pass
330
331
332class SysModuleStub(object):
333  def __init__(self):
334    self.platform = ''
335
336
337class ThermalThrottleModuleStub(object):
338  class ThermalThrottleStub(object):
339    def __init__(self, adb):
340      pass
341
342  def __init__(self):
343    self.ThermalThrottle = ThermalThrottleModuleStub.ThermalThrottleStub
344