Python contextlib 模块，redirect_stderr() 实例源码

def test_write_to_file(self):
        f, path = tempfile.mkstemp(suffix='csv')
        self.json_response = {
            'total': 1,
            'issues': [test_data.singleSprintStory()]
        }
        lines = None
        try:
            with redirect_stderr(self.std_err):
                with redirect_stdout(self.std_out):
                    prog.main(['-w', 'blah', 'cycletime', '--no-progress', '-o', path, 'TEST'])
            with open(path, 'r') as o:
                lines = o.readlines()
        finally:
            os.unlink(path)
        self.assertEqual(2, len(lines))

def test_filter_by_date_argparse(self):
        '''The velocity commands date filter validates input argument.'''
        self.json_response = {
            'total': 1,
            'issues': [test_data.singleSprintStory()]
        }
        with redirect_stderr(self.std_err):
            with redirect_stdout(self.std_out):
                prog.main(['-w', 'blah', 'velocity', '--no-progress', '--filter-by-date', '11/01/2017', 'TEST'])

        with self.assertRaises(SystemExit) as ctx:
            with redirect_stderr(self.std_err):
                with redirect_stdout(self.std_out):
                    prog.main(['-w', 'blah', 'velocity', '--no-progress', '--filter-by-date', 'not a date', 'TEST'])

        exc = ctx.exception
        self.assertEqual(exc.code, 2)
        self.assertRegex_(self.std_err.getvalue(), r'velocity: error:')

def redirect_stderr(x):
    """ Redirects stderr to another file-like object.

        This is some compatibility code to support Python 3.4.
    """
    if hasattr(contextlib, 'redirect_stderr'):
        result = contextlib.redirect_stderr
    else:
        @contextlib.contextmanager
        def result(x):
            """ Stand-in for Python 3.5's `redirect_stderr`.

                Notes: Non-reentrant, non-threadsafe
            """
            old_stderr = sys.stderr
            sys.stderr = x
            yield
            sys.stder = old_stderr

    return result(x)

###############################################################################

def out(self, cmd, text=None):
        """Test if text is in output of command"""
        # for TRAVIS use maximal two cores
        if (self.njobs and '--njobs' not in cmd and
                cmd.split()[0] in ('correlate', 'stretch')):
            cmd = cmd + ' --njobs ' + self.njobs
        # disabling the logger is necessary, because the logging
        # configuration cannot be changed easily on subsequent calls
        # of yam in this test suite
        if self.verbose and cmd.split()[0] in ('correlate', 'stack', 'stretch'):
            if '-v' not in cmd:
                cmd = cmd + ' -vvv'
            logging.getLogger('yam').disabled = False
        elif self.verbose:
            logging.getLogger('yam').disabled = True
        if self.verbose:
            tqdm.tqdm.write('> yam ' + cmd)
        # catching all output, print only if tests are run with -v
        try:
            with io.StringIO() as f:
                with redirect_stdout(f), redirect_stderr(f):
                    try:
                        self.script(cmd.split())
                    except SystemExit:
                        pass
                output = f.getvalue()
            if self.verbose:
                tqdm.tqdm.write(output)
        finally:
            self.pbar.update(1)
        if text is not None:
            self.assertIn(text, output)
        return output

def test_command_without_stdout_ok(self):
        # TODO make it work under windows
        if os.name == 'nt':
            return

        f = io.StringIO()
        with redirect_stdout(f):
            launch_cmd_displays_output(self.cmd_ok, False, False)
        res = f.getvalue()
        self.assertEqual('.', res[:1])

        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            launch_cmd_displays_output(self.cmd_ok, False, False)
        res = f.getvalue()
        self.assertEqual('', res)

def test_command_with_stdout_ok(self):
        # TODO make it work under windows
        if os.name == 'nt':
            return

        f = io.StringIO()
        with redirect_stdout(f):
            launch_cmd_displays_output(self.cmd_ok, True, False)
        res = f.getvalue()
        self.assertEqual('coucou\n\n', res)

        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            launch_cmd_displays_output(self.cmd_ok, True, False)
        res = f.getvalue()
        self.assertEqual('', res)

def test_command_with_stderr_no_stdout_ok(self):
        # TODO make it work under windows
        if os.name == 'nt':
            return

        f = io.StringIO()
        with redirect_stdout(f):
            launch_cmd_displays_output(self.cmd_ok, False, True)
        res = f.getvalue()
        self.assertEqual('.', res[:1])

        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            launch_cmd_displays_output(self.cmd_ok, False, True)
        res = f.getvalue()
        self.assertEqual('', res)

def test_command_without_stderr_and_stdout_err(self):
        # TODO make it work under windows
        if os.name == 'nt':
            return

        f = io.StringIO()
        with redirect_stdout(f):
            launch_cmd_displays_output(self.cmd_nook, False, False)
        res = f.getvalue()
        self.assertEqual('\n', res)

        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            launch_cmd_displays_output(self.cmd_nook, False, False)
        res = f.getvalue()
        self.assertEqual('', res)

