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 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
36 from osm_lcm
.tests
import test_db_descriptors
as descriptors
38 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
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
57 "global": {"loglevel": "DEBUG"},
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"),
69 "uri": "http://{}:{}/openmano".format(
70 getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
72 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
73 "logger_name": "lcm.ROclient",
80 class TestMyNS(asynctest
.TestCase
):
81 async def _n2vc_DeployCharms(
93 for status
, message
in (
94 ("maintenance", "installing sofwware"),
97 # call callback after some time
98 asyncio
.sleep(5, loop
=self
.loop
)
99 callback(model_name
, application_name
, status
, message
, *callback_args
)
102 def _n2vc_FormatApplicationName(*args
):
105 yield "app_name-{}".format(num_calls
)
108 def _n2vc_CreateExecutionEnvironment(
109 self
, namespace
, reuse_ee_id
, db_dict
, *args
, **kwargs
111 k_list
= namespace
.split(".")
112 ee_id
= k_list
[1] + "."
114 for k
in k_list
[2:4]:
120 def _ro_status(self
, *args
, **kwargs
):
121 print("Args > {}".format(args
))
122 print("kwargs > {}".format(kwargs
))
125 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_update_action_text
)
128 if kwargs
.get("delete"):
129 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_delete_action_text
)
133 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_ns_text
)
135 # if ip address provided, replace descriptor
136 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
138 ip_addresses_list
= ip_addresses
.split(",")
139 for vnf
in ro_ns_desc
["vnfs"]:
140 if not ip_addresses_list
:
142 vnf
["ip_address"] = ip_addresses_list
[0]
143 for vm
in vnf
["vms"]:
144 if not ip_addresses_list
:
146 vm
["ip_address"] = ip_addresses_list
.pop(0)
150 for net
in ro_ns_desc
["nets"]:
151 if net
["status"] != "ACTIVE":
152 net
["status"] = "ACTIVE"
155 for vnf
in ro_ns_desc
["vnfs"]:
156 for vm
in vnf
["vms"]:
157 if vm
["status"] != "ACTIVE":
158 vm
["status"] = "ACTIVE"
161 def _ro_deploy(self
, *args
, **kwargs
):
162 return {"action_id": args
[1]["action_id"], "nsr_id": args
[0], "status": "ok"}
164 def _return_uuid(self
, *args
, **kwargs
):
167 async def setUp(self
):
170 if not getenv("OSMLCMTEST_DB_NOMOCK"):
171 # Cleanup singleton Database instance
172 Database
.instance
= None
174 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
175 self
.db
.create_list("vnfds", yaml
.safe_load(descriptors
.db_vnfds_text
))
178 yaml
.safe_load(descriptors
.db_vnfds_revisions_text
),
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
))
184 yaml
.safe_load(descriptors
.db_vim_accounts_text
),
188 yaml
.safe_load(descriptors
.db_k8sclusters_text
),
191 "nslcmops", yaml
.safe_load(descriptors
.db_nslcmops_text
)
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
)
197 self
.msg
= asynctest
.Mock(MsgKafka())
200 if not getenv("OSMLCMTEST_FS_NOMOCK"):
201 self
.fs
= asynctest
.Mock(
202 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
204 self
.fs
.get_params
.return_value
= {
205 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
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
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
= ("", [])
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
)
223 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
224 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
225 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
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
_n
2vc
= asynctest
.CoroutineMock()
234 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
235 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
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"
244 # allow several versions of n2vc
245 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(
246 side_effect
=self
._n
2vc
_FormatApplicationName
()
248 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(
249 side_effect
=self
._n
2vc
_DeployCharms
251 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
252 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
254 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(
257 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(
260 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
261 side_effect
=self
._return
_uuid
263 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
264 side_effect
=self
._return
_uuid
266 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(
267 return_value
="completed"
269 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(
270 return_value
={"result": "ok", "pubkey": pub_key
}
272 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(
275 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
276 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key")
278 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(
283 if not getenv("OSMLCMTEST_RO_NOMOCK"):
284 self
.my_ns
.RO
= asynctest
.Mock(
285 NgRoClient(self
.loop
, **lcm_config
["ro_config"])
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
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
)
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")
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")
318 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
320 # self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
321 # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
322 # "operationState": "COMPLETED"},
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()
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")
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")
346 # # TODO add more checks of called methods
347 # # TODO add a terminate
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",
355 # "execution-environment-ref": "not_used_ee"
359 # "id": ee_descriptor_id,
360 # "juju": {"charm": "simple"},
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",
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
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
389 expected_value
= "COMPLETED"
390 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
393 self
.assertEqual(return_value
, expected_value
)
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
405 expected_value
= "Error"
406 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
409 self
.assertEqual(return_value
, expected_value
)
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")
416 # TODO: Add more higher-lever tests here, for example:
417 # scale-out/scale-in operations with success/error result
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(
427 self
.assertEqual(return_value
, expected_value
)
428 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
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"]
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(
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(
445 self
.assertEqual(return_value
, expected_value
)
446 self
.my_ns
.k8sclusterjuju
.scale
.assert_called_once()
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"]
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(
458 self
.assertEqual(return_value
, expected_value
)
459 self
.my_ns
.k8sclusterjuju
.scale
.assert_called()
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
]:
473 in value
["lcm-operations-configuration"][
474 "operate-vnf-op-config"
477 for k
, v
in enumerate(
478 value
["lcm-operations-configuration"][
479 "operate-vnf-op-config"
483 v
.get("execution-environment-list")
484 and "juju" in v
["execution-environment-list"][k
]
486 expected_value
= self
.db
.get_list("nsrs")[i
][
489 await self
.my_ns
._on
_update
_n
2vc
_db
(
490 "nsrs", {"_id": nsr_id
}, "_admin.deployed.VCA.0", {}
492 return_value
= self
.db
.get_list("nsrs")[i
]["vcaStatus"]
493 self
.assertEqual(return_value
, expected_value
)
495 # Test _retry_or_skip_suboperation()
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
})
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
)
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
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
][
529 "member_vnf_index": vnf_index
,
530 "primitive": primitive
,
531 "primitive_params": primitive_params
,
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
537 "member_vnf_index": vnf_index
,
539 "primitive_params": primitive_params
,
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
)
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
)
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
})
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
,
563 self
.my_ns
._update
_suboperation
_status
(
564 db_nslcmop
, op_index
, operationState
, detailed_status
566 self
.db
.set_one
.assert_called_once_with(
569 update_dict
=expected_update_dict
,
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
})
577 num_ops_before
= len(db_nslcmop
.get("_admin", {}).get("operations", [])) - 1
579 vdu_count_index
= None
582 mapped_primitive_params
= {
585 "data-type": "STRING",
587 "default-value": "<touch_filename2>",
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
(
603 mapped_primitive_params
,
608 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
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
(
619 mapped_primitive_params
,
624 self
.assertEqual(op_index_zero
, 0)
626 # Add a 'RO' suboperation
627 RO_nsr_id
= "1234567890"
632 "member-vnf-index": "1",
633 "osm_vdu_id": "dataVM",
636 op_index
= self
.my_ns
._add
_suboperation
(
643 mapped_primitive_params
,
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
)
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
658 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
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"
674 "data-type": "STRING",
676 "default-value": "<touch_filename2>",
682 # Delete all sub-operations to be sure this is a new sub-operation
683 del db_nslcmop
["_admin"]["operations"]
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
690 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
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
698 self
.assertTrue(op_index_existing
>= 0)
700 # Change operationState 'manually' for this sub-operation
701 db_nslcmop
["_admin"]["operations"][op_index_existing
][
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
710 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
712 # RO sub-operation test:
713 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
714 RO_nsr_id
= "1234567890"
719 "member-vnf-index": "1",
720 "osm_vdu_id": "dataVM",
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
726 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
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
732 self
.assertTrue(op_index_existing_RO
>= 0)
734 # Change operationState 'manually' for this RO sub-operation
735 db_nslcmop
["_admin"]["operations"][op_index_existing_RO
][
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
744 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
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"}
753 db_vnfrs
= {"multikdu": db_vnfr
}
754 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
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
=("", "")
764 self
.my_ns
.k8sclusterhelm3
.get_services
= asynctest
.CoroutineMock(
767 await self
.my_ns
.deploy_kdus(
768 logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
770 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
771 db_nsr
= self
.db
.get_list("nsrs")[1]
774 db_nsr
["_admin"]["deployed"],
775 "K8s entry not created at '_admin.deployed'",
777 self
.assertIsInstance(
778 db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
781 len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
784 "kdu-instance": "k8s_id",
785 "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
786 "k8scluster-type": "helm-chart-v3",
788 "member-vnf-index": "multikdu",
790 "kdu-deployment-name": None,
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
)
798 nsr_kdu_model_result
in expected_kdu_model
799 or expected_kdu_model
in nsr_kdu_model_result
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
)
807 nsr_kdu_model_result
in expected_kdu_model
808 or expected_kdu_model
in nsr_kdu_model_result
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
):
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(
825 self
.assertEqual(return_value
, expected_value
)
826 with self
.assertRaises(Exception) as context
:
827 self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
829 "database exception Not found entry with filter"
830 in str(context
.exception
)
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"]
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(
848 expected_value
= "COMPLETED"
849 self
.assertEqual(return_value
, expected_value
)
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"]
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(
863 expected_value
= "FAILED"
864 self
.assertEqual(return_value
, expected_value
)
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",
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",
883 # unset={"vdur.status": None})
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")
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})
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")
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"})
925 # # modify vnfd descriptor to include terminate_primitive
926 # terminate_primitive = [{
928 # "parameter": [{"name": "filename", "value": "terminate_filename"}],
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})
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")
948 async def test_update(self
):
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"
956 mock_charm_hash
.reset_mock()
957 mock_juju_bundle
.reset_mock()
959 mock_charm_upgrade
.reset_mock()
960 mock_software_version
.reset_mock()
964 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
965 "Charm package changed, sw-version is not changed.",
970 q_filter
={"_id": vnfd_id
},
971 update_dict
={"_admin.revision": 3, "kdu": []},
976 q_filter
={"_id": vnfd_id
+ ":1"},
977 update_dict
={"_admin.revision": 1, "kdu": []},
981 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
984 mock_charm_hash
= Mock(autospec
=True)
985 mock_charm_hash
.return_value
= True
987 mock_juju_bundle
= Mock(return_value
=None)
989 mock_software_version
= Mock(autospec
=True)
990 mock_software_version
.side_effect
= ["1.0", "1.0"]
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
997 fs
= Mock(autospec
=True)
998 fs
.path
.__add
__ = Mock()
999 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1000 fs
.sync
.side_effect
= [None, None]
1002 instance
= self
.my_ns
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"
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
1015 "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
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")
1029 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1033 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
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",
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
)
1047 return_ns_operational_state
, expected_ns_operational_state
1049 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1054 i
=2, t
="Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"
1058 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 1}
1062 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1065 mock_charm_hash
= Mock(autospec
=True)
1066 mock_charm_hash
.return_value
= True
1068 mock_juju_bundle
= Mock(return_value
=None)
1069 mock_software_version
= Mock(autospec
=True)
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
1076 fs
= Mock(autospec
=True)
1077 fs
.path
.__add
__ = Mock()
1078 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1079 fs
.sync
.side_effect
= [None, None]
1081 instance
= self
.my_ns
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"
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
1095 await instance
.update(nsr_id
, nslcmop_id
)
1097 return_operation_state
= self
.db
.get_one(
1098 "nslcmops", {"_id": nslcmop_id
}
1099 ).get("operationState")
1101 return_operation_error
= self
.db
.get_one(
1102 "nslcmops", {"_id": nslcmop_id
}
1103 ).get("errorMessage")
1105 return_ns_operational_state
= self
.db
.get_one(
1106 "nsrs", {"_id": nsr_id
}
1107 ).get("operational-status")
1109 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1113 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
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()
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
)
1127 return_ns_operational_state
, expected_ns_operational_state
1129 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1135 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed, "
1136 "Charm package is not changed, sw-version is not changed.",
1140 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 3}
1145 q_filter
={"_id": vnfd_id
+ ":1"},
1146 update_dict
={"_admin.revision": 1},
1150 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1153 mock_charm_hash
= Mock(autospec
=True)
1154 mock_charm_hash
.return_value
= False
1156 mock_juju_bundle
= Mock(return_value
=None)
1158 mock_software_version
= Mock(autospec
=True)
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"]
1166 fs
= Mock(autospec
=True)
1167 fs
.path
.__add
__ = Mock()
1168 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1169 fs
.sync
.side_effect
= [None, None]
1171 instance
= self
.my_ns
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"
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
1185 await instance
.update(nsr_id
, nslcmop_id
)
1187 return_operation_state
= self
.db
.get_one(
1188 "nslcmops", {"_id": nslcmop_id
}
1189 ).get("operationState")
1191 return_operation_error
= self
.db
.get_one(
1192 "nslcmops", {"_id": nslcmop_id
}
1193 ).get("errorMessage")
1195 return_ns_operational_state
= self
.db
.get_one(
1196 "nsrs", {"_id": nsr_id
}
1197 ).get("operational-status")
1199 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1203 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
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",
1212 self
.assertEqual(fs
.sync
.call_count
, 2)
1213 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1215 mock_juju_bundle
.assert_not_called()
1216 mock_charm_upgrade
.assert_not_called()
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
)
1222 return_ns_operational_state
, expected_ns_operational_state
1224 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1230 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed, "
1231 "Charm package exists, sw-version changed.",
1236 q_filter
={"_id": vnfd_id
},
1237 update_dict
={"_admin.revision": 3, "software-version": "3.0"},
1242 q_filter
={"_id": vnfd_id
+ ":1"},
1243 update_dict
={"_admin.revision": 1},
1248 q_filter
={"_id": vnfr_id
},
1249 update_dict
={"revision": 1},
1252 mock_charm_hash
= Mock(autospec
=True)
1253 mock_charm_hash
.return_value
= False
1255 mock_juju_bundle
= Mock(return_value
=None)
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
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",
1268 fs
= Mock(autospec
=True)
1269 fs
.path
.__add
__ = Mock()
1270 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1271 fs
.sync
.side_effect
= [None, None]
1273 instance
= self
.my_ns
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"
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
1287 await instance
.update(nsr_id
, nslcmop_id
)
1289 return_operation_state
= self
.db
.get_one(
1290 "nslcmops", {"_id": nslcmop_id
}
1291 ).get("operationState")
1293 return_operation_error
= self
.db
.get_one(
1294 "nslcmops", {"_id": nslcmop_id
}
1295 ).get("errorMessage")
1297 return_ns_operational_state
= self
.db
.get_one(
1298 "nsrs", {"_id": nsr_id
}
1299 ).get("operational-status")
1301 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1305 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1309 self
.assertEqual(fs
.sync
.call_count
, 2)
1310 mock_charm_hash
.assert_not_called()
1312 mock_juju_bundle
.assert_not_called()
1313 mock_charm_upgrade
.assert_not_called()
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
)
1319 return_ns_operational_state
, expected_ns_operational_state
1321 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1327 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
1328 "Charm package exists, sw-version not changed, juju-bundle exists",
1333 q_filter
={"_id": vnfd_id
},
1335 "_admin.revision": 3,
1336 "software-version": "1.0",
1337 "kdu.0.juju-bundle": "stable/native-kdu",
1343 q_filter
={"_id": vnfd_id
+ ":1"},
1345 "_admin.revision": 1,
1346 "software-version": "1.0",
1347 "kdu.0.juju-bundle": "stable/native-kdu",
1352 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1355 mock_charm_hash
= Mock(autospec
=True)
1356 mock_charm_hash
.return_value
= True
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",
1364 fs
= Mock(autospec
=True)
1365 fs
.path
.__add
__ = Mock()
1366 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1367 fs
.sync
.side_effect
= [None, None]
1369 instance
= self
.my_ns
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"
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
):
1381 await instance
.update(nsr_id
, nslcmop_id
)
1383 return_operation_state
= self
.db
.get_one(
1384 "nslcmops", {"_id": nslcmop_id
}
1385 ).get("operationState")
1387 return_operation_error
= self
.db
.get_one(
1388 "nslcmops", {"_id": nslcmop_id
}
1389 ).get("errorMessage")
1391 return_ns_operational_state
= self
.db
.get_one(
1392 "nsrs", {"_id": nsr_id
}
1393 ).get("operational-status")
1395 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1399 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
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)
1407 mock_charm_upgrade
.assert_not_called()
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
)
1413 return_ns_operational_state
, expected_ns_operational_state
1415 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1421 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
1422 "Charm package exists, sw-version not changed, charm-upgrade failed",
1427 q_filter
={"_id": vnfd_id
},
1429 "_admin.revision": 3,
1430 "software-version": "1.0",
1437 q_filter
={"_id": vnfd_id
+ ":1"},
1439 "_admin.revision": 1,
1440 "software-version": "1.0",
1446 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1449 mock_charm_hash
= Mock(autospec
=True)
1450 mock_charm_hash
.return_value
= True
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
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",
1463 fs
= Mock(autospec
=True)
1464 fs
.path
.__add
__ = Mock()
1465 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1466 fs
.sync
.side_effect
= [None, None]
1468 instance
= self
.my_ns
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"
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
):
1480 await instance
.update(nsr_id
, nslcmop_id
)
1482 return_operation_state
= self
.db
.get_one(
1483 "nslcmops", {"_id": nslcmop_id
}
1484 ).get("operationState")
1486 return_operation_error
= self
.db
.get_one(
1487 "nslcmops", {"_id": nslcmop_id
}
1488 ).get("errorMessage")
1490 return_ns_operational_state
= self
.db
.get_one(
1491 "nsrs", {"_id": nsr_id
}
1492 ).get("operational-status")
1494 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1498 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
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)
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
)
1510 return_ns_operational_state
, expected_ns_operational_state
1512 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1516 def test_ns_update_helper_methods(self
):
1519 mock_path
.mock_reset()
1520 mock_checksumdir
.mock_reset()
1523 i
=1, t
="Find software version, VNFD does not have have software version"
1525 # Testing method find_software_version
1527 db_vnfd
= self
.db
.get_one(
1528 "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
1530 expected_result
= "1.0"
1531 result
= find_software_version(db_vnfd
)
1533 result
, expected_result
, "Default sw version should be 1.0"
1537 i
=2, t
="Find software version, VNFD includes software version"
1539 # Testing method find_software_version
1541 db_vnfd
= self
.db
.get_one(
1542 "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
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")
1549 with self
.subTest(i
=3, t
="Check charm hash, Hash did not change"):
1550 # Testing method check_charm_hash_changed
1552 current_path
, target_path
= "/tmp/charm1", "/tmp/charm1"
1553 fs
= Mock(autospec
=True)
1554 fs
.path
.__add
__ = Mock()
1555 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1557 mock_path
= Mock(autospec
=True)
1558 mock_path
.exists
.side_effect
= [True, True]
1560 mock_checksumdir
= Mock(autospec
=True)
1561 mock_checksumdir
.dirhash
.side_effect
= ["hash_value", "hash_value"]
1563 instance
= self
.my_ns
1564 expected_result
= False
1566 with patch
.object(instance
, "fs", fs
), patch(
1567 "checksumdir.dirhash", mock_checksumdir
.dirhash
1568 ), patch("os.path.exists", mock_path
.exists
):
1570 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1572 result
, expected_result
, "Wrong charm hash control value"
1574 self
.assertEqual(mock_path
.exists
.call_count
, 2)
1575 self
.assertEqual(mock_checksumdir
.dirhash
.call_count
, 2)
1579 with self
.subTest(i
=4, t
="Check charm hash, Hash has changed"):
1580 # Testing method check_charm_hash_changed
1582 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1583 fs
= Mock(autospec
=True)
1584 fs
.path
.__add
__ = Mock()
1585 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1587 mock_path
= Mock(autospec
=True)
1588 mock_path
.exists
.side_effect
= [True, True]
1590 mock_checksumdir
= Mock(autospec
=True)
1591 mock_checksumdir
.dirhash
.side_effect
= ["hash_value", "another_hash_value"]
1593 instance
= self
.my_ns
1594 expected_result
= True
1596 with patch
.object(instance
, "fs", fs
), patch(
1597 "checksumdir.dirhash", mock_checksumdir
.dirhash
1598 ), patch("os.path.exists", mock_path
.exists
):
1600 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1602 result
, expected_result
, "Wrong charm hash control value"
1604 self
.assertEqual(mock_path
.exists
.call_count
, 2)
1605 self
.assertEqual(mock_checksumdir
.dirhash
.call_count
, 2)
1609 with self
.subTest(i
=5, t
="Check charm hash, Charm path does not exists"):
1610 # Testing method check_charm_hash_changed
1612 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1613 fs
= Mock(autospec
=True)
1614 fs
.path
.__add
__ = Mock()
1615 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1617 mock_path
= Mock(autospec
=True)
1618 mock_path
.exists
.side_effect
= [True, False]
1620 mock_checksumdir
= Mock(autospec
=True)
1621 mock_checksumdir
.dirhash
.side_effect
= ["hash_value", "hash_value"]
1623 instance
= self
.my_ns
1625 with patch
.object(instance
, "fs", fs
), patch(
1626 "checksumdir.dirhash", mock_checksumdir
.dirhash
1627 ), patch("os.path.exists", mock_path
.exists
):
1629 with self
.assertRaises(LcmException
):
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)
1637 with self
.subTest(i
=6, t
="Check juju bundle existence"):
1638 # Testing method check_juju_bundle_existence
1640 test_vnfd1
= self
.db
.get_one(
1641 "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
1643 test_vnfd2
= self
.db
.get_one(
1644 "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
1647 expected_result
= None
1648 result
= check_juju_bundle_existence(test_vnfd1
)
1649 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1651 expected_result
= "stable/native-kdu"
1652 result
= check_juju_bundle_existence(test_vnfd2
)
1653 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1655 with self
.subTest(i
=7, t
="Check charm artifacts"):
1656 # Testing method check_juju_bundle_existence
1659 "folder": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
1660 "pkg-dir": "hackfest_3charmed_vnfd",
1662 charm_name
= "simple"
1663 charm_type
= "lxc_proxy_charm"
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
1670 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1674 "folder": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
1676 charm_name
= "basic"
1681 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/Scripts/helm-charts/basic"
1683 result
= get_charm_artifact_path(
1684 base_folder
, charm_name
, charm_type
, revision
1686 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1689 if __name__
== "__main__":