X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_instance_topics.py;h=a883bdc299b679162455ab7e9b3367e0ef211e84;hb=refs%2Fchanges%2F51%2F13451%2F1;hp=0d4adec3d32aa3434cc3251b7e2498a9fe3ed35e;hpb=c54e3be9ab71ec448b99c000adda9f8f2f60cabd;p=osm%2FNBI.git diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py index 0d4adec3..a883bdc2 100644 --- a/osm_nbi/tests/test_instance_topics.py +++ b/osm_nbi/tests/test_instance_topics.py @@ -49,13 +49,11 @@ class TestNsLcmOpTopic(unittest.TestCase): self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) self.nslcmop_topic.check_quota = Mock(return_value=None) # skip quota - self.db.create_list( - "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader) - ) - self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader)) - self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader)) - self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader)) - self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) + self.db.create_list("vim_accounts", yaml.safe_load(db_vim_accounts_text)) + self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) + self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) + self.db.create_list("vnfrs", yaml.safe_load(db_vnfrs_text)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) self.db.create = Mock(return_value="created_id") self.nsd = self.db.get_list("nsds")[0] self.nsd_id = self.nsd["_id"] @@ -389,8 +387,8 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) def test_get_vnfd_from_vnf_member_revision(self): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0] - test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text)[0] + test_vnfd = yaml.safe_load(db_vnfds_text) self.db.get_one.side_effect = [test_vnfr, test_vnfd] _ = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"]) self.assertEqual( @@ -405,9 +403,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): ) def test_get_vnfd_from_vnf_member_no_revision(self): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0] + test_vnfr = yaml.safe_load(db_vnfrs_text)[0] test_vnfr["revision"] = 3 - test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader) + test_vnfd = yaml.safe_load(db_vnfds_text) self.db.get_one.side_effect = [test_vnfr, test_vnfd] _ = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"]) self.assertEqual( @@ -434,9 +432,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): session = {} with self.subTest(i=1, t="VNF instance does not belong to NS"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) test_nsr[0]["constituent-vnfr-ref"][ 0 ] = "99d90b0c-faff-4b9f-bccd-017f33985984" @@ -460,9 +458,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): ) with self.subTest(i=2, t="Ns update request validated with no exception"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) self.db.create_list("vnfrs", test_vnfr) self.db.create_list("nsrs", test_nsr) nsrs = self.db.get_list("nsrs")[1] @@ -498,9 +496,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): ) with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) self.db.create_list("vnfrs", test_vnfr) self.db.create_list("nsrs", test_nsr) nsrs = self.db.get_list("nsrs")[2] @@ -524,9 +522,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): with self.subTest( i=5, t="Ns update REMOVE_VNF request validated with no exception" ): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) self.db.create_list("vnfrs", test_vnfr) self.db.create_list("nsrs", test_nsr) nsrs = self.db.get_list("nsrs")[1] @@ -550,11 +548,9 @@ class TestNsrTopic(unittest.TestCase): self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None) self.nsr_topic.check_quota = Mock(return_value=None) # skip quota - self.db.create_list( - "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader) - ) - self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader)) - self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader)) + self.db.create_list("vim_accounts", yaml.safe_load(db_vim_accounts_text)) + self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) + self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) self.db.create = Mock(return_value="created_id") self.nsd = self.db.get_list("nsds")[0] self.nsd_id = self.nsd["_id"] @@ -615,6 +611,10 @@ class TestNsrTopic(unittest.TestCase): ) else: nsr_id = created_item["nsr-id-ref"] + self.assertTrue( + "nsState" not in created_item["_admin"], + "Database VNF record must not contain '_admin.nsState'", + ) elif _call[0][0] == "nsrs": created_nsrs.append(created_item) @@ -624,6 +624,11 @@ class TestNsrTopic(unittest.TestCase): ) else: nsr_id = created_item["_id"] + + self.assertTrue( + created_item["_admin"]["nsState"] == "NOT_INSTANTIATED", + "Database record must contain '_admin.nsState=NOT INSTANTIATE'", + ) else: assert True, "created an unknown record {} at database".format( _call[0][0] @@ -638,10 +643,6 @@ class TestNsrTopic(unittest.TestCase): created_item["_admin"], "Database record must contain '_admin.created'", ) - self.assertTrue( - created_item["_admin"]["nsState"] == "NOT_INSTANTIATED", - "Database record must contain '_admin.nstate=NOT INSTANTIATE'", - ) self.assertEqual( len(created_vnfrs), 2, "created a mismatch number of vnfr at database" @@ -717,7 +718,7 @@ class TestNsrTopic(unittest.TestCase): } filter_q = {} for refresh_status in ("true", "false"): - self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) actual_nsr = self.db.get_list("nsrs")[0] nsr_id = actual_nsr["_id"] filter_q["vcaStatus-refresh"] = refresh_status @@ -739,7 +740,7 @@ class TestNsrTopic(unittest.TestCase): } filter_q = {"vcaStatus-refresh": "true"} time_delta = 120 - self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) nsr = self.db.get_list("nsrs")[0] # When vcaStatus-refresh is true @@ -774,7 +775,7 @@ class TestNsrTopic(unittest.TestCase): ) def test_delete_ns(self): - self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) self.nsr = self.db.get_list("nsrs")[0] self.nsr_id = self.nsr["_id"] self.db_set_one = self.db.set_one