]>
Commit | Line | Data |
---|---|---|
d2e6a577 FG |
1 | import os |
2 | try: | |
3 | from cStringIO import StringIO | |
4 | except ImportError: # pragma: no cover | |
5 | from io import StringIO # pragma: no cover | |
6 | from textwrap import dedent | |
7 | import pytest | |
8 | from ceph_volume import configuration, exceptions | |
9 | ||
b32b8144 FG |
10 | tabbed_conf = """ |
11 | [global] | |
12 | default = 0 | |
13 | other_h = 1 # comment | |
14 | other_c = 1 ; comment | |
15 | colon = ; | |
16 | hash = # | |
17 | """ | |
18 | ||
d2e6a577 FG |
19 | |
20 | class TestConf(object): | |
21 | ||
22 | def setup(self): | |
23 | self.conf_file = StringIO(dedent(""" | |
24 | [foo] | |
25 | default = 0 | |
26 | """)) | |
27 | ||
28 | def test_get_non_existing_list(self): | |
29 | cfg = configuration.Conf() | |
30 | cfg.is_valid = lambda: True | |
eafe8130 | 31 | cfg.read_conf(self.conf_file) |
d2e6a577 FG |
32 | assert cfg.get_list('global', 'key') == [] |
33 | ||
34 | def test_get_non_existing_list_get_default(self): | |
35 | cfg = configuration.Conf() | |
36 | cfg.is_valid = lambda: True | |
eafe8130 | 37 | cfg.read_conf(self.conf_file) |
d2e6a577 FG |
38 | assert cfg.get_list('global', 'key', ['a']) == ['a'] |
39 | ||
40 | def test_get_rid_of_comments(self): | |
41 | cfg = configuration.Conf() | |
42 | cfg.is_valid = lambda: True | |
43 | conf_file = StringIO(dedent(""" | |
44 | [foo] | |
45 | default = 0 # this is a comment | |
46 | """)) | |
47 | ||
eafe8130 | 48 | cfg.read_conf(conf_file) |
d2e6a577 FG |
49 | assert cfg.get_list('foo', 'default') == ['0'] |
50 | ||
51 | def test_gets_split_on_commas(self): | |
52 | cfg = configuration.Conf() | |
53 | cfg.is_valid = lambda: True | |
54 | conf_file = StringIO(dedent(""" | |
55 | [foo] | |
56 | default = 0,1,2,3 # this is a comment | |
57 | """)) | |
58 | ||
eafe8130 | 59 | cfg.read_conf(conf_file) |
d2e6a577 FG |
60 | assert cfg.get_list('foo', 'default') == ['0', '1', '2', '3'] |
61 | ||
62 | def test_spaces_and_tabs_are_ignored(self): | |
63 | cfg = configuration.Conf() | |
64 | cfg.is_valid = lambda: True | |
65 | conf_file = StringIO(dedent(""" | |
66 | [foo] | |
67 | default = 0, 1, 2 ,3 # this is a comment | |
68 | """)) | |
69 | ||
eafe8130 | 70 | cfg.read_conf(conf_file) |
d2e6a577 FG |
71 | assert cfg.get_list('foo', 'default') == ['0', '1', '2', '3'] |
72 | ||
73 | ||
74 | class TestLoad(object): | |
75 | ||
b32b8144 FG |
76 | def test_load_from_path(self, tmpdir): |
77 | conf_path = os.path.join(str(tmpdir), 'ceph.conf') | |
78 | with open(conf_path, 'w') as conf: | |
79 | conf.write(tabbed_conf) | |
80 | result = configuration.load(conf_path) | |
81 | assert result.get('global', 'default') == '0' | |
82 | ||
83 | def test_load_with_colon_comments(self, tmpdir): | |
84 | conf_path = os.path.join(str(tmpdir), 'ceph.conf') | |
85 | with open(conf_path, 'w') as conf: | |
86 | conf.write(tabbed_conf) | |
87 | result = configuration.load(conf_path) | |
88 | assert result.get('global', 'other_c') == '1' | |
89 | ||
90 | def test_load_with_hash_comments(self, tmpdir): | |
91 | conf_path = os.path.join(str(tmpdir), 'ceph.conf') | |
92 | with open(conf_path, 'w') as conf: | |
93 | conf.write(tabbed_conf) | |
94 | result = configuration.load(conf_path) | |
95 | assert result.get('global', 'other_h') == '1' | |
96 | ||
d2e6a577 FG |
97 | def test_path_does_not_exist(self): |
98 | with pytest.raises(exceptions.ConfigurationError): | |
99 | conf = configuration.load('/path/does/not/exist/ceph.con') | |
100 | conf.is_valid() | |
101 | ||
102 | def test_unable_to_read_configuration(self, tmpdir, capsys): | |
103 | ceph_conf = os.path.join(str(tmpdir), 'ceph.conf') | |
104 | with open(ceph_conf, 'w') as config: | |
105 | config.write(']broken] config\n[[') | |
b32b8144 FG |
106 | with pytest.raises(RuntimeError): |
107 | configuration.load(ceph_conf) | |
d2e6a577 | 108 | stdout, stderr = capsys.readouterr() |
494da23a | 109 | assert 'File contains no section headers' in stderr |
b32b8144 FG |
110 | |
111 | @pytest.mark.parametrize('commented', ['colon','hash']) | |
112 | def test_coment_as_a_value(self, tmpdir, commented): | |
113 | conf_path = os.path.join(str(tmpdir), 'ceph.conf') | |
114 | with open(conf_path, 'w') as conf: | |
115 | conf.write(tabbed_conf) | |
116 | result = configuration.load(conf_path) | |
117 | assert result.get('global', commented) == '' |