ryu/.eggs/pbr-5.3.1-py2.7.egg/pbr/tests/base.py

   1 # Copyright 2010-2011 OpenStack Foundation
   2 # Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 # Copyright (C) 2013 Association of Universities for Research in Astronomy
  16 #                    (AURA)
  17 #
  18 # Redistribution and use in source and binary forms, with or without
  19 # modification, are permitted provided that the following conditions are met:
  20 #
  21 #     1. Redistributions of source code must retain the above copyright
  22 #        notice, this list of conditions and the following disclaimer.
  23 #
  24 #     2. Redistributions in binary form must reproduce the above
  25 #        copyright notice, this list of conditions and the following
  26 #        disclaimer in the documentation and/or other materials provided
  27 #        with the distribution.
  28 #
  29 #     3. The name of AURA and its representatives may not be used to
  30 #        endorse or promote products derived from this software without
  31 #        specific prior written permission.
  32 #
  33 # THIS SOFTWARE IS PROVIDED BY AURA ``AS IS'' AND ANY EXPRESS OR IMPLIED
  34 # WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  35 # MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  36 # DISCLAIMED. IN NO EVENT SHALL AURA BE LIABLE FOR ANY DIRECT, INDIRECT,
  37 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
  38 # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS
  39
  40 """Common utilities used in testing"""
  41
  42 import os
  43 import shutil
  44 import subprocess
  45 import sys
  46
  47 import fixtures
  48 import testresources
  49 import testtools
  50 from testtools import content
  51
  52 from pbr import options
  53
  54
  55 class DiveDir(fixtures.Fixture):
  56     """Dive into given directory and return back on cleanup.
  57
  58     :ivar path: The target directory.
  59     """
  60
  61     def __init__(self, path):
  62         self.path = path
  63
  64     def setUp(self):
  65         super(DiveDir, self).setUp()
  66         self.addCleanup(os.chdir, os.getcwd())
  67         os.chdir(self.path)
  68
  69
  70 class BaseTestCase(testtools.TestCase, testresources.ResourcedTestCase):
  71
  72     def setUp(self):
  73         super(BaseTestCase, self).setUp()
  74         test_timeout = os.environ.get('OS_TEST_TIMEOUT', 30)
  75         try:
  76             test_timeout = int(test_timeout)
  77         except ValueError:
  78             # If timeout value is invalid, fail hard.
  79             print("OS_TEST_TIMEOUT set to invalid value"
  80                   " defaulting to no timeout")
  81             test_timeout = 0
  82         if test_timeout > 0:
  83             self.useFixture(fixtures.Timeout(test_timeout, gentle=True))
  84
  85         if os.environ.get('OS_STDOUT_CAPTURE') in options.TRUE_VALUES:
  86             stdout = self.useFixture(fixtures.StringStream('stdout')).stream
  87             self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
  88         if os.environ.get('OS_STDERR_CAPTURE') in options.TRUE_VALUES:
  89             stderr = self.useFixture(fixtures.StringStream('stderr')).stream
  90             self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
  91         self.log_fixture = self.useFixture(
  92             fixtures.FakeLogger('pbr'))
  93
  94         # Older git does not have config --local, so create a temporary home
  95         # directory to permit using git config --global without stepping on
  96         # developer configuration.
  97         self.useFixture(fixtures.TempHomeDir())
  98         self.useFixture(fixtures.NestedTempfile())
  99         self.useFixture(fixtures.FakeLogger())
 100         # TODO(lifeless) we should remove PBR_VERSION from the environment.
 101         # rather than setting it, because thats not representative - we need to
 102         # test non-preversioned codepaths too!
 103         self.useFixture(fixtures.EnvironmentVariable('PBR_VERSION', '0.0'))
 104
 105         self.temp_dir = self.useFixture(fixtures.TempDir()).path
 106         self.package_dir = os.path.join(self.temp_dir, 'testpackage')
 107         shutil.copytree(os.path.join(os.path.dirname(__file__), 'testpackage'),
 108                         self.package_dir)
 109         self.addCleanup(os.chdir, os.getcwd())
 110         os.chdir(self.package_dir)
 111         self.addCleanup(self._discard_testpackage)
 112         # Tests can opt into non-PBR_VERSION by setting preversioned=False as
 113         # an attribute.
 114         if not getattr(self, 'preversioned', True):
 115             self.useFixture(fixtures.EnvironmentVariable('PBR_VERSION'))
 116             setup_cfg_path = os.path.join(self.package_dir, 'setup.cfg')
 117             with open(setup_cfg_path, 'rt') as cfg:
 118                 content = cfg.read()
 119             content = content.replace(u'version = 0.1.dev', u'')
 120             with open(setup_cfg_path, 'wt') as cfg:
 121                 cfg.write(content)
 122
 123     def _discard_testpackage(self):
 124         # Remove pbr.testpackage from sys.modules so that it can be freshly
 125         # re-imported by the next test
 126         for k in list(sys.modules):
 127             if (k == 'pbr_testpackage' or
 128                     k.startswith('pbr_testpackage.')):
 129                 del sys.modules[k]
 130
 131     def run_pbr(self, *args, **kwargs):
 132         return self._run_cmd('pbr', args, **kwargs)
 133
 134     def run_setup(self, *args, **kwargs):
 135         return self._run_cmd(sys.executable, ('setup.py',) + args, **kwargs)
 136
 137     def _run_cmd(self, cmd, args=[], allow_fail=True, cwd=None):
 138         """Run a command in the root of the test working copy.
 139
 140         Runs a command, with the given argument list, in the root of the test
 141         working copy--returns the stdout and stderr streams and the exit code
 142         from the subprocess.
 143
 144         :param cwd: If falsy run within the test package dir, otherwise run
 145             within the named path.
 146         """
 147         cwd = cwd or self.package_dir
 148         result = _run_cmd([cmd] + list(args), cwd=cwd)
 149         if result[2] and not allow_fail:
 150             raise Exception("Command failed retcode=%s" % result[2])
 151         return result
 152
 153
 154 class CapturedSubprocess(fixtures.Fixture):
 155     """Run a process and capture its output.
 156
 157     :attr stdout: The output (a string).
 158     :attr stderr: The standard error (a string).
 159     :attr returncode: The return code of the process.
 160
 161     Note that stdout and stderr are decoded from the bytestrings subprocess
 162     returns using error=replace.
 163     """
 164
 165     def __init__(self, label, *args, **kwargs):
 166         """Create a CapturedSubprocess.
 167
 168         :param label: A label for the subprocess in the test log. E.g. 'foo'.
 169         :param *args: The *args to pass to Popen.
 170         :param **kwargs: The **kwargs to pass to Popen.
 171         """
 172         super(CapturedSubprocess, self).__init__()
 173         self.label = label
 174         self.args = args
 175         self.kwargs = kwargs
 176         self.kwargs['stderr'] = subprocess.PIPE
 177         self.kwargs['stdin'] = subprocess.PIPE
 178         self.kwargs['stdout'] = subprocess.PIPE
 179
 180     def setUp(self):
 181         super(CapturedSubprocess, self).setUp()
 182         proc = subprocess.Popen(*self.args, **self.kwargs)
 183         out, err = proc.communicate()
 184         self.out = out.decode('utf-8', 'replace')
 185         self.err = err.decode('utf-8', 'replace')
 186         self.addDetail(self.label + '-stdout', content.text_content(self.out))
 187         self.addDetail(self.label + '-stderr', content.text_content(self.err))
 188         self.returncode = proc.returncode
 189         if proc.returncode:
 190             raise AssertionError('Failed process %s' % proc.returncode)
 191         self.addCleanup(delattr, self, 'out')
 192         self.addCleanup(delattr, self, 'err')
 193         self.addCleanup(delattr, self, 'returncode')
 194
 195
 196 def _run_cmd(args, cwd):
 197     """Run the command args in cwd.
 198
 199     :param args: The command to run e.g. ['git', 'status']
 200     :param cwd: The directory to run the comamnd in.
 201     :return: ((stdout, stderr), returncode)
 202     """
 203     p = subprocess.Popen(
 204         args, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 205         stderr=subprocess.PIPE, cwd=cwd)
 206     streams = tuple(s.decode('latin1').strip() for s in p.communicate())
 207     for stream_content in streams:
 208         print(stream_content)
 209     return (streams) + (p.returncode,)
 210
 211
 212 def _config_git():
 213     _run_cmd(
 214         ['git', 'config', '--global', 'user.email', 'example@example.com'],
 215         None)
 216     _run_cmd(
 217         ['git', 'config', '--global', 'user.name', 'OpenStack Developer'],
 218         None)
 219     _run_cmd(
 220         ['git', 'config', '--global', 'user.signingkey',
 221          'example@example.com'], None)