webkitpy: Override TEMPDIR for Xvfb.

third_party/WebKit/Tools/Scripts/webkitpy/common/system/executive_mock.py

 # Copyright (C) 2011 Google Inc. All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are
 # met:
 #
 #    * Redistributions of source code must retain the above copyright
 # notice, this list of conditions and the following disclaimer.
 #    * Redistributions in binary form must reproduce the above
 # copyright notice, this list of conditions and the following disclaimer
 # in the documentation and/or other materials provided with the
 # distribution.
 #    * Neither the name of Google Inc. nor the names of its
 # contributors may be used to endorse or promote products derived from
 # this software without specific prior written permission.
 #
 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+import collections
 import logging
 import os
 import StringIO
 
 from webkitpy.common.system.executive import ScriptError
 
 _log = logging.getLogger(__name__)
 
 
 class MockProcess(object):
@@ skipping 14 matching lines @@
             return None
         return self.returncode
 
     def communicate(self, *_):
         return (self.stdout.getvalue(), self.stderr.getvalue())
 
     def kill(self):
         return
 
 
+MockCall = collections.namedtuple('MockCall', ('args', 'env'))
+
+
 class MockExecutive(object):
     PIPE = 'MOCK PIPE'
     STDOUT = 'MOCK STDOUT'
     DEVNULL = 'MOCK_DEVNULL'
 
     @staticmethod
     def ignore_error(error):
         pass
 
     def __init__(self, should_log=False, should_throw=False,
                  output='MOCK output of child process', stderr='',
                  exit_code=0, exception=None, run_command_fn=None):
         self._should_log = should_log
         self._should_throw = should_throw
         # FIXME: Once executive wraps os.getpid() we can just use a static pid for "this" process.
         self._running_pids = {'test-webkitpy': os.getpid()}
         self._output = output
         self._stderr = stderr
         self._exit_code = exit_code
         self._exception = exception
         self._run_command_fn = run_command_fn
         self._proc = None
-        self.calls = []
+        self.full_calls = []
+
+    def _append_call(self, args, env=None):
+        if env:
+            env = env.copy()
+        self.full_calls.append(MockCall(
+            args=args,
+            env=env.copy() if env is not None else None,
+        ))
 
     def check_running_pid(self, pid):
         return pid in self._running_pids.values()
 
     def running_pids(self, process_name_filter):
         running_pids = []
         for process_name, process_pid in self._running_pids.iteritems():
             if process_name_filter(process_name):
                 running_pids.append(process_pid)
 
@@ skipping 10 matching lines @@
                     args,
                     cwd=None,
                     input=None,  # pylint: disable=redefined-builtin
                     timeout_seconds=False,
                     error_handler=None,
                     return_exit_code=False,
                     return_stderr=True,
                     decode_output=False,
                     env=None,
                     debug_logging=False):
-        self.calls.append(args)
+        self._append_call(args, env=env)
 
         assert isinstance(args, list) or isinstance(args, tuple)
 
         if self._should_log:
             env_string = ''
             if env:
                 env_string = ', env=%s' % env
             input_string = ''
             if input:
                 input_string = ', input=%s' % input
@@ skipping 26 matching lines @@
         pass
 
     def kill_process(self, pid):
         pass
 
     def interrupt(self, pid):
         pass
 
     def popen(self, args, cwd=None, env=None, **_):
         assert all(isinstance(arg, basestring) for arg in args)
-        self.calls.append(args)
+        self._append_call(args, env=env)
         if self._should_log:
             cwd_string = ''
             if cwd:
                 cwd_string = ', cwd=%s' % cwd
             env_string = ''
             if env:
                 env_string = ', env=%s' % env
             _log.info('MOCK popen: %s%s%s', args, cwd_string, env_string)
         if not self._proc:
             self._proc = MockProcess(self._output)
         return self._proc
 
     def call(self, args, **_):
         assert all(isinstance(arg, basestring) for arg in args)
-        self.calls.append(args)
+        self._append_call(args)
         _log.info('Mock call: %s', args)
 
     def run_in_parallel(self, commands):
         assert len(commands)
 
-        num_previous_calls = len(self.calls)
+        num_previous_calls = len(self.full_calls)
         command_outputs = []
         for cmd_line, cwd in commands:
             assert all(isinstance(arg, basestring) for arg in cmd_line)
             command_outputs.append([0, self.run_command(cmd_line, cwd=cwd), ''])
 
-        new_calls = self.calls[num_previous_calls:]
-        self.calls = self.calls[:num_previous_calls]
-        self.calls.append(new_calls)
+        new_calls = self.full_calls[num_previous_calls:]
+        self.full_calls = self.full_calls[:num_previous_calls]
+        self.full_calls.append(new_calls)
         return command_outputs
 
     def map(self, thunk, arglist, processes=None):
         return map(thunk, arglist)
 
     def process_dump(self):
         return []
 
+    @property
+    def calls(self):
+        def get_args(v):
+            if isinstance(v, list):
+                return [get_args(e) for e in v]
+            elif isinstance(v, MockCall):
+                return v.args
+            else:
+                return TypeError('Unknown full_calls type: %s' % (type(v).__name__,))
+        return get_args(self.full_calls)

[Dirk Pranke, 2017/05/04 00:05:48]

Can't this just be:

  @property
  def calls(self):
    return [fc.args for fc in self.full_calls]

If full_calls isn't a list of MockCalls, you'll get some sort of assertion here,
and isn't that good enough?

[dnj, 2017/05/04 00:20:37]

No, b/c "run_in_parallel" adds lists of MockCall to the "full_calls" list, so
it's at least one level recursive.

[Dirk Pranke, 2017/05/04 00:43:46]

Ah, I see. That's ugly. Can you add a TODO and/or file a bug so we can get rid
of that?

[dnj, 2017/05/04 16:07:54]

Done.

+
 
 def mock_git_commands(vals, strict=False):
     def run_fn(args):
         sub_command = args[1]
         if strict and sub_command not in vals:
             raise AssertionError('{} not found in sub-command list {}'.format(
                 sub_command, vals))
         return vals.get(sub_command, '')
     return MockExecutive(run_command_fn=run_fn)