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"]
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(
)
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(
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"
)
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]
)
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]
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]
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"]
}
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
}
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
)
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