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
6 # http://www.apache.org/licenses/LICENSE-2.0
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
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: alfonso.tiernosepulveda@telefonica.com
19 import asynctest
# pip3 install asynctest --user
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
32 from osm_lcm
.tests
import test_db_descriptors
as descriptors
34 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
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
53 "global": {"loglevel": "DEBUG"},
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"),
65 "uri": "http://{}:{}/openmano".format(
66 getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
68 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
69 "logger_name": "lcm.ROclient",
76 class TestMyNS(asynctest
.TestCase
):
77 async def _n2vc_DeployCharms(
89 for status
, message
in (
90 ("maintenance", "installing sofwware"),
93 # call callback after some time
94 asyncio
.sleep(5, loop
=self
.loop
)
95 callback(model_name
, application_name
, status
, message
, *callback_args
)
98 def _n2vc_FormatApplicationName(*args
):
101 yield "app_name-{}".format(num_calls
)
104 def _n2vc_CreateExecutionEnvironment(
105 self
, namespace
, reuse_ee_id
, db_dict
, *args
, **kwargs
107 k_list
= namespace
.split(".")
108 ee_id
= k_list
[1] + "."
110 for k
in k_list
[2:4]:
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
126 ro_ns_desc
= yaml
.load(descriptors
.ro_ns_text
, Loader
=yaml
.Loader
)
128 # if ip address provided, replace descriptor
129 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
131 ip_addresses_list
= ip_addresses
.split(",")
132 for vnf
in ro_ns_desc
["vnfs"]:
133 if not ip_addresses_list
:
135 vnf
["ip_address"] = ip_addresses_list
[0]
136 for vm
in vnf
["vms"]:
137 if not ip_addresses_list
:
139 vm
["ip_address"] = ip_addresses_list
.pop(0)
143 for net
in ro_ns_desc
["nets"]:
144 if net
["status"] != "ACTIVE":
145 net
["status"] = "ACTIVE"
148 for vnf
in ro_ns_desc
["vnfs"]:
149 for vm
in vnf
["vms"]:
150 if vm
["status"] != "ACTIVE":
151 vm
["status"] = "ACTIVE"
154 def _ro_deploy(self
, *args
, **kwargs
):
155 return {"action_id": args
[1]["action_id"], "nsr_id": args
[0], "status": "ok"}
157 def _return_uuid(self
, *args
, **kwargs
):
160 async def setUp(self
):
163 if not getenv("OSMLCMTEST_DB_NOMOCK"):
164 # Cleanup singleton Database instance
165 Database
.instance
= None
167 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
169 "vnfds", yaml
.load(descriptors
.db_vnfds_text
, Loader
=yaml
.Loader
)
172 "nsds", yaml
.load(descriptors
.db_nsds_text
, Loader
=yaml
.Loader
)
175 "nsrs", yaml
.load(descriptors
.db_nsrs_text
, Loader
=yaml
.Loader
)
179 yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
),
183 yaml
.load(descriptors
.db_k8sclusters_text
, Loader
=yaml
.Loader
),
186 "nslcmops", yaml
.load(descriptors
.db_nslcmops_text
, Loader
=yaml
.Loader
)
189 "vnfrs", yaml
.load(descriptors
.db_vnfrs_text
, Loader
=yaml
.Loader
)
191 self
.db_vim_accounts
= yaml
.load(
192 descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
196 self
.msg
= asynctest
.Mock(MsgKafka())
199 if not getenv("OSMLCMTEST_FS_NOMOCK"):
200 self
.fs
= asynctest
.Mock(
201 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
203 self
.fs
.get_params
.return_value
= {
204 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
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
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
= ("", [])
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
)
222 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
223 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
224 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
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
_n
2vc
= asynctest
.CoroutineMock()
233 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
234 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
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"
243 # allow several versions of n2vc
244 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(
245 side_effect
=self
._n
2vc
_FormatApplicationName
()
247 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(
248 side_effect
=self
._n
2vc
_DeployCharms
250 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
251 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
253 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(
256 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(
259 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
260 side_effect
=self
._return
_uuid
262 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
263 side_effect
=self
._return
_uuid
265 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(
266 return_value
="completed"
268 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(
269 return_value
={"result": "ok", "pubkey": pub_key
}
271 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(
274 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
275 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key")
277 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(
282 if not getenv("OSMLCMTEST_RO_NOMOCK"):
283 self
.my_ns
.RO
= asynctest
.Mock(
284 NgRoClient(self
.loop
, **lcm_config
["ro_config"])
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
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
)
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")
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")
317 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
319 # self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
320 # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
321 # "operationState": "COMPLETED"},
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()
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")
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")
345 # # TODO add more checks of called methods
346 # # TODO add a terminate
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",
354 # "execution-environment-ref": "not_used_ee"
358 # "id": ee_descriptor_id,
359 # "juju": {"charm": "simple"},
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",
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
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")
383 # TODO: Add more higher-lever tests here, for example:
384 # scale-out/scale-in operations with success/error result
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(
394 self
.assertEqual(return_value
, expected_value
)
395 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
397 # Test scale() for native kdu
398 # this also includes testing _scale_kdu()
399 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["ns"]
400 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["instantiate"]
402 self
.my_ns
.k8sclusterjuju
.scale
= asynctest
.mock
.CoroutineMock()
403 self
.my_ns
.k8sclusterjuju
.exec_primitive
= asynctest
.mock
.CoroutineMock()
404 self
.my_ns
.k8sclusterjuju
.get_scale_count
= asynctest
.mock
.CoroutineMock(
407 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
408 expected_value
= "COMPLETED"
409 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
412 self
.assertEqual(return_value
, expected_value
)
413 self
.my_ns
.k8sclusterjuju
.scale
.assert_called_once()
415 # Test scale() for native kdu with 2 resource
416 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["ns"]
417 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["instantiate"]
419 self
.my_ns
.k8sclusterjuju
.get_scale_count
.return_value
= 2
420 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
421 expected_value
= "COMPLETED"
422 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
425 self
.assertEqual(return_value
, expected_value
)
426 self
.my_ns
.k8sclusterjuju
.scale
.assert_called()
428 async def test_vca_status_refresh(self
):
429 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
430 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
431 await self
.my_ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
432 expected_value
= dict()
433 return_value
= dict()
434 vnf_descriptors
= self
.db
.get_list("vnfds")
435 for i
, _
in enumerate(vnf_descriptors
):
436 for j
, value
in enumerate(vnf_descriptors
[i
]["df"]):
437 if "lcm-operations-configuration" in vnf_descriptors
[i
]["df"][j
]:
440 in value
["lcm-operations-configuration"][
441 "operate-vnf-op-config"
444 for k
, v
in enumerate(
445 value
["lcm-operations-configuration"][
446 "operate-vnf-op-config"
450 v
.get("execution-environment-list")
451 and "juju" in v
["execution-environment-list"][k
]
453 expected_value
= self
.db
.get_list("nsrs")[i
][
456 await self
.my_ns
._on
_update
_n
2vc
_db
(
457 "nsrs", {"_id": nsr_id
}, "_admin.deployed.VCA.0", {}
459 return_value
= self
.db
.get_list("nsrs")[i
]["vcaStatus"]
460 self
.assertEqual(return_value
, expected_value
)
462 # Test _retry_or_skip_suboperation()
464 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
465 # - if marked as anything but 'COMPLETED', the suboperation index is expected
466 def test_scale_retry_or_skip_suboperation(self
):
467 # Load an alternative 'nslcmops' YAML for this test
468 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
469 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
471 # Test when 'operationState' is 'COMPLETED'
472 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = "COMPLETED"
473 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
474 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
475 self
.assertEqual(return_value
, expected_value
)
476 # Test when 'operationState' is not 'COMPLETED'
477 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = None
478 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
479 expected_value
= op_index
480 self
.assertEqual(return_value
, expected_value
)
482 # Test _find_suboperation()
483 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
484 def test_scale_find_suboperation(self
):
485 # Load an alternative 'nslcmops' YAML for this test
486 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
487 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
488 # Find this sub-operation
490 vnf_index
= db_nslcmop
["_admin"]["operations"][op_index
]["member_vnf_index"]
491 primitive
= db_nslcmop
["_admin"]["operations"][op_index
]["primitive"]
492 primitive_params
= db_nslcmop
["_admin"]["operations"][op_index
][
496 "member_vnf_index": vnf_index
,
497 "primitive": primitive
,
498 "primitive_params": primitive_params
,
500 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
501 self
.assertEqual(found_op_index
, op_index
)
502 # Test with not-matching params
504 "member_vnf_index": vnf_index
,
506 "primitive_params": primitive_params
,
508 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
509 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
512 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
513 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
515 # Test _update_suboperation_status()
516 def test_scale_update_suboperation_status(self
):
517 self
.db
.set_one
= asynctest
.Mock()
518 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
519 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
521 # Force the initial values to be distinct from the updated ones
522 q_filter
= {"_id": db_nslcmop
["_id"]}
523 # Test to change 'operationState' and 'detailed-status'
524 operationState
= "COMPLETED"
525 detailed_status
= "Done"
526 expected_update_dict
= {
527 "_admin.operations.0.operationState": operationState
,
528 "_admin.operations.0.detailed-status": detailed_status
,
530 self
.my_ns
._update
_suboperation
_status
(
531 db_nslcmop
, op_index
, operationState
, detailed_status
533 self
.db
.set_one
.assert_called_once_with(
536 update_dict
=expected_update_dict
,
540 def test_scale_add_suboperation(self
):
541 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
542 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
544 num_ops_before
= len(db_nslcmop
.get("_admin", {}).get("operations", [])) - 1
546 vdu_count_index
= None
549 mapped_primitive_params
= {
552 "data-type": "STRING",
554 "default-value": "<touch_filename2>",
559 operationState
= "PROCESSING"
560 detailed_status
= "In progress"
561 operationType
= "PRE-SCALE"
562 # Add a 'pre-scale' suboperation
563 op_index_after
= self
.my_ns
._add
_suboperation
(
570 mapped_primitive_params
,
575 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
577 # Delete all suboperations and add the same operation again
578 del db_nslcmop
["_admin"]["operations"]
579 op_index_zero
= self
.my_ns
._add
_suboperation
(
586 mapped_primitive_params
,
591 self
.assertEqual(op_index_zero
, 0)
593 # Add a 'RO' suboperation
594 RO_nsr_id
= "1234567890"
599 "member-vnf-index": "1",
600 "osm_vdu_id": "dataVM",
603 op_index
= self
.my_ns
._add
_suboperation
(
610 mapped_primitive_params
,
617 db_RO_nsr_id
= db_nslcmop
["_admin"]["operations"][op_index
]["RO_nsr_id"]
618 self
.assertEqual(op_index
, 1)
619 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
621 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
622 op_index_invalid
= self
.my_ns
._add
_suboperation
(
623 None, None, None, None, None, None, None, None, None, None, None
625 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
627 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
628 # check the possible return values:
629 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
630 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
631 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
632 def test_scale_check_or_add_scale_suboperation(self
):
633 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
634 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
635 operationType
= "PRE-SCALE"
641 "data-type": "STRING",
643 "default-value": "<touch_filename2>",
649 # Delete all sub-operations to be sure this is a new sub-operation
650 del db_nslcmop
["_admin"]["operations"]
652 # Add a new sub-operation
653 # For new sub-operations, operationState is set to 'PROCESSING' by default
654 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
655 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
657 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
659 # Use the same parameters again to match the already added sub-operation
660 # which has status 'PROCESSING' (!= 'COMPLETED') by default
661 # The expected return value is a non-negative number
662 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
663 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
665 self
.assertTrue(op_index_existing
>= 0)
667 # Change operationState 'manually' for this sub-operation
668 db_nslcmop
["_admin"]["operations"][op_index_existing
][
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
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
675 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
677 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
679 # RO sub-operation test:
680 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
681 RO_nsr_id
= "1234567890"
686 "member-vnf-index": "1",
687 "osm_vdu_id": "dataVM",
690 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
691 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
693 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
695 # Use the same parameters again to match the already added RO sub-operation
696 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
697 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
699 self
.assertTrue(op_index_existing_RO
>= 0)
701 # Change operationState 'manually' for this RO sub-operation
702 db_nslcmop
["_admin"]["operations"][op_index_existing_RO
][
705 # Then use the same parameters again to match the already added sub-operation,
706 # which now has status 'COMPLETED'
707 # The expected return value is SUBOPERATION_STATUS_SKIP
708 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
709 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
711 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
713 async def test_deploy_kdus(self
):
714 nsr_id
= descriptors
.test_ids
["TEST-KDU"]["ns"]
715 nslcmop_id
= descriptors
.test_ids
["TEST-KDU"]["instantiate"]
716 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
717 db_vnfr
= self
.db
.get_one(
718 "vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "multikdu"}
720 db_vnfrs
= {"multikdu": db_vnfr
}
721 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
725 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
= asynctest
.mock
.Mock()
726 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
.return_value
= "k8s_id"
727 self
.my_ns
.k8sclusterhelm3
.install
= asynctest
.CoroutineMock()
728 self
.my_ns
.k8sclusterhelm3
.synchronize_repos
= asynctest
.CoroutineMock(
729 return_value
=("", "")
731 self
.my_ns
.k8sclusterhelm3
.get_services
= asynctest
.CoroutineMock(
734 await self
.my_ns
.deploy_kdus(
735 logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
737 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
738 db_nsr
= self
.db
.get_list("nsrs")[1]
741 db_nsr
["_admin"]["deployed"],
742 "K8s entry not created at '_admin.deployed'",
744 self
.assertIsInstance(
745 db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
748 len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
751 "kdu-instance": "k8s_id",
752 "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
753 "k8scluster-type": "helm-chart-v3",
755 "member-vnf-index": "multikdu",
757 "kdu-deployment-name": None,
760 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][0])
761 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
762 expected_kdu_model
= "stable/openldap:1.2.1"
763 self
.assertEqual(nsr_result
, k8s_instace_info
)
765 nsr_kdu_model_result
in expected_kdu_model
766 or expected_kdu_model
in nsr_kdu_model_result
768 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][1])
769 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
770 k8s_instace_info
["kdu-name"] = "mongo"
771 expected_kdu_model
= "stable/mongodb"
772 self
.assertEqual(nsr_result
, k8s_instace_info
)
774 nsr_kdu_model_result
in expected_kdu_model
775 or expected_kdu_model
in nsr_kdu_model_result
778 # async def test_instantiate_pdu(self):
779 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
780 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
781 # # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
782 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
783 # update_dict={"ip-address": "10.205.1.46",
784 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
785 # "vdur.0.pdu-type": "PDU-TYPE-1",
786 # "vdur.0.ip-address": "10.205.1.46",
788 # unset={"vdur.status": None})
789 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
790 # update_dict={"ip-address": "10.205.1.47",
791 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
792 # "vdur.0.pdu-type": "PDU-TYPE-1",
793 # "vdur.0.ip-address": "10.205.1.47",
795 # unset={"vdur.status": None})
797 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
798 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
799 # self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
800 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
801 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
802 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
803 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
805 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
806 # async def test_terminate_without_configuration(self):
807 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
808 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
809 # # set instantiation task as completed
810 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
811 # update_dict={"operationState": "COMPLETED"})
812 # self.db.set_one("nsrs", {"_id": nsr_id},
813 # update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
815 # await self.my_ns.terminate(nsr_id, nslcmop_id)
816 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
817 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
818 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
819 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
820 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
821 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
822 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
823 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
824 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
825 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
826 # for vnfr in db_vnfrs_list:
827 # self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
829 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
830 # async def test_terminate_primitive(self):
831 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
832 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
833 # # set instantiation task as completed
834 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
835 # update_dict={"operationState": "COMPLETED"})
837 # # modify vnfd descriptor to include terminate_primitive
838 # terminate_primitive = [{
840 # "parameter": [{"name": "filename", "value": "terminate_filename"}],
843 # db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
844 # self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
845 # {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
847 # await self.my_ns.terminate(nsr_id, nslcmop_id)
848 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
849 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
850 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
851 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
852 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
853 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
854 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
855 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
856 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
859 if __name__
== "__main__":