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