def test_command_without_stderr_but_stdout_err(self):
        # TODO make it work under windows
        if os.name == 'nt':
            return

        f = io.StringIO()
        with redirect_stdout(f):
            launch_cmd_displays_output(self.cmd_nook, True, False)
        res = f.getvalue()
        self.assertEqual('\n', res)

        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            launch_cmd_displays_output(self.cmd_nook, True, False)
        res = f.getvalue()
        self.assertEqual('', res)

def test_command_with_stderr_no_stdout_err_loop(self):
        # TODO make it work under windows
        if os.name == 'nt':
            return

        f = io.StringIO()
        with redirect_stdout(f):
            launch_cmd_displays_output(['wget', '--debug', '--tries', '3', 'http://doesnotexist'], False, True)
        res = f.getvalue()
        expected = re.compile('.*\.\.\. and more.*', re.MULTILINE)
        self.assertRegex(res, expected)

        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            launch_cmd_displays_output(self.cmd_nook, False, True)
        res = f.getvalue()
        self.assertEqual('', res)

def redirect_stderr(x):
    """ Redirects stderr to another file-like object.

        This is some compatibility code to support Python 3.4.
    """
    if hasattr(contextlib, 'redirect_stderr'):
        result = contextlib.redirect_stderr
    else:
        @contextlib.contextmanager
        def result(x):
            """ Stand-in for Python 3.5's `redirect_stderr`.

                Notes: Non-reentrant, non-threadsafe
            """
            old_stderr = sys.stderr
            sys.stderr = x
            yield
            sys.stder = old_stderr

    return result(x)

###############################################################################

def run_command_inline(argv, funct, *args, **kwargs):
    argv_save = sys.argv
    environ_save = EnvironmentSnapshot()
    captured_stdout = StringIO()
    captured_stderr = StringIO()
    sys.argv = argv
    exitcode = None
    with redirect_stdout(captured_stdout):
        with redirect_stderr(captured_stderr):
            try:
                exitcode = funct(*args, **kwargs)
            except SystemExit as e:
                exitcode = e.code
            finally:
                # restore environment, command-line arguments, and the native
                # modules system
                environ_save.load()
                sys.argv = argv_save
                fixtures.init_native_modules_system()

    return (exitcode,
            captured_stdout.getvalue(),
            captured_stderr.getvalue())

def test_stores_credentials(self):
        with redirect_stdout(self.std_out):
            with redirect_stderr(self.std_err):
                prog.main(['-w','blah','-u','usera', 'cycletime', 'IIQCB'])

        self.assertEqual('blah', keyring.get_keyring().entries['qjira-sp_usera'])

def test_not_authorized_clears_credentials(self):
        self.assertEqual('xyzzy', keyring.get_keyring().entries['qjira-sp_userb'])
        self.raise401 = True

        with self.assertRaises(HTTPError) as ctx:
            with redirect_stdout(self.std_out):
                with redirect_stderr(self.std_err):
                    prog.main(['-w','xyzzy','-u','userb', 'cycletime', 'IIQCB'])

        exc = ctx.exception
        self.assertEqual(exc.response.status_code, 401)

        with self.assertRaises(KeyError):
            keyring.get_keyring().entries['qjira-sp_userb']

def test_progress_shown(self):
        re_1of1 = re.compile('Retrieved 1 issue')
        self.json_response = {
            'total': 1,
            'issues': [test_data.singleSprintStory()]
        }
        with redirect_stdout(self.std_out):
            with redirect_stderr(self.std_err):
                prog.main([ '-w', 'blah','cycletime', 'TEST'])

        self.assertRegex_(self.std_err.getvalue(), re_1of1)

def test_progress_hidden(self):
        re_1of1 = re.compile('Retrieved 1 issue')
        self.json_response = {
            'total': 1,
            'issues': [test_data.singleSprintStory()]
        }
        with redirect_stderr(self.std_err):
            with redirect_stdout(self.std_out):
                prog.main(['-w', 'blah', 'cycletime', '--no-progress', 'TEST'])

        self.assertNotRegex_(self.std_err.getvalue(), re_1of1)

def test_command_jql_require_jql(self):
        with self.assertRaises(SystemExit) as ctx:
            with redirect_stderr(self.std_err):
                prog.main([ '-w', 'blah', 'jql', '--no-progress'])
        exc = ctx.exception
        self.assertEqual(exc.code, 2)
        self.assertRegex_(self.std_err.getvalue(), r'jql: error:')

def test_error(self):
        with redirect_stderr(self.std_err):
            Log.error('hello')
        output = self.std_err.getvalue()
        self.assertEqual('[ERROR] hello\n', output)

def test_info(self):
        Log.debugLevel = 1
        with redirect_stderr(self.std_err):
            Log.info('hello')
        self.assertEqual('[INFO] hello\n', self.std_err.getvalue())

def test_debug(self):
        Log.debugLevel = 2
        with redirect_stderr(self.std_err):
            Log.debug('hello')
        self.assertEqual('[DEBUG] hello\n', self.std_err.getvalue())

