]>
Commit | Line | Data |
---|---|---|
494da23a TL |
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 | |
d2e6a577 | 9 | import pytest |
494da23a | 10 | import sys |
d2e6a577 | 11 | from ceph_volume import terminal |
494da23a | 12 | from ceph_volume.log import setup_console |
d2e6a577 FG |
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' | |
494da23a TL |
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] |