# contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
##
+from contextlib import contextmanager
import unittest
+from time import time
from unittest.mock import Mock, mock_open # patch, MagicMock
from osm_common.dbbase import DbException
from osm_nbi.engine import EngineException
from osm_common.dbmemory import DbMemory
from osm_common.fsbase import FsBase
from osm_common.msgbase import MsgBase
+from osm_common import dbbase
from http import HTTPStatus
from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
from osm_nbi.tests.test_db_descriptors import (
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.db.set_one = Mock(return_value={"updated": 1})
self.nsd = self.db.get_list("nsds")[0]
self.nsd_id = self.nsd["_id"]
self.nsr = self.db.get_list("nsrs")[0]
self.vim_id = self.vim["_id"]
def test_create_instantiate(self):
+ self.db.set_one = Mock(return_value={"updated": 1})
session = {
"force": False,
"admin": False,
"Engine exception bad http_code with {}".format(indata_copy),
)
+ def test_update_remove_vnf(self):
+ vnfr_id = self.db.get_list("vnfrs")[0]["_id"]
+ session = {}
+ self.db.set_one(
+ "nsrs",
+ {"_id": self.nsr_id},
+ {"_admin.nsState": "INSTANTIATED"},
+ )
+ indata = {
+ "lcmOperationType": "update",
+ "updateType": "REMOVE_VNF",
+ "nsInstanceId": self.nsr_id,
+ "removeVnfInstanceId": vnfr_id,
+ }
+
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsr_project],
+ "method": "write",
+ }
+ rollback = []
+ headers = {}
+
+ nslcmop_id, _ = self.nslcmop_topic.new(
+ rollback, session, indata, kwargs=None, headers=headers
+ )
+
+ 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", "nslcmops entry must be created at database"
+ )
+ created_nslcmop = _call[0][1]
+ self.assertEqual(
+ self.nsr_id,
+ created_nslcmop["nsInstanceId"],
+ "mismatch between nsId '_id' in created nslcmop and database nsr",
+ )
+ self.assertTrue(
+ created_nslcmop["lcmOperationType"] == "update",
+ "Database record must contain 'lcmOperationType=update'",
+ )
+ self.assertTrue(
+ created_nslcmop["operationParams"]["updateType"] == "REMOVE_VNF",
+ "Database record must contain 'updateType=REMOVE_VNF'",
+ )
+
+ def test_migrate(self):
+ _ = self.db.get_list("vnfrs")[0]["_id"]
+ session = {}
+ self.db.set_one(
+ "nsrs",
+ {"_id": self.nsr_id},
+ {"_admin.nsState": "INSTANTIATED"},
+ )
+ session = {
+ "force": False,
+ "admin": False,
+ "public": False,
+ "project_id": [self.nsr_project],
+ "method": "write",
+ }
+ rollback = []
+ headers = {}
+
+ with self.subTest(i=1, t="Migration for Specific VM"):
+ indata = {
+ "lcmOperationType": "migrate",
+ "nsInstanceId": self.nsr_id,
+ "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
+ )
+
+ 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", "nslcmops entry must be created at database"
+ )
+ created_nslcmop = _call[0][1]
+ self.assertEqual(
+ self.nsr_id,
+ created_nslcmop["nsInstanceId"],
+ "mismatch between nsId '_id' in created nslcmop and database nsr",
+ )
+ self.assertTrue(
+ created_nslcmop["lcmOperationType"] == "migrate",
+ "Database record must contain 'lcmOperationType=migrate'",
+ )
+ with self.subTest(i=2, t="Migration of all VDUs in a VNF"):
+ indata = {
+ "lcmOperationType": "migrate",
+ "nsInstanceId": self.nsr_id,
+ "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
+ }
+ nslcmop_id, _ = self.nslcmop_topic.new(
+ rollback, session, indata, kwargs=None, headers=headers
+ )
+
+ self.assertEqual(
+ self.db.create.call_count,
+ 2,
+ "database create not called, or called more than once",
+ )
+ _call = self.db.create.call_args_list[0]
+ self.assertEqual(
+ _call[0][0], "nslcmops", "nslcmops entry must be created at database"
+ )
+ created_nslcmop = _call[0][1]
+ self.assertEqual(
+ self.nsr_id,
+ created_nslcmop["nsInstanceId"],
+ "mismatch between nsId '_id' in created nslcmop and database nsr",
+ )
+ self.assertTrue(
+ created_nslcmop["lcmOperationType"] == "migrate",
+ "Database record must contain 'lcmOperationType=migrate'",
+ )
+ with self.subTest(i=3, t="Migration failure - vduId not provided in vdu "):
+ indata = {
+ "lcmOperationType": "migrate",
+ "nsInstanceId": self.nsr_id,
+ "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
+ )
+ self.assertTrue(
+ "Format error at 'vdu' ''vduId' is a required property'"
+ in str(e.exception)
+ )
+
+
+class TestNsLcmOpTopicWithMock(unittest.TestCase):
+ def setUp(self):
+ self.db = Mock(dbbase.DbBase())
+ self.fs = Mock(FsBase())
+ self.fs.get_params.return_value = {"./fake/folder"}
+ self.fs.file_open = mock_open()
+ self.msg = Mock(MsgBase())
+ # create class
+ self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
+
+ def test_get_vnfd_from_vnf_member_revision(self):
+ 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(
+ 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.safe_load(db_vnfrs_text)[0]
+ test_vnfr["revision"] = 3
+ 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(
+ 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):
+ try:
+ yield None
+ except exception_type:
+ raise self.failureException("{} raised".format(exception_type.__name__))
+
+ def test_check_ns_update_operation(self):
+ self.db = DbMemory()
+ self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
+ session = {}
+
+ with self.subTest(i=1, t="VNF instance does not belong to NS"):
+ test_vnfr = yaml.safe_load(db_vnfrs_text)
+ test_vnfr[0]["revision"] = 2
+ test_nsr = yaml.safe_load(db_nsrs_text)
+ test_nsr[0]["constituent-vnfr-ref"][
+ 0
+ ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
+ self.db.create_list("vnfrs", test_vnfr)
+ self.db.create_list("nsrs", test_nsr)
+ nsrs = self.db.get_list("nsrs")[0]
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertRaises(EngineException) as expected_exception:
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+ self.assertEqual(
+ str(expected_exception.exception),
+ "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
+ " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
+ )
+
+ with self.subTest(i=2, t="Ns update request validated with no exception"):
+ test_vnfr = yaml.safe_load(db_vnfrs_text)
+ test_vnfr[0]["revision"] = 2
+ 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]
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertNotRaises(EngineException):
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+
+ with self.subTest(
+ i=3, t="Ns update request rejected because of too small timeout"
+ ):
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ "timeout_ns_update": 50,
+ }
+ with self.assertRaises(EngineException) as expected_exception:
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+ self.assertEqual(
+ str(expected_exception.exception),
+ "Error in validating ns-update request: 50 second is not enough "
+ "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
+ )
+
+ with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"):
+ test_vnfr = yaml.safe_load(db_vnfrs_text)
+ test_vnfr[0]["revision"] = 2
+ 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]
+ indata = {
+ "updateType": "CHANGE_VNFPKG",
+ "changeVnfPackageData": {
+ "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
+ },
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertRaises(EngineException) as expected_exception:
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+ self.assertEqual(
+ str(expected_exception.exception),
+ "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
+ "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
+ "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
+ )
+
+ with self.subTest(
+ i=5, t="Ns update REMOVE_VNF request validated with no exception"
+ ):
+ test_vnfr = yaml.safe_load(db_vnfrs_text)
+ test_vnfr[0]["revision"] = 2
+ 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]
+ indata = {
+ "updateType": "REMOVE_VNF",
+ "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
+ "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
+ }
+ with self.assertNotRaises(EngineException):
+ self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
+
class TestNsrTopic(unittest.TestCase):
def setUp(self):
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"]
self.assertEqual(
len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
)
+
+ self.assertEqual(
+ created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"],
+ 1,
+ "vdur first interface position does not match",
+ )
+
+ self.assertEqual(
+ created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"],
+ 2,
+ "vdur second interface position does not match",
+ )
+
self.assertEqual(
len(created_nsrs), 1, "Only one nsrs must be created at database"
)
"Expected '{}' at exception text".format(expect_text),
)
+ def test_show_instance(self):
+ 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.safe_load(db_nsrs_text))
+ actual_nsr = self.db.get_list("nsrs")[0]
+ nsr_id = actual_nsr["_id"]
+ 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."
+ )
+
+ 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"}
+ time_delta = 120
+ self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text))
+ nsr = self.db.get_list("nsrs")[0]
+
+ # When vcaStatus-refresh is 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"
+ 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."
+ )
+
+ # When vcaStatus-refresh is false but modified time is less than threshold
+ 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.",
+ )
+
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