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 uuid import uuid4
  31
  32 from osm_lcm.tests import test_db_descriptors as descriptors
  33
  34 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  35
  36 """ Perform unittests using asynctest of osm_lcm.ns module
  37 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
  38     OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
  39     OSMLCMTEST_NS_NAME: change name of NS
  40     OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
  41     OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
  42     OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
  43     OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
  44     OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
  45     OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
  46     OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
  47     OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
  48     OSMLCM_VCA_XXX: configuration of N2VC
  49     OSMLCM_RO_XXX: configuration of RO
  50 """
  51
  52 lcm_config = {
  53     "global": {"loglevel": "DEBUG"},
  54     "timeout": {},
  55     "VCA": {  # TODO replace with os.get_env to get other configurations
  56         "host": getenv("OSMLCM_VCA_HOST", "vca"),
  57         "port": getenv("OSMLCM_VCA_PORT", 17070),
  58         "user": getenv("OSMLCM_VCA_USER", "admin"),
  59         "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
  60         "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
  61         "ca_cert": getenv("OSMLCM_VCA_CACERT", None),
  62         "apiproxy": getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1"),
  63     },
  64     "ro_config": {
  65         "uri": "http://{}:{}/openmano".format(
  66             getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
  67         ),
  68         "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
  69         "logger_name": "lcm.ROclient",
  70         "loglevel": "DEBUG",
  71         "ng": True,
  72     },
  73 }
  74
  75
  76 class TestMyNS(asynctest.TestCase):
  77     async def _n2vc_DeployCharms(
  78         self,
  79         model_name,
  80         application_name,
  81         vnfd,
  82         charm_path,
  83         params={},
  84         machine_spec={},
  85         callback=None,
  86         *callback_args
  87     ):
  88         if callback:
  89             for status, message in (
  90                 ("maintenance", "installing sofwware"),
  91                 ("active", "Ready!"),
  92             ):
  93                 # call callback after some time
  94                 asyncio.sleep(5, loop=self.loop)
  95                 callback(model_name, application_name, status, message, *callback_args)
  96
  97     @staticmethod
  98     def _n2vc_FormatApplicationName(*args):
  99         num_calls = 0
 100         while True:
 101             yield "app_name-{}".format(num_calls)
 102             num_calls += 1
 103
 104     def _n2vc_CreateExecutionEnvironment(
 105         self, namespace, reuse_ee_id, db_dict, *args, **kwargs
 106     ):
 107         k_list = namespace.split(".")
 108         ee_id = k_list[1] + "."
 109         if len(k_list) >= 2:
 110             for k in k_list[2:4]:
 111                 ee_id += k[:8]
 112         else:
 113             ee_id += "_NS_"
 114         return ee_id, {}
 115
 116     def _ro_status(self, *args, **kwargs):
 117         print("Args > {}".format(args))
 118         print("kwargs > {}".format(kwargs))
 119         if kwargs.get("delete"):
 120             ro_ns_desc = yaml.load(
 121                 descriptors.ro_delete_action_text, Loader=yaml.Loader
 122             )
 123             while True:
 124                 yield ro_ns_desc
 125
 126         ro_ns_desc = yaml.load(descriptors.ro_ns_text, Loader=yaml.Loader)
 127
 128         # if ip address provided, replace descriptor
 129         ip_addresses = getenv("OSMLCMTEST_NS_IPADDRESS", "")
 130         if ip_addresses:
 131             ip_addresses_list = ip_addresses.split(",")
 132             for vnf in ro_ns_desc["vnfs"]:
 133                 if not ip_addresses_list:
 134                     break
 135                 vnf["ip_address"] = ip_addresses_list[0]
 136                 for vm in vnf["vms"]:
 137                     if not ip_addresses_list:
 138                         break
 139                     vm["ip_address"] = ip_addresses_list.pop(0)
 140
 141         while True:
 142             yield ro_ns_desc
 143             for net in ro_ns_desc["nets"]:
 144                 if net["status"] != "ACTIVE":
 145                     net["status"] = "ACTIVE"
 146                     break
 147             else:
 148                 for vnf in ro_ns_desc["vnfs"]:
 149                     for vm in vnf["vms"]:
 150                         if vm["status"] != "ACTIVE":
 151                             vm["status"] = "ACTIVE"
 152                             break
 153
 154     def _ro_deploy(self, *args, **kwargs):
 155         return {"action_id": args[1]["action_id"], "nsr_id": args[0], "status": "ok"}
 156
 157     def _return_uuid(self, *args, **kwargs):
 158         return str(uuid4())
 159
 160     async def setUp(self):
 161
 162         # Mock DB
 163         if not getenv("OSMLCMTEST_DB_NOMOCK"):
 164             # Cleanup singleton Database instance
 165             Database.instance = None
 166
 167             self.db = Database({"database": {"driver": "memory"}}).instance.db
 168             self.db.create_list(
 169                 "vnfds", yaml.load(descriptors.db_vnfds_text, Loader=yaml.Loader)
 170             )
 171             self.db.create_list(
 172                 "nsds", yaml.load(descriptors.db_nsds_text, Loader=yaml.Loader)
 173             )
 174             self.db.create_list(
 175                 "nsrs", yaml.load(descriptors.db_nsrs_text, Loader=yaml.Loader)
 176             )
 177             self.db.create_list(
 178                 "vim_accounts",
 179                 yaml.load(descriptors.db_vim_accounts_text, Loader=yaml.Loader),
 180             )
 181             self.db.create_list(
 182                 "k8sclusters",
 183                 yaml.load(descriptors.db_k8sclusters_text, Loader=yaml.Loader),
 184             )
 185             self.db.create_list(
 186                 "nslcmops", yaml.load(descriptors.db_nslcmops_text, Loader=yaml.Loader)
 187             )
 188             self.db.create_list(
 189                 "vnfrs", yaml.load(descriptors.db_vnfrs_text, Loader=yaml.Loader)
 190             )
 191             self.db_vim_accounts = yaml.load(
 192                 descriptors.db_vim_accounts_text, Loader=yaml.Loader
 193             )
 194
 195         # Mock kafka
 196         self.msg = asynctest.Mock(MsgKafka())
 197
 198         # Mock filesystem
 199         if not getenv("OSMLCMTEST_FS_NOMOCK"):
 200             self.fs = asynctest.Mock(
 201                 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
 202             )
 203             self.fs.get_params.return_value = {
 204                 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
 205             }
 206             self.fs.file_open = asynctest.mock_open()
 207             # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock()  # called on a python "with"
 208             # self.fs.file_open.return_value.__enter__.return_value.read.return_value = ""   # empty file
 209
 210         # Mock TaskRegistry
 211         self.lcm_tasks = asynctest.Mock(TaskRegistry())
 212         self.lcm_tasks.lock_HA.return_value = True
 213         self.lcm_tasks.waitfor_related_HA.return_value = None
 214         self.lcm_tasks.lookfor_related.return_value = ("", [])
 215
 216         # Mock VCA - K8s
 217         if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
 218             ns.K8sJujuConnector = asynctest.MagicMock(ns.K8sJujuConnector)
 219             ns.K8sHelmConnector = asynctest.MagicMock(ns.K8sHelmConnector)
 220             ns.K8sHelm3Connector = asynctest.MagicMock(ns.K8sHelm3Connector)
 221
 222         if not getenv("OSMLCMTEST_VCA_NOMOCK"):
 223             ns.N2VCJujuConnector = asynctest.MagicMock(ns.N2VCJujuConnector)
 224             ns.LCMHelmConn = asynctest.MagicMock(ns.LCMHelmConn)
 225
 226         # Create NsLCM class
 227         self.my_ns = ns.NsLcm(self.msg, self.lcm_tasks, lcm_config, self.loop)
 228         self.my_ns.fs = self.fs
 229         self.my_ns.db = self.db
 230         self.my_ns._wait_dependent_n2vc = asynctest.CoroutineMock()
 231
 232         # Mock logging
 233         if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
 234             self.my_ns.logger = asynctest.Mock(self.my_ns.logger)
 235
 236         # Mock VCA - N2VC
 237         if not getenv("OSMLCMTEST_VCA_NOMOCK"):
 238             pub_key = getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
 239             # self.my_ns.n2vc = asynctest.Mock(N2VC())
 240             self.my_ns.n2vc.GetPublicKey.return_value = getenv(
 241                 "OSMLCM_VCA_PUBKEY", "public_key"
 242             )
 243             # allow several versions of n2vc
 244             self.my_ns.n2vc.FormatApplicationName = asynctest.Mock(
 245                 side_effect=self._n2vc_FormatApplicationName()
 246             )
 247             self.my_ns.n2vc.DeployCharms = asynctest.CoroutineMock(
 248                 side_effect=self._n2vc_DeployCharms
 249             )
 250             self.my_ns.n2vc.create_execution_environment = asynctest.CoroutineMock(
 251                 side_effect=self._n2vc_CreateExecutionEnvironment
 252             )
 253             self.my_ns.n2vc.install_configuration_sw = asynctest.CoroutineMock(
 254                 return_value=pub_key
 255             )
 256             self.my_ns.n2vc.get_ee_ssh_public__key = asynctest.CoroutineMock(
 257                 return_value=pub_key
 258             )
 259             self.my_ns.n2vc.exec_primitive = asynctest.CoroutineMock(
 260                 side_effect=self._return_uuid
 261             )
 262             self.my_ns.n2vc.exec_primitive = asynctest.CoroutineMock(
 263                 side_effect=self._return_uuid
 264             )
 265             self.my_ns.n2vc.GetPrimitiveStatus = asynctest.CoroutineMock(
 266                 return_value="completed"
 267             )
 268             self.my_ns.n2vc.GetPrimitiveOutput = asynctest.CoroutineMock(
 269                 return_value={"result": "ok", "pubkey": pub_key}
 270             )
 271             self.my_ns.n2vc.delete_execution_environment = asynctest.CoroutineMock(
 272                 return_value=None
 273             )
 274             self.my_ns.n2vc.get_public_key = asynctest.CoroutineMock(
 275                 return_value=getenv("OSMLCM_VCA_PUBKEY", "public_key")
 276             )
 277             self.my_ns.n2vc.delete_namespace = asynctest.CoroutineMock(
 278                 return_value=None
 279             )
 280
 281         # Mock RO
 282         if not getenv("OSMLCMTEST_RO_NOMOCK"):
 283             self.my_ns.RO = asynctest.Mock(
 284                 NgRoClient(self.loop, **lcm_config["ro_config"])
 285             )
 286             # TODO first time should be empty list, following should return a dict
 287             # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
 288             self.my_ns.RO.deploy = asynctest.CoroutineMock(
 289                 self.my_ns.RO.deploy, side_effect=self._ro_deploy
 290             )
 291             # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status)
 292             # self.my_ns.RO.create_action = asynctest.CoroutineMock(self.my_ns.RO.create_action,
 293             #                                                      return_value={"vm-id": {"vim_result": 200,
 294             #                                                                              "description": "done"}})
 295             self.my_ns.RO.delete = asynctest.CoroutineMock(self.my_ns.RO.delete)
 296
 297     # @asynctest.fail_on(active_handles=True)   # all async tasks must be completed
 298     # async def test_instantiate(self):
 299     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 300     #     nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 301     #     # print("Test instantiate started")
 302
 303     #     # delete deployed information of database
 304     #     if not getenv("OSMLCMTEST_DB_NOMOCK"):
 305     #         if self.db.get_list("nsrs")[0]["_admin"].get("deployed"):
 306     #             del self.db.get_list("nsrs")[0]["_admin"]["deployed"]
 307     #         for db_vnfr in self.db.get_list("vnfrs"):
 308     #             db_vnfr.pop("ip_address", None)
 309     #             for db_vdur in db_vnfr["vdur"]:
 310     #                 db_vdur.pop("ip_address", None)
 311     #                 db_vdur.pop("mac_address", None)
 312     #         if getenv("OSMLCMTEST_RO_VIMID"):
 313     #             self.db.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
 314     #         if getenv("OSMLCMTEST_RO_VIMID"):
 315     #             self.db.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
 316
 317     #     await self.my_ns.instantiate(nsr_id, nslcmop_id)
 318
 319     #     self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
 320     #                                               {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
 321     #                                                "operationState": "COMPLETED"},
 322     #                                               loop=self.loop)
 323     #     self.lcm_tasks.lock_HA.assert_called_once_with('ns', 'nslcmops', nslcmop_id)
 324     #     if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
 325     #         self.assertTrue(self.my_ns.logger.debug.called, "Debug method not called")
 326     #         self.my_ns.logger.error.assert_not_called()
 327     #         self.my_ns.logger.exception().assert_not_called()
 328
 329     #     if not getenv("OSMLCMTEST_DB_NOMOCK"):
 330     #         self.assertTrue(self.db.set_one.called, "db.set_one not called")
 331     #         db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 332     #         db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
 333     #         self.assertEqual(db_nsr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
 334     #         for vnfr in db_vnfrs_list:
 335     #             self.assertEqual(vnfr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
 336
 337     #     if not getenv("OSMLCMTEST_VCA_NOMOCK"):
 338     #         # check intial-primitives called
 339     #         self.assertTrue(self.my_ns.n2vc.exec_primitive.called,
 340     #                         "Exec primitive not called for initial config primitive")
 341     #         for _call in self.my_ns.n2vc.exec_primitive.call_args_list:
 342     #             self.assertIn(_call[1]["primitive_name"], ("config", "touch"),
 343     #                           "called exec primitive with a primitive different than config or touch")
 344
 345     #     # TODO add more checks of called methods
 346     #     # TODO add a terminate
 347
 348     # async def test_instantiate_ee_list(self):
 349     #     # Using modern IM where configuration is in the new format of execution_environment_list
 350     #     ee_descriptor_id = "charm_simple"
 351     #     non_used_initial_primitive = {
 352     #         "name": "not_to_be_called",
 353     #         "seq": 3,
 354     #         "execution-environment-ref": "not_used_ee"
 355     #     }
 356     #     ee_list = [
 357     #         {
 358     #             "id": ee_descriptor_id,
 359     #             "juju": {"charm": "simple"},
 360
 361     #         },
 362     #     ]
 363
 364     #     self.db.set_one(
 365     #         "vnfds",
 366     #         q_filter={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"},
 367     #         update_dict={"vnf-configuration.0.execution-environment-list": ee_list,
 368     #                      "vnf-configuration.0.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id,
 369     #                      "vnf-configuration.0.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id,
 370     #                      "vnf-configuration.0.initial-config-primitive.2": non_used_initial_primitive,
 371     #                      "vnf-configuration.0.config-primitive.0.execution-environment-ref": ee_descriptor_id,
 372     #                      "vnf-configuration.0.config-primitive.0.execution-environment-primitive": "touch_charm",
 373     #                      },
 374     #         unset={"vnf-configuration.juju": None})
 375     #     await self.test_instantiate()
 376     #     # this will check that the initial-congig-primitive 'not_to_be_called' is not called
 377
 378     # Test scale() and related methods
 379     @asynctest.fail_on(active_handles=True)  # all async tasks must be completed
 380     async def test_scale(self):
 381         # print("Test scale started")
 382
 383         # TODO: Add more higher-lever tests here, for example:
 384         # scale-out/scale-in operations with success/error result
 385
 386         # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
 387         nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 388         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 389         await self.my_ns.scale(nsr_id, nslcmop_id)
 390         expected_value = "FAILED"
 391         return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
 392             "operationState"
 393         )
 394         self.assertEqual(return_value, expected_value)
 395         # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
 396
 397     async def test_vca_status_refresh(self):
 398         nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 399         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 400         await self.my_ns.vca_status_refresh(nsr_id, nslcmop_id)
 401         expected_value = dict()
 402         return_value = dict()
 403         vnf_descriptors = self.db.get_list("vnfds")
 404         for i, _ in enumerate(vnf_descriptors):
 405             for j, value in enumerate(vnf_descriptors[i]["df"]):
 406                 if "lcm-operations-configuration" in vnf_descriptors[i]["df"][j]:
 407                     if (
 408                         "day1-2"
 409                         in value["lcm-operations-configuration"][
 410                             "operate-vnf-op-config"
 411                         ]
 412                     ):
 413                         for k, v in enumerate(
 414                             value["lcm-operations-configuration"][
 415                                 "operate-vnf-op-config"
 416                             ]["day1-2"]
 417                         ):
 418                             if "juju" in v["execution-environment-list"][k]:
 419                                 expected_value = self.db.get_list("nsrs")[i][
 420                                     "vcaStatus"
 421                                 ]
 422                                 await self.my_ns._on_update_n2vc_db(
 423                                     "nsrs", {"_id": nsr_id}, "_admin.deployed.VCA.0", {}
 424                                 )
 425                                 return_value = self.db.get_list("nsrs")[i]["vcaStatus"]
 426         self.assertEqual(return_value, expected_value)
 427
 428     # Test _retry_or_skip_suboperation()
 429     # Expected result:
 430     # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
 431     # - if marked as anything but 'COMPLETED', the suboperation index is expected
 432     def test_scale_retry_or_skip_suboperation(self):
 433         # Load an alternative 'nslcmops' YAML for this test
 434         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 435         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 436         op_index = 2
 437         # Test when 'operationState' is 'COMPLETED'
 438         db_nslcmop["_admin"]["operations"][op_index]["operationState"] = "COMPLETED"
 439         return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
 440         expected_value = self.my_ns.SUBOPERATION_STATUS_SKIP
 441         self.assertEqual(return_value, expected_value)
 442         # Test when 'operationState' is not 'COMPLETED'
 443         db_nslcmop["_admin"]["operations"][op_index]["operationState"] = None
 444         return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
 445         expected_value = op_index
 446         self.assertEqual(return_value, expected_value)
 447
 448     # Test _find_suboperation()
 449     # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
 450     def test_scale_find_suboperation(self):
 451         # Load an alternative 'nslcmops' YAML for this test
 452         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 453         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 454         # Find this sub-operation
 455         op_index = 2
 456         vnf_index = db_nslcmop["_admin"]["operations"][op_index]["member_vnf_index"]
 457         primitive = db_nslcmop["_admin"]["operations"][op_index]["primitive"]
 458         primitive_params = db_nslcmop["_admin"]["operations"][op_index][
 459             "primitive_params"
 460         ]
 461         match = {
 462             "member_vnf_index": vnf_index,
 463             "primitive": primitive,
 464             "primitive_params": primitive_params,
 465         }
 466         found_op_index = self.my_ns._find_suboperation(db_nslcmop, match)
 467         self.assertEqual(found_op_index, op_index)
 468         # Test with not-matching params
 469         match = {
 470             "member_vnf_index": vnf_index,
 471             "primitive": "",
 472             "primitive_params": primitive_params,
 473         }
 474         found_op_index = self.my_ns._find_suboperation(db_nslcmop, match)
 475         self.assertEqual(found_op_index, self.my_ns.SUBOPERATION_STATUS_NOT_FOUND)
 476         # Test with None
 477         match = None
 478         found_op_index = self.my_ns._find_suboperation(db_nslcmop, match)
 479         self.assertEqual(found_op_index, self.my_ns.SUBOPERATION_STATUS_NOT_FOUND)
 480
 481     # Test _update_suboperation_status()
 482     def test_scale_update_suboperation_status(self):
 483         self.db.set_one = asynctest.Mock()
 484         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 485         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 486         op_index = 0
 487         # Force the initial values to be distinct from the updated ones
 488         q_filter = {"_id": db_nslcmop["_id"]}
 489         # Test to change 'operationState' and 'detailed-status'
 490         operationState = "COMPLETED"
 491         detailed_status = "Done"
 492         expected_update_dict = {
 493             "_admin.operations.0.operationState": operationState,
 494             "_admin.operations.0.detailed-status": detailed_status,
 495         }
 496         self.my_ns._update_suboperation_status(
 497             db_nslcmop, op_index, operationState, detailed_status
 498         )
 499         self.db.set_one.assert_called_once_with(
 500             "nslcmops",
 501             q_filter=q_filter,
 502             update_dict=expected_update_dict,
 503             fail_on_empty=False,
 504         )
 505
 506     def test_scale_add_suboperation(self):
 507         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 508         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 509         vnf_index = "1"
 510         num_ops_before = len(db_nslcmop.get("_admin", {}).get("operations", [])) - 1
 511         vdu_id = None
 512         vdu_count_index = None
 513         vdu_name = None
 514         primitive = "touch"
 515         mapped_primitive_params = {
 516             "parameter": [
 517                 {
 518                     "data-type": "STRING",
 519                     "name": "filename",
 520                     "default-value": "<touch_filename2>",
 521                 }
 522             ],
 523             "name": "touch",
 524         }
 525         operationState = "PROCESSING"
 526         detailed_status = "In progress"
 527         operationType = "PRE-SCALE"
 528         # Add a 'pre-scale' suboperation
 529         op_index_after = self.my_ns._add_suboperation(
 530             db_nslcmop,
 531             vnf_index,
 532             vdu_id,
 533             vdu_count_index,
 534             vdu_name,
 535             primitive,
 536             mapped_primitive_params,
 537             operationState,
 538             detailed_status,
 539             operationType,
 540         )
 541         self.assertEqual(op_index_after, num_ops_before + 1)
 542
 543         # Delete all suboperations and add the same operation again
 544         del db_nslcmop["_admin"]["operations"]
 545         op_index_zero = self.my_ns._add_suboperation(
 546             db_nslcmop,
 547             vnf_index,
 548             vdu_id,
 549             vdu_count_index,
 550             vdu_name,
 551             primitive,
 552             mapped_primitive_params,
 553             operationState,
 554             detailed_status,
 555             operationType,
 556         )
 557         self.assertEqual(op_index_zero, 0)
 558
 559         # Add a 'RO' suboperation
 560         RO_nsr_id = "1234567890"
 561         RO_scaling_info = [
 562             {
 563                 "type": "create",
 564                 "count": 1,
 565                 "member-vnf-index": "1",
 566                 "osm_vdu_id": "dataVM",
 567             }
 568         ]
 569         op_index = self.my_ns._add_suboperation(
 570             db_nslcmop,
 571             vnf_index,
 572             vdu_id,
 573             vdu_count_index,
 574             vdu_name,
 575             primitive,
 576             mapped_primitive_params,
 577             operationState,
 578             detailed_status,
 579             operationType,
 580             RO_nsr_id,
 581             RO_scaling_info,
 582         )
 583         db_RO_nsr_id = db_nslcmop["_admin"]["operations"][op_index]["RO_nsr_id"]
 584         self.assertEqual(op_index, 1)
 585         self.assertEqual(RO_nsr_id, db_RO_nsr_id)
 586
 587         # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
 588         op_index_invalid = self.my_ns._add_suboperation(
 589             None, None, None, None, None, None, None, None, None, None, None
 590         )
 591         self.assertEqual(op_index_invalid, self.my_ns.SUBOPERATION_STATUS_NOT_FOUND)
 592
 593     # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
 594     # check the possible return values:
 595     # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
 596     # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
 597     # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
 598     def test_scale_check_or_add_scale_suboperation(self):
 599         nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 600         db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 601         operationType = "PRE-SCALE"
 602         vnf_index = "1"
 603         primitive = "touch"
 604         primitive_params = {
 605             "parameter": [
 606                 {
 607                     "data-type": "STRING",
 608                     "name": "filename",
 609                     "default-value": "<touch_filename2>",
 610                 }
 611             ],
 612             "name": "touch",
 613         }
 614
 615         # Delete all sub-operations to be sure this is a new sub-operation
 616         del db_nslcmop["_admin"]["operations"]
 617
 618         # Add a new sub-operation
 619         # For new sub-operations, operationState is set to 'PROCESSING' by default
 620         op_index_new = self.my_ns._check_or_add_scale_suboperation(
 621             db_nslcmop, vnf_index, primitive, primitive_params, operationType
 622         )
 623         self.assertEqual(op_index_new, self.my_ns.SUBOPERATION_STATUS_NEW)
 624
 625         # Use the same parameters again to match the already added sub-operation
 626         # which has status 'PROCESSING' (!= 'COMPLETED') by default
 627         # The expected return value is a non-negative number
 628         op_index_existing = self.my_ns._check_or_add_scale_suboperation(
 629             db_nslcmop, vnf_index, primitive, primitive_params, operationType
 630         )
 631         self.assertTrue(op_index_existing >= 0)
 632
 633         # Change operationState 'manually' for this sub-operation
 634         db_nslcmop["_admin"]["operations"][op_index_existing][
 635             "operationState"
 636         ] = "COMPLETED"
 637         # Then use the same parameters again to match the already added sub-operation,
 638         # which now has status 'COMPLETED'
 639         # The expected return value is SUBOPERATION_STATUS_SKIP
 640         op_index_skip = self.my_ns._check_or_add_scale_suboperation(
 641             db_nslcmop, vnf_index, primitive, primitive_params, operationType
 642         )
 643         self.assertEqual(op_index_skip, self.my_ns.SUBOPERATION_STATUS_SKIP)
 644
 645         # RO sub-operation test:
 646         # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
 647         RO_nsr_id = "1234567890"
 648         RO_scaling_info = [
 649             {
 650                 "type": "create",
 651                 "count": 1,
 652                 "member-vnf-index": "1",
 653                 "osm_vdu_id": "dataVM",
 654             }
 655         ]
 656         op_index_new_RO = self.my_ns._check_or_add_scale_suboperation(
 657             db_nslcmop, vnf_index, None, None, "SCALE-RO", RO_nsr_id, RO_scaling_info
 658         )
 659         self.assertEqual(op_index_new_RO, self.my_ns.SUBOPERATION_STATUS_NEW)
 660
 661         # Use the same parameters again to match the already added RO sub-operation
 662         op_index_existing_RO = self.my_ns._check_or_add_scale_suboperation(
 663             db_nslcmop, vnf_index, None, None, "SCALE-RO", RO_nsr_id, RO_scaling_info
 664         )
 665         self.assertTrue(op_index_existing_RO >= 0)
 666
 667         # Change operationState 'manually' for this RO sub-operation
 668         db_nslcmop["_admin"]["operations"][op_index_existing_RO][
 669             "operationState"
 670         ] = "COMPLETED"
 671         # Then use the same parameters again to match the already added sub-operation,
 672         # which now has status 'COMPLETED'
 673         # The expected return value is SUBOPERATION_STATUS_SKIP
 674         op_index_skip_RO = self.my_ns._check_or_add_scale_suboperation(
 675             db_nslcmop, vnf_index, None, None, "SCALE-RO", RO_nsr_id, RO_scaling_info
 676         )
 677         self.assertEqual(op_index_skip_RO, self.my_ns.SUBOPERATION_STATUS_SKIP)
 678
 679     async def test_deploy_kdus(self):
 680         nsr_id = descriptors.test_ids["TEST-KDU"]["ns"]
 681         nslcmop_id = descriptors.test_ids["TEST-KDU"]["instantiate"]
 682         db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 683         db_vnfr = self.db.get_one(
 684             "vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "multikdu"}
 685         )
 686         db_vnfrs = {"multikdu": db_vnfr}
 687         db_vnfd = self.db.get_one("vnfds", {"_id": db_vnfr["vnfd-id"]})
 688         db_vnfds = [db_vnfd]
 689         task_register = {}
 690         logging_text = "KDU"
 691         self.my_ns.k8sclusterhelm3.generate_kdu_instance_name = asynctest.mock.Mock()
 692         self.my_ns.k8sclusterhelm3.generate_kdu_instance_name.return_value = "k8s_id"
 693         self.my_ns.k8sclusterhelm3.install = asynctest.CoroutineMock()
 694         self.my_ns.k8sclusterhelm3.synchronize_repos = asynctest.CoroutineMock(
 695             return_value=("", "")
 696         )
 697         self.my_ns.k8sclusterhelm3.get_services = asynctest.CoroutineMock(
 698             return_value=([])
 699         )
 700         await self.my_ns.deploy_kdus(
 701             logging_text, nsr_id, nslcmop_id, db_vnfrs, db_vnfds, task_register
 702         )
 703         await asyncio.wait(list(task_register.keys()), timeout=100)
 704         db_nsr = self.db.get_list("nsrs")[1]
 705         self.assertIn(
 706             "K8s",
 707             db_nsr["_admin"]["deployed"],
 708             "K8s entry not created at '_admin.deployed'",
 709         )
 710         self.assertIsInstance(
 711             db_nsr["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
 712         )
 713         self.assertEqual(
 714             len(db_nsr["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
 715         )
 716         k8s_instace_info = {
 717             "kdu-instance": "k8s_id",
 718             "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
 719             "k8scluster-type": "helm-chart-v3",
 720             "kdu-name": "ldap",
 721             "member-vnf-index": "multikdu",
 722             "namespace": None,
 723         }
 724
 725         nsr_result = copy.deepcopy(db_nsr["_admin"]["deployed"]["K8s"][0])
 726         nsr_kdu_model_result = nsr_result.pop("kdu-model")
 727         expected_kdu_model = "stable/openldap:1.2.1"
 728         self.assertEqual(nsr_result, k8s_instace_info)
 729         self.assertTrue(
 730             nsr_kdu_model_result in expected_kdu_model
 731             or expected_kdu_model in nsr_kdu_model_result
 732         )
 733         nsr_result = copy.deepcopy(db_nsr["_admin"]["deployed"]["K8s"][1])
 734         nsr_kdu_model_result = nsr_result.pop("kdu-model")
 735         k8s_instace_info["kdu-name"] = "mongo"
 736         expected_kdu_model = "stable/mongodb"
 737         self.assertEqual(nsr_result, k8s_instace_info)
 738         self.assertTrue(
 739             nsr_kdu_model_result in expected_kdu_model
 740             or expected_kdu_model in nsr_kdu_model_result
 741         )
 742
 743     # async def test_instantiate_pdu(self):
 744     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 745     #     nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
 746     #     # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
 747     #     self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
 748     #                     update_dict={"ip-address": "10.205.1.46",
 749     #                                  "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
 750     #                                  "vdur.0.pdu-type": "PDU-TYPE-1",
 751     #                                  "vdur.0.ip-address": "10.205.1.46",
 752     #                                  },
 753     #                     unset={"vdur.status": None})
 754     #     self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
 755     #                     update_dict={"ip-address": "10.205.1.47",
 756     #                                  "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
 757     #                                  "vdur.0.pdu-type": "PDU-TYPE-1",
 758     #                                  "vdur.0.ip-address": "10.205.1.47",
 759     #                                  },
 760     #                     unset={"vdur.status": None})
 761
 762     #     await self.my_ns.instantiate(nsr_id, nslcmop_id)
 763     #     db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 764     #     self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
 765     #     self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
 766     #     self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
 767     #     self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
 768     #     self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
 769
 770     # @asynctest.fail_on(active_handles=True)   # all async tasks must be completed
 771     # async def test_terminate_without_configuration(self):
 772     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 773     #     nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
 774     #     # set instantiation task as completed
 775     #     self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
 776     #                      update_dict={"operationState": "COMPLETED"})
 777     #     self.db.set_one("nsrs", {"_id": nsr_id},
 778     #                     update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
 779
 780     #     await self.my_ns.terminate(nsr_id, nslcmop_id)
 781     #     db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 782     #     self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
 783     #     db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 784     #     self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 785     #     self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 786     #     self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
 787     #     self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
 788     #     self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
 789     #     self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
 790     #     db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
 791     #     for vnfr in db_vnfrs_list:
 792     #         self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
 793
 794     # @asynctest.fail_on(active_handles=True)   # all async tasks must be completed
 795     # async def test_terminate_primitive(self):
 796     #     nsr_id = descriptors.test_ids["TEST-A"]["ns"]
 797     #     nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
 798     #     # set instantiation task as completed
 799     #     self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
 800     #                      update_dict={"operationState": "COMPLETED"})
 801
 802     #     # modify vnfd descriptor to include terminate_primitive
 803     #     terminate_primitive = [{
 804     #         "name": "touch",
 805     #         "parameter": [{"name": "filename", "value": "terminate_filename"}],
 806     #         "seq": '1'
 807     #     }]
 808     #     db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
 809     #     self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
 810     #                     {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
 811
 812     #     await self.my_ns.terminate(nsr_id, nslcmop_id)
 813     #     db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 814     #     self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
 815     #     db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 816     #     self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 817     #     self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
 818     #     self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
 819     #     self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
 820     #     self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
 821     #     self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
 822
 823
 824 if __name__ == "__main__":
 825     asynctest.main()