Attachment 407317 Details for Bug 215712 – Patch

[patch] Patch

bug-215712-20200826114413.patch (text/plain), 31.00 KB, created by Jonathan Bedard on 2020-08-26 11:44:14 PDT

(hide)

Description:

Filename:

MIME Type:

Creator: Jonathan Bedard

Created: 2020-08-26 11:44:14 PDT

Size: 31.00 KB

patch

obsolete

>Index: Tools/ChangeLog
>===================================================================
>--- Tools/ChangeLog	(revision 266177)
>+++ Tools/ChangeLog	(working copy)
>@@ -1,3 +1,37 @@
>+2020-08-26  Jonathan Bedard  <jbedard@apple.com>
>+
>+       [webkitcorepy] Standard Popen mocking API
>+        https://bugs.webkit.org/show_bug.cgi?id=215712
>+        <rdar://problem/67501911>
>+
>+        Reviewed by Dewei Zhu.
>+
>+        * Scripts/libraries/webkitcorepy/README.md: Add mocks.Subprocess documentation.
>+        * Scripts/libraries/webkitcorepy/webkitcorepy/__init__.py: Bump version.
>+        * Scripts/libraries/webkitcorepy/webkitcorepy/mocks/__init__.py:  Export Subprocess, ProcessCompletion.
>+        * Scripts/libraries/webkitcorepy/webkitcorepy/mocks/popen.py: Added.
>+        (Popen): Mock Popen object (for both Python 2 and 3). This object extracts a ProcessCompletion object from the
>+        mocked Subprocess stack and behaives according to the content of that object.
>+        * Scripts/libraries/webkitcorepy/webkitcorepy/mocks/subprocess.py: Added.
>+        (PopenBase): Base class to share code between Python 2 and 3 constructor.
>+        (Popen): Mock Popen object (for both Python 2 and 3). This object extracts a ProcessCompletion object from the
>+        mocked Subprocess stack and behaives according to the content of that object.
>+        * Scripts/libraries/webkitcorepy/webkitcorepy/mocks/subprocess.py: Added.
>+        (ProcessCompletion): Class which captures the result of a mock Popen call.
>+        (Subprocess):
>+        (Subprocess.Router): Object which routes a set of subprocess arguments to a specific ProcessCompletion or callback which
>+        generates a ProcessCompletion.
>+        (Subprocess.completion_generator_for): Given a file, search through the mocked Subprocess stack to find the
>+        first completion for the provided file.
>+        (Subprocess.completion_for): Extract file name from arguments, find the completion generator, and call
>+        it with the provieded arguments, working directory and stdin.
>+        (Subprocess.__init__):
>+        * Scripts/libraries/webkitcorepy/webkitcorepy/tests/mocks/subprocess_unittest.py: Added.
>+        (MockSubprocess):
>+        (MockCheckOutput):
>+        (MockCheckCall):
>+        (MockRun):
>+
> 2020-08-26  Youenn Fablet  <youenn@apple.com>
> 
>         enumerateDevices should expose audiooutput devices that are tied to an audio input device
>Index: Tools/Scripts/libraries/webkitcorepy/README.md
>===================================================================
>--- Tools/Scripts/libraries/webkitcorepy/README.md	(revision 266170)
>+++ Tools/Scripts/libraries/webkitcorepy/README.md	(working copy)
>@@ -76,3 +76,35 @@ from webkitcorepy import run
> 
> result = run([sys.executable, '-c', 'print("message")'], capture_output=True, encoding='utf-8')
> ```
>+
>+Mocking of subprocess commands:
>+```
>+from webkitcorepy import mocks, run
>+
>+with mocks.Subprocess(
>+    'ls', completion=mocks.ProcessCompletion(returncode=0, stdout='file1.txt\nfile2.txt\n'),
>+):
>+    result = run(['ls'], capture_output=True, encoding='utf-8')
>+    assert result.returncode == 0
>+    assert result.stdout == 'file1.txt\nfile2.txt\n'
>+```
>+The mocking system for subprocess also supports other subprocess APIs based on Popen:
>+```
>+with mocks.Subprocess(
>+    'ls', completion=mocks.ProcessCompletion(returncode=0, stdout='file1.txt\nfile2.txt\n'),
>+):
>+    assert subprocess.check_output(['ls']) == b'file1.txt\nfile2.txt\n'
>+    assert subprocess.check_call(['ls']) == 0
>+```
>+For writing integration tests, the mocking system for subprocess supports mocking multiple process calls at the same time:
>+```
>+with mocks.Subprocess(
>+    mocks.Subprocess.Router('command-a', 'argument', completion=mocks.ProcessCompletion(returncode=0)),
>+    mocks.Subprocess.Router('command-b', completion=mocks.ProcessCompletion(returncode=-1)),
>+):
>+    result = run(['command-a', 'argument'])
>+    assert result.returncode == 0
>+
>+    result = run(['command-b'])
>+    assert result.returncode == -1
>+```
>Index: Tools/Scripts/libraries/webkitcorepy/webkitcorepy/__init__.py
>===================================================================
>--- Tools/Scripts/libraries/webkitcorepy/webkitcorepy/__init__.py	(revision 266170)
>+++ Tools/Scripts/libraries/webkitcorepy/webkitcorepy/__init__.py	(working copy)
>@@ -35,7 +35,7 @@ from webkitcorepy.timeout import Timeout
> from webkitcorepy.subprocess_utils import TimeoutExpired, CompletedProcess, run
> from webkitcorepy.output_capture import LoggerCapture, OutputCapture, OutputDuplicate
> 
>-version = Version(0, 4, 0)
>+version = Version(0, 4, 1)
> 
> from webkitcorepy.autoinstall import Package, AutoInstall
> if sys.version_info > (3, 0):
>Index: Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/__init__.py
>===================================================================
>--- Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/__init__.py	(revision 266170)
>+++ Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/__init__.py	(working copy)
>@@ -22,3 +22,4 @@
> 
> from webkitcorepy.mocks.context_stack import ContextStack
> from webkitcorepy.mocks.time_ import Time
>+from webkitcorepy.mocks.subprocess import ProcessCompletion, Subprocess
>Index: Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/popen.py
>===================================================================
>--- Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/popen.py	(nonexistent)
>+++ Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/popen.py	(working copy)
>@@ -0,0 +1,237 @@
>+# Copyright (C) 2020 Apple 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:
>+# 1.  Redistributions of source code must retain the above copyright
>+#    notice, this list of conditions and the following disclaimer.
>+# 2.  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.
>+#
>+# THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS 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 APPLE INC. OR ITS 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 os
>+import io
>+import subprocess
>+import signal
>+import sys
>+import time
>+
>+from webkitcorepy import log, string_utils, Timeout, TimeoutExpired, unicode
>+from webkitcorepy.mocks import Subprocess
>+
>+# This file is mocked version of the subprocess.Popen object. This object differs slightly between Python 2 and 3.
>+# This object is not a complete mock of subprocess.Popen, but it does enable thorough testing of code which uses Popen,
>+# in particular, the subprocess.check_call, subprocess.check_output and subprocess.run commands.
>+#
>+# If you find yourself needing to edit this file, lean heavily on the testing in
>+# webkitcorepy.tests.mocks.subprocess_unittest and Python's implementation of Popen.
>+
>+class PopenBase(object):
>+    NEXT_PID = os.getpid() + 1
>+
>+    def __init__(self, args, bufsize=None, cwd=None, stdin=None, stdout=None, stderr=None):
>+        self._completion = None
>+        self._communication_started = False
>+        if bufsize is None:
>+            bufsize = -1
>+        if not isinstance(bufsize, int):
>+            raise TypeError("bufsize must be an integer")
>+
>+        self._args = args
>+        self._cwd = cwd
>+
>+        self.returncode = None
>+
>+        self.stdin = string_utils.BytesIO() if stdin is None or stdin == -1 else stdin
>+        self.stdout = string_utils.BytesIO() if stdout == -1 else stdout
>+        self._stdout_type = bytes if stdout == -1 else str
>+        self.stderr = string_utils.BytesIO() if stderr == -1 else stderr
>+        self._stderr_type = bytes if stderr == -1 else str
>+
>+        Subprocess.completion_generator_for(args[0])
>+
>+        self.pid = self.NEXT_PID
>+        self.NEXT_PID += 1
>+
>+        self._start_time = time.time()
>+
>+
>+class Popen(PopenBase):
>+    if sys.version_info > (3, 0):
>+        def __init__(self, args, bufsize=None, executable=None,
>+                     stdin=None, stdout=None, stderr=None,
>+                     preexec_fn=None, close_fds=True,
>+                     shell=False, cwd=None, env=None, universal_newlines=None,
>+                     startupinfo=None, creationflags=0,
>+                     restore_signals=True, start_new_session=False,
>+                     pass_fds=(), encoding=None, errors=None, text=None):
>+
>+            super(Popen, self).__init__(args, bufsize=bufsize, cwd=cwd, stdin=stdin, stdout=stdout, stderr=stderr)
>+
>+            if pass_fds and not close_fds:
>+                log.warn("pass_fds overriding close_fds.")
>+
>+            for arg in args:
>+                if not isinstance(arg, (str, bytes, os.PathLike)):
>+                    raise TypeError(
>+                        'expected {}, {} or os.PathLike object, not {}',
>+                        str, bytes, type(arg),
>+                    )
>+
>+            self.args = args
>+            self.encoding = encoding
>+            self.errors = errors
>+
>+            if (text is not None and universal_newlines is not None and bool(universal_newlines) != bool(text)):
>+                raise subprocess.SubprocessError('Cannot disambiguate when both text and universal_newlines are supplied but different. Pass one or the other.')
>+
>+            self.text_mode = encoding or errors or text or universal_newlines
>+
>+            if self.stdin is not None and self.text_mode:
>+                self.stdin = io.TextIOWrapper(self.stdin, write_through=True, line_buffering=(bufsize == 1), encoding=encoding, errors=errors)
>+            if self.stdout is not None and self.text_mode:
>+                self.stdout = io.TextIOWrapper(self.stdout, encoding=encoding, errors=errors)
>+                self._stdout_type = str
>+            if self.stderr is not None and self.text_mode:
>+                self.stderr = io.TextIOWrapper(self.stderr, encoding=encoding, errors=errors)
>+                self._stderr_type = str
>+
>+        def communicate(self, input=None, timeout=None):
>+            if self._communication_started and input:
>+                raise ValueError('Cannot send input after starting communication')
>+
>+            self._communication_started = True
>+            if input:
>+                self.stdin.write(string_utils.encode(input))
>+            self.wait(timeout=timeout)
>+            return self.stdout.read() if self.stdout else None, self.stderr.read() if self.stderr else None
>+
>+        def wait(self, timeout=None):
>+            if self.poll() is not None:
>+                return
>+
>+            if timeout and (self._completion.elapsed is None or timeout < self._completion.elapsed):
>+                raise TimeoutExpired(self._args, timeout)
>+
>+            if self._completion.elapsed is None:
>+                raise ValueError('Running a command that hangs without a timeout')
>+
>+            if self._completion.elapsed:
>+                time.sleep(self._completion.elapsed)
>+
>+            self.returncode = self._completion.returncode
>+            if self.stdout:
>+                self.stdout.seek(0)
>+            if self.stderr:
>+                self.stderr.seek(0)
>+
>+        def __enter__(self):
>+            return self
>+
>+        def __exit__(self, exc_type, exc_value, traceback):
>+            if self.stdout:
>+                self.stdout.close()
>+            if self.stderr:
>+                self.stderr.close()
>+            if self.stdin:
>+                self.stdin.close()
>+
>+    else:
>+        def __init__(self, args, bufsize=-1, executable=None,
>+                     stdin=None, stdout=None, stderr=None,
>+                     preexec_fn=None, close_fds=True,
>+                     shell=False, cwd=None, env=None, universal_newlines=None,
>+                     startupinfo=None, creationflags=0):
>+
>+            super(Popen, self).__init__(args, bufsize=bufsize, cwd=cwd, stdin=stdin, stdout=stdout, stderr=stderr)
>+
>+            for index in range(len(args)):
>+                if not isinstance(args[index], (str, unicode)):
>+                    raise TypeError('execv() arg {} must contain only strings'.format(index + 1))
>+
>+            self.text_mode = universal_newlines
>+
>+        def communicate(self, input=None):
>+            if self._communication_started and input:
>+                raise ValueError('Cannot send input after starting communication')
>+
>+            self._communication_started = True
>+            if input:
>+                self.stdin.write(input)
>+            self.wait()
>+            return self.stdout.read() if self.stdout else None, self.stderr.read() if self.stderr else None
>+
>+        def wait(self):
>+            if self.poll() is not None:
>+                return
>+
>+            # Need to check the timeout context
>+            timeout = Timeout.difference()
>+            if timeout and (self._completion.elapsed is None or timeout < self._completion.elapsed):
>+                raise TimeoutExpired(self._args, timeout)
>+
>+            if self._completion.elapsed is None:
>+                raise ValueError('Running a command that hangs without a timeout')
>+
>+            if self._completion.elapsed:
>+                time.sleep(self._completion.elapsed)
>+
>+            self.returncode = self._completion.returncode
>+            if self.stdout:
>+                self.stdout.seek(0)
>+            if self.stderr:
>+                self.stderr.seek(0)
>+
>+    @property
>+    def universal_newlines(self):
>+        return self.text_mode
>+
>+    @universal_newlines.setter
>+    def universal_newlines(self, universal_newlines):
>+        self.text_mode = bool(universal_newlines)
>+
>+    def poll(self):
>+        if not self._completion:
>+            self.stdin.seek(0)
>+            self._completion = Subprocess.completion_for(*self._args, cwd=self._cwd, input=self.stdin.read())
>+
>+            (self.stdout or sys.stdout).write(string_utils.decode(self._completion.stdout, target_type=self._stdout_type))
>+            (self.stdout or sys.stdout).flush()
>+
>+            (self.stderr or sys.stderr).write(string_utils.decode(self._completion.stderr, target_type=self._stderr_type))
>+            (self.stderr or sys.stderr).flush()
>+
>+        if self.returncode is not None and time.time() >= self._start_time + self._completion.elapsed:
>+            self.returncode = self._completion.returncode
>+            if self.stdout:
>+                self.stdout.seek(0)
>+            if self.stderr:
>+                self.stderr.seek(0)
>+
>+        return self.returncode
>+
>+    def send_signal(self, sig):
>+        if self.returncode is not None:
>+            return
>+
>+        if sig not in [signal.SIGTERM, signal.SIGKILL]:
>+            raise ValueError('Mock Popen object cannot handle signal {}'.format(sig))
>+        log.critical('Mock process {} send signal {}'.format(self.pid, sig))
>+        self.returncode = -1
>+
>+    def terminate(self):
>+        self.send_signal(signal.SIGTERM)
>+
>+    def kill(self):
>+        self.send_signal(signal.SIGKILL)
>Index: Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/subprocess.py
>===================================================================
>--- Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/subprocess.py	(nonexistent)
>+++ Tools/Scripts/libraries/webkitcorepy/webkitcorepy/mocks/subprocess.py	(working copy)
>@@ -0,0 +1,185 @@
>+# Copyright (C) 2020 Apple 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:
>+# 1.  Redistributions of source code must retain the above copyright
>+#    notice, this list of conditions and the following disclaimer.
>+# 2.  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.
>+#
>+# THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS 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 APPLE INC. OR ITS 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 re
>+import sys
>+
>+from functools import cmp_to_key
>+from webkitcorepy import string_utils, unicode
>+from webkitcorepy.mocks import ContextStack
>+
>+
>+class ProcessCompletion(object):
>+    def __init__(self, returncode=None, stdout=None, stderr=None, elapsed=0):
>+        self.returncode = 1 if returncode is None else returncode
>+        self.stdout = string_utils.encode(stdout) if stdout else b''
>+        self.stderr = string_utils.encode(stderr) if stderr else b''
>+        self.elapsed = elapsed
>+
>+
>+class Subprocess(ContextStack):
>+    """
>+    Organize ProcessCompletions so calls to subprocess functions will return a ProcessCompletion for
>+    a set of arguments or trigger a ProcessCompletion generator. mocks.Subprocess makes an attempt to
>+    prioritize Router objects for a given set of arguments such that the most specific applicable router
>+    is prefered.
>+
>+    Example usage mocking a single command:
>+        with mocks.Subprocess(
>+            'ls', completion=mocks.ProcessCompletion(returncode=0, stdout='file1.txt\nfile2.txt\n'),
>+        ):
>+            result = run(['ls'], capture_output=True, encoding='utf-8')
>+            assert result.returncode == 0
>+            assert result.stdout == 'file1.txt\nfile2.txt\n'
>+
>+    Example usage mocking a set of commands:
>+        with mocks.Subprocess(
>+            mocks.Subprocess.Router('command-a', 'argument', completion=mocks.ProcessCompletion(returncode=0)),
>+            mocks.Subprocess.Router('command-b', completion=mocks.ProcessCompletion(returncode=-1)),
>+        ):
>+            result = run(['command-a', 'argument'])
>+            assert result.returncode == 0
>+
>+            result = run(['command-b'])
>+            assert result.returncode == -1
>+    """
>+    top = None
>+
>+    class Router(object):
>+        def __init__(self, *args, **kwargs):
>+            completion = kwargs.pop('completion', ProcessCompletion())
>+            cwd = kwargs.pop('cwd', None)
>+            input = kwargs.pop('input', None)
>+            generator = kwargs.pop('generator', None)
>+            if kwargs.keys():
>+                raise TypeError('__init__() got an unexpected keyword argument {}'.format(kwargs.keys()[0]))
>+
>+            if isinstance(args, str) or isinstance(args, unicode):
>+                self.args = [args]
>+            elif not args:
>+                raise ValueError('Arguments must be provided to completion router')
>+            else:
>+                self.args = args
>+
>+            self.generator = generator or (lambda *args, **kwargs: completion)
>+            self.cwd = cwd
>+            self.input = string_utils.encode(input) if input else None
>+
>+        def matches(self, *args, **kwargs):
>+            cwd = kwargs.pop('cwd', None)
>+            input = kwargs.pop('input', None)
>+            if kwargs.keys():
>+                raise TypeError('matches() got an unexpected keyword argument {}'.format(kwargs.keys()[0]))
>+
>+            if len(self.args) > len(args):
>+                return False
>+
>+            for count in range(len(self.args)):
>+                if self.args[count] is None:
>+                    return False
>+
>+                if self.args[count] == args[count]:
>+                    continue
>+                elif hasattr(self.args[count], 'match') and self.args[count].match(args[count]):
>+                    continue
>+                elif re.match(self.args[count], args[count]):
>+                    continue
>+                return False
>+
>+            if self.cwd is not None and cwd != self.cwd:
>+                return False
>+            if self.input is not None and input != self.input:
>+                return False
>+            return True
>+
>+        def __call__(self, *args, **kwargs):
>+            cwd = kwargs.pop('cwd', None)
>+            input = kwargs.pop('input', None)
>+            if kwargs.keys():
>+                raise TypeError('__call__() got an unexpected keyword argument {}'.format(kwargs.keys()[0]))
>+            return self.generator(*args, cwd=cwd, input=input)
>+
>+        @classmethod
>+        def compare(cls, a, b):
>+            for candidate in [
>+                len(b.args) - len(a.args),
>+                0 if type(a.cwd) == type(b.cwd) else -1 if a.cwd else 1,
>+                0 if type(a.input) == type(b.input) else -1 if a.input else 1,
>+            ]:
>+                if candidate:
>+                    return candidate
>+            return 0
>+
>+    @classmethod
>+    def completion_generator_for(cls, program):
>+        current = cls.top
>+        candidates = []
>+        while current:
>+            for completion in current.completions:
>+                if completion.args[0] == program:
>+                    candidates.append(completion)
>+                if current.ordered:
>+                    break
>+            current = current.previous
>+
>+        if candidates:
>+            return candidates
>+
>+        if sys.version_info > (3, 0):
>+            raise FileNotFoundError("No such file or directory: '{path}': '{path}'".format(path=program))
>+        raise OSError('[Errno 2] No such file or directory')
>+
>+    @classmethod
>+    def completion_for(cls, *args, **kwargs):
>+        candidates = [
>+            candidate for candidate in cls.completion_generator_for(args[0]) if candidate.matches(*args, **kwargs)
>+        ]
>+        if not candidates:
>+            raise AssertionError('Provided arguments to {} do not match a provided completion'.format(args[0]))
>+
>+        completion = candidates[0]
>+        current = cls.top
>+        while current:
>+            if current.ordered and completion is current.completions[0]:
>+                current.completions.pop(0)
>+                break
>+            current = current.previous
>+        return completion(*args, **kwargs)
>+
>+    def __init__(self, *args, **kwargs):
>+        if all([isinstance(arg, Subprocess.Router) for arg in args]):
>+            self.ordered = kwargs.pop('ordered', False)
>+            if kwargs.keys():
>+                raise TypeError('__init__() got an unexpected keyword argument {}'.format(kwargs.keys()[0]))
>+            self.completions = list(args) if self.ordered else sorted(args, key=cmp_to_key(self.Router.compare))
>+        elif any([isinstance(arg, Subprocess.Router) for arg in args]):
>+            raise TypeError('mocks.Subprocess arguments must be of a consistent type')
>+        else:
>+            self.ordered = False
>+            self.completions = [Subprocess.Router(*args, **kwargs)]
>+
>+        super(Subprocess, self).__init__(cls=Subprocess)
>+
>+        # Allow mock to be managed via autoinstall
>+        from mock import patch
>+        from webkitcorepy.mocks.popen import Popen
>+        self.patches.append(patch('subprocess.Popen', new=Popen))
>Index: Tools/Scripts/libraries/webkitcorepy/webkitcorepy/tests/mocks/subprocess_unittest.py
>===================================================================
>--- Tools/Scripts/libraries/webkitcorepy/webkitcorepy/tests/mocks/subprocess_unittest.py	(nonexistent)
>+++ Tools/Scripts/libraries/webkitcorepy/webkitcorepy/tests/mocks/subprocess_unittest.py	(working copy)
>@@ -0,0 +1,155 @@
>+# Copyright (C) 2020 Apple 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:
>+# 1.  Redistributions of source code must retain the above copyright
>+#    notice, this list of conditions and the following disclaimer.
>+# 2.  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.
>+#
>+# THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS 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 APPLE INC. OR ITS 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 subprocess
>+import unittest
>+
>+from webkitcorepy import BytesIO, OutputCapture, mocks, run
>+
>+
>+class MockSubprocess(unittest.TestCase):
>+    LS = mocks.Subprocess.Router(
>+        'ls',
>+        completion=mocks.ProcessCompletion(returncode=0, stdout='file1.txt\nfile2.txt\n'),
>+    )
>+    SLEEP = mocks.Subprocess.Router(
>+        'sleep',
>+        generator=lambda *args, **kwargs: mocks.ProcessCompletion(returncode=0, elapsed=int(args[1])),
>+    )
>+
>+    def test_documentation(self):
>+        with OutputCapture():
>+            with mocks.Subprocess(
>+                    mocks.Subprocess.Router('command-a', 'argument', completion=mocks.ProcessCompletion(returncode=0)),
>+                    mocks.Subprocess.Router('command-b', completion=mocks.ProcessCompletion(returncode=-1)),
>+            ):
>+                result = run(['command-a', 'argument'])
>+                assert result.returncode == 0
>+
>+                result = run(['command-b'])
>+                assert result.returncode == -1
>+
>+            with mocks.Subprocess(
>+                    'ls', completion=mocks.ProcessCompletion(returncode=0, stdout='file1.txt\nfile2.txt\n'),
>+            ):
>+                assert subprocess.check_output(['ls']) == b'file1.txt\nfile2.txt\n'
>+                assert subprocess.check_call(['ls']) == 0
>+
>+            with mocks.Subprocess(
>+                    mocks.Subprocess.Router('command-a', 'argument', completion=mocks.ProcessCompletion(returncode=0)),
>+                    mocks.Subprocess.Router('command-b', completion=mocks.ProcessCompletion(returncode=-1)),
>+            ):
>+                result = run(['command-a', 'argument'])
>+                assert result.returncode == 0
>+
>+                result = run(['command-b'])
>+                assert result.returncode == -1
>+
>+    def test_no_file(self):
>+        with mocks.Subprocess(self.LS, self.SLEEP):
>+            with self.assertRaises(OSError):
>+                run(['invalid-file'])
>+
>+    def test_implied_router(self):
>+        with mocks.Subprocess('command', completion=mocks.ProcessCompletion(returncode=0)):
>+            self.assertEqual(run(['command']).returncode, 0)
>+
>+            with self.assertRaises(OSError):
>+                run(['invalid-file'])
>+
>+    def test_popen(self):
>+        with mocks.Time:
>+            with mocks.Subprocess(self.LS, self.SLEEP):
>+                ls = subprocess.Popen(['ls'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
>+                ls.wait()
>+                self.assertEqual(0, ls.poll())
>+                self.assertEqual(b'file1.txt\nfile2.txt\n', ls.stdout.read())
>+
>+                sleep = subprocess.Popen(['sleep', '1'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
>+                self.assertEqual(None, sleep.poll())
>+                sleep.wait()
>+                self.assertEqual(0, sleep.poll())
>+
>+    def test_ordered(self):
>+        with OutputCapture(), mocks.Subprocess(
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=0)),
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=1)),
>+            ordered=True,
>+        ):
>+            self.assertEqual(run(['command']).returncode, 0)
>+            self.assertEqual(run(['command']).returncode, 1)
>+            with self.assertRaises(OSError):
>+                run(['command'])
>+
>+    def test_argument_priority(self):
>+        with OutputCapture(), mocks.Subprocess(
>+            mocks.Subprocess.Router('command', '--help', completion=mocks.ProcessCompletion(returncode=0)),
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=1)),
>+        ):
>+            self.assertEqual(run(['command']).returncode, 1)
>+            self.assertEqual(run(['command', '--help']).returncode, 0)
>+
>+    def test_cwd_priority(self):
>+        with OutputCapture(), mocks.Subprocess(
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=0), cwd='/example'),
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=1)),
>+        ):
>+            self.assertEqual(run(['command']).returncode, 1)
>+            self.assertEqual(run(['command'], cwd='/example').returncode, 0)
>+
>+    def test_input_priority(self):
>+        with OutputCapture(), mocks.Subprocess(
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=0), input='stdin'),
>+            mocks.Subprocess.Router('command', completion=mocks.ProcessCompletion(returncode=1)),
>+        ):
>+            self.assertEqual(run(['command']).returncode, 1)
>+            self.assertEqual(run(['command'], input='stdin').returncode, 0)
>+            self.assertEqual(run(['command'], stdin=BytesIO(b'stdin')).returncode, 0)
>+
>+
>+class MockCheckOutput(unittest.TestCase):
>+    def test_popen(self):
>+        with mocks.Subprocess(MockSubprocess.LS):
>+            result = subprocess.check_output(['ls'])
>+            self.assertEqual(result, b'file1.txt\nfile2.txt\n')
>+
>+
>+class MockCheckCall(unittest.TestCase):
>+    def test_popen(self):
>+        with OutputCapture() as captured:
>+            with mocks.Subprocess(MockSubprocess.LS):
>+                result = subprocess.check_call(['ls'])
>+                self.assertEqual(result, 0)
>+
>+        self.assertEqual(captured.stdout.getvalue(), 'file1.txt\nfile2.txt\n')
>+
>+
>+class MockRun(unittest.TestCase):
>+    def test_popen(self):
>+        with OutputCapture() as captured:
>+            with mocks.Subprocess(MockSubprocess.LS):
>+                result = run(['ls'])
>+                self.assertEqual(result.returncode, 0)
>+                self.assertEqual(result.stdout, None)
>+                self.assertEqual(result.stderr, None)
>+
>+        self.assertEqual(captured.stdout.getvalue(), 'file1.txt\nfile2.txt\n')

Actions: View | Formatted Diff | Diff

Attachments on bug 215712: 406963 | 407245 | 407317 | 407328