osm_lcm/tests/test_ns.py

   1 #
   2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   3 # not use this file except in compliance with the License. You may obtain
   4 # a copy of the License at
   5 #
   6 #         http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  11 # License for the specific language governing permissions and limitations
  12 # under the License.
  13 #
  14 # For those usages not covered by the Apache License, Version 2.0 please
  15 # contact: alfonso.tiernosepulveda@telefonica.com
  16 ##
  17
  18
  19 import asynctest  # pip3 install asynctest --user
  20 import asyncio
  21 from copy import deepcopy
  22 import yaml
  23 import copy
  24 from n2vc.exceptions import N2VCException
  25 from os import getenv
  26 from osm_lcm import ns
  27 from osm_common.msgkafka import MsgKafka
  28
  29 from osm_lcm.data_utils.lcm_config import LcmCfg
  30 from osm_lcm.lcm_utils import TaskRegistry
  31 from osm_lcm.ng_ro import NgRoClient
  32 from osm_lcm.data_utils.database.database import Database
  33 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
  34 from osm_lcm.data_utils.vca import Relation, EERelation, DeployedVCA
  35 from osm_lcm.data_utils.vnfd import find_software_version
  36 from osm_lcm.lcm_utils import check_juju_bundle_existence, get_charm_artifact_path
  37 from osm_lcm.lcm_utils import LcmException
  38 from uuid import uuid4
  39 from unittest.mock import Mock, patch
  40
  41 from osm_lcm.tests import test_db_descriptors as descriptors
  42
  43 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  44
  45 """ Perform unittests using asynctest of osm_lcm.ns module
  46 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
  47     OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
  48     OSMLCMTEST_NS_NAME: change name of NS
  49     OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
  50     OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
  51     OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
  52     OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
  53     OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
  54     OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
  55     OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
  56     OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
  57     OSMLCM_VCA_XXX: configuration of N2VC
  58     OSMLCM_RO_XXX: configuration of RO
  59 """
  60
  61 lcm_config_dict = {
  62     "global": {"loglevel": "DEBUG"},
  63     "timeout": {},
  64     "VCA": {  # TODO replace with os.get_env to get other configurations
  65         "host": getenv("OSMLCM_VCA_HOST", "vca"),
  66         "port": getenv("OSMLCM_VCA_PORT", 17070),
  67         "user": getenv("OSMLCM_VCA_USER", "admin"),
  68         "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
  69         "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
  70         "ca_cert": getenv("OSMLCM_VCA_CACERT", None),
  71         "apiproxy": getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1"),
  72     },
  73     "RO": {
  74         "uri": "http://{}:{}/openmano".format(
  75             getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
  76         ),
  77         "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
  78         "logger_name": "lcm.ROclient",
  79         "loglevel": "DEBUG",
  80         "ng": True,
  81     },
  82 }
  83
  84 lcm_config = LcmCfg()
  85 lcm_config.set_from_dict(lcm_config_dict)
  86 lcm_config.transform()
  87
  88 nsr_id = descriptors.test_ids["TEST-A"]["ns"]
  89 nslcmop_id = descriptors.test_ids["TEST-A"]["update"]
  90 vnfr_id = "6421c7c9-d865-4fb4-9a13-d4275d243e01"
  91 vnfd_id = "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
  92 update_fs = Mock(autospec=True)
  93 update_fs.path.__add__ = Mock()
  94 update_fs.path.side_effect = ["/", "/", "/", "/"]
  95 update_fs.sync.side_effect = [None, None]
  96
  97
  98 def callable(a):
  99     return a
 100
 101
 102 class TestBaseNS(asynctest.TestCase):
 103     async def _n2vc_DeployCharms(
 104         self,
 105         model_name,
 106         application_name,
 107         vnfd,
 108         charm_path,
 109         params={},
 110         machine_spec={},
 111         callback=None,
 112         *callback_args,
 113     ):
 114         if callback:
 115             for status, message in (
 116                 ("maintenance", "installing sofwware"),
 117                 ("active", "Ready!"),
 118             ):
 119                 # call callback after some time
 120                 asyncio.sleep(5)
 121                 callback(model_name, application_name, status, message, *callback_args)
 122
 123     @staticmethod
 124     def _n2vc_FormatApplicationName(*args):
 125         num_calls = 0
 126         while True:
 127             yield "app_name-{}".format(num_calls)
 128             num_calls += 1
 129
 130     def _n2vc_CreateExecutionEnvironment(
 131         self, namespace, reuse_ee_id, db_dict, *args, **kwargs
 132     ):
 133         k_list = namespace.split(".")
 134         ee_id = k_list[1] + "."
 135         if len(k_list) >= 2:
 136             for k in k_list[2:4]:
 137                 ee_id += k[:8]
 138         else:
 139             ee_id += "_NS_"
 140         return ee_id, {}
 141
 142     def _ro_status(self, *args, **kwargs):
 143         print("Args > {}".format(args))
 144         print("kwargs > {}".format(kwargs))
 145         if args:
 146             if "update" in args:
 147                 ro_ns_desc = yaml.safe_load(descriptors.ro_update_action_text)
 148                 while True:
 149                     yield ro_ns_desc
 150         if kwargs.get("delete"):
 151             ro_ns_desc = yaml.safe_load(descriptors.ro_delete_action_text)
 152             while True:
 153                 yield ro_ns_desc
 154
 155         ro_ns_desc = yaml.safe_load(descriptors.ro_ns_text)
 156
 157         # if ip address provided, replace descriptor
 158         ip_addresses = getenv("OSMLCMTEST_NS_IPADDRESS", "")
 159         if ip_addresses:
 160             ip_addresses_list = ip_addresses.split(",")
 161             for vnf in ro_ns_desc["vnfs"]:
 162                 if not ip_addresses_list:
 163                     break
 164                 vnf["ip_address"] = ip_addresses_list[0]
 165                 for vm in vnf["vms"]:
 166                     if not ip_addresses_list:
 167                         break
 168                     vm["ip_address"] = ip_addresses_list.pop(0)
 169
 170         while True:
 171             yield ro_ns_desc
 172             for net in ro_ns_desc["nets"]:
 173                 if net["status"] != "ACTIVE":
 174                     net["status"] = "ACTIVE"
 175                     break
 176             else:
 177                 for vnf in ro_ns_desc["vnfs"]:
 178                     for vm in vnf["vms"]:
 179                         if vm["status"] != "ACTIVE":
 180                             vm["status"] = "ACTIVE"
 181                             break
 182
 183     def _ro_deploy(self, *args, **kwargs):
 184         return {"action_id": args[1]["action_id"], "nsr_id": args[0], "status": "ok"}
 185
 186     def _return_uuid(self, *args, **kwargs):
 187         return str(uuid4())
 188
 189     async def setUp(self):
 190         self.mock_db()
 191         self.mock_kafka()
 192         self.mock_filesystem()
 193         self.mock_task_registry()
 194         self.mock_vca_k8s()
 195         self.create_nslcm_class()
 196         self.mock_logging()
 197         self.mock_vca_n2vc()
 198         self.mock_ro()
 199
 200     def mock_db(self):
 201         if not getenv("OSMLCMTEST_DB_NOMOCK"):
 202             # Cleanup singleton Database instance
 203             Database.instance = None
 204
 205             self.db = Database({"database": {"driver": "memory"}}).instance.db
 206             self.db.create_list("vnfds", yaml.safe_load(descriptors.db_vnfds_text))
 207             self.db.create_list(
 208                 "vnfds_revisions", yaml.safe_load(descriptors.db_vnfds_revisions_text)
 209             )
 210             self.db.create_list("nsds", yaml.safe_load(descriptors.db_nsds_text))
 211             self.db.create_list("nsrs", yaml.safe_load(descriptors.db_nsrs_text))
 212             self.db.create_list(
 213                 "vim_accounts", yaml.safe_load(descriptors.db_vim_accounts_text)
 214             )
 215             self.db.create_list(
 216                 "k8sclusters", yaml.safe_load(descriptors.db_k8sclusters_text)
 217             )
 218             self.db.create_list(
 219                 "nslcmops", yaml.safe_load(descriptors.db_nslcmops_text)
 220             )
 221             self.db.create_list("vnfrs", yaml.safe_load(descriptors.db_vnfrs_text))
 222             self.db_vim_accounts = yaml.safe_load(descriptors.db_vim_accounts_text)
 223
 224     def mock_kafka(self):
 225         self.msg = asynctest.Mock(MsgKafka())
 226
 227     def mock_filesystem(self):
 228         if not getenv("OSMLCMTEST_FS_NOMOCK"):
 229             self.fs = asynctest.Mock(
 230                 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
 231             )
 232             self.fs.get_params.return_value = {
 233                 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
 234             }
 235             self.fs.file_open = asynctest.mock_open()
 236             # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock()  # called on a python "with"
 237             # self.fs.file_open.return_value.__enter__.return_value.read.return_value = ""   # empty file
 238
 239     def mock_task_registry(self):
 240         self.lcm_tasks = asynctest.Mock(TaskRegistry())
 241         self.lcm_tasks.lock_HA.return_value = True
 242         self.lcm_tasks.waitfor_related_HA.return_value = None
 243         self.lcm_tasks.lookfor_related.return_value = ("", [])
 244
 245     def mock_vca_k8s(self):
 246         if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
 247             ns.K8sJujuConnector = asynctest.MagicMock(ns.K8sJujuConnector)
 248             # ns.K8sHelmConnector = asynctest.MagicMock(ns.K8sHelmConnector)
 249             ns.K8sHelm3Connector = asynctest.MagicMock(ns.K8sHelm3Connector)
 250
 251         if not getenv("OSMLCMTEST_VCA_NOMOCK"):
 252             ns.N2VCJujuConnector = asynctest.MagicMock(ns.N2VCJujuConnector)
 253             ns.LCMHelmConn = asynctest.MagicMock(ns.LCMHelmConn)
 254
 255     def create_nslcm_class(self):
 256         self.my_ns = ns.NsLcm(self.msg, self.lcm_tasks, lcm_config)
 257         self.my_ns.fs = self.fs
 258         self.my_ns.db = self.db
 259         self.my_ns._wait_dependent_n2vc = asynctest.CoroutineMock()
 260
 261     def mock_logging(self):
 262         if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
 263             self.my_ns.logger = asynctest.Mock(self.my_ns.logger)
 264
 265     def mock_vca_n2vc(self):
 266         if not getenv("OSMLCMTEST_VCA_NOMOCK"):
 267             pub_key = getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
 268             # self.my_ns.n2vc = asynctest.Mock(N2VC())
 269             self.my_ns.n2vc.GetPublicKey.return_value = getenv(
 270                 "OSMLCM_VCA_PUBKEY", "public_key"
 271             )
 272             # allow several versions of n2vc
 273             self.my_ns.n2vc.FormatApplicationName = asynctest.Mock(
 274                 side_effect=self._n2vc_FormatApplicationName()
 275             )
 276             self.my_ns.n2vc.DeployCharms = asynctest.CoroutineMock(
 277                 side_effect=self._n2vc_DeployCharms
 278             )
 279             self.my_ns.n2vc.create_execution_environment = asynctest.CoroutineMock(
 280                 side_effect=self._n2vc_CreateExecutionEnvironment
 281             )
 282             self.my_ns.n2vc.install_configuration_sw = asynctest.CoroutineMock(
 283                 return_value=pub_key
 284             )
 285             self.my_ns.n2vc.get_ee_ssh_public__key = asynctest.CoroutineMock(
 286                 return_value=pub_key
 287             )
 288             self.my_ns.n2vc.exec_primitive = asynctest.CoroutineMock(
 289                 side_effect=self._return_uuid
 290             )
 291             self.my_ns.n2vc.exec_primitive = asynctest.CoroutineMock(
 292                 side_effect=self._return_uuid
 293             )
 294             self.my_ns.n2vc.GetPrimitiveStatus = asynctest.CoroutineMock(
 295                 return_value="completed"
 296             )
 297             self.my_ns.n2vc.GetPrimitiveOutput = asynctest.CoroutineMock(
 298                 return_value={"result": "ok", "pubkey": pub_key}
 299             )
 300             self.my_ns.n2vc.delete_execution_environment = asynctest.CoroutineMock(
 301                 return_value=None
 302             )
 303             self.my_ns.n2vc.get_public_key = asynctest.CoroutineMock(
 304                 return_value=getenv("OSMLCM_VCA_PUBKEY", "public_key")
 305             )
 306             self.my_ns.n2vc.delete_namespace = asynctest.CoroutineMock(
 307                 return_value=None
 308             )
 309             self.my_ns.n2vc.register_execution_environment = asynctest.CoroutineMock(
 310                 return_value="model-name.application-name.k8s"
 311             )
 312
 313     def mock_ro(self):
 314         if not getenv("OSMLCMTEST_RO_NOMOCK"):
 315             self.my_ns.RO = asynctest.Mock(NgRoClient(**lcm_config.RO.to_dict()))
 316             # TODO first time should be empty list, following should return a dict
 317             # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
 318             self.my_ns.RO.deploy = asynctest.CoroutineMock(
 319                 self.my_ns.RO.deploy, side_effect=self._ro_deploy
 320             )
 321             # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status)
 322             # self.my_ns.RO.create_action = asynctest.CoroutineMock(self.my_ns.RO.create_action,
 323             #                                                      return_value={"vm-id": {"vim_result": 200,
 324             #                                                                              "description": "done"}})
 325             self.my_ns.RO.delete = asynctest.CoroutineMock(self.my_ns.RO.delete)
 326
 327
 328 class TestMyNS(TestBaseNS):
 329     @asynctest.fail_on(active_handles=True)
 330     async def test_start_stop_rebuild_pass(self):
 331         nsr_id = descriptors.test_ids["TEST-OP-VNF"]["ns"]
 332         nslcmop_id = descriptors.test_ids["TEST-OP-VNF"]["nslcmops"]
 333         vnf_id = descriptors.test_ids["TEST-OP-VNF"]["vnfrs"]
 334         additional_param = {"count-index": "0"}
 335         operation_type = "start"
 336         await self.my_ns.rebuild_start_stop(
 337             nsr_id, nslcmop_id, vnf_id, additional_param, operation_type
 338         )
 339         expected_value = "COMPLETED"
 340         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 341             "operationState"
 342         )
 343         self.assertEqual(return_value, expected_value)
 344
 345     @asynctest.fail_on(active_handles=True)
 346     async def test_start_stop_rebuild_fail(self):
 347         nsr_id = descriptors.test_ids["TEST-OP-VNF"]["ns"]
 348         nslcmop_id = descriptors.test_ids["TEST-OP-VNF"]["nslcmops1"]
 349         vnf_id = descriptors.test_ids["TEST-OP-VNF"]["vnfrs"]
 350         additional_param = {"count-index": "0"}
 351         operation_type = "stop"
 352         await self.my_ns.rebuild_start_stop(
 353             nsr_id, nslcmop_id, vnf_id, additional_param, operation_type
 354         )
 355         expected_value = "Error"
 356         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 357             "operationState"
 358         )
 359         self.assertEqual(return_value, expected_value)
 360
 361     # Test scale() and related methods
 362     @asynctest.fail_on(active_handles=True)  # all async tasks must be completed
 363     async def test_scale(self):
 364         # print("Test scale started")
 365
 366         # TODO: Add more higher-lever tests here, for example:
 367         # scale-out/scale-in operations with success/error result
 368
 369         # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
 370         nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 371         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 372         await self.my_ns.scale(nsr_id, nslcmop_id)
 373         expected_value = "FAILED"
 374         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 375             "operationState"
 376         )
 377         self.assertEqual(return_value, expected_value)
 378         # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
 379
 380         # Test scale() for native kdu
 381         # this also includes testing _scale_kdu()
 382         nsr_id = descriptors.test_ids["TEST-NATIVE-KDU"]["ns"]
 383         nslcmop_id = descriptors.test_ids["TEST-NATIVE-KDU"]["instantiate"]
 384
 385         self.my_ns.k8sclusterjuju.scale = asynctest.mock.CoroutineMock()
 386         self.my_ns.k8sclusterjuju.exec_primitive = asynctest.mock.CoroutineMock()
 387         self.my_ns.k8sclusterjuju.get_scale_count = asynctest.mock.CoroutineMock(
 388             return_value=1
 389         )
 390         await self.my_ns.scale(nsr_id, nslcmop_id)
 391         expected_value = "COMPLETED"
 392         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 393             "operationState"
 394         )
 395         self.assertEqual(return_value, expected_value)
 396         self.my_ns.k8sclusterjuju.scale.assert_called_once()
 397
 398         # Test scale() for native kdu with 2 resource
 399         nsr_id = descriptors.test_ids["TEST-NATIVE-KDU-2"]["ns"]
 400         nslcmop_id = descriptors.test_ids["TEST-NATIVE-KDU-2"]["instantiate"]
 401
 402         self.my_ns.k8sclusterjuju.get_scale_count.return_value = 2
 403         await self.my_ns.scale(nsr_id, nslcmop_id)
 404         expected_value = "COMPLETED"
 405         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 406             "operationState"
 407         )
 408         self.assertEqual(return_value, expected_value)
 409         self.my_ns.k8sclusterjuju.scale.assert_called()
 410
 411     async def test_vca_status_refresh(self):
 412         nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 413         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 414         await self.my_ns.vca_status_refresh(nsr_id, nslcmop_id)
 415         expected_value = dict()
 416         return_value = dict()
 417         vnf_descriptors = self.db.get_list("vnfds")
 418         for i, _ in enumerate(vnf_descriptors):
 419             for j, value in enumerate(vnf_descriptors[i]["df"]):
 420                 if "lcm-operations-configuration" in vnf_descriptors[i]["df"][j]:
 421                     if (
 422                         "day1-2"
 423                         in value["lcm-operations-configuration"][
 424                             "operate-vnf-op-config"
 425                         ]
 426                     ):
 427                         for k, v in enumerate(
 428                             value["lcm-operations-configuration"][
 429                                 "operate-vnf-op-config"
 430                             ]["day1-2"]
 431                         ):
 432                             if (
 433                                 v.get("execution-environment-list")
 434                                 and "juju" in v["execution-environment-list"][0]
 435                             ):
 436                                 expected_value = self.db.get_list("nsrs")[i][
 437                                     "vcaStatus"
 438                                 ]
 439                                 await self.my_ns._on_update_n2vc_db(
 440                                     "nsrs",
 441                                     {"_id": nsr_id},
 442                                     "_admin.deployed.VCA.{}".format(k),
 443                                     {},
 444                                 )
 445                                 return_value = self.db.get_list("nsrs")[i]["vcaStatus"]
 446         self.assertEqual(return_value, expected_value)
 447
 448     # Test _retry_or_skip_suboperation()
 449     # Expected result:
 450     # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
 451     # - if marked as anything but 'COMPLETED', the suboperation index is expected
 452     def test_scale_retry_or_skip_suboperation(self):
 453         # Load an alternative 'nslcmops' YAML for this test
 454         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 455         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 456         op_index = 2
 457         # Test when 'operationState' is 'COMPLETED'
 458         db_nslcmop["_admin"]["operations"][op_index]["operationState"] = "COMPLETED"
 459         return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
 460         expected_value = self.my_ns.SUBOPERATION_STATUS_SKIP
 461         self.assertEqual(return_value, expected_value)
 462         # Test when 'operationState' is not 'COMPLETED'
 463         db_nslcmop["_admin"]["operations"][op_index]["operationState"] = None
 464         return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
 465         expected_value = op_index
 466         self.assertEqual(return_value, expected_value)
 467
 468     # Test _find_suboperation()
 469     # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
 470     def test_scale_find_suboperation(self):
 471         # Load an alternative 'nslcmops' YAML for this test
 472         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 473         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 474         # Find this sub-operation
 475         op_index = 2
 476         vnf_index = db_nslcmop["_admin"]["operations"][op_index]["member_vnf_index"]
 477         primitive = db_nslcmop["_admin"]["operations"][op_index]["primitive"]
 478         primitive_params = db_nslcmop["_admin"]["operations"][op_index][
 479             "primitive_params"
 480         ]
 481         match = {
 482             "member_vnf_index": vnf_index,
 483             "primitive": primitive,
 484             "primitive_params": primitive_params,
 485         }
 486         found_op_index = self.my_ns._find_suboperation(db_nslcmop, match)
 487         self.assertEqual(found_op_index, op_index)
 488         # Test with not-matching params
 489         match = {
 490             "member_vnf_index": vnf_index,
 491             "primitive": "",
 492             "primitive_params": primitive_params,
 493         }
 494         found_op_index = self.my_ns._find_suboperation(db_nslcmop, match)
 495         self.assertEqual(found_op_index, self.my_ns.SUBOPERATION_STATUS_NOT_FOUND)
 496         # Test with None
 497         match = None
 498         found_op_index = self.my_ns._find_suboperation(db_nslcmop, match)
 499         self.assertEqual(found_op_index, self.my_ns.SUBOPERATION_STATUS_NOT_FOUND)
 500
 501     # Test _update_suboperation_status()
 502     def test_scale_update_suboperation_status(self):
 503         self.db.set_one = asynctest.Mock()
 504         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 505         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 506         op_index = 0
 507         # Force the initial values to be distinct from the updated ones
 508         q_filter = {"_id": db_nslcmop["_id"]}
 509         # Test to change 'operationState' and 'detailed-status'
 510         operationState = "COMPLETED"
 511         detailed_status = "Done"
 512         expected_update_dict = {
 513             "_admin.operations.0.operationState": operationState,
 514             "_admin.operations.0.detailed-status": detailed_status,
 515         }
 516         self.my_ns._update_suboperation_status(
 517             db_nslcmop, op_index, operationState, detailed_status
 518         )
 519         self.db.set_one.assert_called_once_with(
 520             "nslcmops",
 521             q_filter=q_filter,
 522             update_dict=expected_update_dict,
 523             fail_on_empty=False,
 524         )
 525
 526     def test_scale_add_suboperation(self):
 527         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 528         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 529         vnf_index = "1"
 530         num_ops_before = len(db_nslcmop.get("_admin", {}).get("operations", [])) - 1
 531         vdu_id = None
 532         vdu_count_index = None
 533         vdu_name = None
 534         primitive = "touch"
 535         mapped_primitive_params = {
 536             "parameter": [
 537                 {
 538                     "data-type": "STRING",
 539                     "name": "filename",
 540                     "default-value": "<touch_filename2>",
 541                 }
 542             ],
 543             "name": "touch",
 544         }
 545         operationState = "PROCESSING"
 546         detailed_status = "In progress"
 547         operationType = "PRE-SCALE"
 548         # Add a 'pre-scale' suboperation
 549         op_index_after = self.my_ns._add_suboperation(
 550             db_nslcmop,
 551             vnf_index,
 552             vdu_id,
 553             vdu_count_index,
 554             vdu_name,
 555             primitive,
 556             mapped_primitive_params,
 557             operationState,
 558             detailed_status,
 559             operationType,
 560         )
 561         self.assertEqual(op_index_after, num_ops_before + 1)
 562
 563         # Delete all suboperations and add the same operation again
 564         del db_nslcmop["_admin"]["operations"]
 565         op_index_zero = self.my_ns._add_suboperation(
 566             db_nslcmop,
 567             vnf_index,
 568             vdu_id,
 569             vdu_count_index,
 570             vdu_name,
 571             primitive,
 572             mapped_primitive_params,
 573             operationState,
 574             detailed_status,
 575             operationType,
 576         )
 577         self.assertEqual(op_index_zero, 0)
 578
 579         # Add a 'RO' suboperation
 580         RO_nsr_id = "1234567890"
 581         RO_scaling_info = [
 582             {
 583                 "type": "create",
 584                 "count": 1,
 585                 "member-vnf-index": "1",
 586                 "osm_vdu_id": "dataVM",
 587             }
 588         ]
 589         op_index = self.my_ns._add_suboperation(
 590             db_nslcmop,
 591             vnf_index,
 592             vdu_id,
 593             vdu_count_index,
 594             vdu_name,
 595             primitive,
 596             mapped_primitive_params,
 597             operationState,
 598             detailed_status,
 599             operationType,
 600             RO_nsr_id,
 601             RO_scaling_info,
 602         )
 603         db_RO_nsr_id = db_nslcmop["_admin"]["operations"][op_index]["RO_nsr_id"]
 604         self.assertEqual(op_index, 1)
 605         self.assertEqual(RO_nsr_id, db_RO_nsr_id)
 606
 607         # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
 608         op_index_invalid = self.my_ns._add_suboperation(
 609             None, None, None, None, None, None, None, None, None, None, None
 610         )
 611         self.assertEqual(op_index_invalid, self.my_ns.SUBOPERATION_STATUS_NOT_FOUND)
 612
 613     # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
 614     # check the possible return values:
 615     # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
 616     # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
 617     # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
 618     def test_scale_check_or_add_scale_suboperation(self):
 619         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 620         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 621         operationType = "PRE-SCALE"
 622         vnf_index = "1"
 623         primitive = "touch"
 624         primitive_params = {
 625             "parameter": [
 626                 {
 627                     "data-type": "STRING",
 628                     "name": "filename",
 629                     "default-value": "<touch_filename2>",
 630                 }
 631             ],
 632             "name": "touch",
 633         }
 634
 635         # Delete all sub-operations to be sure this is a new sub-operation
 636         del db_nslcmop["_admin"]["operations"]
 637
 638         # Add a new sub-operation
 639         # For new sub-operations, operationState is set to 'PROCESSING' by default
 640         op_index_new = self.my_ns._check_or_add_scale_suboperation(
 641             db_nslcmop, vnf_index, primitive, primitive_params, operationType
 642         )
 643         self.assertEqual(op_index_new, self.my_ns.SUBOPERATION_STATUS_NEW)
 644
 645         # Use the same parameters again to match the already added sub-operation
 646         # which has status 'PROCESSING' (!= 'COMPLETED') by default
 647         # The expected return value is a non-negative number
 648         op_index_existing = self.my_ns._check_or_add_scale_suboperation(
 649             db_nslcmop, vnf_index, primitive, primitive_params, operationType
 650         )
 651         self.assertTrue(op_index_existing >= 0)
 652
 653         # Change operationState 'manually' for this sub-operation
 654         db_nslcmop["_admin"]["operations"][op_index_existing][
 655             "operationState"
 656         ] = "COMPLETED"
 657         # Then use the same parameters again to match the already added sub-operation,
 658         # which now has status 'COMPLETED'
 659         # The expected return value is SUBOPERATION_STATUS_SKIP
 660         op_index_skip = self.my_ns._check_or_add_scale_suboperation(
 661             db_nslcmop, vnf_index, primitive, primitive_params, operationType
 662         )
 663         self.assertEqual(op_index_skip, self.my_ns.SUBOPERATION_STATUS_SKIP)
 664
 665         # RO sub-operation test:
 666         # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
 667         RO_nsr_id = "1234567890"
 668         RO_scaling_info = [
 669             {
 670                 "type": "create",
 671                 "count": 1,
 672                 "member-vnf-index": "1",
 673                 "osm_vdu_id": "dataVM",
 674             }
 675         ]
 676         op_index_new_RO = self.my_ns._check_or_add_scale_suboperation(
 677             db_nslcmop, vnf_index, None, None, "SCALE-RO", RO_nsr_id, RO_scaling_info
 678         )
 679         self.assertEqual(op_index_new_RO, self.my_ns.SUBOPERATION_STATUS_NEW)
 680
 681         # Use the same parameters again to match the already added RO sub-operation
 682         op_index_existing_RO = self.my_ns._check_or_add_scale_suboperation(
 683             db_nslcmop, vnf_index, None, None, "SCALE-RO", RO_nsr_id, RO_scaling_info
 684         )
 685         self.assertTrue(op_index_existing_RO >= 0)
 686
 687         # Change operationState 'manually' for this RO sub-operation
 688         db_nslcmop["_admin"]["operations"][op_index_existing_RO][
 689             "operationState"
 690         ] = "COMPLETED"
 691         # Then use the same parameters again to match the already added sub-operation,
 692         # which now has status 'COMPLETED'
 693         # The expected return value is SUBOPERATION_STATUS_SKIP
 694         op_index_skip_RO = self.my_ns._check_or_add_scale_suboperation(
 695             db_nslcmop, vnf_index, None, None, "SCALE-RO", RO_nsr_id, RO_scaling_info
 696         )
 697         self.assertEqual(op_index_skip_RO, self.my_ns.SUBOPERATION_STATUS_SKIP)
 698
 699     async def test_deploy_kdus(self):
 700         nsr_id = descriptors.test_ids["TEST-KDU"]["ns"]
 701         nslcmop_id = descriptors.test_ids["TEST-KDU"]["instantiate"]
 702         db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 703         db_vnfr = self.db.get_one(
 704             "vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "multikdu"}
 705         )
 706         db_vnfrs = {"multikdu": db_vnfr}
 707         db_vnfd = self.db.get_one("vnfds", {"_id": db_vnfr["vnfd-id"]})
 708         db_vnfds = [db_vnfd]
 709         task_register = {}
 710         logging_text = "KDU"
 711         self.my_ns.k8sclusterhelm3.generate_kdu_instance_name = asynctest.mock.Mock()
 712         self.my_ns.k8sclusterhelm3.generate_kdu_instance_name.return_value = "k8s_id"
 713         self.my_ns.k8sclusterhelm3.install = asynctest.CoroutineMock()
 714         self.my_ns.k8sclusterhelm3.synchronize_repos = asynctest.CoroutineMock(
 715             return_value=("", "")
 716         )
 717         self.my_ns.k8sclusterhelm3.get_services = asynctest.CoroutineMock(
 718             return_value=([])
 719         )
 720         await self.my_ns.deploy_kdus(
 721             logging_text, nsr_id, nslcmop_id, db_vnfrs, db_vnfds, task_register
 722         )
 723         await asyncio.wait(list(task_register.keys()), timeout=100)
 724         db_nsr = self.db.get_list("nsrs")[1]
 725         self.assertIn(
 726             "K8s",
 727             db_nsr["_admin"]["deployed"],
 728             "K8s entry not created at '_admin.deployed'",
 729         )
 730         self.assertIsInstance(
 731             db_nsr["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
 732         )
 733         self.assertEqual(
 734             len(db_nsr["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
 735         )
 736         k8s_instace_info = {
 737             "kdu-instance": "k8s_id",
 738             "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
 739             "k8scluster-type": "helm-chart-v3",
 740             "kdu-name": "ldap",
 741             "member-vnf-index": "multikdu",
 742             "namespace": None,
 743             "kdu-deployment-name": None,
 744         }
 745
 746         nsr_result = copy.deepcopy(db_nsr["_admin"]["deployed"]["K8s"][0])
 747         nsr_kdu_model_result = nsr_result.pop("kdu-model")
 748         expected_kdu_model = "stable/openldap:1.2.1"
 749         self.assertEqual(nsr_result, k8s_instace_info)
 750         self.assertTrue(
 751             nsr_kdu_model_result in expected_kdu_model
 752             or expected_kdu_model in nsr_kdu_model_result
 753         )
 754         nsr_result = copy.deepcopy(db_nsr["_admin"]["deployed"]["K8s"][1])
 755         nsr_kdu_model_result = nsr_result.pop("kdu-model")
 756         k8s_instace_info["kdu-name"] = "mongo"
 757         expected_kdu_model = "stable/mongodb"
 758         self.assertEqual(nsr_result, k8s_instace_info)
 759         self.assertTrue(
 760             nsr_kdu_model_result in expected_kdu_model
 761             or expected_kdu_model in nsr_kdu_model_result
 762         )
 763
 764     # Test remove_vnf() and related methods
 765     @asynctest.fail_on(active_handles=True)  # all async tasks must be completed
 766     async def test_remove_vnf(self):
 767         # Test REMOVE_VNF
 768         nsr_id = descriptors.test_ids["TEST-UPDATE"]["ns"]
 769         nslcmop_id = descriptors.test_ids["TEST-UPDATE"]["removeVnf"]
 770         vnf_instance_id = descriptors.test_ids["TEST-UPDATE"]["vnf"]
 771         mock_wait_ng_ro = asynctest.CoroutineMock()
 772         with patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro):
 773             await self.my_ns.update(nsr_id, nslcmop_id)
 774             expected_value = "COMPLETED"
 775             return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 776                 "operationState"
 777             )
 778             self.assertEqual(return_value, expected_value)
 779             with self.assertRaises(Exception) as context:
 780                 self.db.get_one("vnfrs", {"_id": vnf_instance_id})
 781             self.assertTrue(
 782                 "database exception Not found entry with filter"
 783                 in str(context.exception)
 784             )
 785
 786     # test vertical scale executes sucessfully
 787     # @patch("osm_lcm.ng_ro.status.response")
 788     @asynctest.fail_on(active_handles=True)
 789     async def test_vertical_scaling(self):
 790         nsr_id = descriptors.test_ids["TEST-V-SCALE"]["ns"]
 791         nslcmop_id = descriptors.test_ids["TEST-V-SCALE"]["instantiate"]
 792
 793         # calling the vertical scale fucntion
 794         # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status("update"))
 795         mock_wait_ng_ro = asynctest.CoroutineMock()
 796         with patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro):
 797             await self.my_ns.vertical_scale(nsr_id, nslcmop_id)
 798             return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 799                 "operationState"
 800             )
 801             expected_value = "COMPLETED"
 802             self.assertNotEqual(return_value, expected_value)
 803
 804     # test vertical scale executes fail
 805     @asynctest.fail_on(active_handles=True)
 806     async def test_vertical_scaling_fail(self):
 807         # get th nsr nad nslcmops id from descriptors
 808         nsr_id = descriptors.test_ids["TEST-V-SCALE"]["ns"]
 809         nslcmop_id = descriptors.test_ids["TEST-V-SCALE"]["instantiate-1"]
 810
 811         # calling the vertical scale fucntion
 812         await self.my_ns.vertical_scale(nsr_id, nslcmop_id)
 813         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 814             "operationState"
 815         )
 816         expected_value = "FAILED"
 817         self.assertEqual(return_value, expected_value)
 818
 819     # async def test_instantiate_pdu(self):
 820     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 821     #     nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 822     #     # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
 823     #     self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
 824     #                     update_dict={"ip-address": "10.205.1.46",
 825     #                                  "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
 826     #                                  "vdur.0.pdu-type": "PDU-TYPE-1",
 827     #                                  "vdur.0.ip-address": "10.205.1.46",
 828     #                                  },
 829     #                     unset={"vdur.status": None})
 830     #     self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
 831     #                     update_dict={"ip-address": "10.205.1.47",
 832     #                                  "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
 833     #                                  "vdur.0.pdu-type": "PDU-TYPE-1",
 834     #                                  "vdur.0.ip-address": "10.205.1.47",
 835     #                                  },
 836     #                     unset={"vdur.status": None})
 837
 838     #     await self.my_ns.instantiate(nsr_id, nslcmop_id)
 839     #     db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 840     #     self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
 841     #     self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
 842     #     self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
 843     #     self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
 844     #     self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
 845
 846     # @asynctest.fail_on(active_handles=True)   # all async tasks must be completed
 847     # async def test_terminate_without_configuration(self):
 848     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 849     #     nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
 850     #     # set instantiation task as completed
 851     #     self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
 852     #                      update_dict={"operationState": "COMPLETED"})
 853     #     self.db.set_one("nsrs", {"_id": nsr_id},
 854     #                     update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
 855
 856     #     await self.my_ns.terminate(nsr_id, nslcmop_id)
 857     #     db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 858     #     self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
 859     #     db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 860     #     self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 861     #     self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 862     #     self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
 863     #     self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
 864     #     self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
 865     #     self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
 866     #     db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
 867     #     for vnfr in db_vnfrs_list:
 868     #         self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
 869
 870     # @asynctest.fail_on(active_handles=True)   # all async tasks must be completed
 871     # async def test_terminate_primitive(self):
 872     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 873     #     nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
 874     #     # set instantiation task as completed
 875     #     self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
 876     #                      update_dict={"operationState": "COMPLETED"})
 877
 878     #     # modify vnfd descriptor to include terminate_primitive
 879     #     terminate_primitive = [{
 880     #         "name": "touch",
 881     #         "parameter": [{"name": "filename", "value": "terminate_filename"}],
 882     #         "seq": '1'
 883     #     }]
 884     #     db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
 885     #     self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
 886     #                     {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
 887
 888     #     await self.my_ns.terminate(nsr_id, nslcmop_id)
 889     #     db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 890     #     self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
 891     #     db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 892     #     self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 893     #     self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 894     #     self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
 895     #     self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
 896     #     self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
 897     #     self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
 898
 899     @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
 900     @patch(
 901         "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
 902     )
 903     @patch("osm_lcm.data_utils.vnfd.find_software_version")
 904     @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
 905     async def test_update_change_vnfpkg_sw_version_not_changed(
 906         self,
 907         mock_juju_bundle,
 908         mock_software_version,
 909         mock_charm_upgrade,
 910         mock_charm_hash,
 911     ):
 912         """Update type: CHANGE_VNFPKG, latest_vnfd revision changed,
 913         Charm package changed, sw-version is not changed"""
 914         self.db.set_one(
 915             "vnfds",
 916             q_filter={"_id": vnfd_id},
 917             update_dict={"_admin.revision": 3, "kdu": []},
 918         )
 919
 920         self.db.set_one(
 921             "vnfds_revisions",
 922             q_filter={"_id": vnfd_id + ":1"},
 923             update_dict={"_admin.revision": 1, "kdu": []},
 924         )
 925
 926         self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
 927
 928         mock_charm_hash.return_value = True
 929         mock_software_version.side_effect = ["1.0", "1.0"]
 930
 931         task = asyncio.Future()
 932         task.set_result(("COMPLETED", "some_output"))
 933         mock_charm_upgrade.return_value = task
 934
 935         instance = self.my_ns
 936         fs = deepcopy(update_fs)
 937         instance.fs = fs
 938
 939         expected_operation_state = "COMPLETED"
 940         expected_operation_error = ""
 941         expected_vnfr_revision = 3
 942         expected_ns_state = "INSTANTIATED"
 943         expected_ns_operational_state = "running"
 944
 945         await instance.update(nsr_id, nslcmop_id)
 946         return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 947             "operationState"
 948         )
 949         return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 950             "errorMessage"
 951         )
 952         return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
 953             "operational-status"
 954         )
 955         return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
 956             "revision"
 957         )
 958         return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
 959         mock_charm_hash.assert_called_with(
 960             f"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
 961             f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
 962         )
 963         self.assertEqual(fs.sync.call_count, 2)
 964         self.assertEqual(return_ns_state, expected_ns_state)
 965         self.assertEqual(return_operation_state, expected_operation_state)
 966         self.assertEqual(return_operation_error, expected_operation_error)
 967         self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
 968         self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
 969
 970     @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
 971     @patch(
 972         "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
 973     )
 974     @patch("osm_lcm.data_utils.vnfd.find_software_version")
 975     @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
 976     async def test_update_change_vnfpkg_vnfd_revision_not_changed(
 977         self,
 978         mock_juju_bundle,
 979         mock_software_version,
 980         mock_charm_upgrade,
 981         mock_charm_hash,
 982     ):
 983         """Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"""
 984         self.db.set_one(
 985             "vnfds", q_filter={"_id": vnfd_id}, update_dict={"_admin.revision": 1}
 986         )
 987         self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
 988
 989         mock_charm_hash.return_value = True
 990
 991         task = asyncio.Future()
 992         task.set_result(("COMPLETED", "some_output"))
 993         mock_charm_upgrade.return_value = task
 994
 995         instance = self.my_ns
 996
 997         expected_operation_state = "COMPLETED"
 998         expected_operation_error = ""
 999         expected_vnfr_revision = 1
