Test_20_dumps_and_loads   A
last analyzed

Complexity

Total Complexity 10

Size/Duplication

Total Lines 59
Duplicated Lines 15.25 %

Importance

Changes 0
Metric Value
c 0
b 0
f 0
dl 9
loc 59
rs 10
wmc 10

8 Methods

Rating   Name   Duplication   Size   Complexity  
A test_44_loads_w_type__template() 0 11 2
A test_49_loads_w_validation_error() 5 5 1
A test_42_loads_w_type_not_exist() 0 4 1
A test_48_loads_w_validation() 3 10 1
A test_40_loads_wo_type() 0 3 1
A test_30_dumps_and_loads() 0 3 1
A test_46_loads_w_type__broken_template() 0 10 2
A test_30_dumps_and_loads__w_options() 0 4 1

How to fix   Duplicated Code   

Duplicated Code

Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.

Common duplication problems, and corresponding solutions are:

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
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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