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 # patch, MagicMock
from osm_common.dbbase import DbException
from osm_nbi.engine import EngineException
from osm_common.dbmemory import DbMemory
"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"]
+ _ = self.db.get_list("vnfrs")[0]["_id"]
session = {}
self.db.set_one(
"nsrs",
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.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_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 / blobdiff