def basic_test(self, test_name):
        with self.recorder.use_cassette('Test{}.system'.format(test_name)):
            browser = load_yaml_config(
                self.db, self.subreddit,
                Path('./test/configs/{}Config.yaml'.format(test_name)))
            temp_stderr = StringIO()
            with redirect_stderr(temp_stderr):
                browser.run()
            assert temp_stderr.getvalue() == ""
            assert all(
                interaction.used
                for interaction in self.recorder.current_cassette.interactions)
        rows = self.db.execute('SELECT * FROM actions').fetchall()
        assert len(rows) == 1
        return browser

def test_log_exception(self):
        # Traceback and exception should be written to stderr.
        exc_type = NotImplementedError
        exc_message = 'hello'
        try:
            raise exc_type(exc_message)
        except Exception as caught:
            exc = caught
        stderr = io.StringIO()
        with contextlib.redirect_stderr(stderr):
            self.server.log_exception(exc)
        logged = stderr.getvalue()
        self.assertIn(exc_type.__name__, logged)
        self.assertIn(exc_message, logged)
        self.assertIn('Traceback', logged)

def test_log(self):
        message = "something happened"
        stderr = io.StringIO()
        with contextlib.redirect_stderr(stderr):
            self.server.log(message)
        self.assertEqual(stderr.getvalue(), message + "\n")

def test_bench_equality():
    """[Model Selection] Test benchmark correspondence with eval."""

    with open(os.devnull, 'w') as f, redirect_stderr(f):
        evl = Evaluator(mape_scorer, cv=5)
        evl.fit(X, y, estimators={'pr': [OLS()], 'no': [OLS()]},
                param_dicts={}, preprocessing={'pr': [Scale()], 'no': []})

        out = benchmark(X, y, mape_scorer, 5, {'pr': [OLS()], 'no': [OLS()]},
                        {'pr': [Scale()], 'no': []}, None)

    np.testing.assert_approx_equal(out['test_score-m']['no.ols'],
                                   evl.results['test_score-m']['no.ols'])

def run(cls, kls, proba, preprocessing, **kwargs):
    """Function for executing specified test."""
    model_selection = kwargs.pop('model_selection', None)
    if kls == 'subsemble':
        p = kwargs['partitions']
    else:
        p = 1

    ests = ESTS[(proba, preprocessing)]
    prep = PREPROCESSING if preprocessing else None

    data = Data(kls, proba, preprocessing, **kwargs)

    X, y = data.get_data((LEN, WIDTH), MOD)
    (F, wf), _ = data.ground_truth(X, y, p)

    with open(os.devnull, 'w') as f, redirect_stderr(f):
        ens = cls()
        ens.add(kls, ests, prep, proba=proba, dtype=np.float64, **kwargs)

        if model_selection:
            ens.model_selection = True

        ens.fit(X, y)

        pred, _ = ens.transform(X, y)

    np.testing.assert_array_equal(F, pred)

def test_load():
    """[Utils] Check that load handles exceptions gracefully"""

    config.set_ivals(0.1, 0.1)

    with open(os.devnull, 'w') as f, redirect_stderr(f):
        np.testing.assert_raises(
            ParallelProcessingError,
            utils.load, os.path.join(os.getcwd(), 'nonexist'))

def test_check_cache():
    """[Base] Test check cache."""
    tmp = config.get_prefix() + "test"
    os.mkdir(tmp)
    with open(os.devnull, 'w') as f, redirect_stderr(f):
        subprocess.Popen("echo this is a test >> " + tmp +
                         "/test.txt", shell=True)
        config.clear_cache(config.get_tmpdir())

def test_invalid_config(self):
        """Test an existing configuration file but invalid"""
        c = Config(base_dir + '/static/config_invalid.ini')
        self.assertFalse(c.read())
        self.assertGreater(len(c.errors), 0)
        self.assertTrue('project_name' in c.errors)
        self.assertEqual('Missing', c.errors['project_name'])
        self.assertTrue('php.version' in c.errors)
        self.assertEqual('the value "8.0" is unacceptable.', c.errors['php.version'])

        # Don't go further with python < 3.5
        try:
            from contextlib import redirect_stderr
        except Exception:
            return

        f = io.StringIO()
        with redirect_stderr(f):
            c.display_errors()
        res = f.getvalue()

        regex = re.compile('Failed validating .*config_invalid.ini', re.MULTILINE)
        self.assertRegex(res, regex)

        regex = re.compile('the value ".*8.0.*" is unacceptable', re.MULTILINE)
        self.assertRegex(res, regex)

def push(self, command):
        """Push some code."""
        s = StringIO()
        with redirect_stdout(s), redirect_stderr(s):
            if super(Shell, self).push(command):
                self.player.notify('...')
        s.seek(0)
        self.player.notify(s.read())

def rst2dtree(rst):

    err = io.StringIO()
    with contextlib.redirect_stderr(err):
        try:
            dtree = docutils.core.publish_doctree(rst)
        except docutils.utils.SystemMessage as e:
            dtree = None
            print("parsing failed", e)
    return err.getvalue(), dtree