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_marshal(self):
differ = JsonDiffer()
d = {
delete: 3,
'$delete': 4,
insert: 4,
'$$something': 1
}
dm = differ.marshal(d)
self.assertEqual(d, differ.unmarshal(dm))
def test_compact_syntax(self, scenario):
a, b = scenario
differ = JsonDiffer(syntax='compact')
d = differ.diff(a, b)
self.assertEqual(b, differ.patch(a, d))
dm = differ.marshal(d)
self.assertEqual(d, differ.unmarshal(dm))
def test_symmetric_syntax(self, scenario):
a, b = scenario
differ = JsonDiffer(syntax='symmetric')
d = differ.diff(a, b)
self.assertEqual(b, differ.patch(a, d))
self.assertEqual(a, differ.unpatch(b, d))
dm = differ.marshal(d)
self.assertEqual(d, differ.unmarshal(dm))
sync_index_pattern = load_kibana_index_pattern(path)
# load generated index pattern
gen_index_pattern = json.load(args.gen_index_pattern)["objects"][0]
gen_index_pattern["attributes"].pop("title") # detects title set in sync'd pattern
# decode fields, they are json
sync_index_pattern["attributes"]["fields"] = json.loads(sync_index_pattern["attributes"].pop("fields"))
gen_index_pattern["attributes"]["fields"] = json.loads(gen_index_pattern["attributes"].pop("fields"))
exit_val = 0
exit_val = max(exit_val, iterate(sync_index_pattern["id"], "", gen_index_pattern, sync_index_pattern))
# double check that there is no difference
if exit_val == 0:
d = jsondiff.JsonDiffer(syntax='symmetric').diff(sync_index_pattern, gen_index_pattern)
if d:
print("index patterns differ: ", d)
return 5
print("up-to-date")
return exit_val
if len(v1) > 0 and isinstance(v1[0], dict):
for item in v1:
qkey = find_key(item)
if qkey == "":
print("Script is missing type to compare {}".format(item))
return 3
item2 = find_item(v2, qkey, item[qkey])
ret_val = max(ret_val, iterate(val_id, build_key(key, "{}={}".format(qkey, item[qkey])), item, item2,
apm_v1=apm_v1))
else:
v1, v2 = sorted(v1), sorted(v2)
for item1, item2 in zip(v1, v2):
ret_val = max(ret_val, iterate(val_id, key, *json_val(item1, item2), apm_v1=apm_v1))
else:
d = jsondiff.JsonDiffer(syntax='symmetric').diff(*json_val(v1, v2))
if d:
if key == "attributes.title" or key == "attributes.fields.name=transaction.marks.*.*":
return ret_val
ret_val = 2
print("Difference for id '{}' for key '{}'".format(val_id, key))
try:
print(json.dumps(d, indent=4))
except:
print(d)
v1_label, v2_label = "APM Server", "Kibana"
if not apm_v1:
v1_label, v2_label = v2_label, v1_label
print("Value in {}: {!r}".format(v1_label, v1))
print("Value in {}: {!r}".format(v2_label, v2))
print("---")
return ret_val
def __print_preview(old_json, new_json):
error_print('\n---------------- Preview (Beta) ----------------')
if not new_json:
error_print('\nSource configuration is empty. No changes will be made.')
return False
# perform diff operation
# to simplify output, add one shared key in src and dest configuration
new_json['@base'] = ''
old_json['@base'] = ''
differ = JsonDiffer(syntax='explicit')
res = differ.diff(old_json, new_json)
keys = str(res.keys())
if res == {} or (('update' not in keys) and ('insert' not in keys)):
error_print('\nTarget configuration already contains all key-values in source. No changes will be made.')
return False
# format result printing
for action, changes in res.items():
if action.label == 'delete':
continue # we do not delete KVs while importing/exporting
elif action.label == 'insert':
error_print('\nAdding:')
for key, adding in changes.items():
record = {'key': key}
for attribute, value in adding.items():
record[str(attribute)] = str(value)