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