from unittest import TestCase from nested_lookup import nested_lookup, nested_update from nested_lookup import nested_delete, nested_alter class BaseLookUpApi(TestCase): def setUp(self): self.sample_data1 = { "build_version": { "model_name": "MacBook Pro", "build_version": { "processor_name": "Intel Core i7", "processor_speed": "2.7 GHz", "core_details": { "build_version": "4", "l2_cache(per_core)": "256 KB", }, }, "number_of_cores": "4", "memory": "256 KB", }, "os_details": {"product_version": "10.13.6", "build_version": "17G65"}, "name": "Test", "date": "YYYY-MM-DD HH:MM:SS", } self.sample_data2 = { "hardware_details": { "model_name": "MacBook Pro", "processor_details": [ {"processor_name": "Intel Core i7", "processor_speed": "2.7 GHz"}, {"total_number_of_cores": "4", "l2_cache(per_core)": "256 KB"}, ], "total_number_of_cores": "5", "memory": "16 GB", } } self.sample_data3 = { "values": [ { "checks": [ { "monitoring_zones": [ "mzdfw", "mzfra", "mzhkg", "mziad", "mzlon", "mzord", "mzsyd", ] } ] } ] } self.sample_data4 = { "modelversion": "1.1.0", "vorgangsID": "1", "versorgungsvorschlagDatum": 1510558834978, "eingangsdatum": 1510558834978, "plz": 82269, "vertragsteile": [ { "typ": "1", "beitragsDaten": { "endalter": 85, "brutto": 58.76, "netto": 58.76, "zahlungsrhythmus": "MONATLICH", "plz": 86899, }, "beginn": 1512082800000, "lebenslang": "True", "ueberschussverwendung": { "ueberschussverwendung": "2", "indexoption": "3", }, "deckung": [ { "typ": "2", "art": "1", "leistung": {"value": 7500242424.0, "einheit": "2"}, "leistungsRhythmus": "1", } ], "zuschlagNachlass": [], }, { "typ": "1", "beitragsDaten": { "endalter": 85, "brutto": 0.6, "netto": 0.6, "zahlungsrhythmus": "1", }, "zuschlagNachlass": [], }, ], } class TestNestedDelete(BaseLookUpApi): def test_sample_data1(self): result = { "os_details": {"product_version": "10.13.6"}, "name": "Test", "date": "YYYY-MM-DD HH:MM:SS", } self.assertEqual(result, nested_delete(self.sample_data1, "build_version")) def test_sample_data2(self): result = { "hardware_details": { "model_name": "MacBook Pro", "total_number_of_cores": "5", "memory": "16 GB", } } self.assertEqual(result, nested_delete(self.sample_data2, "processor_details")) def test_sample_data3(self): result = {"values": [{"checks": [{}]}]} self.assertEqual(result, nested_delete(self.sample_data3, "monitoring_zones")) class TestNestedUpdate(BaseLookUpApi): def test_sample_data1(self): result = { "build_version": "Test1", "os_details": {"product_version": "10.13.6", "build_version": "Test1"}, "name": "Test", "date": "YYYY-MM-DD HH:MM:SS", } self.assertEqual( result, nested_update(self.sample_data1, "build_version", "Test1") ) def test_sample_data1_list_input_treat_list_as_element_true(self): result = { "build_version": ["Test5", "Test6", "Test7"], "os_details": { "product_version": "10.13.6", "build_version": ["Test5", "Test6", "Test7"], }, "name": "Test", "date": "YYYY-MM-DD HH:MM:SS", } self.assertEqual( result, nested_update( self.sample_data1, "build_version", ["Test5", "Test6", "Test7"], treat_as_element=True, ), ) def test_nested_update_in_place_false(self): """ ested_update should mutate and return a copy of the original document """ before_id = id(self.sample_data1) result = nested_update( self.sample_data1, "build_version", "Test2", in_place=False ) after_id = id(result) # the object ids should _not_ match. self.assertNotEqual(before_id, after_id) def test_nested_update_in_place_true(self): """ nested_update should mutate and return the original document """ before_id = id(self.sample_data1) result = nested_update( self.sample_data1, "build_version", "Test2", in_place=True ) after_id = id(result) # the object ids should match. self.assertEqual(before_id, after_id) def test_nested_update_in_place_true_list_input(self): doc = self.sample_data4 # get all instances of the given element findings = nested_lookup("plz", doc, False, True) # alter those instances updated_findings = list() for key, val in findings.items(): for elem in val: updated_findings.append(elem + 300) # update those instances with the altered results doc_updated = nested_update( doc, "plz", updated_findings, treat_as_element=False ) elem1 = doc_updated["plz"] # 85269 # 87199 elem2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] self.assertEqual(elem1, 82569) self.assertEqual(elem2, 87199) def test_nested_update_in_place_false_list_input(self): doc = self.sample_data4 # get all instances of the given element findings = nested_lookup("plz", doc, False, True) # alter those instances updated_findings = list() for key, val in findings.items(): for elem in val: updated_findings.append(elem + 300) # update those instances with the altered results doc_updated = nested_update( doc, "plz", updated_findings, in_place=False, treat_as_element=False ) elem1 = doc_updated["plz"] elem2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] self.assertEqual(elem1, 82569) self.assertEqual(elem2, 87199) def test_nested_update_in_place_false_list_input_as_element_false(self): doc = self.sample_data4 # get all instances of the given element list_input = [1, 2, 3, 4, 5] # update those instances with the altered results doc_updated = nested_update( doc, "plz", list_input, in_place=False, treat_as_element=False ) elem1 = doc_updated["plz"] elem2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] # should not work without specifying "treat_list_as_element = True" # in nested_update self.assertNotEqual(elem1, list_input) self.assertNotEqual(elem2, list_input) def test_nested_update_in_place_false_list_input_as_element_true(self): doc = self.sample_data4 # get all instances of the given element list_input = [1, 2, 3, 4, 5] # update those instances with the altered results doc_updated = nested_update( doc, "plz", list_input, in_place=False, treat_as_element=True ) elem1 = doc_updated["plz"] elem2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] # should not work without specifying "treat_list_as_element = True" # in nested_update self.assertEqual(elem1, list_input) self.assertEqual(elem2, list_input) def test_nested_update_in_place_true_list_input_as_element_false(self): doc = self.sample_data4 # get all instances of the given element list_input = [1, 2, 3, 4, 5] # update those instances with the altered results doc_updated = nested_update( doc, "plz", list_input, in_place=True, treat_as_element=False ) elem1 = doc_updated["plz"] elem2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] # should not work without specifying "treat_list_as_element = True" # in nested_update self.assertNotEqual(elem1, list_input) self.assertNotEqual(elem2, list_input) def test_nested_update_in_place_true_list_input_as_element_true(self): doc = self.sample_data4 # get all instances of the given element list_input = [1, 2, 3, 4, 5] # update those instances with the altered results doc_updated = nested_update( doc, "plz", list_input, in_place=True, treat_as_element=True ) elem1 = doc_updated["plz"] elem2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] # should not work without specifying "treat_list_as_element = True" # in nested_update self.assertEqual(elem1, list_input) self.assertEqual(elem2, list_input) def test_nested_delete_in_place_false(self): """ nested_delete should mutate and return a copy of the original document """ before_id = id(self.sample_data1) result = nested_delete(self.sample_data1, "build_version", in_place=False) after_id = id(result) # the object ids should _not_ match. self.assertNotEqual(before_id, after_id) def test_nested_delete_in_place_true(self): """nested_delete should mutate and return the original document""" before_id = id(self.sample_data1) result = nested_delete(self.sample_data1, "build_version", in_place=True) after_id = id(result) # the object ids should match. self.assertEqual(before_id, after_id) def test_nested_update_taco_for_example(self): document = [{"taco": 42}, {"salsa": [{"burrito": {"taco": 69}}]}] updated_document = nested_update( document, "taco", [100, 200], treat_as_element=False ) self.assertEqual(updated_document[0]["taco"], 100) # The multi-update version only works for scalar input, # if you need to adress a list of dicts, you have to # manually iterate over those and pass them to nested_update # one by one self.assertEqual(updated_document[1]["salsa"][0]["burrito"]["taco"], 200) def test_nested_update_raise_error(self): doc = self.sample_data4 # get all instances of the given element list_input = 1 # update those instances with the altered results self.assertRaises( Exception, nested_update, doc, "plz", list_input, in_place=True, treat_as_element=False, ) def test_sample_data2(self): result = { "hardware_details": { "model_name": "MacBook Pro", "processor_details": {"test_key1": "test_value1"}, "total_number_of_cores": "5", "memory": "16 GB", } } self.assertEqual( result, nested_update( self.sample_data2, "processor_details", {"test_key1": "test_value1"} ), ) def test_sample_data3(self): result = {"values": [{"checks": {"key1": ["value1"], "key2": "value2"}}]} self.assertEqual( result, nested_update( self.sample_data3, "checks", {"key1": ["value1"], "key2": "value2"} ), ) def test_sample_data4(self): result = { "modelversion": {"key1": ["value1"], "key2": "value2"}, "vorgangsID": "1", "versorgungsvorschlagDatum": 1510558834978, "eingangsdatum": 1510558834978, "plz": 82269, "vertragsteile": [ { "typ": "1", "beitragsDaten": { "endalter": 85, "brutto": 58.76, "netto": 58.76, "zahlungsrhythmus": "MONATLICH", "plz": 86899, }, "beginn": 1512082800000, "lebenslang": "True", "ueberschussverwendung": { "ueberschussverwendung": "2", "indexoption": "3", }, "deckung": [ { "typ": "2", "art": "1", "leistung": {"value": 7500242424.0, "einheit": "2"}, "leistungsRhythmus": "1", } ], "zuschlagNachlass": [], }, { "typ": "1", "beitragsDaten": { "endalter": 85, "brutto": 0.6, "netto": 0.6, "zahlungsrhythmus": "1", }, "zuschlagNachlass": [], }, ], } self.assertEqual( result, nested_update( self.sample_data4, "modelversion", {"key1": ["value1"], "key2": "value2"}, ), ) class TestNestedAlter(BaseLookUpApi): def test_nested_alter_in_place_true(self): # callback functions def callback(data): return str(data) + "###" doc_updated = nested_alter( self.sample_data4, "vorgangsID", callback, in_place=True ) vorgangsid = doc_updated["vorgangsID"] self.assertEqual(vorgangsid, "1###") def test_nested_alter_in_place_false(self): # callback functions def callback(data): return str(data) + "###" doc_updated = nested_alter( self.sample_data4, "vorgangsID", callback, in_place=False ) vorgangsid = doc_updated["vorgangsID"] # should not work without specifying # "treat_list_as_element = True" in nested_update self.assertEqual(vorgangsid, "1###") def test_nested_alter_list_input_in_place_true(self): # callback functions def callback(data): return str(data) + "###" doc_updated = nested_alter( self.sample_data4, ["plz", "vorgangsID"], callback, in_place=True ) plz1 = doc_updated["plz"] plz2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] vorgangsid = doc_updated["vorgangsID"] # should not work without specifying # "treat_list_as_element = True" in nested_update self.assertEqual(plz1, "82269###") self.assertEqual(plz2, "86899###") self.assertEqual(vorgangsid, "1###") def test_nested_alter_list_input_with_args_in_place_true(self): # callback functions def callback(data, str1, str2): return str(data) + str1 + str2 doc_updated = nested_alter( self.sample_data4, ["plz", "vorgangsID"], callback, function_parameters=["abc", "def"], in_place=True, ) plz1 = doc_updated["plz"] plz2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] vorgangsid = doc_updated["vorgangsID"] # should not work without specifying # "treat_list_as_element = True" in nested_update self.assertEqual(plz1, "82269abcdef") self.assertEqual(plz2, "86899abcdef") self.assertEqual(vorgangsid, "1abcdef") def test_nested_alter_list_input_with_args_in_place_false(self): # callback functions def callback(data, str1, str2): return str(data) + str1 + str2 doc_updated = nested_alter( self.sample_data4, ["plz", "vorgangsID"], callback, function_parameters=["abc", "def"], in_place=False, ) plz1 = doc_updated["plz"] plz2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] vorgangsid = doc_updated["vorgangsID"] # should not work without specifying # "treat_list_as_element = True" in nested_update self.assertEqual(plz1, "82269abcdef") self.assertEqual(plz2, "86899abcdef") self.assertEqual(vorgangsid, "1abcdef") def test_nested_alter_list_input_in_place_false(self): # callback functions def callback(data): return str(data) + "###" doc_updated = nested_alter( self.sample_data4, ["plz", "vorgangsID"], callback, in_place=False ) plz1 = doc_updated["plz"] plz2 = doc_updated["vertragsteile"][0]["beitragsDaten"]["plz"] vorgangsid = doc_updated["vorgangsID"] # should not work without specifying # "treat_list_as_element = True" in nested_update self.assertEqual(plz1, "82269###") self.assertEqual(plz2, "86899###") self.assertEqual(vorgangsid, "1###") def test_nested_alter_taco_for_example(self): documents = [{"taco": 42}, {"salsa": [{"burrito": {"taco": 69}}]}] # write a callback function which processes a scalar value. # Be aware about the possible types which can be passed to # the callback functions. # In this example we can be sure that only int will be passed, # in production you should check the type because it could be # anything. def callback(data): return data + 10 # (data/100*10) # The alter-version only works for scalar input (one dict), # if you need to adress a list of dicts, you have to # manually iterate over those and pass them to # nested_update one by one out = [] for elem in documents: altered_document = nested_alter(elem, "taco", callback) out.append(altered_document) self.maxDiff = None self.assertEqual(out[0]["taco"], 52) self.assertEqual(out[1]["salsa"][0]["burrito"]["taco"], 79) def test_nested_alter_work_with_right_order(self): document = {"taco": 42, "salsa": [{"burrito":{"key":20}}], "key":50} def callback(data): return data + 100 altered_document = nested_alter(document, "key", callback, in_place=True) self.assertEqual(altered_document["salsa"][0]["burrito"]["key"], 120) self.assertEqual(altered_document["key"], 150) def test_sample_data4(self): result = { "modelversion": "1.1.0", "vorgangsID": "1", "versorgungsvorschlagDatum": 1510558834978, "eingangsdatum": 1510558834978, "plz": 82270, "vertragsteile": [ { "typ": "1", "beitragsDaten": { "endalter": 85, "brutto": 58.76, "netto": 58.76, "zahlungsrhythmus": "MONATLICH", "plz": 86900, }, "beginn": 1512082800000, "lebenslang": "True", "ueberschussverwendung": { "ueberschussverwendung": "2", "indexoption": "3", }, "deckung": [ { "typ": "2", "art": "1", "leistung": {"value": 7500242424.0, "einheit": "2"}, "leistungsRhythmus": "1", } ], "zuschlagNachlass": [], }, { "typ": "1", "beitragsDaten": { "endalter": 85, "brutto": 0.6, "netto": 0.6, "zahlungsrhythmus": "1", }, "zuschlagNachlass": [], }, ], } # add +1 to all plz def callback(data): return data + 1 self.maxDiff = None self.assertEqual(result, nested_alter(self.sample_data4, "plz", callback))