1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
# -*- mode:python; tab-width:4; indent-tabs-mode:nil; coding:utf-8 -*-
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):
help = "this is the subcommand help"
def __init__(self, argv):
self.argv = argv
def main(self):
pass
class BadSubCommand(object):
def __init__(self, argv):
self.argv = argv
def main(self):
raise SystemExit(100)
class TestSubhelp(object):
def test_no_sub_command_help(self):
assert terminal.subhelp({}) == ''
def test_single_level_help(self):
result = terminal.subhelp({'sub': SubCommand})
assert 'this is the subcommand help' in result
def test_has_title_header(self):
result = terminal.subhelp({'sub': SubCommand})
assert 'Available subcommands:' in result
def test_command_with_no_help(self):
class SubCommandNoHelp(object):
pass
result = terminal.subhelp({'sub': SubCommandNoHelp})
assert result == ''
class TestDispatch(object):
def test_no_subcommand_found(self):
result = terminal.dispatch({'sub': SubCommand}, argv=[])
assert result is None
def test_no_main_found(self):
class NoMain(object):
def __init__(self, argv):
pass
result = terminal.dispatch({'sub': NoMain}, argv=['sub'])
assert result is None
def test_subcommand_found_and_dispatched(self):
with pytest.raises(SystemExit) as error:
terminal.dispatch({'sub': SubCommand}, argv=['sub'])
assert str(error.value) == '0'
def test_subcommand_found_and_dispatched_with_errors(self):
with pytest.raises(SystemExit) as error:
terminal.dispatch({'sub': BadSubCommand}, argv=['sub'])
assert str(error.value) == '100'
@pytest.fixture
def stream():
def make_stream(buffer, encoding):
# mock a stdout with given encoding
if sys.version_info >= (3, 0):
stderr = sys.stderr
stream = io.TextIOWrapper(buffer,
encoding=encoding,
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
# the inner buffer for "encoding" attribute in this case
stream.encoding = encoding
return stream
return make_stream
class TestWriteUnicode(object):
def setup_method(self):
self.octpus_and_squid_en = u'octpus and squid'
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_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")
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]
|
