Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_36_load_w_validation(self):
cnf_path = os.path.join(self.workdir, "cnf.json")
scm_path = os.path.join(self.workdir, "scm.json")
TT.dump(CNF_0, cnf_path)
TT.dump(SCM_0, scm_path)
cnf_2 = TT.load(cnf_path, ac_context={}, ac_schema=scm_path)
self.assertEqual(cnf_2["name"], CNF_0["name"])
self.assertEqual(cnf_2["a"], CNF_0["a"])
self.assertEqual(cnf_2["b"]["b"], CNF_0["b"]["b"])
self.assertEqual(cnf_2["b"]["c"], CNF_0["b"]["c"])
def test_32_dump_and_load__w_options(self):
TT.dump(self.dic, self.a_path, indent=2)
self.assertTrue(os.path.exists(self.a_path))
TT.dump(self.upd, self.b_path, indent=2)
self.assertTrue(os.path.exists(self.b_path))
res = TT.load(self.a_path, parse_int=int)
dic = copy.deepcopy(self.dic)
self.assert_dicts_equal(res, dic)
res = TT.load(self.g_path, parse_int=int)
exp = copy.deepcopy(self.exp)
self.assert_dicts_equal(res, exp)
res = TT.load([self.a_path, self.b_path], parse_int=int)
exp = copy.deepcopy(self.exp)
self.assert_dicts_equal(res, exp)
def test_40_load_w_query(self):
cnf_path = os.path.join(self.workdir, "cnf.json")
TT.dump(CNF_0, cnf_path)
try:
if TT.query.jmespath:
self.assertEqual(TT.load(cnf_path, ac_query="a"), 1)
self.assertEqual(TT.load(cnf_path, ac_query="b.b"), [1, 2])
self.assertEqual(TT.load(cnf_path, ac_query="b.b[1]"), 2)
self.assertEqual(TT.load(cnf_path, ac_query="b.b[1:]"), [2])
self.assertEqual(TT.load(cnf_path, ac_query="b.b[::-1]"),
[2, 1])
self.assertEqual(TT.load(cnf_path, ac_query="length(b.b)"), 2)
except (NameError, AttributeError):
pass # jmespath is not available.
def test_32_dump_and_load__w_options(self):
TT.dump(self.dic, self.a_path, indent=2)
self.assertTrue(os.path.exists(self.a_path))
TT.dump(self.upd, self.b_path, indent=2)
self.assertTrue(os.path.exists(self.b_path))
res = TT.load(self.a_path, parse_int=int)
dic = copy.deepcopy(self.dic)
self.assert_dicts_equal(res, dic)
res = TT.load(self.g_path, parse_int=int)
exp = copy.deepcopy(self.exp)
self.assert_dicts_equal(res, exp)
res = TT.load([self.a_path, self.b_path], parse_int=int)
exp = copy.deepcopy(self.exp)
self.assert_dicts_equal(res, exp)
def test_31_dump_and_load__to_from_stream(self):
with TT.open(self.a_path, mode='w') as strm:
TT.dump(self.dic, strm)
self.assertTrue(os.path.exists(self.a_path))
with TT.open(self.a_path) as strm:
res = TT.load(strm, ac_parser="json")
self.assert_dicts_equal(res, self.dic)
def test_30_dump_and_load(self):
TT.dump(self.dic, self.a_path)
TT.dump(self.upd, self.b_path)
self.assertTrue(os.path.exists(self.a_path))
self.assertTrue(os.path.exists(self.b_path))
res = TT.load(self.a_path)
self.assert_dicts_equal(res, self.dic)
res = TT.load(self.g_path)
self.assert_dicts_equal(res, self.exp)
res = TT.load([self.a_path, self.b_path])
self.assert_dicts_equal(res, self.exp)
def _load_diff(args, extra_opts):
"""
:param args: :class:`argparse.Namespace` object
:param extra_opts: Map object given to API.load as extra options
"""
try:
diff = API.load(args.inputs, args.itype,
ac_ignore_missing=args.ignore_missing,
ac_merge=args.merge,
ac_template=args.template,
ac_schema=args.schema,
**extra_opts)
except API.UnknownProcessorTypeError:
_exit_with_output("Wrong input type '%s'" % args.itype, 1)
except API.UnknownFileTypeError:
_exit_with_output("No appropriate backend was found for given file "
"type='%s', inputs=%s" % (args.itype,
", ".join(args.inputs)),
1)
_exit_if_load_failure(diff,
"Failed to load: args=%s" % ", ".join(args.inputs))
return diff