"Database record must contain 'updateType=REMOVE_VNF'",
)
+ 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"},
+ )
+ 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):