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