Completed
Push — master ( 6ef2f5...b75cb1 )
by Satoru
53s
created

Test_22_single_load   A

Complexity

Total Complexity 5

Size/Duplication

Total Lines 23
Duplicated Lines 0 %

Importance

Changes 1
Bugs 0 Features 1
Metric Value
c 1
b 0
f 1
dl 0
loc 23
rs 10
wmc 5

4 Methods

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