1 | # |
||
2 | # Copyright (C) 2012 - 2017 Satoru SATOH <ssato at redhat.com> |
||
3 | # License: MIT |
||
4 | # |
||
5 | # pylint: disable=missing-docstring, invalid-name, no-member |
||
6 | from __future__ import absolute_import |
||
7 | |||
8 | import copy |
||
9 | import logging |
||
10 | import io |
||
11 | import os |
||
12 | import os.path |
||
13 | import unittest |
||
14 | |||
15 | import anyconfig.api as TT |
||
16 | import anyconfig.backends |
||
17 | import anyconfig.compat |
||
18 | import anyconfig.dicts |
||
19 | import anyconfig.template |
||
20 | import tests.common |
||
21 | |||
22 | from tests.common import CNF_0, SCM_0, CNF_1, dicts_equal, selfdir |
||
23 | |||
24 | |||
25 | # suppress logging messages. |
||
26 | TT.LOGGER.setLevel(logging.CRITICAL) |
||
27 | |||
28 | CNF_TMPL_0 = """name: {{ name|default('a') }} |
||
29 | a: {{ a }} |
||
30 | b: |
||
31 | b: |
||
32 | {% for x in b.b -%} |
||
33 | - {{ x }} |
||
34 | {% endfor %} |
||
35 | c: {{ b.c }} |
||
36 | """ |
||
37 | |||
38 | CNF_TMPL_1 = """a: {{ a }} |
||
39 | b: |
||
40 | b: |
||
41 | {% for x in b.b -%} |
||
42 | - {{ x }} |
||
43 | {% endfor %} |
||
44 | c: {{ b.c }} |
||
45 | |||
46 | name: {{ name }} |
||
47 | """ |
||
48 | |||
49 | CNF_TMPL_2 = """a: {{ a }} |
||
50 | b: |
||
51 | b: |
||
52 | {% for x in b.b -%} |
||
53 | - {{ x }} |
||
54 | {% endfor %} |
||
55 | d: {{ b.d }} |
||
56 | e: 0 |
||
57 | """ |
||
58 | |||
59 | CNF_XML_1 = {'config': {'@attrs': {'name': 'foo'}, |
||
60 | 'a': '0', |
||
61 | 'b': {'@attrs': {'id': 'b0'}, '@text': 'bbb'}, |
||
62 | 'c': None, |
||
63 | 'sect0': {'d': 'x, y, z'}, |
||
64 | 'list1': [{'item': '0'}, {'item': '1'}, |
||
65 | {'item': '2'}], |
||
66 | 'list2': {'@attrs': {'id': 'list2'}, |
||
67 | '@children': [{'item': 'i'}, |
||
68 | {'item': 'j'}]}}} |
||
69 | |||
70 | NULL_CNTNR = TT.anyconfig.dicts.convert_to({}) |
||
71 | |||
72 | |||
73 | class MyODict(anyconfig.compat.OrderedDict): |
||
74 | pass |
||
75 | |||
76 | |||
77 | def _is_file_object(obj): |
||
78 | try: |
||
79 | return isinstance(obj, file) |
||
80 | except NameError: # python 3.x |
||
81 | return isinstance(obj, io.IOBase) |
||
82 | |||
83 | |||
84 | class Test_10_find_loader(unittest.TestCase): |
||
85 | |||
86 | def _assert_isinstance(self, obj, cls, msg=False): |
||
87 | self.assertTrue(isinstance(obj, cls), msg or repr(obj)) |
||
88 | |||
89 | def test_10_find_loader__w_parser_type_or_instance_or_by_file(self): |
||
90 | cpath = "dummy.conf" |
||
91 | for psr in anyconfig.backends.PARSERS: |
||
92 | self._assert_isinstance(TT.find_loader(cpath, psr.type()), psr) |
||
93 | self._assert_isinstance(TT.find_loader(cpath, psr()), psr) |
||
94 | for ext in psr.extensions(): |
||
95 | self._assert_isinstance(TT.find_loader("dummy." + ext), psr, |
||
96 | "ext=%s, psr=%r" % (ext, psr)) |
||
97 | |||
98 | def test_30_find_loader__unknown_parser_type(self): |
||
99 | self.assertRaises(TT.UnknownProcessorTypeError, |
||
100 | TT.find_loader, "a.cnf", "type_not_exist") |
||
101 | |||
102 | def test_40_find_loader__unknown_file_type(self): |
||
103 | self.assertRaises(TT.UnknownFileTypeError, |
||
104 | TT.find_loader, "dummy.ext_not_found") |
||
105 | |||
106 | |||
107 | class TestBase(unittest.TestCase): |
||
108 | |||
109 | cnf = dic = dict(a=1, b=dict(b=[0, 1], c="C"), name="a") |
||
110 | upd = dict(a=2, b=dict(b=[1, 2, 3, 4, 5], d="D"), e=0) |
||
111 | |||
112 | def assert_dicts_equal(self, dic, ref, ordered=False): |
||
113 | self.assertTrue(dicts_equal(dic, ref, ordered=ordered), |
||
114 | "%r%s vs.%s%r" % (dic, os.linesep, os.linesep, ref)) |
||
115 | |||
116 | |||
117 | class Test_20_dumps_and_loads(TestBase): |
||
118 | |||
119 | def test_30_dumps_and_loads(self): |
||
120 | res = TT.loads(TT.dumps(self.cnf, "json"), "json") |
||
121 | self.assert_dicts_equal(res, self.cnf) |
||
122 | |||
123 | def test_30_dumps_and_loads__w_options(self): |
||
124 | res = TT.loads(TT.dumps(self.cnf, "json", indent=2), "json", |
||
125 | ensure_ascii=False) |
||
126 | self.assert_dicts_equal(res, self.cnf) |
||
127 | |||
128 | def test_40_loads_wo_type(self): |
||
129 | cnf_s = "requires:bash,zsh" |
||
130 | self.assertTrue(TT.loads(cnf_s) is None) |
||
131 | |||
132 | def test_42_loads_w_type_not_exist(self): |
||
133 | a_s = "requires:bash,zsh" |
||
134 | self.assertRaises(TT.UnknownProcessorTypeError, |
||
135 | TT.loads, a_s, "type_not_exist") |
||
136 | |||
137 | def test_44_loads_w_type__template(self): |
||
138 | if not anyconfig.template.SUPPORTED: |
||
139 | return |
||
140 | |||
141 | a_s = "requires: [{{ requires|join(', ') }}]" |
||
142 | reqs = dict(requires=["bash", "zsh"]) |
||
143 | |||
144 | a1 = TT.loads(a_s, ac_parser="yaml", ac_template=True, |
||
145 | ac_context=reqs) |
||
146 | |||
147 | self.assertEqual(a1["requires"], reqs["requires"]) |
||
148 | |||
149 | def test_46_loads_w_type__broken_template(self): |
||
150 | if not anyconfig.template.SUPPORTED: |
||
151 | return |
||
152 | |||
153 | a = dict(requires="{% }}", ) |
||
154 | a_s = 'requires: "{% }}"' |
||
155 | a1 = TT.loads(a_s, ac_parser="yaml", ac_template=True, |
||
156 | ac_context={}) |
||
157 | |||
158 | self.assertEqual(a1["requires"], a["requires"]) |
||
159 | |||
160 | def test_48_loads_w_validation(self): |
||
161 | cnf_s = TT.dumps(CNF_0, "json") |
||
162 | scm_s = TT.dumps(SCM_0, "json") |
||
163 | cnf_2 = TT.loads(cnf_s, ac_parser="json", ac_context={}, |
||
164 | ac_schema=scm_s) |
||
165 | |||
166 | self.assertEqual(cnf_2["name"], CNF_0["name"]) |
||
167 | View Code Duplication | self.assertEqual(cnf_2["a"], CNF_0["a"]) |
|
0 ignored issues
–
show
Duplication
introduced
by
![]() |
|||
168 | self.assertEqual(cnf_2["b"]["b"], CNF_0["b"]["b"]) |
||
169 | self.assertEqual(cnf_2["b"]["c"], CNF_0["b"]["c"]) |
||
170 | |||
171 | def test_49_loads_w_validation_error(self): |
||
172 | cnf_s = """{"a": "aaa"}""" |
||
173 | scm_s = TT.dumps(SCM_0, "json") |
||
174 | cnf_2 = TT.loads(cnf_s, ac_parser="json", ac_schema=scm_s) |
||
175 | self.assertTrue(cnf_2 is None, cnf_2) |
||
176 | |||
177 | |||
178 | class Test_22_single_load(TestBase): |
||
179 | |||
180 | a_path = os.path.join(selfdir(), "00-cnf.json") |
||
181 | cnf = CNF_1 |
||
182 | pathlib = anyconfig.compat.pathlib |
||
183 | |||
184 | def test_10__single_load(self): |
||
185 | res = TT.single_load(self.a_path) |
||
186 | self.assert_dicts_equal(res, self.cnf) |
||
187 | |||
188 | def test_12__single_load__ac_parser(self): |
||
189 | res = TT.single_load(self.a_path, ac_parser="json") |
||
190 | self.assert_dicts_equal(res, self.cnf) |
||
191 | |||
192 | def test_20__single_load__stream(self): |
||
193 | res = TT.single_load(open(self.a_path), ac_parser="json") |
||
194 | self.assert_dicts_equal(res, self.cnf) |
||
195 | |||
196 | def test_30__single_load__pathlib(self): |
||
197 | if self.pathlib: |
||
198 | pobj = self.pathlib.Path(self.a_path) |
||
199 | res = TT.single_load(pobj) |
||
200 | self.assert_dicts_equal(res, self.cnf) |
||
201 | |||
202 | |||
203 | class TestBaseWithIO(TestBase): |
||
204 | |||
205 | def setUp(self): |
||
206 | self.workdir = tests.common.setup_workdir() |
||
207 | self.a_path = os.path.join(self.workdir, "a.json") |
||
208 | self.exp = copy.deepcopy(self.dic) |
||
209 | |||
210 | def tearDown(self): |
||
211 | tests.common.cleanup_workdir(self.workdir) |
||
212 | |||
213 | |||
214 | class Test_30_single_load(TestBaseWithIO): |
||
215 | |||
216 | def test_10_dump_and_single_load(self): |
||
217 | TT.dump(self.cnf, self.a_path) |
||
218 | self.assertTrue(os.path.exists(self.a_path)) |
||
219 | |||
220 | res = TT.single_load(self.a_path) |
||
221 | self.assert_dicts_equal(res, self.cnf) |
||
222 | |||
223 | def test_11_dump_and_single_load__to_from_stream(self): |
||
224 | TT.dump(self.cnf, TT.open(self.a_path, mode='w')) |
||
225 | self.assertTrue(os.path.exists(self.a_path)) |
||
226 | |||
227 | res = TT.single_load(TT.open(self.a_path)) |
||
228 | self.assert_dicts_equal(res, self.cnf) |
||
229 | |||
230 | def test_12_dump_and_single_load__no_parser(self): |
||
231 | self.assertRaises(TT.UnknownFileTypeError, |
||
232 | TT.single_load, "dummy.ext_not_exist") |
||
233 | |||
234 | def test_14_single_load__ignore_missing(self): |
||
235 | cpath = os.path.join(os.curdir, "conf_file_should_not_exist") |
||
236 | assert not os.path.exists(cpath) |
||
237 | |||
238 | self.assertEqual(TT.single_load(cpath, "ini", ac_ignore_missing=True), |
||
239 | NULL_CNTNR) |
||
240 | |||
241 | def test_15_single_load__fail_to_render_template(self): |
||
242 | if not anyconfig.template.SUPPORTED: |
||
243 | return |
||
244 | |||
245 | cnf_s = "name: '{{ name'" # Should cause template renering error. |
||
246 | cpath = os.path.join(self.workdir, "a.yaml") |
||
247 | TT.open(cpath, mode='w').write(cnf_s) |
||
248 | |||
249 | cnf = TT.single_load(cpath, ac_template=True, ac_context=dict(a=1)) |
||
250 | self.assertEqual(cnf["name"], "{{ name") |
||
251 | |||
252 | def test_16_single_load__template(self): |
||
253 | if not anyconfig.template.SUPPORTED: |
||
254 | return |
||
255 | |||
256 | cpath = os.path.join(self.workdir, "a.yaml") |
||
257 | TT.open(cpath, mode='w').write(CNF_TMPL_0) |
||
258 | |||
259 | cnf = TT.single_load(cpath, ac_template=True, ac_context=self.cnf) |
||
260 | self.assert_dicts_equal(cnf, self.cnf) |
||
261 | |||
262 | spath = os.path.join(self.workdir, "scm.json") |
||
263 | TT.dump(dict(type="integer"), spath) # Validation should fail. |
||
264 | |||
265 | cnf2 = TT.single_load(cpath, ac_template=True, ac_context=self.cnf, |
||
266 | ac_schema=spath) |
||
267 | self.assertTrue(cnf2 is None) |
||
268 | |||
269 | def test_18_single_load__templates(self): |
||
270 | if not anyconfig.template.SUPPORTED: |
||
271 | return |
||
272 | |||
273 | a_path = os.path.join(self.workdir, "a.yml") |
||
274 | b_path = os.path.join(self.workdir, "b.yml") |
||
275 | a2_path = os.path.join(self.workdir, "x/y/z", "a.yml") |
||
276 | |||
277 | open(a_path, 'w').write("{% include 'b.yml' %}") |
||
278 | open(b_path, 'w').write(CNF_TMPL_0) |
||
279 | os.makedirs(os.path.dirname(a2_path)) |
||
280 | open(a2_path, 'w').write("a: 'xyz'") |
||
281 | |||
282 | cnf1 = TT.single_load(a_path, ac_template=True, ac_context=self.cnf) |
||
283 | self.assertTrue(dicts_equal(self.cnf, cnf1), str(cnf1)) |
||
284 | |||
285 | cnf2 = TT.single_load(a2_path, ac_template=True) |
||
286 | self.assertEqual(cnf2["a"], "xyz") |
||
287 | |||
288 | def test_19_dump_and_single_load_with_validation(self): |
||
289 | cnf = CNF_0 |
||
290 | scm = SCM_0 |
||
291 | |||
292 | cnf_path = os.path.join(self.workdir, "cnf_19.json") |
||
293 | scm_path = os.path.join(self.workdir, "scm_19.json") |
||
294 | |||
295 | TT.dump(cnf, cnf_path) |
||
296 | TT.dump(scm, scm_path) |
||
297 | self.assertTrue(os.path.exists(cnf_path)) |
||
298 | self.assertTrue(os.path.exists(scm_path)) |
||
299 | |||
300 | cnf_1 = TT.single_load(cnf_path, ac_schema=scm_path) |
||
301 | |||
302 | self.assertFalse(cnf_1 is None) # Validation should succeed. |
||
303 | self.assertTrue(dicts_equal(cnf_1, cnf), cnf_1) |
||
304 | |||
305 | cnf_2 = cnf.copy() |
||
306 | cnf_2["a"] = "aaa" # It's type should be integer not string. |
||
307 | cnf_2_path = os.path.join(self.workdir, "cnf_19_2.json") |
||
308 | TT.dump(cnf_2, cnf_2_path) |
||
309 | self.assertTrue(os.path.exists(cnf_2_path)) |
||
310 | |||
311 | cnf_3 = TT.single_load(cnf_2_path, ac_schema=scm_path) |
||
312 | self.assertTrue(cnf_3 is None) # Validation should fail. |
||
313 | |||
314 | def test_20_dump_and_single_load__w_ordered_option(self): |
||
315 | TT.dump(self.cnf, self.a_path) |
||
316 | self.assertTrue(os.path.exists(self.a_path)) |
||
317 | |||
318 | # It works w/ JSON backend but some backend cannot keep the order of |
||
319 | # items and the tests might fail. |
||
320 | res = TT.single_load(self.a_path, ac_ordered=True) |
||
321 | self.assert_dicts_equal(res, self.cnf, ordered=True) |
||
322 | self.assertTrue(isinstance(res, anyconfig.compat.OrderedDict)) |
||
323 | |||
324 | def test_22_dump_and_single_load__w_ac_dict_option(self): |
||
325 | TT.dump(self.cnf, self.a_path) |
||
326 | self.assertTrue(os.path.exists(self.a_path)) |
||
327 | |||
328 | res = TT.single_load(self.a_path, ac_dict=MyODict) |
||
329 | self.assert_dicts_equal(res, self.cnf, ordered=True) |
||
330 | self.assertTrue(isinstance(res, MyODict)) |
||
331 | |||
332 | |||
333 | class Test_32_single_load(unittest.TestCase): |
||
334 | |||
335 | cnf = CNF_XML_1 |
||
336 | |||
337 | def setUp(self): |
||
338 | self.workdir = tests.common.setup_workdir() |
||
339 | |||
340 | def tearDown(self): |
||
341 | tests.common.cleanup_workdir(self.workdir) |
||
342 | |||
343 | def _load_and_dump_with_opened_files(self, filename, rmode='r', wmode='w', |
||
344 | **oopts): |
||
345 | cpath = os.path.join(self.workdir, filename) |
||
346 | |||
347 | with TT.open(cpath, 'w', **oopts) as out: |
||
348 | TT.dump(self.cnf, out) |
||
349 | self.assertTrue(_is_file_object(out)) |
||
350 | self.assertEqual(out.mode, wmode) |
||
351 | |||
352 | with TT.open(cpath, 'rb', **oopts) as inp: |
||
353 | cnf1 = TT.single_load(inp) |
||
354 | self.assertTrue(_is_file_object(inp)) |
||
355 | self.assertEqual(inp.mode, rmode) |
||
356 | cpair = (self.cnf, cnf1) |
||
357 | self.assertTrue(dicts_equal(*cpair), "%r vs. %r" % cpair) |
||
358 | |||
359 | def test_10_open_json_file(self): |
||
360 | View Code Duplication | self._load_and_dump_with_opened_files("a.json") |
|
0 ignored issues
–
show
|
|||
361 | |||
362 | def test_20_open_xml_file(self): |
||
363 | if "xml" in anyconfig.backends.list_types(): |
||
364 | self._load_and_dump_with_opened_files("a.xml", 'rb', 'wb') |
||
365 | |||
366 | def test_30_open_bson_file(self): |
||
367 | if "bson" in anyconfig.backends.list_types(): |
||
368 | self._load_and_dump_with_opened_files("a.bson", 'rb', 'wb') |
||
369 | |||
370 | def test_40_open_yaml_file(self): |
||
371 | if "yaml" in anyconfig.backends.list_types(): |
||
372 | self._load_and_dump_with_opened_files("a.yaml") |
||
373 | View Code Duplication | self._load_and_dump_with_opened_files("a.yml") |
|
0 ignored issues
–
show
|
|||
374 | |||
375 | |||
376 | class Test_34_single_load(TestBaseWithIO): |
||
377 | |||
378 | def test_10_single_load_w_validation(self): |
||
379 | cnf_path = os.path.join(self.workdir, "cnf.json") |
||
380 | scm_path = os.path.join(self.workdir, "scm.json") |
||
381 | TT.dump(CNF_0, cnf_path) |
||
382 | TT.dump(SCM_0, scm_path) |
||
383 | |||
384 | cnf_2 = TT.single_load(cnf_path, ac_context={}, ac_schema=scm_path) |
||
385 | |||
386 | self.assertEqual(cnf_2["name"], CNF_0["name"]) |
||
387 | self.assertEqual(cnf_2["a"], CNF_0["a"]) |
||
388 | self.assertEqual(cnf_2["b"]["b"], CNF_0["b"]["b"]) |
||
389 | self.assertEqual(cnf_2["b"]["c"], CNF_0["b"]["c"]) |
||
390 | |||
391 | def test_20_single_load_w_query(self): |
||
392 | cpath = os.path.join(self.workdir, "cnf.json") |
||
393 | TT.dump(CNF_0, cpath) |
||
394 | |||
395 | try: |
||
396 | if TT.query.jmespath: |
||
397 | View Code Duplication | self.assertEqual(TT.single_load(cpath, ac_query="a"), 1) |
|
0 ignored issues
–
show
|
|||
398 | self.assertEqual(TT.single_load(cpath, ac_query="b.b"), [1, 2]) |
||
399 | self.assertEqual(TT.single_load(cpath, ac_query="b.b[1]"), 2) |
||
400 | self.assertEqual(TT.single_load(cpath, ac_query="b.b[1:]"), |
||
401 | [2]) |
||
402 | self.assertEqual(TT.single_load(cpath, ac_query="b.b[::-1]"), |
||
403 | [2, 1]) |
||
404 | self.assertEqual(TT.single_load(cpath, ac_query="length(b.b)"), |
||
405 | 2) |
||
406 | except (NameError, AttributeError): |
||
407 | pass # jmespath is not available. |
||
408 | |||
409 | |||
410 | class TestBaseWithIOMultiFiles(TestBaseWithIO): |
||
411 | |||
412 | def setUp(self): |
||
413 | super(TestBaseWithIOMultiFiles, self).setUp() |
||
414 | self.b_path = os.path.join(self.workdir, "b.json") |
||
415 | self.g_path = os.path.join(self.workdir, "*.json") |
||
416 | |||
417 | exp = copy.deepcopy(self.upd) # Assume MS_DICTS strategy was used. |
||
418 | exp["b"]["c"] = self.dic["b"]["c"] |
||
419 | exp["name"] = self.dic["name"] |
||
420 | self.exp = exp |
||
421 | |||
422 | |||
423 | class Test_40_multi_load_with_strategies(TestBaseWithIOMultiFiles): |
||
424 | |||
425 | def _check_multi_load_with_strategy(self, exp, merge=TT.MS_DICTS): |
||
426 | TT.dump(self.dic, self.a_path) |
||
427 | TT.dump(self.upd, self.b_path) |
||
428 | |||
429 | self.assertTrue(os.path.exists(self.a_path)) |
||
430 | self.assertTrue(os.path.exists(self.b_path)) |
||
431 | |||
432 | View Code Duplication | res0 = TT.multi_load(self.g_path, ac_merge=merge) |
|
0 ignored issues
–
show
|
|||
433 | res1 = TT.multi_load([self.g_path, self.b_path], ac_merge=merge) |
||
434 | |||
435 | self.assertTrue(res0) |
||
436 | self.assertTrue(res1) |
||
437 | |||
438 | self.assert_dicts_equal(res0, exp) |
||
439 | self.assert_dicts_equal(res1, exp) |
||
440 | |||
441 | def test_10_default_merge_strategy(self): |
||
442 | exp = copy.deepcopy(self.upd) |
||
443 | exp["b"]["c"] = self.dic["b"]["c"] |
||
444 | exp["name"] = self.dic["name"] |
||
445 | |||
446 | self._check_multi_load_with_strategy(exp, merge=None) |
||
447 | self._check_multi_load_with_strategy(exp) |
||
448 | |||
449 | def test_20_merge_dicts_and_lists(self): |
||
450 | exp = copy.deepcopy(self.upd) |
||
451 | exp["b"]["b"] = [0] + self.upd["b"]["b"] |
||
452 | exp["b"]["c"] = self.dic["b"]["c"] |
||
453 | exp["name"] = self.dic["name"] |
||
454 | self._check_multi_load_with_strategy(exp, merge=TT.MS_DICTS_AND_LISTS) |
||
455 | |||
456 | def test_30_merge_with_replace(self): |
||
457 | exp = copy.deepcopy(self.upd) |
||
458 | exp["name"] = self.dic["name"] |
||
459 | self._check_multi_load_with_strategy(exp, merge=TT.MS_REPLACE) |
||
460 | |||
461 | def test_40_merge_wo_replace(self): |
||
462 | exp = copy.deepcopy(self.dic) |
||
463 | exp["e"] = self.upd["e"] |
||
464 | self._check_multi_load_with_strategy(exp, merge=TT.MS_NO_REPLACE) |
||
465 | |||
466 | def test_60_wrong_merge_strategy(self): |
||
467 | cpath = os.path.join(self.workdir, "a.json") |
||
468 | TT.dump(dict(a=1, b=2), cpath) |
||
469 | try: |
||
470 | TT.multi_load([cpath, cpath], ac_merge="merge_st_not_exist") |
||
471 | raise RuntimeError("Wrong merge strategy was not handled!") |
||
472 | except ValueError: |
||
473 | self.assertTrue(1 == 1) # To suppress warn of pylint. |
||
474 | |||
475 | |||
476 | class Test_42_multi_load(TestBaseWithIOMultiFiles): |
||
477 | |||
478 | def test_10_multi_load__empty_path_list(self): |
||
479 | self.assertEqual(TT.multi_load([]), NULL_CNTNR) |
||
480 | |||
481 | def test_20_dump_and_multi_load__mixed_file_types(self): |
||
482 | c_path = os.path.join(self.workdir, "c.yml") |
||
483 | |||
484 | TT.dump(self.dic, self.a_path) # JSON |
||
485 | try: |
||
486 | TT.dump(self.upd, c_path) # YAML |
||
487 | except (TT.UnknownProcessorTypeError, TT.UnknownFileTypeError): |
||
488 | return # YAML backend is not available in this env. |
||
489 | |||
490 | self.assertTrue(os.path.exists(self.a_path)) |
||
491 | self.assertTrue(os.path.exists(c_path)) |
||
492 | |||
493 | res = TT.multi_load([self.a_path, c_path]) |
||
494 | self.assert_dicts_equal(res, self.exp) |
||
495 | |||
496 | def test_30_dump_and_multi_load__to_from_stream(self): |
||
497 | TT.dump(self.dic, self.a_path) |
||
498 | TT.dump(self.upd, self.b_path) |
||
499 | |||
500 | res = TT.multi_load([TT.open(self.a_path), TT.open(self.b_path)]) |
||
501 | self.assert_dicts_equal(res, self.exp) |
||
502 | |||
503 | def test_40_multi_load__ignore_missing(self): |
||
504 | cpath = os.path.join(os.curdir, "conf_file_should_not_exist") |
||
505 | assert not os.path.exists(cpath) |
||
506 | |||
507 | self.assertEqual(TT.multi_load([cpath], ac_parser="ini", |
||
508 | ac_ignore_missing=True), |
||
509 | NULL_CNTNR) |
||
510 | # It will be remove after 'ignore_missing' was deprecated and removed. |
||
511 | self.assertEqual(TT.multi_load([cpath], ac_parser="ini", |
||
512 | ignore_missing=True), |
||
513 | NULL_CNTNR) |
||
514 | |||
515 | def test_50_multi_load__templates(self): |
||
516 | if not anyconfig.template.SUPPORTED: |
||
517 | return |
||
518 | |||
519 | ctx = self.dic.copy() |
||
520 | TT.merge(ctx, self.upd, ac_merge=TT.MS_DICTS) |
||
521 | |||
522 | a_path = self.a_path.replace(".json", ".yml") |
||
523 | b_path = self.b_path.replace(".json", ".yml") |
||
524 | g_path = self.g_path.replace(".json", ".yml") |
||
525 | |||
526 | TT.open(a_path, mode='w').write(CNF_TMPL_1) |
||
527 | TT.open(b_path, mode='w').write(CNF_TMPL_2) |
||
528 | |||
529 | opts = dict(ac_merge=TT.MS_DICTS, ac_template=True, ac_context=ctx) |
||
530 | try: |
||
531 | res0 = TT.multi_load(g_path, **opts) |
||
532 | res1 = TT.multi_load([g_path, b_path], **opts) |
||
533 | except (TT.UnknownProcessorTypeError, TT.UnknownFileTypeError): |
||
534 | return |
||
535 | |||
536 | self.assert_dicts_equal(res0, self.exp) |
||
537 | self.assert_dicts_equal(res1, self.exp) |
||
538 | |||
539 | def test_60_multi_load__w_ac_dict_option(self): |
||
540 | TT.dump(self.dic, self.a_path) |
||
541 | TT.dump(self.upd, self.b_path) |
||
542 | |||
543 | res = TT.multi_load(self.g_path, ac_dict=MyODict) |
||
544 | self.assert_dicts_equal(res, self.exp) |
||
545 | self.assertTrue(isinstance(res, MyODict)) |
||
546 | |||
547 | |||
548 | class Test_44_multi_load(TestBase): |
||
549 | |||
550 | def test_10_multi_load_w_validation_for_partial_single_config_files(self): |
||
551 | cpaths = [os.path.join(selfdir(), "00-00-cnf.json"), |
||
552 | os.path.join(selfdir(), "00-01-cnf.json"), |
||
553 | os.path.join(selfdir(), "00-02-cnf.json")] |
||
554 | spath = os.path.join(selfdir(), "00-scm.json") |
||
555 | |||
556 | cnf = TT.multi_load(cpaths, ac_schema=spath) |
||
557 | ref = TT.multi_load(cpaths) |
||
558 | self.assert_dicts_equal(cnf, ref, ordered=False) |
||
559 | |||
560 | |||
561 | class Test_50_load_and_dump(TestBaseWithIOMultiFiles): |
||
562 | |||
563 | def test_30_dump_and_load(self): |
||
564 | TT.dump(self.dic, self.a_path) |
||
565 | TT.dump(self.upd, self.b_path) |
||
566 | |||
567 | self.assertTrue(os.path.exists(self.a_path)) |
||
568 | self.assertTrue(os.path.exists(self.b_path)) |
||
569 | |||
570 | View Code Duplication | res = TT.load(self.a_path) |
|
0 ignored issues
–
show
|
|||
571 | self.assert_dicts_equal(res, self.dic) |
||
572 | |||
573 | res = TT.load(self.g_path) |
||
574 | self.assert_dicts_equal(res, self.exp) |
||
575 | |||
576 | res = TT.load([self.a_path, self.b_path]) |
||
577 | self.assert_dicts_equal(res, self.exp) |
||
578 | |||
579 | def test_31_dump_and_load__to_from_stream(self): |
||
580 | with TT.open(self.a_path, mode='w') as strm: |
||
581 | TT.dump(self.dic, strm) |
||
582 | |||
583 | self.assertTrue(os.path.exists(self.a_path)) |
||
584 | |||
585 | with TT.open(self.a_path) as strm: |
||
586 | res = TT.load(strm, ac_parser="json") |
||
587 | self.assert_dicts_equal(res, self.dic) |
||
588 | |||
589 | def test_32_dump_and_load__w_options(self): |
||
590 | TT.dump(self.dic, self.a_path, indent=2) |
||
591 | self.assertTrue(os.path.exists(self.a_path)) |
||
592 | |||
593 | TT.dump(self.upd, self.b_path, indent=2) |
||
594 | View Code Duplication | self.assertTrue(os.path.exists(self.b_path)) |
|
0 ignored issues
–
show
|
|||
595 | |||
596 | res = TT.load(self.a_path, parse_int=int) |
||
597 | dic = copy.deepcopy(self.dic) |
||
598 | self.assert_dicts_equal(res, dic) |
||
599 | |||
600 | res = TT.load(self.g_path, parse_int=int) |
||
601 | exp = copy.deepcopy(self.exp) |
||
602 | self.assert_dicts_equal(res, exp) |
||
603 | |||
604 | res = TT.load([self.a_path, self.b_path], parse_int=int) |
||
605 | exp = copy.deepcopy(self.exp) |
||
606 | self.assert_dicts_equal(res, exp) |
||
607 | View Code Duplication | ||
0 ignored issues
–
show
|
|||
608 | def test_34_load__ignore_missing(self): |
||
609 | cpath = os.path.join(os.curdir, "conf_file_should_not_exist") |
||
610 | assert not os.path.exists(cpath) |
||
611 | |||
612 | self.assertEqual(TT.load([cpath], ac_parser="ini", |
||
613 | ignore_missing=True), |
||
614 | NULL_CNTNR) |
||
615 | |||
616 | def test_36_load_w_validation(self): |
||
617 | cnf_path = os.path.join(self.workdir, "cnf.json") |
||
618 | scm_path = os.path.join(self.workdir, "scm.json") |
||
619 | TT.dump(CNF_0, cnf_path) |
||
620 | TT.dump(SCM_0, scm_path) |
||
621 | |||
622 | cnf_2 = TT.load(cnf_path, ac_context={}, ac_schema=scm_path) |
||
623 | |||
624 | self.assertEqual(cnf_2["name"], CNF_0["name"]) |
||
625 | self.assertEqual(cnf_2["a"], CNF_0["a"]) |
||
626 | self.assertEqual(cnf_2["b"]["b"], CNF_0["b"]["b"]) |
||
627 | self.assertEqual(cnf_2["b"]["c"], CNF_0["b"]["c"]) |
||
628 | |||
629 | def test_38_load_w_validation_yaml(self): |
||
630 | cnf_path = os.path.join(self.workdir, "cnf.yml") |
||
631 | View Code Duplication | scm_path = os.path.join(self.workdir, "scm.yml") |
|
0 ignored issues
–
show
|
|||
632 | TT.dump(CNF_0, cnf_path) |
||
633 | TT.dump(SCM_0, scm_path) |
||
634 | |||
635 | cnf_2 = TT.load(cnf_path, ac_context={}, ac_schema=scm_path) |
||
636 | |||
637 | self.assertEqual(cnf_2["name"], CNF_0["name"]) |
||
638 | self.assertEqual(cnf_2["a"], CNF_0["a"]) |
||
639 | self.assertEqual(cnf_2["b"]["b"], CNF_0["b"]["b"]) |
||
640 | self.assertEqual(cnf_2["b"]["c"], CNF_0["b"]["c"]) |
||
641 | |||
642 | def test_39_single_load__w_validation(self): |
||
643 | (cnf, scm) = (CNF_0, SCM_0) |
||
644 | cpath = os.path.join(self.workdir, "cnf.json") |
||
645 | spath = os.path.join(self.workdir, "scm.json") |
||
646 | |||
647 | TT.dump(cnf, cpath) |
||
648 | TT.dump(scm, spath) |
||
649 | |||
650 | cnf1 = TT.single_load(cpath, ac_schema=spath) |
||
651 | self.assert_dicts_equal(cnf, cnf1) |
||
652 | |||
653 | def test_40_load_w_query(self): |
||
654 | cnf_path = os.path.join(self.workdir, "cnf.json") |
||
655 | TT.dump(CNF_0, cnf_path) |
||
656 | |||
657 | try: |
||
658 | if TT.query.jmespath: |
||
659 | self.assertEqual(TT.load(cnf_path, ac_query="a"), 1) |
||
660 | self.assertEqual(TT.load(cnf_path, ac_query="b.b"), [1, 2]) |
||
661 | self.assertEqual(TT.load(cnf_path, ac_query="b.b[1]"), 2) |
||
662 | self.assertEqual(TT.load(cnf_path, ac_query="b.b[1:]"), [2]) |
||
663 | self.assertEqual(TT.load(cnf_path, ac_query="b.b[::-1]"), |
||
664 | [2, 1]) |
||
665 | self.assertEqual(TT.load(cnf_path, ac_query="length(b.b)"), 2) |
||
666 | except (NameError, AttributeError): |
||
667 | pass # jmespath is not available. |
||
668 | |||
669 | # vim:sw=4:ts=4:et: |
||
670 |