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))
python类redirect_stderr()的实例源码
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
