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