system_stub.py revision 5f1c94371a64b3196d4be9466099bb892df9b88e 
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  pathsep = os.pathsep
201
202  def __init__(self, sys_module=sys):
203    self.path = OsModuleStub.OsPathModuleStub(sys_module)
204    self.environ = OsModuleStub.OsEnvironModuleStub()
205    self.display = ':0'
206    self.local_app_data = None
207    self.sys_path = None
208    self.program_files = None
209    self.program_files_x86 = None
210    self.devnull = os.devnull
211
212  def access(self, path, _):
213    return path in self.path.files
214
215  def getenv(self, name, value=None):
216    if name == 'DISPLAY':
217      env = self.display
218    elif name == 'LOCALAPPDATA':
219      env = self.local_app_data
220    elif name == 'PATH':
221      env = self.sys_path
222    elif name == 'PROGRAMFILES':
223      env = self.program_files
224    elif name == 'PROGRAMFILES(X86)':
225      env = self.program_files_x86
226    else:
227      raise NotImplementedError('Unsupported getenv')
228    return env if env else value
229
230
231class PerfControlModuleStub(object):
232  class PerfControlStub(object):
233    def __init__(self, adb):
234      pass
235
236  def __init__(self):
237    self.PerfControl = PerfControlModuleStub.PerfControlStub
238
239
240class RawInputFunctionStub(object):
241  def __init__(self):
242    self.input = ''
243
244  def __call__(self, name, *args, **kwargs):
245    return self.input
246
247
248class SubprocessModuleStub(object):
249  class PopenStub(object):
250    def __init__(self):
251      self.communicate_result = ('', '')
252
253    def __call__(self, args, **kwargs):
254      return self
255
256    def communicate(self):
257      return self.communicate_result
258
259  def __init__(self):
260    self.Popen = SubprocessModuleStub.PopenStub()
261    self.PIPE = None
262
263  def call(self, *args, **kwargs):
264    pass
265
266
267class SysModuleStub(object):
268  def __init__(self):
269    self.platform = ''
270
271
272class ThermalThrottleModuleStub(object):
273  class ThermalThrottleStub(object):
274    def __init__(self, adb):
275      pass
276
277  def __init__(self):
278    self.ThermalThrottle = ThermalThrottleModuleStub.ThermalThrottleStub
279