• 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    }
28    self.adb_commands = None
29    self.os = None
30    self.subprocess = None
31    self.sys = None
32
33    self._base_module = base_module
34    self._overrides = {}
35
36    for module_name in module_list:
37      self._overrides[module_name] = getattr(base_module, module_name, None)
38      setattr(self, module_name, stubs[module_name]())
39      setattr(base_module, module_name, getattr(self, module_name))
40
41    if self.os and self.sys:
42      self.os.path.sys = self.sys
43
44  def __del__(self):
45    assert not len(self._overrides)
46
47  def Restore(self):
48    for module_name, original_module in self._overrides.iteritems():
49      setattr(self._base_module, module_name, original_module)
50    self._overrides = {}
51
52
53class AdbCommandsModuleStub(object):
54  class AdbCommandsStub(object):
55    def __init__(self, module, device):
56      self._module = module
57      self._device = device
58      self.is_root_enabled = True
59
60    def RunShellCommand(self, args):
61      if isinstance(args, basestring):
62        args = shlex.split(args)
63      handler = self._module.shell_command_handlers[args[0]]
64      return handler(args)
65
66    def IsRootEnabled(self):
67      return self.is_root_enabled
68
69    def RestartAdbdOnDevice(self):
70      pass
71
72    def IsUserBuild(self):
73      return False
74
75    def WaitForDevicePm(self):
76      pass
77
78  def __init__(self):
79    self.attached_devices = []
80    self.shell_command_handlers = {}
81
82    def AdbCommandsStubConstructor(device=None):
83      return AdbCommandsModuleStub.AdbCommandsStub(self, device)
84    self.AdbCommands = AdbCommandsStubConstructor
85
86  @staticmethod
87  def IsAndroidSupported():
88    return True
89
90  def GetAttachedDevices(self):
91    return self.attached_devices
92
93  def SetupPrebuiltTools(self, _):
94    return True
95
96  def CleanupLeftoverProcesses(self):
97    pass
98
99
100class CloudStorageModuleStub(object):
101  INTERNAL_BUCKET = None
102  PUBLIC_BUCKET = None
103
104  class CloudStorageError(Exception):
105    pass
106
107  def __init__(self):
108    self.remote_paths = []
109    self.local_file_hashes = {}
110    self.local_hash_files = {}
111
112  def List(self, _):
113    return self.remote_paths
114
115  def Insert(self, bucket, remote_path, local_path):
116    pass
117
118  def CalculateHash(self, file_path):
119    return self.local_file_hashes[file_path]
120
121  def ReadHash(self, hash_path):
122    return self.local_hash_files[hash_path]
123
124
125class OpenFunctionStub(object):
126  class FileStub(object):
127    def __init__(self, data):
128      self._data = data
129
130    def __enter__(self):
131      return self
132
133    def __exit__(self, *args):
134      pass
135
136    def read(self, size=None):
137      if size:
138        return self._data[:size]
139      else:
140        return self._data
141
142  def __init__(self):
143    self.files = {}
144
145  def __call__(self, name, *args, **kwargs):
146    return OpenFunctionStub.FileStub(self.files[name])
147
148
149class OsModuleStub(object):
150  class OsEnvironModuleStub(object):
151    def get(self, _):
152      return None
153
154  class OsPathModuleStub(object):
155    def __init__(self, sys_module):
156      self.sys = sys_module
157      self.files = []
158
159    def exists(self, path):
160      return path in self.files
161
162    def isfile(self, path):
163      return path in self.files
164
165    def join(self, *paths):
166      def IsAbsolutePath(path):
167        if self.sys.platform.startswith('win'):
168          return re.match('[a-zA-Z]:\\\\', path)
169        else:
170          return path.startswith('/')
171
172      # Per Python specification, if any component is an absolute path,
173      # discard previous components.
174      for index, path in reversed(list(enumerate(paths))):
175        if IsAbsolutePath(path):
176          paths = paths[index:]
177          break
178
179      if self.sys.platform.startswith('win'):
180        tmp = os.path.join(*paths)
181        return tmp.replace('/', '\\')
182      else:
183        tmp = os.path.join(*paths)
184        return tmp.replace('\\', '/')
185
186    @staticmethod
187    def expanduser(path):
188      return os.path.expanduser(path)
189
190    @staticmethod
191    def dirname(path):
192      return os.path.dirname(path)
193
194    @staticmethod
195    def splitext(path):
196      return os.path.splitext(path)
197
198  X_OK = os.X_OK
199
200  def __init__(self, sys_module=sys):
201    self.path = OsModuleStub.OsPathModuleStub(sys_module)
202    self.environ = OsModuleStub.OsEnvironModuleStub()
203    self.display = ':0'
204    self.local_app_data = None
205    self.program_files = None
206    self.program_files_x86 = None
207    self.devnull = os.devnull
208
209  def access(self, path, _):
210    return path in self.path.files
211
212  def getenv(self, name):
213    if name == 'DISPLAY':
214      return self.display
215    elif name == 'LOCALAPPDATA':
216      return self.local_app_data
217    elif name == 'PROGRAMFILES':
218      return self.program_files
219    elif name == 'PROGRAMFILES(X86)':
220      return self.program_files_x86
221    raise Exception('Unsupported getenv')
222
223
224class PerfControlModuleStub(object):
225  class PerfControlStub(object):
226    def __init__(self, adb):
227      pass
228
229  def __init__(self):
230    self.PerfControl = PerfControlModuleStub.PerfControlStub
231
232
233class RawInputFunctionStub(object):
234  def __init__(self):
235    self.input = ''
236
237  def __call__(self, name, *args, **kwargs):
238    return self.input
239
240
241class SubprocessModuleStub(object):
242  class PopenStub(object):
243    def __init__(self):
244      self.communicate_result = ('', '')
245
246    def __call__(self, args, **kwargs):
247      return self
248
249    def communicate(self):
250      return self.communicate_result
251
252  def __init__(self):
253    self.Popen = SubprocessModuleStub.PopenStub()
254    self.PIPE = None
255
256  def call(self, *args, **kwargs):
257    pass
258
259
260class SysModuleStub(object):
261  def __init__(self):
262    self.platform = ''
263
264
265class ThermalThrottleModuleStub(object):
266  class ThermalThrottleStub(object):
267    def __init__(self, adb):
268      pass
269
270  def __init__(self):
271    self.ThermalThrottle = ThermalThrottleModuleStub.ThermalThrottleStub
272