from contextlib import contextmanager
import unittest
from time import time
-from unittest.mock import Mock, mock_open # patch, MagicMock
+from unittest.mock import Mock, mock_open
from osm_common.dbbase import DbException
from osm_nbi.engine import EngineException
from osm_common.dbmemory import DbMemory
from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
from osm_nbi.tests.test_db_descriptors import (
db_vim_accounts_text,
+ db_paas_accounts_text,
db_nsds_text,
db_vnfds_text,
db_nsrs_text,
self.vim = self.db.get_list("vim_accounts")[0]
self.vim_id = self.vim["_id"]
-
- def test_create_instantiate(self):
- self.db.set_one = Mock(return_value={"updated": 1})
- session = {
+ self.db.create_list(
+ "paas", yaml.load(db_paas_accounts_text, Loader=yaml.Loader)
+ )
+ self.paas_ids = [
+ paas_element["_id"] for paas_element in self.db.get_list("paas")
+ ]
+ self.paas_id = self.paas_ids[0]
+ self.vnfrs = {
+ vnfr["_id"]: vnfr["member-vnf-index-ref"]
+ for vnfr in self.db.get_list("vnfrs")
+ }
+ self.number_of_vnfr = len(self.vnfrs)
+ self.new_session = {
"force": False,
"admin": False,
"public": False,
"project_id": [self.nsr_project],
"method": "write",
}
+
+ def check_operation_params(self, operation_params):
+ self.assertEqual(self.nsd_id, operation_params["nsdId"])
+ self.assertEqual(self.nsr_id, operation_params["nsInstanceId"])
+ self.assertEqual("name", operation_params["nsName"])
+
+ def test_create_instantiate_with_vim_account(self):
+ self.db.set_one = Mock(return_value={"updated": 1})
indata = {
"nsdId": self.nsd_id,
"nsInstanceId": self.nsr_id,
headers = {}
nslcmop_id, _ = self.nslcmop_topic.new(
- rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers
+ rollback,
+ self.new_session,
+ indata=deepcopy(indata),
+ kwargs=None,
+ headers=headers,
)
# check nslcmop is created at database
"rollback mismatch with created/set items at database",
)
+ operation_params = created_nslcmop["operationParams"]
+ self.check_operation_params(operation_params)
+ self.assertEqual(self.vim_id, operation_params["vimAccountId"])
+ self.assertNotIn("paasAccountId", operation_params)
+
+ self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr)
+
+ for call in self.db.set_one.call_args_list:
+ topic, vnfr_id, update = call[0]
+ self.assertEqual(topic, "vnfrs")
+ self.assertIn(vnfr_id["_id"], self.vnfrs)
+ self.assertEqual(self.vim_id, update["vim-account-id"])
+ self.assertNotIn("paas-account-id", update)
+
# test parameters with error
bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
test_set = (
with self.assertRaises(expect_exc, msg=message) as e:
self.nslcmop_topic.new(
rollback,
- session,
+ self.new_session,
indata=deepcopy(indata),
kwargs=kwargs_,
headers=headers,
"Expected '{}' at exception text".format(expect_text),
)
+ def test_create_instantiate_with_paas_account(self):
+ self.db.set_one = Mock(return_value={"updated": 1})
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsInstanceId": self.nsr_id,
+ "nsName": "name",
+ "paasAccountId": self.paas_id,
+ "lcmOperationType": "instantiate",
+ }
+
+ nslcmop_id, _ = self.nslcmop_topic.new(
+ [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
+ )
+
+ # check nslcmop is created at database
+ self.assertEqual(
+ self.db.create.call_count,
+ 1,
+ "database create not called, or called more than once",
+ )
+ _call = self.db.create.call_args_list[0]
+ self.assertEqual(
+ _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
+ )
+
+ created_nslcmop = _call[0][1]
+ self.assertEqual(
+ nslcmop_id,
+ created_nslcmop["_id"],
+ "mismatch between return id and database '_id'",
+ )
+
+ operation_params = created_nslcmop["operationParams"]
+ self.check_operation_params(operation_params)
+ self.assertEqual(self.paas_id, operation_params["paasAccountId"])
+ self.assertNotIn("vimAccountId", operation_params)
+
+ self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr)
+
+ for call in self.db.set_one.call_args_list:
+ topic, vnfr_id, update = call[0]
+ self.assertEqual(topic, "vnfrs")
+ self.assertIn(vnfr_id["_id"], self.vnfrs)
+ self.assertEqual(self.paas_id, update["paas-account-id"])
+ self.assertNotIn("vim-account-id", update)
+
+ def test_create_instantiate_invalid_paas_account_raises_exception(self):
+ self.db.set_one = Mock(return_value={"updated": 1})
+ invalid_paas_id = "88d90b0c-faff-4b9f-bccd-017f33985984"
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsInstanceId": self.nsr_id,
+ "nsName": "name",
+ "paasAccountId": invalid_paas_id,
+ "lcmOperationType": "instantiate",
+ }
+
+ with self.assertRaises(EngineException):
+ nslcmop_id, _ = self.nslcmop_topic.new(
+ [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
+ )
+ self.db.set_one.assert_not_called()
+
+ def test_create_instantiate_with_paas_account_in_vnf(self):
+ self.db.set_one = Mock(return_value={"updated": 1})
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsInstanceId": self.nsr_id,
+ "nsName": "name",
+ "paasAccountId": self.paas_ids[0],
+ "lcmOperationType": "instantiate",
+ "vnf": [{"member-vnf-index": "1", "paasAccountId": self.paas_ids[1]}],
+ }
+
+ nslcmop_id, _ = self.nslcmop_topic.new(
+ [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
+ )
+
+ # check nslcmop is created at database
+ self.assertEqual(
+ self.db.create.call_count,
+ 1,
+ "database create not called, or called more than once",
+ )
+ _call = self.db.create.call_args_list[0]
+ self.assertEqual(
+ _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
+ )
+
+ created_nslcmop = _call[0][1]
+ self.assertEqual(
+ nslcmop_id,
+ created_nslcmop["_id"],
+ "mismatch between return id and database '_id'",
+ )
+
+ operation_params = created_nslcmop["operationParams"]
+ self.check_operation_params(operation_params)
+ self.assertEqual(self.paas_id, operation_params["paasAccountId"])
+ self.assertNotIn("vimAccountId", operation_params)
+ expected_paas_id = ""
+
+ self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr)
+ for call in self.db.set_one.call_args_list:
+ topic, vnfr_id, update = call[0]
+ self.assertEqual(topic, "vnfrs")
+ self.assertIn(vnfr_id["_id"], self.vnfrs)
+ vnf_index_ref = self.vnfrs[vnfr_id["_id"]]
+ if vnf_index_ref == "1":
+ expected_paas_id = self.paas_ids[1]
+ elif vnf_index_ref == "2":
+ expected_paas_id = self.paas_ids[0]
+ else:
+ assert False
+ self.assertEqual(expected_paas_id, update["paas-account-id"])
+ self.assertNotIn("vim-account-id", update)
+
+ def test_create_instantiate_invalid_paas_account_in_vnf_raises_exception(self):
+ self.db.set_one = Mock(return_value={"updated": 1})
+ invalid_paas_id = "88d90b0c-faff-4b9f-bccd-017f33985984"
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsInstanceId": self.nsr_id,
+ "nsName": "name",
+ "paasAccountId": self.paas_ids[0],
+ "lcmOperationType": "instantiate",
+ "vnf": [{"member-vnf-index": "1", "paasAccountId": invalid_paas_id}],
+ }
+
+ with self.assertRaises(EngineException):
+ self.nslcmop_topic.new(
+ [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
+ )
+ self.db.set_one.assert_not_called()
+
def test_check_ns_operation_action(self):
nsrs = self.db.get_list("nsrs")[0]
session = {}
vnfr_id = self.db.get_list("vnfrs")[0]["_id"]
session = {}
self.db.set_one(
- "nsrs",
- {"_id": self.nsr_id},
- {"_admin.nsState": "INSTANTIATED"},
+ "nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"}
)
indata = {
"lcmOperationType": "update",
"updateType": "REMOVE_VNF",
"nsInstanceId": self.nsr_id,
- "removeVnfInstanceId": vnfr_id
+ "removeVnfInstanceId": vnfr_id,
}
session = {
)
def test_migrate(self):
- vnfr_id = self.db.get_list("vnfrs")[0]["_id"]
session = {}
self.db.set_one(
- "nsrs",
- {"_id": self.nsr_id},
- {"_admin.nsState": "INSTANTIATED"},
+ "nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"}
)
session = {
"force": False,
indata = {
"lcmOperationType": "migrate",
"nsInstanceId": self.nsr_id,
- "migrateToHost":"sample02",
- "vdu": {
- "vduCountIndex": 0,
- "vduId": "mgmtVM"
- },
- "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
+ "migrateToHost": "sample02",
+ "vdu": {"vduCountIndex": 0, "vduId": "mgmtVM"},
+ "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
}
nslcmop_id, _ = self.nslcmop_topic.new(
rollback, session, indata, kwargs=None, headers=headers
indata = {
"lcmOperationType": "migrate",
"nsInstanceId": self.nsr_id,
- "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
+ "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
}
nslcmop_id, _ = self.nslcmop_topic.new(
rollback, session, indata, kwargs=None, headers=headers
indata = {
"lcmOperationType": "migrate",
"nsInstanceId": self.nsr_id,
- "migrateToHost":"sample02",
- "vdu": {
- "vduCountIndex": 0
- },
- "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
+ "migrateToHost": "sample02",
+ "vdu": {"vduCountIndex": 0},
+ "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
}
with self.assertRaises(Exception) as e:
nslcmop_id, _ = self.nslcmop_topic.new(
- rollback, session, indata, kwargs=None, headers=headers
+ rollback, session, indata, kwargs=None, headers=headers
+ )
+ self.assertTrue(
+ "Format error at 'vdu' ''vduId' is a required property'"
+ in str(e.exception)
)
- self.assertTrue("Format error at 'vdu' ''vduId' is a required property'" in str(e.exception))
class TestNsLcmOpTopicWithMock(unittest.TestCase):
test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
self.db.get_one.side_effect = [test_vnfr, test_vnfd]
- vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
- self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
- self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds', "Incorrect second DB lookup")
+ self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"])
+ self.assertEqual(
+ self.db.get_one.call_args_list[0][0][0],
+ "vnfrs",
+ "Incorrect first DB lookup",
+ )
+ self.assertEqual(
+ self.db.get_one.call_args_list[1][0][0],
+ "vnfds",
+ "Incorrect second DB lookup",
+ )
def test_get_vnfd_from_vnf_member_no_revision(self):
test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
- test_vnfr['revision'] = 3
+ test_vnfr["revision"] = 3
test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
self.db.get_one.side_effect = [test_vnfr, test_vnfd]
- vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
- self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
- self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup")
+ self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"])
+ self.assertEqual(
+ self.db.get_one.call_args_list[0][0][0],
+ "vnfrs",
+ "Incorrect first DB lookup",
+ )
+ self.assertEqual(
+ self.db.get_one.call_args_list[1][0][0],
+ "vnfds_revisions",
+ "Incorrect second DB lookup",
+ )
@contextmanager
def assertNotRaises(self, exception_type):
"VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
)
- with self.subTest(i=5, t="Ns update REMOVE_VNF request validated with no exception"):
+ 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[0]["revision"] = 2
test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
with self.assertNotRaises(EngineException):
self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+
class TestNsrTopic(unittest.TestCase):
def setUp(self):
self.db = DbMemory()
self.vim = self.db.get_list("vim_accounts")[0]
self.vim_id = self.vim["_id"]
- def test_create(self):
- session = {
- "force": False,
- "admin": False,
- "public": False,
- "project_id": [self.nsd_project],
- "method": "write",
- }
- indata = {
- "nsdId": self.nsd_id,
- "nsName": "name",
- "vimAccountId": self.vim_id,
- "additionalParamsForVnf": [
- {
- "member-vnf-index": "hackfest_vnf1",
- "additionalParams": {"touch_filename": "file"},
- },
- {
- "member-vnf-index": "hackfest_vnf2",
- "additionalParams": {"touch_filename": "file"},
- },
- ],
- }
- rollback = []
- headers = {}
-
- self.nsr_topic.new(
- rollback, session, indata=indata, kwargs=None, headers=headers
- )
+ self.paas_id = "3d93ca4e-b007-4d94-bbdc-61911078b864"
- # check vnfrs and nsrs created in whatever order
- created_vnfrs = []
- created_nsrs = []
- nsr_id = None
+ def stock_db_creations(self, created_vnfrs, created_nsrs):
for _call in self.db.create.call_args_list:
assert len(_call[0]) >= 2, "called db.create with few parameters"
created_item = _call[0][1]
if _call[0][0] == "vnfrs":
created_vnfrs.append(created_item)
- self.assertIn(
- "member-vnf-index-ref",
- created_item,
- "Created item must contain member-vnf-index-ref section",
- )
- if nsr_id:
- self.assertEqual(
- nsr_id,
- created_item["nsr-id-ref"],
- "bad reference id from vnfr to nsr",
- )
- else:
- nsr_id = created_item["nsr-id-ref"]
-
elif _call[0][0] == "nsrs":
created_nsrs.append(created_item)
- if nsr_id:
- self.assertEqual(
- nsr_id, created_item["_id"], "bad reference id from vnfr to nsr"
- )
- else:
- nsr_id = created_item["_id"]
else:
- assert True, "created an unknown record {} at database".format(
+ assert False, "created an unknown record {} at database".format(
_call[0][0]
)
+ self.check_admin_field_in_db_call(created_item)
- self.assertTrue(
- created_item["_admin"].get("projects_read"),
- "Database record must contain '_amdin.projects_read'",
- )
- self.assertIn(
- "created",
- 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'",
- )
+ def check_admin_field_in_db_call(self, created_item):
+ self.assertTrue(
+ created_item["_admin"].get("projects_read"),
+ "Database record must contain '_amdin.projects_read'",
+ )
+ self.assertIn(
+ "created",
+ 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'",
+ )
+ def check_vnfr_content(self, vnfr, nsr_id):
+ self.assertEqual(vnfr["vim-account-id"], None)
+ self.assertEqual(vnfr["paas-account-id"], None)
+ self.assertIn(
+ "member-vnf-index-ref",
+ vnfr,
+ "Created item must contain member-vnf-index-ref section",
+ )
self.assertEqual(
- len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
+ nsr_id, vnfr["nsr-id-ref"], "bad reference id from vnfr to nsr"
)
+ self.check_vdur_interfaces(vnfr)
+ def check_vdur_interfaces(self, vnfr):
self.assertEqual(
- created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"],
+ vnfr["vdur"][0]["interfaces"][0]["position"],
1,
"vdur first interface position does not match",
)
-
self.assertEqual(
- created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"],
+ vnfr["vdur"][0]["interfaces"][1]["position"],
2,
"vdur second interface position does not match",
)
+ def check_vnfrs(self, created_vnfrs, nsrs_id):
+ self.assertEqual(
+ len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
+ )
+ for vnfr in created_vnfrs:
+ self.check_vnfr_content(vnfr, nsrs_id)
+
+ def check_vnfrs_ref_in_nsr(self, nsrs, created_vnfrs):
+ self.assertEqual(len(nsrs["constituent-vnfr-ref"]), len(created_vnfrs))
+ self.assertNotEqual(created_vnfrs[0]["_id"], created_vnfrs[1]["_id"])
+ for vnfr in created_vnfrs:
+ vnfr_id = vnfr["_id"]
+ self.assertIn(vnfr_id, nsrs["constituent-vnfr-ref"])
+
+ def check_nsrs(self, created_nsrs, created_vnfrs, vim_id, paas_id):
self.assertEqual(
len(created_nsrs), 1, "Only one nsrs must be created at database"
)
+ nsrs = created_nsrs[0]
+ self.assertEqual(nsrs["vimdatacenter"], vim_id)
+ self.assertEqual(nsrs["paasdatacenter"], paas_id)
+ self.check_vnfrs_ref_in_nsr(nsrs, created_vnfrs)
+
+ def check_created_nsrs_and_vnfrs(
+ self, created_nsrs, created_vnfrs, rollback, vim_id, paas_id
+ ):
+
+ nsrs_id = created_nsrs[0]["_id"]
+
self.assertEqual(
len(rollback),
len(created_vnfrs) + 1,
"rollback mismatch with created items at database",
)
+ self.check_nsrs(created_nsrs, created_vnfrs, vim_id, paas_id)
+ self.check_vnfrs(created_vnfrs, nsrs_id)
+
+ def test_create_with_vim_account(self):
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsd_project],
+ "method": "write",
+ }
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsName": "name",
+ "vimAccountId": self.vim_id,
+ "additionalParamsForVnf": [
+ {
+ "member-vnf-index": "hackfest_vnf1",
+ "additionalParams": {"touch_filename": "file"},
+ },
+ {
+ "member-vnf-index": "hackfest_vnf2",
+ "additionalParams": {"touch_filename": "file"},
+ },
+ ],
+ }
+ rollback = []
+ headers = {}
+
+ self.nsr_topic.new(
+ rollback, session, indata=indata, kwargs=None, headers=headers
+ )
+
+ # check vnfrs and nsrs created in whatever order
+ created_vnfrs = []
+ created_nsrs = []
+ self.stock_db_creations(created_vnfrs, created_nsrs)
+ self.check_created_nsrs_and_vnfrs(
+ created_nsrs, created_vnfrs, rollback, self.vim_id, None
+ )
+
+ def test_create_with_vim_account_raise_exception(self):
# test parameters with error
bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsd_project],
+ "method": "write",
+ }
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsName": "name",
+ "vimAccountId": self.vim_id,
+ "additionalParamsForVnf": [
+ {
+ "member-vnf-index": "hackfest_vnf1",
+ "additionalParams": {"touch_filename": "file"},
+ },
+ {
+ "member-vnf-index": "hackfest_vnf2",
+ "additionalParams": {"touch_filename": "file"},
+ },
+ ],
+ }
+ rollback = []
+ headers = {}
test_set = (
# TODO add "nsd"
(
self.assertTrue(e.exception.http_code == expect_code)
if expect_text_list:
for expect_text in expect_text_list:
- self.assertIn(expect_text, str(e.exception).lower(),
- "Expected '{}' at exception text".format(expect_text))
+ self.assertIn(
+ expect_text,
+ str(e.exception).lower(),
+ "Expected '{}' at exception text".format(expect_text),
+ )
+
+ def test_create_with_paas_account(self):
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsd_project],
+ "method": "write",
+ }
+ indata = {
+ "nsdId": self.nsd_id,
+ "nsName": "name",
+ "paasAccountId": self.paas_id,
+ "additionalParamsForVnf": [
+ {
+ "member-vnf-index": "hackfest_vnf1",
+ "additionalParams": {"touch_filename": "file"},
+ },
+ {
+ "member-vnf-index": "hackfest_vnf2",
+ "additionalParams": {"touch_filename": "file"},
+ },
+ ],
+ }
+ rollback = []
+ headers = {}
+
+ self.nsr_topic.new(
+ rollback, session, indata=indata, kwargs=None, headers=headers
+ )
+
+ created_vnfrs = []
+ created_nsrs = []
+ self.stock_db_creations(created_vnfrs, created_nsrs)
+ self.check_created_nsrs_and_vnfrs(
+ created_nsrs, created_vnfrs, rollback, None, self.paas_id
+ )
def test_show_instance(self):
- session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsd_project],
+ "method": "write",
+ }
filter_q = {}
for refresh_status in ("true", "false"):
self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
actual_nsr = self.db.get_list("nsrs")[0]
nsr_id = actual_nsr["_id"]
- filter_q['vcaStatus-refresh'] = refresh_status
+ filter_q["vcaStatus-refresh"] = refresh_status
expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q)
self.nsr_topic.delete(session, nsr_id)
actual_nsr.pop("_admin")
expected_nsr.pop("_admin")
- self.assertEqual(expected_nsr, actual_nsr, "Database nsr and show() nsr do not match.")
+ self.assertEqual(
+ expected_nsr, actual_nsr, "Database nsr and show() nsr do not match."
+ )
def test_vca_status_refresh(self):
- session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
- filter_q = {'vcaStatus-refresh': 'true'}
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsd_project],
+ "method": "write",
+ }
+ filter_q = {"vcaStatus-refresh": "true"}
time_delta = 120
self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
nsr = self.db.get_list("nsrs")[0]
# When vcaStatus-refresh is true
- filter_q['vcaStatus-refresh'] = "true"
+ filter_q["vcaStatus-refresh"] = "true"
self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
msg_args = self.msg.write.call_args[0]
self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
self.assertGreater(nsr["_admin"]["modified"], time() - time_delta)
# When vcaStatus-refresh is false but modified time is within threshold
- filter_q['vcaStatus-refresh'] = "false"
+ filter_q["vcaStatus-refresh"] = "false"
time_now = time()
nsr["_admin"]["modified"] = time_now
self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
msg_args = self.msg.write.call_args[1]
self.assertEqual(msg_args, {}, "Message should not be sent.")
- self.assertEqual(nsr["_admin"]["modified"], time_now, "Modified time should not be changed.")
+ self.assertEqual(
+ nsr["_admin"]["modified"], time_now, "Modified time should not be changed."
+ )
# When vcaStatus-refresh is false but modified time is less than threshold
- filter_q['vcaStatus-refresh'] = "false"
- nsr["_admin"]["modified"] = time() - (2*time_delta)
+ filter_q["vcaStatus-refresh"] = "false"
+ nsr["_admin"]["modified"] = time() - (2 * time_delta)
self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
msg_args = self.msg.write.call_args[0]
self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
self.nsr_topic.delete(session, nsr["_id"])
- self.assertGreater(nsr["_admin"]["modified"], time() - time_delta, "Modified time is not changed.")
+ self.assertGreater(
+ nsr["_admin"]["modified"],
+ time() - time_delta,
+ "Modified time is not changed.",
+ )
def test_delete_ns(self):
self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
projects / osm / NBI.git / commitdiff