"Fossies" - the Fresh Open Source Software Archive

Member "buildbot-2.3.1/buildbot/test/util/gpo.py" (23 May 2019, 4110 Bytes) of package /linux/misc/buildbot-2.3.1.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "gpo.py": 2.0.1_vs_2.1.0.

    1 # This file is part of Buildbot.  Buildbot is free software: you can
    2 # redistribute it and/or modify it under the terms of the GNU General Public
    3 # License as published by the Free Software Foundation, version 2.
    4 #
    5 # This program is distributed in the hope that it will be useful, but WITHOUT
    6 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
    7 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
    8 # details.
    9 #
   10 # You should have received a copy of the GNU General Public License along with
   11 # this program; if not, write to the Free Software Foundation, Inc., 51
   12 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
   13 #
   14 # Copyright Buildbot Team Members
   15 
   16 from twisted.internet import defer
   17 from twisted.internet import utils
   18 
   19 
   20 def _check_env_is_expected(test, expected_env, env):
   21     if expected_env is None:
   22         return
   23 
   24     env = env or {}
   25     for var, value in expected_env.items():
   26         test.assertEqual(env.get(var), value,
   27                          'Expected environment to have %s = %r' % (var, value))
   28 
   29 
   30 class Expect:
   31     _stdout = b""
   32     _stderr = b""
   33     _exit = 0
   34     _path = None
   35     _env = None
   36 
   37     def __init__(self, bin, *args):
   38         self._bin = bin
   39         self._args = args
   40 
   41     def stdout(self, stdout):
   42         assert(isinstance(stdout, bytes))
   43         self._stdout = stdout
   44         return self
   45 
   46     def stderr(self, stderr):
   47         assert(isinstance(stderr, bytes))
   48         self._stderr = stderr
   49         return self
   50 
   51     def exit(self, exit):
   52         self._exit = exit
   53         return self
   54 
   55     def path(self, path):
   56         self._path = path
   57         return self
   58 
   59     def env(self, env):
   60         self._env = env
   61         return self
   62 
   63     def check(self, test, bin, path, args, env):
   64         test.assertDictEqual(
   65             dict(bin=bin, path=path, args=tuple(args)),
   66             dict(bin=self._bin, path=self._path, args=self._args), "unexpected command run")
   67 
   68         _check_env_is_expected(test, self._env, env)
   69         return (self._stdout, self._stderr, self._exit)
   70 
   71     def __repr__(self):
   72         return "<gpo.Expect(bin=%s, args=%s)>" % (self._bin, self._args)
   73 
   74 
   75 class GetProcessOutputMixin:
   76     longMessage = True
   77 
   78     def setUpGetProcessOutput(self):
   79         self._gpo_patched = False
   80         self._expected_commands = []
   81         self._gpo_expect_env = {}
   82 
   83     def assertAllCommandsRan(self):
   84         self.assertEqual(self._expected_commands, [],
   85                          "assert all expected commands were run")
   86 
   87     @defer.inlineCallbacks
   88     def patched_getProcessOutput(self, bin, args, env=None,
   89                                  errortoo=False, path=None):
   90         stdout, stderr, exit = \
   91             yield self.patched_getProcessOutputAndValue(bin, args, env=env,
   92                                                         path=path)
   93         if errortoo:
   94             return stdout + stderr
   95         if stderr:
   96             raise IOError("got stderr: %r" % (stderr,))
   97         return stdout
   98 
   99     def patched_getProcessOutputAndValue(self, bin, args, env=None,
  100                                          path=None):
  101         _check_env_is_expected(self, self._gpo_expect_env, env)
  102 
  103         if not self._expected_commands:
  104             self.fail("got command %s %s when no further commands were expected"
  105                       % (bin, args))
  106 
  107         expect = self._expected_commands.pop(0)
  108         return defer.succeed(expect.check(self, bin, path, args, env))
  109 
  110     def _patch_gpo(self):
  111         if not self._gpo_patched:
  112             self.patch(utils, "getProcessOutput",
  113                        self.patched_getProcessOutput)
  114             self.patch(utils, "getProcessOutputAndValue",
  115                        self.patched_getProcessOutputAndValue)
  116             self._gpo_patched = True
  117 
  118     def addGetProcessOutputExpectEnv(self, d):
  119         self._gpo_expect_env.update(d)
  120 
  121     def expectCommands(self, *exp):
  122         """
  123         Add to the expected commands, along with their results.  Each
  124         argument should be an instance of L{Expect}.
  125         """
  126         self._patch_gpo()
  127         self._expected_commands.extend(exp)