ceph/src/ceph-volume/ceph_volume/tests/test_terminal.py

   1 # -*- mode:python; tab-width:4; indent-tabs-mode:nil; coding:utf-8 -*-
   2
   3 import codecs
   4 import io
   5 try:
   6     from io import StringIO
   7 except ImportError:
   8     from StringIO import StringIO
   9 import pytest
  10 import sys
  11 from ceph_volume import terminal
  12 from ceph_volume.log import setup_console
  13
  14
  15 class SubCommand(object):
  16
  17     help = "this is the subcommand help"
  18
  19     def __init__(self, argv):
  20         self.argv = argv
  21
  22     def main(self):
  23         pass
  24
  25
  26 class BadSubCommand(object):
  27
  28     def __init__(self, argv):
  29         self.argv = argv
  30
  31     def main(self):
  32         raise SystemExit(100)
  33
  34
  35 class TestSubhelp(object):
  36
  37     def test_no_sub_command_help(self):
  38         assert terminal.subhelp({}) == ''
  39
  40     def test_single_level_help(self):
  41         result = terminal.subhelp({'sub': SubCommand})
  42
  43         assert 'this is the subcommand help' in result
  44
  45     def test_has_title_header(self):
  46         result = terminal.subhelp({'sub': SubCommand})
  47         assert 'Available subcommands:' in result
  48
  49     def test_command_with_no_help(self):
  50         class SubCommandNoHelp(object):
  51             pass
  52         result = terminal.subhelp({'sub': SubCommandNoHelp})
  53         assert result == ''
  54
  55
  56 class TestDispatch(object):
  57
  58     def test_no_subcommand_found(self):
  59         result = terminal.dispatch({'sub': SubCommand}, argv=[])
  60         assert result is None
  61
  62     def test_no_main_found(self):
  63         class NoMain(object):
  64
  65             def __init__(self, argv):
  66                 pass
  67         result = terminal.dispatch({'sub': NoMain}, argv=['sub'])
  68         assert result is None
  69
  70     def test_subcommand_found_and_dispatched(self):
  71         with pytest.raises(SystemExit) as error:
  72             terminal.dispatch({'sub': SubCommand}, argv=['sub'])
  73         assert str(error.value) == '0'
  74
  75     def test_subcommand_found_and_dispatched_with_errors(self):
  76         with pytest.raises(SystemExit) as error:
  77             terminal.dispatch({'sub': BadSubCommand}, argv=['sub'])
  78         assert str(error.value) == '100'
  79
  80
  81 @pytest.fixture
  82 def stream():
  83     def make_stream(buffer, encoding):
  84         # mock a stdout with given encoding
  85         if sys.version_info >= (3, 0):
  86             stderr = sys.stderr
  87             stream = io.TextIOWrapper(buffer,
  88                                       encoding=encoding,
  89                                       errors=stderr.errors,
  90                                       newline=stderr.newlines,
  91                                       line_buffering=stderr.line_buffering)
  92         else:
  93             stream = codecs.getwriter(encoding)(buffer)
  94             # StreamWriter does not have encoding attached to it, it will ask
  95             # the inner buffer for "encoding" attribute in this case
  96             stream.encoding = encoding
  97         return stream
  98     return make_stream
  99
 100
 101 class TestWriteUnicode(object):
 102
 103     def setup(self):
 104         self.octpus_and_squid_en = u'octpus and squid'
 105         self.octpus_and_squid_zh = u'章鱼和鱿鱼'
 106         self.message = self.octpus_and_squid_en + self.octpus_and_squid_zh
 107         setup_console()
 108
 109     def test_stdout_writer(self, capsys):
 110         # should work with whatever stdout is
 111         terminal.stdout(self.message)
 112         _, err = capsys.readouterr()
 113         assert self.octpus_and_squid_en in err
 114         assert self.octpus_and_squid_zh in err
 115
 116     @pytest.mark.parametrize('encoding', ['ascii', 'utf8'])
 117     def test_writer_log(self, stream, encoding, monkeypatch, caplog):
 118         writer = StringIO()
 119         terminal._Write(_writer=writer).raw(self.message)
 120         writer.flush()
 121         writer.seek(0)
 122         output = writer.readlines()[0]
 123         assert self.octpus_and_squid_en in output
 124
 125     @pytest.mark.parametrize('encoding', ['utf8'])
 126     def test_writer(self, encoding, stream, monkeypatch, capsys, caplog):
 127         buffer = io.BytesIO()
 128         writer = stream(buffer, encoding)
 129         terminal._Write(_writer=writer).raw(self.message)
 130         writer.flush()
 131         writer.seek(0)
 132         val = buffer.getvalue()
 133         assert self.octpus_and_squid_en.encode(encoding) in val
 134
 135     def test_writer_uses_log_on_unicodeerror(self, stream, monkeypatch, capture):
 136
 137         if sys.version_info > (3,):
 138             pytest.skip("Something breaks inside of pytest's capsys")
 139         monkeypatch.setattr(terminal.terminal_logger, 'info', capture)
 140         buffer = io.BytesIO()
 141         writer = stream(buffer, 'ascii')
 142         terminal._Write(_writer=writer).raw(self.message)
 143         assert self.octpus_and_squid_en in capture.calls[0]['args'][0]