| #!/usr/bin/env python2.7 | |
| # Copyright 2019 The Fuchsia Authors. All rights reserved. | |
| # Use of this source code is governed by a BSD-style license that can be | |
| # found in the LICENSE file. | |
| import errno | |
| import fnmatch | |
| import os | |
| import subprocess | |
| from StringIO import StringIO | |
| import test_env | |
| from lib.host import Host | |
| from process_fake import FakeProcess | |
| class FakeHost(Host): | |
| """Fake host that avoids interacting with the real system. | |
| Attributes: | |
| history: List of commands whose execution has been faked. | |
| log: Saves messages printed by echo, error, or choose. | |
| selection: Pre-selected option for a future call to choose(). | |
| """ | |
| ENVVARS = {} | |
| def __init__(self): | |
| super(FakeHost, self).__init__() | |
| self._log = [] | |
| self._dirs = set() | |
| self._files = {} | |
| self._links = {} | |
| self._envvars = {} | |
| self._processes = {} | |
| self._elapsed = 0.0 | |
| self._cwd = '/' | |
| self.tracing |= os.getenv(Host.TRACE_ENVVAR) == '1' | |
| if self.tracing: | |
| # Makes the output a bit prettier... | |
| print('') | |
| @property | |
| def processes(self): | |
| """Commands whose creation and/or execution has been faked.""" | |
| return self._processes | |
| @property | |
| def elapsed(self): | |
| return self._elapsed | |
| @property | |
| def cwd(self): | |
| return self._cwd | |
| @cwd.setter | |
| def cwd(self, cwd): | |
| self._cwd = cwd | |
| # Fake filesystem routines | |
| def getcwd(self): | |
| """Fake implementation overriding Host.getcwd.""" | |
| return self._cwd | |
| def _dereference(self, pathname): | |
| while pathname in self._links: | |
| pathname = self._links[pathname] | |
| return pathname | |
| def isdir(self, pathname): | |
| """Fake implementation overriding Host.isdir.""" | |
| return self._dereference(pathname) in self._dirs | |
| def isfile(self, pathname): | |
| """Fake implementation overriding Host.isfile.""" | |
| return self._dereference(pathname) in self._files | |
| def glob(self, pattern): | |
| """Fake implementation overriding Host.glob.""" | |
| return sorted(fnmatch.filter(self._files, pattern)) | |
| class File(StringIO): | |
| """A file-like object that can be used in "with" statements.""" | |
| def __init__(self): | |
| self._contents = None | |
| def open(self, mode): | |
| """Recreates the StringIO base object to simulate opening.""" | |
| StringIO.__init__(self) | |
| if mode == 'w' or mode == 'w+': | |
| return | |
| self.write(self._contents) | |
| if mode == 'r' or mode == 'r+': | |
| self.seek(0, 0) | |
| def __enter__(self): | |
| return self | |
| def __exit__(self, exc_type, exc_value, exc_traceback): | |
| if not exc_type: | |
| self.seek(0, 0) | |
| self._contents = self.read() | |
| def open(self, pathname, mode='r', on_error=None, missing_ok=False): | |
| """Opens a fake file for reading and/or writing.""" | |
| self.trace(' opening: {}'.format(pathname)) | |
| pathname = self._dereference(pathname) | |
| assert pathname not in self._dirs, 'Directory exists: {}'.format( | |
| pathname) | |
| if pathname in self._files: | |
| file = self._files[pathname] | |
| elif mode != 'r' and mode != 'r+': | |
| file = FakeHost.File() | |
| self._files[pathname] = file | |
| elif missing_ok: | |
| return None | |
| elif on_error: | |
| self.error(*on_error) | |
| else: | |
| self.error('Failed to open {}.'.format(pathname)) | |
| file.open(mode) | |
| return file | |
| def mkdir(self, pathname): | |
| """Fake implementation overriding BuildEnv.mkdir.""" | |
| self.trace('creating: {}'.format(pathname)) | |
| pathname = self._dereference(pathname) | |
| assert pathname not in self._files, 'File exists: {}'.format(pathname) | |
| self._dirs.add(pathname) | |
| self.create_process(['mkdir', '-p', pathname]) | |
| def link(self, pathname, linkname): | |
| """Fake implementation overriding BuildEnv.link.""" | |
| self.trace(' linking: {}'.format(linkname)) | |
| self.trace(' to: {}'.format(pathname)) | |
| self._links[linkname] = pathname | |
| self.create_process(['ln', '-s', pathname, linkname]) | |
| def remove(self, pathname): | |
| self.trace('removing: {}'.format(pathname)) | |
| pathname = self._dereference(pathname) | |
| if pathname in self._dirs: | |
| self._dirs.remove(pathname) | |
| elif pathname in self._files: | |
| del self._files[pathname] | |
| self.create_process(['rm', '-rf', pathname]) | |
| def _mkdtemp(self): | |
| temp_dir = 'temp_dir' | |
| self.mkdir(temp_dir) | |
| return temp_dir | |
| def add_golden(self, name): | |
| """Add a 'real' golden file in the test directory to the fake filesystem.""" | |
| pathname = os.path.join( | |
| os.path.abspath(os.path.dirname(__file__)), name) | |
| with open(pathname) as f: | |
| golden = f.read() | |
| with self.open(name, 'w') as f: | |
| f.write(golden) | |
| # Other routines | |
| def getenv(self, name): | |
| """Fake of implementation of getenv.""" | |
| if name in self._envvars: | |
| return self._envvars[name] | |
| else: | |
| return FakeHost.ENVVARS.get(name, None) | |
| def setenv(self, name, value): | |
| """Fake of implementation of setenv.""" | |
| self._envvars[name] = value | |
| def create_process(self, args, **kwargs): | |
| """Creates a FakeProcess from subprocess-like parameters. | |
| This method caches the returned object in self._processes, and will | |
| return the same object when given the same parameters. This allows tests | |
| to "pre-create" fake processes and set their responses before invoking | |
| methods under test that create them and use those responses. | |
| """ | |
| cmd = ' '.join(args) | |
| self.trace(' running: {}'.format(cmd)) | |
| process = self._processes.get(cmd, None) | |
| if not process: | |
| process = FakeProcess(self, args, **kwargs) | |
| self._processes[cmd] = process | |
| return process | |
| def sleep(self, duration): | |
| if duration < 0: | |
| return | |
| self.trace('sleeping: {}'.format(duration)) | |
| self._elapsed += duration |