1000         expected_ns_state = "INSTANTIATED"
1001         expected_ns_operational_state = "running"
1002
1003         await instance.update(nsr_id, nslcmop_id)
1004         return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1005             "operationState"
1006         )
1007         return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1008             "errorMessage"
1009         )
1010         return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
1011             "operational-status"
1012         )
1013         return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
1014         return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
1015             "revision"
1016         )
1017         mock_charm_hash.assert_not_called()
1018         mock_software_version.assert_not_called()
1019         mock_juju_bundle.assert_not_called()
1020         mock_charm_upgrade.assert_not_called()
1021         update_fs.sync.assert_not_called()
1022         self.assertEqual(return_ns_state, expected_ns_state)
1023         self.assertEqual(return_operation_state, expected_operation_state)
1024         self.assertEqual(return_operation_error, expected_operation_error)
1025         self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
1026         self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
1027
1028     @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1029     @patch(
1030         "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
1031     )
1032     @patch("osm_lcm.data_utils.vnfd.find_software_version")
1033     @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1034     async def test_update_change_vnfpkg_charm_is_not_changed(
1035         self,
1036         mock_juju_bundle,
1037         mock_software_version,
1038         mock_charm_upgrade,
1039         mock_charm_hash,
1040     ):
1041         """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1042         Charm package is not changed, sw-version is not changed"""
1043         self.db.set_one(
1044             "vnfds",
1045             q_filter={"_id": vnfd_id},
1046             update_dict={"_admin.revision": 3, "kdu": []},
1047         )
1048         self.db.set_one(
1049             "vnfds_revisions",
1050             q_filter={"_id": vnfd_id + ":1"},
1051             update_dict={"_admin.revision": 1, "kdu": []},
1052         )
1053         self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
1054
1055         mock_charm_hash.return_value = False
1056         mock_software_version.side_effect = ["1.0", "1.0"]
1057
1058         instance = self.my_ns
1059         fs = deepcopy(update_fs)
1060         instance.fs = fs
1061         expected_operation_state = "COMPLETED"
1062         expected_operation_error = ""
1063         expected_vnfr_revision = 3
1064         expected_ns_state = "INSTANTIATED"
1065         expected_ns_operational_state = "running"
1066
1067         await instance.update(nsr_id, nslcmop_id)
1068         return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1069             "operationState"
1070         )
1071         return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1072             "errorMessage"
1073         )
1074         return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
1075             "operational-status"
1076         )
1077         return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
1078             "revision"
1079         )
1080         return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
1081         mock_charm_hash.assert_called_with(
1082             f"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1083             f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1084         )
1085         self.assertEqual(fs.sync.call_count, 2)
1086         self.assertEqual(mock_charm_hash.call_count, 1)
1087         mock_juju_bundle.assert_not_called()
1088         mock_charm_upgrade.assert_not_called()
1089         self.assertEqual(return_ns_state, expected_ns_state)
1090         self.assertEqual(return_operation_state, expected_operation_state)
1091         self.assertEqual(return_operation_error, expected_operation_error)
1092         self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
1093         self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
1094
1095     @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1096     @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1097     @patch(
1098         "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
1099     )
1100     @patch("osm_lcm.lcm_utils.get_charm_artifact_path")
1101     async def test_update_change_vnfpkg_sw_version_changed(
1102         self, mock_charm_artifact, mock_charm_upgrade, mock_charm_hash, mock_juju_bundle
1103     ):
1104         """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1105         Charm package exists, sw-version changed."""
1106         self.db.set_one(
1107             "vnfds",
1108             q_filter={"_id": vnfd_id},
1109             update_dict={"_admin.revision": 3, "software-version": "3.0", "kdu": []},
1110         )
1111         self.db.set_one(
1112             "vnfds_revisions",
1113             q_filter={"_id": vnfd_id + ":1"},
1114             update_dict={"_admin.revision": 1, "kdu": []},
1115         )
1116         self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
1117         mock_charm_hash.return_value = False
1118
1119         mock_charm_artifact.side_effect = [
1120             f"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1121             f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1122         ]
1123
1124         instance = self.my_ns
1125         fs = deepcopy(update_fs)
1126         instance.fs = fs
1127         expected_operation_state = "FAILED"
1128         expected_operation_error = "FAILED Checking if existing VNF has charm: Software version change is not supported as VNF instance 6421c7c9-d865-4fb4-9a13-d4275d243e01 has charm."
1129         expected_vnfr_revision = 1
1130         expected_ns_state = "INSTANTIATED"
1131         expected_ns_operational_state = "running"
1132
1133         await instance.update(nsr_id, nslcmop_id)
1134         return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1135             "operationState"
1136         )
1137         return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1138             "errorMessage"
1139         )
1140         return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
1141             "operational-status"
1142         )
1143         return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
1144             "revision"
1145         )
1146         return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
1147         self.assertEqual(fs.sync.call_count, 2)
1148         mock_charm_hash.assert_not_called()
1149         mock_juju_bundle.assert_not_called()
1150         mock_charm_upgrade.assert_not_called()
1151         self.assertEqual(return_ns_state, expected_ns_state)
1152         self.assertEqual(return_operation_state, expected_operation_state)
1153         self.assertEqual(return_operation_error, expected_operation_error)
1154         self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
1155         self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
1156
1157     @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1158     @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1159     @patch(
1160         "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
1161     )
1162     @patch("osm_lcm.data_utils.vnfd.find_software_version")
1163     async def test_update_change_vnfpkg_juju_bundle_exists(
1164         self,
1165         mock_software_version,
1166         mock_charm_upgrade,
1167         mock_charm_hash,
1168         mock_juju_bundle,
1169     ):
1170         """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1171         Charm package exists, sw-version not changed, juju-bundle exists"""
1172         # Upgrade is not allowed with juju bundles, this will cause TypeError
1173         self.db.set_one(
1174             "vnfds",
1175             q_filter={"_id": vnfd_id},
1176             update_dict={
1177                 "_admin.revision": 5,
1178                 "software-version": "1.0",
1179                 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1180             },
1181         )
1182         self.db.set_one(
1183             "vnfds_revisions",
1184             q_filter={"_id": vnfd_id + ":1"},
1185             update_dict={
1186                 "_admin.revision": 1,
1187                 "software-version": "1.0",
1188                 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1189             },
1190         )
1191         self.db.set_one(
1192             "nsrs",
1193             q_filter={"_id": nsr_id},
1194             update_dict={"_admin.deployed.VCA.0.kdu_name": "native-kdu"},
1195         )
1196         self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
1197
1198         mock_charm_hash.side_effect = [True]
1199         mock_software_version.side_effect = ["1.0", "1.0"]
1200         mock_juju_bundle.return_value = True
1201         instance = self.my_ns
1202         fs = deepcopy(update_fs)
1203         instance.fs = fs
1204
1205         expected_vnfr_revision = 1
1206         expected_ns_state = "INSTANTIATED"
1207         expected_ns_operational_state = "running"
1208
1209         await instance.update(nsr_id, nslcmop_id)
1210         return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
1211             "operational-status"
1212         )
1213         return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
1214             "revision"
1215         )
1216         return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
1217         self.assertEqual(fs.sync.call_count, 2)
1218         mock_charm_upgrade.assert_not_called()
1219         mock_charm_hash.assert_not_called()
1220         self.assertEqual(return_ns_state, expected_ns_state)
1221         self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
1222         self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
1223
1224     @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1225     @patch(
1226         "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable=asynctest.Mock(autospec=True)
1227     )
1228     async def test_update_change_vnfpkg_charm_upgrade_failed(
1229         self, mock_charm_upgrade, mock_charm_hash
1230     ):
1231         """ "Update type: CHANGE_VNFPKG, latest_vnfd revision changed"
1232         Charm package exists, sw-version not changed, charm-upgrade failed"""
1233         self.db.set_one(
1234             "vnfds",
1235             q_filter={"_id": vnfd_id},
1236             update_dict={"_admin.revision": 3, "software-version": "1.0", "kdu": []},
1237         )
1238         self.db.set_one(
1239             "vnfds_revisions",
1240             q_filter={"_id": vnfd_id + ":1"},
1241             update_dict={"_admin.revision": 1, "software-version": "1.0", "kdu": []},
1242         )
1243         self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1})
1244
1245         mock_charm_hash.return_value = True
1246
1247         task = asyncio.Future()
1248         task.set_result(("FAILED", "some_error"))
1249         mock_charm_upgrade.return_value = task
1250
1251         instance = self.my_ns
1252         fs = deepcopy(update_fs)
1253         instance.fs = fs
1254         expected_operation_state = "FAILED"
1255         expected_operation_error = "some_error"
1256         expected_vnfr_revision = 1
1257         expected_ns_state = "INSTANTIATED"
1258         expected_ns_operational_state = "running"
1259
1260         await instance.update(nsr_id, nslcmop_id)
1261         return_operation_state = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1262             "operationState"
1263         )
1264         return_operation_error = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
1265             "errorMessage"
1266         )
1267         return_ns_operational_state = self.db.get_one("nsrs", {"_id": nsr_id}).get(
1268             "operational-status"
1269         )
1270         return_vnfr_revision = self.db.get_one("vnfrs", {"_id": vnfr_id}).get(
1271             "revision"
1272         )
1273         return_ns_state = self.db.get_one("nsrs", {"_id": nsr_id}).get("nsState")
1274         self.assertEqual(fs.sync.call_count, 2)
1275         self.assertEqual(mock_charm_hash.call_count, 1)
1276         self.assertEqual(mock_charm_upgrade.call_count, 1)
1277         self.assertEqual(return_ns_state, expected_ns_state)
1278         self.assertEqual(return_operation_state, expected_operation_state)
1279         self.assertEqual(return_operation_error, expected_operation_error)
1280         self.assertEqual(return_ns_operational_state, expected_ns_operational_state)
1281         self.assertEqual(return_vnfr_revision, expected_vnfr_revision)
1282
1283     def test_ns_update_find_sw_version_vnfd_not_includes(self):
1284         """Find software version, VNFD does not have software version"""
1285
1286         db_vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
1287         expected_result = "1.0"
1288         result = find_software_version(db_vnfd)
1289         self.assertEqual(result, expected_result, "Default sw version should be 1.0")
1290
1291     def test_ns_update_find_sw_version_vnfd_includes(self):
1292         """Find software version, VNFD includes software version"""
1293
1294         db_vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
1295         db_vnfd["software-version"] = "3.1"
1296         expected_result = "3.1"
1297         result = find_software_version(db_vnfd)
1298         self.assertEqual(result, expected_result, "VNFD software version is wrong")
1299
1300     @patch("os.path.exists")
1301     @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1302     @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1303     def test_ns_update_check_charm_hash_not_changed(
1304         self, mock_compare_charm_hash, mock_compare_charmdir_hash, mock_path_exists
1305     ):
1306         """Check charm hash, Hash did not change"""
1307
1308         current_path, target_path = "/tmp/charm1", "/tmp/charm1"
1309
1310         fs = Mock()
1311         fs.path.__add__ = Mock()
1312         fs.path.side_effect = [current_path, target_path]
1313         fs.path.__add__.side_effect = [current_path, target_path]
1314
1315         mock_path_exists.side_effect = [True, True]
1316
1317         mock_compare_charmdir_hash.return_value = callable(False)
1318         mock_compare_charm_hash.return_value = callable(False)
1319
1320         instance = self.my_ns
1321         instance.fs = fs
1322         expected_result = False
1323
1324         result = instance.check_charm_hash_changed(current_path, target_path)
1325         self.assertEqual(result, expected_result, "Wrong charm hash control value")
1326         self.assertEqual(mock_path_exists.call_count, 2)
1327         self.assertEqual(mock_compare_charmdir_hash.call_count, 1)
1328         self.assertEqual(mock_compare_charm_hash.call_count, 0)
1329
1330     @patch("os.path.exists")
1331     @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1332     @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1333     def test_ns_update_check_charm_hash_changed(
1334         self, mock_compare_charm_hash, mock_compare_charmdir_hash, mock_path_exists
1335     ):
1336         """Check charm hash, Hash has changed"""
1337
1338         current_path, target_path = "/tmp/charm1", "/tmp/charm2"
1339
1340         fs = Mock()
1341         fs.path.__add__ = Mock()
1342         fs.path.side_effect = [current_path, target_path, current_path, target_path]
1343         fs.path.__add__.side_effect = [
1344             current_path,
1345             target_path,
1346             current_path,
1347             target_path,
1348         ]
1349
1350         mock_path_exists.side_effect = [True, True]
1351         mock_compare_charmdir_hash.return_value = callable(True)
1352         mock_compare_charm_hash.return_value = callable(True)
1353
1354         instance = self.my_ns
1355         instance.fs = fs
1356         expected_result = True
1357
1358         result = instance.check_charm_hash_changed(current_path, target_path)
1359         self.assertEqual(result, expected_result, "Wrong charm hash control value")
1360         self.assertEqual(mock_path_exists.call_count, 2)
1361         self.assertEqual(mock_compare_charmdir_hash.call_count, 1)
1362         self.assertEqual(mock_compare_charm_hash.call_count, 0)
1363
1364     @patch("os.path.exists")
1365     @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1366     @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1367     def test_ns_update_check_no_charm_path(
1368         self, mock_compare_charm_hash, mock_compare_charmdir_hash, mock_path_exists
1369     ):
1370         """Check charm hash, Charm path does not exist"""
1371
1372         current_path, target_path = "/tmp/charm1", "/tmp/charm2"
1373
1374         fs = Mock()
1375         fs.path.__add__ = Mock()
1376         fs.path.side_effect = [current_path, target_path, current_path, target_path]
1377         fs.path.__add__.side_effect = [
1378             current_path,
1379             target_path,
1380             current_path,
1381             target_path,
1382         ]
1383
1384         mock_path_exists.side_effect = [True, False]
1385
1386         mock_compare_charmdir_hash.return_value = callable(False)
1387         mock_compare_charm_hash.return_value = callable(False)
1388         instance = self.my_ns
1389         instance.fs = fs
1390
1391         with self.assertRaises(LcmException):
1392             instance.check_charm_hash_changed(current_path, target_path)
1393             self.assertEqual(mock_path_exists.call_count, 2)
1394             self.assertEqual(mock_compare_charmdir_hash.call_count, 0)
1395             self.assertEqual(mock_compare_charm_hash.call_count, 0)
1396
1397     def test_ns_update_check_juju_bundle_existence_bundle_exists(self):
1398         """Check juju bundle existence"""
1399         test_vnfd2 = self.db.get_one(
1400             "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
1401         )
1402         expected_result = "stable/native-kdu"
1403         result = check_juju_bundle_existence(test_vnfd2)
1404         self.assertEqual(result, expected_result, "Wrong juju bundle name")
1405
1406     def test_ns_update_check_juju_bundle_existence_bundle_does_not_exist(self):
1407         """Check juju bundle existence"""
1408         test_vnfd1 = self.db.get_one("vnfds", {"_id": vnfd_id})
1409         expected_result = None
1410         result = check_juju_bundle_existence(test_vnfd1)
1411         self.assertEqual(result, expected_result, "Wrong juju bundle name")
1412
1413     def test_ns_update_check_juju_bundle_existence_empty_vnfd(self):
1414         """Check juju bundle existence"""
1415         test_vnfd1 = {}
1416         expected_result = None
1417         result = check_juju_bundle_existence(test_vnfd1)
1418         self.assertEqual(result, expected_result, "Wrong juju bundle name")
1419
1420     def test_ns_update_check_juju_bundle_existence_invalid_vnfd(self):
1421         """Check juju bundle existence"""
1422         test_vnfd1 = [{"_id": vnfd_id}]
1423         with self.assertRaises(AttributeError):
1424             check_juju_bundle_existence(test_vnfd1)
1425
1426     def test_ns_update_check_juju_charm_artifacts_base_folder_wth_pkgdir(self):
1427         """Check charm artifacts"""
1428         base_folder = {"folder": vnfd_id, "pkg-dir": "hackfest_3charmed_vnfd"}
1429         charm_name = "simple"
1430         charm_type = "lxc_proxy_charm"
1431         revision = 3
1432         expected_result = f"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple"
1433         result = get_charm_artifact_path(base_folder, charm_name, charm_type, revision)
1434         self.assertEqual(result, expected_result, "Wrong charm artifact path")
1435
1436     def test_ns_update_check_juju_charm_artifacts_base_folder_wthout_pkgdir(self):
1437         """Check charm artifacts, SOL004 packages"""
1438         base_folder = {"folder": vnfd_id}
1439         charm_name = "basic"
1440         charm_type, revision = "", ""
1441         expected_result = f"{vnfd_id}/Scripts/helm-charts/basic"
1442         result = get_charm_artifact_path(base_folder, charm_name, charm_type, revision)
1443         self.assertEqual(result, expected_result, "Wrong charm artifact path")
1444
1445
1446 class TestInstantiateN2VC(TestBaseNS):
1447     async def setUp(self):
1448         await super().setUp()
1449         self.db_nsr = yaml.safe_load(descriptors.db_nsrs_text)[0]
1450         self.db_vnfr = yaml.safe_load(descriptors.db_vnfrs_text)[0]
1451         self.vca_index = 1
1452         self.my_ns._write_configuration_status = Mock()
1453
1454     async def call_instantiate_N2VC(self):
1455         logging_text = "N2VC Instantiation"
1456         config_descriptor = {"config-access": {"ssh-access": {"default-user": "admin"}}}
1457         base_folder = {"pkg-dir": "", "folder": "~"}
1458         stage = ["Stage", "Message"]
1459
1460         await self.my_ns.instantiate_N2VC(
1461             logging_text=logging_text,
1462             vca_index=self.vca_index,
1463             nsi_id="nsi_id",
1464             db_nsr=self.db_nsr,
1465             db_vnfr=self.db_vnfr,
1466             vdu_id=None,
1467             kdu_name=None,
1468             vdu_index=None,
1469             kdu_index=None,
1470             config_descriptor=config_descriptor,
1471             deploy_params={},
1472             base_folder=base_folder,
1473             nslcmop_id="nslcmop_id",
1474             stage=stage,
1475             vca_type="native_charm",
1476             vca_name="vca_name",
1477             ee_config_descriptor={},
1478         )
1479
1480     def check_config_status(self, expected_status):
1481         self.my_ns._write_configuration_status.assert_called_with(
1482             nsr_id=self.db_nsr["_id"], vca_index=self.vca_index, status=expected_status
1483         )
1484
1485     async def call_ns_add_relation(self):
1486         ee_relation = EERelation(
1487             {
1488                 "nsr-id": self.db_nsr["_id"],
1489                 "vdu-profile-id": None,
1490                 "kdu-resource-profile-id": None,
1491                 "vnf-profile-id": "hackfest_vnf1",
1492                 "execution-environment-ref": "f48163a6-c807-47bc-9682-f72caef5af85.alf-c-ab",
1493                 "endpoint": "127.0.0.1",
1494             }
1495         )
1496
1497         relation = Relation("relation-name", ee_relation, ee_relation)
1498         cached_vnfrs = {"hackfest_vnf1": self.db_vnfr}
1499
1500         return await self.my_ns._add_relation(
1501             relation=relation,
1502             vca_type="native_charm",
1503             db_nsr=self.db_nsr,
1504             cached_vnfds={},
1505             cached_vnfrs=cached_vnfrs,
1506         )
1507
1508     async def test_add_relation_ok(self):
1509         await self.call_instantiate_N2VC()
1510         self.check_config_status(expected_status="READY")
1511
1512     async def test_add_relation_returns_false_raises_exception(self):
1513         self.my_ns._add_vca_relations = asynctest.CoroutineMock(return_value=False)
1514
1515         with self.assertRaises(LcmException) as exception:
1516             await self.call_instantiate_N2VC()
1517
1518         exception_msg = "Relations could not be added to VCA."
1519         self.assertTrue(exception_msg in str(exception.exception))
1520         self.check_config_status(expected_status="BROKEN")
1521
1522     async def test_add_relation_raises_lcm_exception(self):
1523         exception_msg = "Relations FAILED"
1524         self.my_ns._add_vca_relations = asynctest.CoroutineMock(
1525             side_effect=LcmException(exception_msg)
1526         )
1527
1528         with self.assertRaises(LcmException) as exception:
1529             await self.call_instantiate_N2VC()
1530
1531         self.assertTrue(exception_msg in str(exception.exception))
1532         self.check_config_status(expected_status="BROKEN")
1533
1534     async def test_n2vc_add_relation_fails_raises_exception(self):
1535         exception_msg = "N2VC failed to add relations"
1536         self.my_ns.n2vc.add_relation = asynctest.CoroutineMock(
1537             side_effect=N2VCException(exception_msg)
1538         )
1539         with self.assertRaises(LcmException) as exception:
1540             await self.call_ns_add_relation()
1541         self.assertTrue(exception_msg in str(exception.exception))
1542
1543     async def test_n2vc_add_relation_ok_returns_true(self):
1544         self.my_ns.n2vc.add_relation = asynctest.CoroutineMock(return_value=None)
1545         self.assertTrue(await self.call_ns_add_relation())
1546
1547
1548 class TestGetVNFRelations(TestBaseNS):
1549     async def setUp(self):
1550         await super().setUp()
1551         self.db_nsd = yaml.safe_load(descriptors.db_nsds_text)[0]
1552
1553     def test_ns_charm_vca_returns_empty_relations(self):
1554         ns_charm_vca = {"member-vnf-index": None, "target_element": "ns"}
1555         nsr_id = self.db_nsd["id"]
1556         deployed_vca = DeployedVCA(nsr_id, ns_charm_vca)
1557
1558         expected_relations = []
1559         self.assertEqual(
1560             expected_relations,
1561             self.my_ns._get_vnf_relations(
1562                 nsr_id=nsr_id, nsd=self.db_nsd, vca=deployed_vca, cached_vnfds={}
1563             ),
1564         )
1565
1566     def test_vnf_returns_relation(self):
1567         vnf_vca = {
1568             "member-vnf-index": "1",
1569             "target_element": "vnf/0",
1570             "ee_descriptor_id": "simple-ee",
1571             "vdu_id": "mgmtVM",
1572         }
1573         nsr_id = self.db_nsd["id"]
1574         deployed_vca = DeployedVCA(nsr_id, vnf_vca)
1575
1576         provider_dict = {
1577             "nsr-id": nsr_id,
1578             "vnf-profile-id": "1",
1579             "vdu-profile-id": "mgmtVM",
1580             "kdu-resource-profile-id": None,
1581             "execution-environment-ref": "simple-ee",
1582             "endpoint": "interface",
1583         }
1584
1585         requirer_dict = {
1586             "nsr-id": nsr_id,
1587             "vnf-profile-id": "1",
1588             "vdu-profile-id": "dataVM",
1589             "kdu-resource-profile-id": None,
1590             "execution-environment-ref": "simple-ee",
1591             "endpoint": "interface",
1592         }
1593
1594         provider = EERelation(provider_dict)
1595         requirer = EERelation(requirer_dict)
1596         relation = Relation("relation", provider, requirer)
1597
1598         relations_found = self.my_ns._get_vnf_relations(
1599             nsr_id=nsr_id, nsd=self.db_nsd, vca=deployed_vca, cached_vnfds={}
1600         )
1601
1602         self.assertEqual(1, len(relations_found))
1603         self.assertEqual(relation, relations_found[0])
1604
1605
1606 if __name__ == "__main__":
1607     asynctest.main()