"""
def __init__(self, string):
- self.stdout = sys.__stdout__
self.appends = ''
self.prepends = ''
- self.isatty = self.stdout.isatty()
+ self.isatty = sys.__stderr__.isatty()
def _set_attributes(self):
"""
class _Write(object):
def __init__(self, _writer=None, prefix='', suffix='', flush=False):
- if _writer is None:
- _writer = sys.stdout
+ # we can't set sys.stderr as the default for _writer. Otherwise
+ # pytest's capturing gets confused
+ if not _writer:
+ _writer = sys.stderr
self._writer = _Write._unicode_output_stream(_writer)
- if _writer is sys.stdout:
- sys.stdout = self._writer
+ sys.stderr = self._writer
self.suffix = suffix
self.prefix = prefix
self.flush = flush
def _unicode_output_stream(stream):
# wrapper for given stream, so it can write unicode without throwing
# exception
- # sys.stdout.encoding is None if !isatty
+ # sys.stderr.encoding is None if !isatty
encoding = stream.encoding or ''
if encoding.upper() in ('UTF-8', 'UTF8'):
# already using unicode encoding, nothing to do
return stream.__class__(stream.buffer, encoding, 'replace',
stream.newlines, stream.line_buffering)
else:
- # in python2, stdout is but a "file"
+ # in python2, stderr is but a "file"
return codecs.getwriter(encoding)(stream, 'replace')
def bold(self, string):
def stderr(msg):
- return _Write(prefix=yellow(' stderr: '), _writer=sys.stderr).raw(msg)
+ return _Write(prefix=yellow(' stderr: ')).raw(msg)
def write(msg):
def error(msg):
- return _Write(prefix=red_arrow, _writer=sys.stderr).raw(msg)
+ return _Write(prefix=red_arrow).raw(msg)
def info(msg):
- return _Write(prefix=blue_arrow, _writer=sys.stderr).raw(msg)
+ return _Write(prefix=blue_arrow).raw(msg)
def debug(msg):
- return _Write(prefix=blue_arrow, _writer=sys.stderr).raw(msg)
+ return _Write(prefix=blue_arrow).raw(msg)
def warning(msg):
- return _Write(prefix=yellow_arrow, _writer=sys.stderr).raw(msg)
+ return _Write(prefix=yellow_arrow).raw(msg)
def success(msg):
"""
Look at every value of every key in the mapper and will output any
``class.help`` possible to return it as a string that will be sent to
- stdout.
+ stderr.
"""
help_text_lines = []
for key, value in mapper.items():
main.Volume(argv=['ceph-volume', '--cluster', 'barnacle', 'lvm', '--help'])
# make sure we aren't causing an actual error
assert error.value.code == 0
- log = caplog.records[1]
+ log = caplog.records[-1]
assert log.message == 'ignoring inability to load ceph.conf'
assert log.levelname == 'ERROR'
main.Volume(argv=['ceph-volume', '--cluster', 'barnacle', 'lvm', '--help'])
# make sure we aren't causing an actual error
assert error.value.code == 0
- log = caplog.records[0]
+ log = caplog.records[-2]
assert log.message == 'Running command: ceph-volume --cluster barnacle lvm --help'
assert log.levelname == 'INFO'
import pytest
+import logging
from ceph_volume.tests.conftest import Factory
from ceph_volume import process
class TestCall(object):
def test_stderr_terminal_and_logfile(self, mock_call, caplog, capsys):
+ caplog.set_level(logging.INFO)
mock_call(stdout='stdout\n', stderr='some stderr message\n')
process.call(['ls'], terminal_verbose=True)
out, err = capsys.readouterr()
assert 'some stderr message' in err
def test_stderr_terminal_and_logfile_off(self, mock_call, caplog, capsys):
+ caplog.set_level(logging.INFO)
mock_call(stdout='stdout\n', stderr='some stderr message\n')
process.call(['ls'], terminal_verbose=False)
out, err = capsys.readouterr()
assert out == ''
def test_verbose_on_failure(self, mock_call, caplog, capsys):
+ caplog.set_level(logging.INFO)
mock_call(stdout='stdout\n', stderr='stderr\n', returncode=1)
process.call(['ls'], terminal_verbose=False, logfile_verbose=False)
out, err = capsys.readouterr()
assert 'Running command: ' in log_lines
assert 'ls' in log_lines
assert 'stderr' in log_lines
- assert 'stdout: stdout' in out
+ assert 'stdout: stdout' in err
+ assert out == ''
def test_silent_verbose_on_failure(self, mock_call, caplog, capsys):
+ caplog.set_level(logging.INFO)
mock_call(stdout='stdout\n', stderr='stderr\n', returncode=1)
process.call(['ls'], verbose_on_failure=False)
out, err = capsys.readouterr()
def make_stream(buffer, encoding):
# mock a stdout with given encoding
if sys.version_info >= (3, 0):
- stdout = sys.stdout
+ stderr = sys.stderr
stream = io.TextIOWrapper(buffer,
encoding=encoding,
- errors=stdout.errors,
- newline=stdout.newlines,
- line_buffering=stdout.line_buffering)
+ errors=stderr.errors,
+ newline=stderr.newlines,
+ line_buffering=stderr.line_buffering)
else:
stream = codecs.getwriter(encoding)(buffer)
# StreamWriter does not have encoding attached to it, it will ask
def test_stdout_writer(self, capsys):
# should work with whatever stdout is
terminal.stdout(self.message)
- out, _ = capsys.readouterr()
- assert self.octpus_and_squid_en in out
+ _, err = capsys.readouterr()
+ assert self.octpus_and_squid_en in err
@pytest.mark.parametrize('encoding', ['ascii', 'utf8'])
def test_writer(self, encoding, stream, monkeypatch, capsys):
- buffer = io.BytesIO()
+ if encoding == 'ascii' and sys.version_info > (3,):
+ pytest.skip("Something breaks inside of pytest's capsys")
# should keep writer alive
with capsys.disabled():
+ buffer = io.BytesIO()
# we want to have access to the sys.stdout's attributes in
# make_stream(), not the ones of pytest.capture.EncodedFile
writer = stream(buffer, encoding)
- monkeypatch.setattr(sys, 'stdout', writer)
+ monkeypatch.setattr(sys, 'stderr', writer)
terminal.stdout(self.message)
- sys.stdout.flush()
- assert self.octpus_and_squid_en.encode(encoding) in buffer.getvalue()
+ writer.flush()
+ val = buffer.getvalue()
+ assert self.octpus_and_squid_en.encode(encoding) in val
monkeypatch.setattr(system.os.path, 'isfile', lambda x: True)
monkeypatch.setattr(system.os.path, 'exists', lambda x: False)
system.which('exedir')
- stdout, stderr = capsys.readouterr()
- assert 'Absolute path not found for executable: exedir' in stderr
- assert 'Ensure $PATH environment variable contains common executable locations' in stderr
+ cap = capsys.readouterr()
+ assert 'Absolute path not found for executable: exedir' in cap.err
+ assert 'Ensure $PATH environment variable contains common executable locations' in cap.err
projects / ceph.git / commitdiff