import codecs
import io
+try:
+ from io import StringIO
+except ImportError:
+ from StringIO import StringIO
import pytest
import sys
from ceph_volume import terminal
+from ceph_volume.log import setup_console
class SubCommand(object):
def setup(self):
self.octpus_and_squid_en = u'octpus and squid'
- octpus_and_squid_zh = u'章鱼和鱿鱼'
- self.message = self.octpus_and_squid_en + octpus_and_squid_zh
+ self.octpus_and_squid_zh = u'章鱼和鱿鱼'
+ self.message = self.octpus_and_squid_en + self.octpus_and_squid_zh
+ setup_console()
def test_stdout_writer(self, capsys):
# should work with whatever stdout is
terminal.stdout(self.message)
_, err = capsys.readouterr()
assert self.octpus_and_squid_en in err
+ assert self.octpus_and_squid_zh in err
@pytest.mark.parametrize('encoding', ['ascii', 'utf8'])
- def test_writer(self, encoding, stream, monkeypatch, capsys):
- if encoding == 'ascii' and sys.version_info > (3,):
+ def test_writer_log(self, stream, encoding, monkeypatch, caplog):
+ writer = StringIO()
+ terminal._Write(_writer=writer).raw(self.message)
+ writer.flush()
+ writer.seek(0)
+ output = writer.readlines()[0]
+ assert self.octpus_and_squid_en in output
+
+ @pytest.mark.parametrize('encoding', ['utf8'])
+ def test_writer(self, encoding, stream, monkeypatch, capsys, caplog):
+ buffer = io.BytesIO()
+ writer = stream(buffer, encoding)
+ terminal._Write(_writer=writer).raw(self.message)
+ writer.flush()
+ writer.seek(0)
+ val = buffer.getvalue()
+ assert self.octpus_and_squid_en.encode(encoding) in val
+
+ def test_writer_uses_log_on_unicodeerror(self, stream, monkeypatch, capture):
+
+ if 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, 'stderr', writer)
- terminal.stdout(self.message)
- writer.flush()
- val = buffer.getvalue()
- assert self.octpus_and_squid_en.encode(encoding) in val
+ monkeypatch.setattr(terminal.terminal_logger, 'info', capture)
+ buffer = io.BytesIO()
+ writer = stream(buffer, 'ascii')
+ terminal._Write(_writer=writer).raw(self.message)
+ assert self.octpus_and_squid_en in capture.calls[0]['args'][0]