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
21 from copy
import deepcopy
25 from osm_lcm
import ns
26 from osm_common
.msgkafka
import MsgKafka
28 from osm_lcm
.data_utils
.lcm_config
import LcmCfg
29 from osm_lcm
.lcm_utils
import TaskRegistry
30 from osm_lcm
.ng_ro
import NgRoClient
31 from osm_lcm
.data_utils
.database
.database
import Database
32 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
33 from osm_lcm
.data_utils
.vnfd
import find_software_version
34 from osm_lcm
.lcm_utils
import check_juju_bundle_existence
, get_charm_artifact_path
35 from osm_lcm
.lcm_utils
import LcmException
36 from uuid
import uuid4
37 from unittest
.mock
import Mock
, patch
39 from osm_lcm
.tests
import test_db_descriptors
as descriptors
41 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
43 """ Perform unittests using asynctest of osm_lcm.ns module
44 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
45 OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
46 OSMLCMTEST_NS_NAME: change name of NS
47 OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
48 OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
49 OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
50 OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
51 OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
52 OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
53 OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
54 OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
55 OSMLCM_VCA_XXX: configuration of N2VC
56 OSMLCM_RO_XXX: configuration of RO
60 "global": {"loglevel": "DEBUG"},
62 "VCA": { # TODO replace with os.get_env to get other configurations
63 "host": getenv("OSMLCM_VCA_HOST", "vca"),
64 "port": getenv("OSMLCM_VCA_PORT", 17070),
65 "user": getenv("OSMLCM_VCA_USER", "admin"),
66 "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
67 "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
68 "ca_cert": getenv("OSMLCM_VCA_CACERT", None),
69 "apiproxy": getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1"),
72 "uri": "http://{}:{}/openmano".format(
73 getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
75 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
76 "logger_name": "lcm.ROclient",
83 lcm_config
.set_from_dict(lcm_config_dict
)
84 lcm_config
.transform()
86 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
87 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["update"]
88 vnfr_id
= "6421c7c9-d865-4fb4-9a13-d4275d243e01"
89 vnfd_id
= "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
90 update_fs
= Mock(autospec
=True)
91 update_fs
.path
.__add
__ = Mock()
92 update_fs
.path
.side_effect
= ["/", "/", "/", "/"]
93 update_fs
.sync
.side_effect
= [None, None]
100 class TestMyNS(asynctest
.TestCase
):
101 async def _n2vc_DeployCharms(
113 for status
, message
in (
114 ("maintenance", "installing sofwware"),
115 ("active", "Ready!"),
117 # call callback after some time
118 asyncio
.sleep(5, loop
=self
.loop
)
119 callback(model_name
, application_name
, status
, message
, *callback_args
)
122 def _n2vc_FormatApplicationName(*args
):
125 yield "app_name-{}".format(num_calls
)
128 def _n2vc_CreateExecutionEnvironment(
129 self
, namespace
, reuse_ee_id
, db_dict
, *args
, **kwargs
131 k_list
= namespace
.split(".")
132 ee_id
= k_list
[1] + "."
134 for k
in k_list
[2:4]:
140 def _ro_status(self
, *args
, **kwargs
):
141 print("Args > {}".format(args
))
142 print("kwargs > {}".format(kwargs
))
145 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_update_action_text
)
148 if kwargs
.get("delete"):
149 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_delete_action_text
)
153 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_ns_text
)
155 # if ip address provided, replace descriptor
156 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
158 ip_addresses_list
= ip_addresses
.split(",")
159 for vnf
in ro_ns_desc
["vnfs"]:
160 if not ip_addresses_list
:
162 vnf
["ip_address"] = ip_addresses_list
[0]
163 for vm
in vnf
["vms"]:
164 if not ip_addresses_list
:
166 vm
["ip_address"] = ip_addresses_list
.pop(0)
170 for net
in ro_ns_desc
["nets"]:
171 if net
["status"] != "ACTIVE":
172 net
["status"] = "ACTIVE"
175 for vnf
in ro_ns_desc
["vnfs"]:
176 for vm
in vnf
["vms"]:
177 if vm
["status"] != "ACTIVE":
178 vm
["status"] = "ACTIVE"
181 def _ro_deploy(self
, *args
, **kwargs
):
182 return {"action_id": args
[1]["action_id"], "nsr_id": args
[0], "status": "ok"}
184 def _return_uuid(self
, *args
, **kwargs
):
187 async def setUp(self
):
190 if not getenv("OSMLCMTEST_DB_NOMOCK"):
191 # Cleanup singleton Database instance
192 Database
.instance
= None
194 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
195 self
.db
.create_list("vnfds", yaml
.safe_load(descriptors
.db_vnfds_text
))
198 yaml
.safe_load(descriptors
.db_vnfds_revisions_text
),
200 self
.db
.create_list("nsds", yaml
.safe_load(descriptors
.db_nsds_text
))
201 self
.db
.create_list("nsrs", yaml
.safe_load(descriptors
.db_nsrs_text
))
204 yaml
.safe_load(descriptors
.db_vim_accounts_text
),
208 yaml
.safe_load(descriptors
.db_k8sclusters_text
),
211 "nslcmops", yaml
.safe_load(descriptors
.db_nslcmops_text
)
213 self
.db
.create_list("vnfrs", yaml
.safe_load(descriptors
.db_vnfrs_text
))
214 self
.db_vim_accounts
= yaml
.safe_load(descriptors
.db_vim_accounts_text
)
217 self
.msg
= asynctest
.Mock(MsgKafka())
220 if not getenv("OSMLCMTEST_FS_NOMOCK"):
221 self
.fs
= asynctest
.Mock(
222 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
224 self
.fs
.get_params
.return_value
= {
225 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
227 self
.fs
.file_open
= asynctest
.mock_open()
228 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
229 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
232 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
233 self
.lcm_tasks
.lock_HA
.return_value
= True
234 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
235 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
238 if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
239 ns
.K8sJujuConnector
= asynctest
.MagicMock(ns
.K8sJujuConnector
)
240 ns
.K8sHelmConnector
= asynctest
.MagicMock(ns
.K8sHelmConnector
)
241 ns
.K8sHelm3Connector
= asynctest
.MagicMock(ns
.K8sHelm3Connector
)
243 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
244 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
245 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
248 self
.my_ns
= ns
.NsLcm(self
.msg
, self
.lcm_tasks
, lcm_config
, self
.loop
)
249 self
.my_ns
.fs
= self
.fs
250 self
.my_ns
.db
= self
.db
251 self
.my_ns
._wait
_dependent
_n
2vc
= asynctest
.CoroutineMock()
254 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
255 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
258 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
259 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
260 # self.my_ns.n2vc = asynctest.Mock(N2VC())
261 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv(
262 "OSMLCM_VCA_PUBKEY", "public_key"
264 # allow several versions of n2vc
265 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(
266 side_effect
=self
._n
2vc
_FormatApplicationName
()
268 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(
269 side_effect
=self
._n
2vc
_DeployCharms
271 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
272 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
274 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(
277 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(
280 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
281 side_effect
=self
._return
_uuid
283 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
284 side_effect
=self
._return
_uuid
286 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(
287 return_value
="completed"
289 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(
290 return_value
={"result": "ok", "pubkey": pub_key
}
292 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(
295 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
296 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key")
298 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(
303 if not getenv("OSMLCMTEST_RO_NOMOCK"):
304 self
.my_ns
.RO
= asynctest
.Mock(
305 NgRoClient(self
.loop
, **lcm_config
.RO
.to_dict())
307 # TODO first time should be empty list, following should return a dict
308 # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
309 self
.my_ns
.RO
.deploy
= asynctest
.CoroutineMock(
310 self
.my_ns
.RO
.deploy
, side_effect
=self
._ro
_deploy
312 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status)
313 # self.my_ns.RO.create_action = asynctest.CoroutineMock(self.my_ns.RO.create_action,
314 # return_value={"vm-id": {"vim_result": 200,
315 # "description": "done"}})
316 self
.my_ns
.RO
.delete
= asynctest
.CoroutineMock(self
.my_ns
.RO
.delete
)
318 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
319 # async def test_instantiate(self):
320 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
321 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
322 # # print("Test instantiate started")
324 # # delete deployed information of database
325 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
326 # if self.db.get_list("nsrs")[0]["_admin"].get("deployed"):
327 # del self.db.get_list("nsrs")[0]["_admin"]["deployed"]
328 # for db_vnfr in self.db.get_list("vnfrs"):
329 # db_vnfr.pop("ip_address", None)
330 # for db_vdur in db_vnfr["vdur"]:
331 # db_vdur.pop("ip_address", None)
332 # db_vdur.pop("mac_address", None)
333 # if getenv("OSMLCMTEST_RO_VIMID"):
334 # self.db.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
335 # if getenv("OSMLCMTEST_RO_VIMID"):
336 # self.db.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
338 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
340 # self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
341 # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
342 # "operationState": "COMPLETED"},
344 # self.lcm_tasks.lock_HA.assert_called_once_with('ns', 'nslcmops', nslcmop_id)
345 # if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
346 # self.assertTrue(self.my_ns.logger.debug.called, "Debug method not called")
347 # self.my_ns.logger.error.assert_not_called()
348 # self.my_ns.logger.exception().assert_not_called()
350 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
351 # self.assertTrue(self.db.set_one.called, "db.set_one not called")
352 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
353 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
354 # self.assertEqual(db_nsr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
355 # for vnfr in db_vnfrs_list:
356 # self.assertEqual(vnfr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
358 # if not getenv("OSMLCMTEST_VCA_NOMOCK"):
359 # # check intial-primitives called
360 # self.assertTrue(self.my_ns.n2vc.exec_primitive.called,
361 # "Exec primitive not called for initial config primitive")
362 # for _call in self.my_ns.n2vc.exec_primitive.call_args_list:
363 # self.assertIn(_call[1]["primitive_name"], ("config", "touch"),
364 # "called exec primitive with a primitive different than config or touch")
366 # # TODO add more checks of called methods
367 # # TODO add a terminate
369 # async def test_instantiate_ee_list(self):
370 # # Using modern IM where configuration is in the new format of execution_environment_list
371 # ee_descriptor_id = "charm_simple"
372 # non_used_initial_primitive = {
373 # "name": "not_to_be_called",
375 # "execution-environment-ref": "not_used_ee"
379 # "id": ee_descriptor_id,
380 # "juju": {"charm": "simple"},
387 # q_filter={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"},
388 # update_dict={"vnf-configuration.0.execution-environment-list": ee_list,
389 # "vnf-configuration.0.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id,
390 # "vnf-configuration.0.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id,
391 # "vnf-configuration.0.initial-config-primitive.2": non_used_initial_primitive,
392 # "vnf-configuration.0.config-primitive.0.execution-environment-ref": ee_descriptor_id,
393 # "vnf-configuration.0.config-primitive.0.execution-environment-primitive": "touch_charm",
395 # unset={"vnf-configuration.juju": None})
396 # await self.test_instantiate()
397 # # this will check that the initial-congig-primitive 'not_to_be_called' is not called
399 @asynctest.fail_on(active_handles
=True)
400 async def test_start_stop_rebuild_pass(self
):
401 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
402 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops"]
403 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
404 additional_param
= {"count-index": "0"}
405 operation_type
= "start"
406 await self
.my_ns
.rebuild_start_stop(
407 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
409 expected_value
= "COMPLETED"
410 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
413 self
.assertEqual(return_value
, expected_value
)
415 @asynctest.fail_on(active_handles
=True)
416 async def test_start_stop_rebuild_fail(self
):
417 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
418 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops1"]
419 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
420 additional_param
= {"count-index": "0"}
421 operation_type
= "stop"
422 await self
.my_ns
.rebuild_start_stop(
423 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
425 expected_value
= "Error"
426 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
429 self
.assertEqual(return_value
, expected_value
)
431 # Test scale() and related methods
432 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
433 async def test_scale(self
):
434 # print("Test scale started")
436 # TODO: Add more higher-lever tests here, for example:
437 # scale-out/scale-in operations with success/error result
439 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
440 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
441 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
442 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
443 expected_value
= "FAILED"
444 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
447 self
.assertEqual(return_value
, expected_value
)
448 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
450 # Test scale() for native kdu
451 # this also includes testing _scale_kdu()
452 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["ns"]
453 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["instantiate"]
455 self
.my_ns
.k8sclusterjuju
.scale
= asynctest
.mock
.CoroutineMock()
456 self
.my_ns
.k8sclusterjuju
.exec_primitive
= asynctest
.mock
.CoroutineMock()
457 self
.my_ns
.k8sclusterjuju
.get_scale_count
= asynctest
.mock
.CoroutineMock(
460 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
461 expected_value
= "COMPLETED"
462 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
465 self
.assertEqual(return_value
, expected_value
)
466 self
.my_ns
.k8sclusterjuju
.scale
.assert_called_once()
468 # Test scale() for native kdu with 2 resource
469 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["ns"]
470 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["instantiate"]
472 self
.my_ns
.k8sclusterjuju
.get_scale_count
.return_value
= 2
473 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
474 expected_value
= "COMPLETED"
475 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
478 self
.assertEqual(return_value
, expected_value
)
479 self
.my_ns
.k8sclusterjuju
.scale
.assert_called()
481 async def test_vca_status_refresh(self
):
482 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
483 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
484 await self
.my_ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
485 expected_value
= dict()
486 return_value
= dict()
487 vnf_descriptors
= self
.db
.get_list("vnfds")
488 for i
, _
in enumerate(vnf_descriptors
):
489 for j
, value
in enumerate(vnf_descriptors
[i
]["df"]):
490 if "lcm-operations-configuration" in vnf_descriptors
[i
]["df"][j
]:
493 in value
["lcm-operations-configuration"][
494 "operate-vnf-op-config"
497 for k
, v
in enumerate(
498 value
["lcm-operations-configuration"][
499 "operate-vnf-op-config"
503 v
.get("execution-environment-list")
504 and "juju" in v
["execution-environment-list"][k
]
506 expected_value
= self
.db
.get_list("nsrs")[i
][
509 await self
.my_ns
._on
_update
_n
2vc
_db
(
510 "nsrs", {"_id": nsr_id
}, "_admin.deployed.VCA.0", {}
512 return_value
= self
.db
.get_list("nsrs")[i
]["vcaStatus"]
513 self
.assertEqual(return_value
, expected_value
)
515 # Test _retry_or_skip_suboperation()
517 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
518 # - if marked as anything but 'COMPLETED', the suboperation index is expected
519 def test_scale_retry_or_skip_suboperation(self
):
520 # Load an alternative 'nslcmops' YAML for this test
521 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
522 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
524 # Test when 'operationState' is 'COMPLETED'
525 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = "COMPLETED"
526 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
527 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
528 self
.assertEqual(return_value
, expected_value
)
529 # Test when 'operationState' is not 'COMPLETED'
530 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = None
531 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
532 expected_value
= op_index
533 self
.assertEqual(return_value
, expected_value
)
535 # Test _find_suboperation()
536 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
537 def test_scale_find_suboperation(self
):
538 # Load an alternative 'nslcmops' YAML for this test
539 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
540 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
541 # Find this sub-operation
543 vnf_index
= db_nslcmop
["_admin"]["operations"][op_index
]["member_vnf_index"]
544 primitive
= db_nslcmop
["_admin"]["operations"][op_index
]["primitive"]
545 primitive_params
= db_nslcmop
["_admin"]["operations"][op_index
][
549 "member_vnf_index": vnf_index
,
550 "primitive": primitive
,
551 "primitive_params": primitive_params
,
553 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
554 self
.assertEqual(found_op_index
, op_index
)
555 # Test with not-matching params
557 "member_vnf_index": vnf_index
,
559 "primitive_params": primitive_params
,
561 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
562 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
565 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
566 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
568 # Test _update_suboperation_status()
569 def test_scale_update_suboperation_status(self
):
570 self
.db
.set_one
= asynctest
.Mock()
571 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
572 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
574 # Force the initial values to be distinct from the updated ones
575 q_filter
= {"_id": db_nslcmop
["_id"]}
576 # Test to change 'operationState' and 'detailed-status'
577 operationState
= "COMPLETED"
578 detailed_status
= "Done"
579 expected_update_dict
= {
580 "_admin.operations.0.operationState": operationState
,
581 "_admin.operations.0.detailed-status": detailed_status
,
583 self
.my_ns
._update
_suboperation
_status
(
584 db_nslcmop
, op_index
, operationState
, detailed_status
586 self
.db
.set_one
.assert_called_once_with(
589 update_dict
=expected_update_dict
,
593 def test_scale_add_suboperation(self
):
594 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
595 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
597 num_ops_before
= len(db_nslcmop
.get("_admin", {}).get("operations", [])) - 1
599 vdu_count_index
= None
602 mapped_primitive_params
= {
605 "data-type": "STRING",
607 "default-value": "<touch_filename2>",
612 operationState
= "PROCESSING"
613 detailed_status
= "In progress"
614 operationType
= "PRE-SCALE"
615 # Add a 'pre-scale' suboperation
616 op_index_after
= self
.my_ns
._add
_suboperation
(
623 mapped_primitive_params
,
628 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
630 # Delete all suboperations and add the same operation again
631 del db_nslcmop
["_admin"]["operations"]
632 op_index_zero
= self
.my_ns
._add
_suboperation
(
639 mapped_primitive_params
,
644 self
.assertEqual(op_index_zero
, 0)
646 # Add a 'RO' suboperation
647 RO_nsr_id
= "1234567890"
652 "member-vnf-index": "1",
653 "osm_vdu_id": "dataVM",
656 op_index
= self
.my_ns
._add
_suboperation
(
663 mapped_primitive_params
,
670 db_RO_nsr_id
= db_nslcmop
["_admin"]["operations"][op_index
]["RO_nsr_id"]
671 self
.assertEqual(op_index
, 1)
672 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
674 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
675 op_index_invalid
= self
.my_ns
._add
_suboperation
(
676 None, None, None, None, None, None, None, None, None, None, None
678 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
680 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
681 # check the possible return values:
682 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
683 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
684 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
685 def test_scale_check_or_add_scale_suboperation(self
):
686 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
687 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
688 operationType
= "PRE-SCALE"
694 "data-type": "STRING",
696 "default-value": "<touch_filename2>",
702 # Delete all sub-operations to be sure this is a new sub-operation
703 del db_nslcmop
["_admin"]["operations"]
705 # Add a new sub-operation
706 # For new sub-operations, operationState is set to 'PROCESSING' by default
707 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
708 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
710 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
712 # Use the same parameters again to match the already added sub-operation
713 # which has status 'PROCESSING' (!= 'COMPLETED') by default
714 # The expected return value is a non-negative number
715 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
716 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
718 self
.assertTrue(op_index_existing
>= 0)
720 # Change operationState 'manually' for this sub-operation
721 db_nslcmop
["_admin"]["operations"][op_index_existing
][
724 # Then use the same parameters again to match the already added sub-operation,
725 # which now has status 'COMPLETED'
726 # The expected return value is SUBOPERATION_STATUS_SKIP
727 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
728 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
730 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
732 # RO sub-operation test:
733 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
734 RO_nsr_id
= "1234567890"
739 "member-vnf-index": "1",
740 "osm_vdu_id": "dataVM",
743 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
744 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
746 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
748 # Use the same parameters again to match the already added RO sub-operation
749 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
750 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
752 self
.assertTrue(op_index_existing_RO
>= 0)
754 # Change operationState 'manually' for this RO sub-operation
755 db_nslcmop
["_admin"]["operations"][op_index_existing_RO
][
758 # Then use the same parameters again to match the already added sub-operation,
759 # which now has status 'COMPLETED'
760 # The expected return value is SUBOPERATION_STATUS_SKIP
761 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
762 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
764 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
766 async def test_deploy_kdus(self
):
767 nsr_id
= descriptors
.test_ids
["TEST-KDU"]["ns"]
768 nslcmop_id
= descriptors
.test_ids
["TEST-KDU"]["instantiate"]
769 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
770 db_vnfr
= self
.db
.get_one(
771 "vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "multikdu"}
773 db_vnfrs
= {"multikdu": db_vnfr
}
774 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
778 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
= asynctest
.mock
.Mock()
779 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
.return_value
= "k8s_id"
780 self
.my_ns
.k8sclusterhelm3
.install
= asynctest
.CoroutineMock()
781 self
.my_ns
.k8sclusterhelm3
.synchronize_repos
= asynctest
.CoroutineMock(
782 return_value
=("", "")
784 self
.my_ns
.k8sclusterhelm3
.get_services
= asynctest
.CoroutineMock(
787 await self
.my_ns
.deploy_kdus(
788 logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
790 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
791 db_nsr
= self
.db
.get_list("nsrs")[1]
794 db_nsr
["_admin"]["deployed"],
795 "K8s entry not created at '_admin.deployed'",
797 self
.assertIsInstance(
798 db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
801 len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
804 "kdu-instance": "k8s_id",
805 "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
806 "k8scluster-type": "helm-chart-v3",
808 "member-vnf-index": "multikdu",
810 "kdu-deployment-name": None,
813 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][0])
814 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
815 expected_kdu_model
= "stable/openldap:1.2.1"
816 self
.assertEqual(nsr_result
, k8s_instace_info
)
818 nsr_kdu_model_result
in expected_kdu_model
819 or expected_kdu_model
in nsr_kdu_model_result
821 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][1])
822 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
823 k8s_instace_info
["kdu-name"] = "mongo"
824 expected_kdu_model
= "stable/mongodb"
825 self
.assertEqual(nsr_result
, k8s_instace_info
)
827 nsr_kdu_model_result
in expected_kdu_model
828 or expected_kdu_model
in nsr_kdu_model_result
831 # Test remove_vnf() and related methods
832 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
833 async def test_remove_vnf(self
):
835 nsr_id
= descriptors
.test_ids
["TEST-UPDATE"]["ns"]
836 nslcmop_id
= descriptors
.test_ids
["TEST-UPDATE"]["removeVnf"]
837 vnf_instance_id
= descriptors
.test_ids
["TEST-UPDATE"]["vnf"]
838 mock_wait_ng_ro
= asynctest
.CoroutineMock()
839 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
840 await self
.my_ns
.update(nsr_id
, nslcmop_id
)
841 expected_value
= "COMPLETED"
842 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
845 self
.assertEqual(return_value
, expected_value
)
846 with self
.assertRaises(Exception) as context
:
847 self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
849 "database exception Not found entry with filter"
850 in str(context
.exception
)
853 # test vertical scale executes sucessfully
854 # @patch("osm_lcm.ng_ro.status.response")
855 @asynctest.fail_on(active_handles
=True)
856 async def test_vertical_scaling(self
):
857 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
858 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate"]
860 # calling the vertical scale fucntion
861 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status("update"))
862 mock_wait_ng_ro
= asynctest
.CoroutineMock()
863 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
864 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
865 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
868 expected_value
= "COMPLETED"
869 self
.assertEqual(return_value
, expected_value
)
871 # test vertical scale executes fail
872 @asynctest.fail_on(active_handles
=True)
873 async def test_vertical_scaling_fail(self
):
874 # get th nsr nad nslcmops id from descriptors
875 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
876 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate-1"]
878 # calling the vertical scale fucntion
879 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
880 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
883 expected_value
= "FAILED"
884 self
.assertEqual(return_value
, expected_value
)
886 # async def test_instantiate_pdu(self):
887 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
888 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
889 # # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
890 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
891 # update_dict={"ip-address": "10.205.1.46",
892 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
893 # "vdur.0.pdu-type": "PDU-TYPE-1",
894 # "vdur.0.ip-address": "10.205.1.46",
896 # unset={"vdur.status": None})
897 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
898 # update_dict={"ip-address": "10.205.1.47",
899 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
900 # "vdur.0.pdu-type": "PDU-TYPE-1",
901 # "vdur.0.ip-address": "10.205.1.47",
903 # unset={"vdur.status": None})
905 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
906 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
907 # self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
908 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
909 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
910 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
911 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
913 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
914 # async def test_terminate_without_configuration(self):
915 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
916 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
917 # # set instantiation task as completed
918 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
919 # update_dict={"operationState": "COMPLETED"})
920 # self.db.set_one("nsrs", {"_id": nsr_id},
921 # update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
923 # await self.my_ns.terminate(nsr_id, nslcmop_id)
924 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
925 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
926 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
927 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
928 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
929 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
930 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
931 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
932 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
933 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
934 # for vnfr in db_vnfrs_list:
935 # self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
937 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
938 # async def test_terminate_primitive(self):
939 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
940 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
941 # # set instantiation task as completed
942 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
943 # update_dict={"operationState": "COMPLETED"})
945 # # modify vnfd descriptor to include terminate_primitive
946 # terminate_primitive = [{
948 # "parameter": [{"name": "filename", "value": "terminate_filename"}],
951 # db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
952 # self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
953 # {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
955 # await self.my_ns.terminate(nsr_id, nslcmop_id)
956 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
957 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
958 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
959 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
960 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
961 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
962 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
963 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
964 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
966 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
968 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
970 @patch("osm_lcm.data_utils.vnfd.find_software_version")
971 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
972 async def test_update_change_vnfpkg_sw_version_not_changed(
975 mock_software_version
,
979 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed,
980 Charm package changed, sw-version is not changed"""
983 q_filter
={"_id": vnfd_id
},
984 update_dict
={"_admin.revision": 3, "kdu": []},
989 q_filter
={"_id": vnfd_id
+ ":1"},
990 update_dict
={"_admin.revision": 1, "kdu": []},
993 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
995 mock_charm_hash
.return_value
= True
996 mock_software_version
.side_effect
= ["1.0", "1.0"]
998 task
= asyncio
.Future()
999 task
.set_result(("COMPLETED", "some_output"))
1000 mock_charm_upgrade
.return_value
= task
1002 instance
= self
.my_ns
1003 fs
= deepcopy(update_fs
)
1006 expected_operation_state
= "COMPLETED"
1007 expected_operation_error
= ""
1008 expected_vnfr_revision
= 3
1009 expected_ns_state
= "INSTANTIATED"
1010 expected_ns_operational_state
= "running"
1012 await instance
.update(nsr_id
, nslcmop_id
)
1013 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1016 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1019 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1020 "operational-status"
1022 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1025 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1026 mock_charm_hash
.assert_called_with(
1027 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1028 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1030 self
.assertEqual(fs
.sync
.call_count
, 2)
1031 self
.assertEqual(return_ns_state
, expected_ns_state
)
1032 self
.assertEqual(return_operation_state
, expected_operation_state
)
1033 self
.assertEqual(return_operation_error
, expected_operation_error
)
1034 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1035 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1037 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1039 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1041 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1042 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1043 async def test_update_change_vnfpkg_vnfd_revision_not_changed(
1046 mock_software_version
,
1050 """Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"""
1052 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 1}
1054 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1056 mock_charm_hash
.return_value
= True
1058 task
= asyncio
.Future()
1059 task
.set_result(("COMPLETED", "some_output"))
1060 mock_charm_upgrade
.return_value
= task
1062 instance
= self
.my_ns
1064 expected_operation_state
= "COMPLETED"
1065 expected_operation_error
= ""
1066 expected_vnfr_revision
= 1
1067 expected_ns_state
= "INSTANTIATED"
1068 expected_ns_operational_state
= "running"
1070 await instance
.update(nsr_id
, nslcmop_id
)
1071 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1074 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1077 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1078 "operational-status"
1080 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1081 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1084 mock_charm_hash
.assert_not_called()
1085 mock_software_version
.assert_not_called()
1086 mock_juju_bundle
.assert_not_called()
1087 mock_charm_upgrade
.assert_not_called()
1088 update_fs
.sync
.assert_not_called()
1089 self
.assertEqual(return_ns_state
, expected_ns_state
)
1090 self
.assertEqual(return_operation_state
, expected_operation_state
)
1091 self
.assertEqual(return_operation_error
, expected_operation_error
)
1092 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1093 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1095 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1097 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1099 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1100 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1101 async def test_update_change_vnfpkg_charm_is_not_changed(
1104 mock_software_version
,
1108 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1109 Charm package is not changed, sw-version is not changed"""
1112 q_filter
={"_id": vnfd_id
},
1113 update_dict
={"_admin.revision": 3, "kdu": []},
1117 q_filter
={"_id": vnfd_id
+ ":1"},
1118 update_dict
={"_admin.revision": 1, "kdu": []},
1120 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1122 mock_charm_hash
.return_value
= False
1123 mock_software_version
.side_effect
= ["1.0", "1.0"]
1125 instance
= self
.my_ns
1126 fs
= deepcopy(update_fs
)
1128 expected_operation_state
= "COMPLETED"
1129 expected_operation_error
= ""
1130 expected_vnfr_revision
= 3
1131 expected_ns_state
= "INSTANTIATED"
1132 expected_ns_operational_state
= "running"
1134 await instance
.update(nsr_id
, nslcmop_id
)
1135 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1138 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1141 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1142 "operational-status"
1144 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1147 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1148 mock_charm_hash
.assert_called_with(
1149 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1150 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1152 self
.assertEqual(fs
.sync
.call_count
, 2)
1153 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1154 mock_juju_bundle
.assert_not_called()
1155 mock_charm_upgrade
.assert_not_called()
1156 self
.assertEqual(return_ns_state
, expected_ns_state
)
1157 self
.assertEqual(return_operation_state
, expected_operation_state
)
1158 self
.assertEqual(return_operation_error
, expected_operation_error
)
1159 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1160 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1162 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1163 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1165 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1167 @patch("osm_lcm.lcm_utils.get_charm_artifact_path")
1168 async def test_update_change_vnfpkg_sw_version_changed(
1169 self
, mock_charm_artifact
, mock_charm_upgrade
, mock_charm_hash
, mock_juju_bundle
1171 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1172 Charm package exists, sw-version changed."""
1175 q_filter
={"_id": vnfd_id
},
1176 update_dict
={"_admin.revision": 3, "software-version": "3.0", "kdu": []},
1180 q_filter
={"_id": vnfd_id
+ ":1"},
1181 update_dict
={"_admin.revision": 1, "kdu": []},
1185 q_filter
={"_id": vnfr_id
},
1186 update_dict
={"revision": 1},
1188 mock_charm_hash
.return_value
= False
1190 mock_charm_artifact
.side_effect
= [
1191 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1192 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1195 instance
= self
.my_ns
1196 fs
= deepcopy(update_fs
)
1198 expected_operation_state
= "FAILED"
1199 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."
1200 expected_vnfr_revision
= 1
1201 expected_ns_state
= "INSTANTIATED"
1202 expected_ns_operational_state
= "running"
1204 await instance
.update(nsr_id
, nslcmop_id
)
1205 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1208 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1211 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1212 "operational-status"
1214 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1217 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1218 self
.assertEqual(fs
.sync
.call_count
, 2)
1219 mock_charm_hash
.assert_not_called()
1220 mock_juju_bundle
.assert_not_called()
1221 mock_charm_upgrade
.assert_not_called()
1222 self
.assertEqual(return_ns_state
, expected_ns_state
)
1223 self
.assertEqual(return_operation_state
, expected_operation_state
)
1224 self
.assertEqual(return_operation_error
, expected_operation_error
)
1225 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1226 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1228 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1229 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1231 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1233 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1234 async def test_update_change_vnfpkg_juju_bundle_exists(
1236 mock_software_version
,
1241 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1242 Charm package exists, sw-version not changed, juju-bundle exists"""
1243 # Upgrade is not allowed with juju bundles, this will cause TypeError
1246 q_filter
={"_id": vnfd_id
},
1248 "_admin.revision": 5,
1249 "software-version": "1.0",
1250 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1255 q_filter
={"_id": vnfd_id
+ ":1"},
1257 "_admin.revision": 1,
1258 "software-version": "1.0",
1259 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1264 q_filter
={"_id": nsr_id
},
1266 "_admin.deployed.VCA.0.kdu_name": "native-kdu",
1269 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1271 mock_charm_hash
.side_effect
= [True]
1272 mock_software_version
.side_effect
= ["1.0", "1.0"]
1273 mock_juju_bundle
.return_value
= True
1274 instance
= self
.my_ns
1275 fs
= deepcopy(update_fs
)
1278 expected_vnfr_revision
= 1
1279 expected_ns_state
= "INSTANTIATED"
1280 expected_ns_operational_state
= "running"
1282 await instance
.update(nsr_id
, nslcmop_id
)
1283 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1284 "operational-status"
1286 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1289 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1290 self
.assertEqual(fs
.sync
.call_count
, 2)
1291 mock_charm_upgrade
.assert_not_called()
1292 mock_charm_hash
.assert_not_called()
1293 self
.assertEqual(return_ns_state
, expected_ns_state
)
1294 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1295 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1297 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1299 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1301 async def test_update_change_vnfpkg_charm_upgrade_failed(
1302 self
, mock_charm_upgrade
, mock_charm_hash
1304 """ "Update type: CHANGE_VNFPKG, latest_vnfd revision changed"
1305 Charm package exists, sw-version not changed, charm-upgrade failed"""
1308 q_filter
={"_id": vnfd_id
},
1310 "_admin.revision": 3,
1311 "software-version": "1.0",
1317 q_filter
={"_id": vnfd_id
+ ":1"},
1319 "_admin.revision": 1,
1320 "software-version": "1.0",
1324 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1326 mock_charm_hash
.return_value
= True
1328 task
= asyncio
.Future()
1329 task
.set_result(("FAILED", "some_error"))
1330 mock_charm_upgrade
.return_value
= task
1332 instance
= self
.my_ns
1333 fs
= deepcopy(update_fs
)
1335 expected_operation_state
= "FAILED"
1336 expected_operation_error
= "some_error"
1337 expected_vnfr_revision
= 1
1338 expected_ns_state
= "INSTANTIATED"
1339 expected_ns_operational_state
= "running"
1341 await instance
.update(nsr_id
, nslcmop_id
)
1342 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1345 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1348 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1349 "operational-status"
1351 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1354 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1355 self
.assertEqual(fs
.sync
.call_count
, 2)
1356 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1357 self
.assertEqual(mock_charm_upgrade
.call_count
, 1)
1358 self
.assertEqual(return_ns_state
, expected_ns_state
)
1359 self
.assertEqual(return_operation_state
, expected_operation_state
)
1360 self
.assertEqual(return_operation_error
, expected_operation_error
)
1361 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1362 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1364 def test_ns_update_find_sw_version_vnfd_not_includes(self
):
1365 """Find software version, VNFD does not have software version"""
1367 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1368 expected_result
= "1.0"
1369 result
= find_software_version(db_vnfd
)
1370 self
.assertEqual(result
, expected_result
, "Default sw version should be 1.0")
1372 def test_ns_update_find_sw_version_vnfd_includes(self
):
1373 """Find software version, VNFD includes software version"""
1375 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1376 db_vnfd
["software-version"] = "3.1"
1377 expected_result
= "3.1"
1378 result
= find_software_version(db_vnfd
)
1379 self
.assertEqual(result
, expected_result
, "VNFD software version is wrong")
1381 @patch("os.path.exists")
1382 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1383 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1384 def test_ns_update_check_charm_hash_not_changed(
1385 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1387 """Check charm hash, Hash did not change"""
1389 current_path
, target_path
= "/tmp/charm1", "/tmp/charm1"
1392 fs
.path
.__add
__ = Mock()
1393 fs
.path
.side_effect
= [current_path
, target_path
]
1394 fs
.path
.__add
__.side_effect
= [current_path
, target_path
]
1396 mock_path_exists
.side_effect
= [True, True]
1398 mock_compare_charmdir_hash
.return_value
= callable(False)
1399 mock_compare_charm_hash
.return_value
= callable(False)
1401 instance
= self
.my_ns
1403 expected_result
= False
1405 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1406 self
.assertEqual(result
, expected_result
, "Wrong charm hash control value")
1407 self
.assertEqual(mock_path_exists
.call_count
, 2)
1408 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 1)
1409 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1411 @patch("os.path.exists")
1412 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1413 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1414 def test_ns_update_check_charm_hash_changed(
1415 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1417 """Check charm hash, Hash has changed"""
1419 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1422 fs
.path
.__add
__ = Mock()
1423 fs
.path
.side_effect
= [current_path
, target_path
, current_path
, target_path
]
1424 fs
.path
.__add
__.side_effect
= [
1431 mock_path_exists
.side_effect
= [True, True]
1432 mock_compare_charmdir_hash
.return_value
= callable(True)
1433 mock_compare_charm_hash
.return_value
= callable(True)
1435 instance
= self
.my_ns
1437 expected_result
= True
1439 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1440 self
.assertEqual(result
, expected_result
, "Wrong charm hash control value")
1441 self
.assertEqual(mock_path_exists
.call_count
, 2)
1442 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 1)
1443 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1445 @patch("os.path.exists")
1446 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1447 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1448 def test_ns_update_check_no_charm_path(
1449 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1451 """Check charm hash, Charm path does not exist"""
1453 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1456 fs
.path
.__add
__ = Mock()
1457 fs
.path
.side_effect
= [current_path
, target_path
, current_path
, target_path
]
1458 fs
.path
.__add
__.side_effect
= [
1465 mock_path_exists
.side_effect
= [True, False]
1467 mock_compare_charmdir_hash
.return_value
= callable(False)
1468 mock_compare_charm_hash
.return_value
= callable(False)
1469 instance
= self
.my_ns
1472 with self
.assertRaises(LcmException
):
1473 instance
.check_charm_hash_changed(current_path
, target_path
)
1474 self
.assertEqual(mock_path_exists
.call_count
, 2)
1475 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 0)
1476 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1478 def test_ns_update_check_juju_bundle_existence_bundle_exists(self
):
1479 """Check juju bundle existence"""
1480 test_vnfd2
= self
.db
.get_one(
1481 "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
1483 expected_result
= "stable/native-kdu"
1484 result
= check_juju_bundle_existence(test_vnfd2
)
1485 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1487 def test_ns_update_check_juju_bundle_existence_bundle_does_not_exist(self
):
1488 """Check juju bundle existence"""
1489 test_vnfd1
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1490 expected_result
= None
1491 result
= check_juju_bundle_existence(test_vnfd1
)
1492 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1494 def test_ns_update_check_juju_bundle_existence_empty_vnfd(self
):
1495 """Check juju bundle existence"""
1497 expected_result
= None
1498 result
= check_juju_bundle_existence(test_vnfd1
)
1499 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1501 def test_ns_update_check_juju_bundle_existence_invalid_vnfd(self
):
1502 """Check juju bundle existence"""
1503 test_vnfd1
= [{"_id": vnfd_id
}]
1504 with self
.assertRaises(AttributeError):
1505 check_juju_bundle_existence(test_vnfd1
)
1507 def test_ns_update_check_juju_charm_artifacts_base_folder_wth_pkgdir(self
):
1508 """Check charm artifacts"""
1511 "pkg-dir": "hackfest_3charmed_vnfd",
1513 charm_name
= "simple"
1514 charm_type
= "lxc_proxy_charm"
1516 expected_result
= f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple"
1517 result
= get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
)
1518 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1520 def test_ns_update_check_juju_charm_artifacts_base_folder_wthout_pkgdir(self
):
1521 """Check charm artifacts, SOL004 packages"""
1525 charm_name
= "basic"
1526 charm_type
, revision
= "", ""
1527 expected_result
= f
"{vnfd_id}/Scripts/helm-charts/basic"
1528 result
= get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
)
1529 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1532 if __name__
== "__main__":