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
.load(
126 descriptors
.ro_update_action_text
, Loader
=yaml
.Loader
130 if kwargs
.get("delete"):
131 ro_ns_desc
= yaml
.load(
132 descriptors
.ro_delete_action_text
, Loader
=yaml
.Loader
137 ro_ns_desc
= yaml
.load(descriptors
.ro_ns_text
, Loader
=yaml
.Loader
)
139 # if ip address provided, replace descriptor
140 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
142 ip_addresses_list
= ip_addresses
.split(",")
143 for vnf
in ro_ns_desc
["vnfs"]:
144 if not ip_addresses_list
:
146 vnf
["ip_address"] = ip_addresses_list
[0]
147 for vm
in vnf
["vms"]:
148 if not ip_addresses_list
:
150 vm
["ip_address"] = ip_addresses_list
.pop(0)
154 for net
in ro_ns_desc
["nets"]:
155 if net
["status"] != "ACTIVE":
156 net
["status"] = "ACTIVE"
159 for vnf
in ro_ns_desc
["vnfs"]:
160 for vm
in vnf
["vms"]:
161 if vm
["status"] != "ACTIVE":
162 vm
["status"] = "ACTIVE"
165 def _ro_deploy(self
, *args
, **kwargs
):
166 return {"action_id": args
[1]["action_id"], "nsr_id": args
[0], "status": "ok"}
168 def _return_uuid(self
, *args
, **kwargs
):
171 async def setUp(self
):
174 if not getenv("OSMLCMTEST_DB_NOMOCK"):
175 # Cleanup singleton Database instance
176 Database
.instance
= None
178 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
180 "vnfds", yaml
.load(descriptors
.db_vnfds_text
, Loader
=yaml
.Loader
)
184 yaml
.load(descriptors
.db_vnfds_revisions_text
, Loader
=yaml
.Loader
),
187 "nsds", yaml
.load(descriptors
.db_nsds_text
, Loader
=yaml
.Loader
)
190 "nsrs", yaml
.load(descriptors
.db_nsrs_text
, Loader
=yaml
.Loader
)
194 yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
),
198 yaml
.load(descriptors
.db_k8sclusters_text
, Loader
=yaml
.Loader
),
201 "nslcmops", yaml
.load(descriptors
.db_nslcmops_text
, Loader
=yaml
.Loader
)
204 "vnfrs", yaml
.load(descriptors
.db_vnfrs_text
, Loader
=yaml
.Loader
)
206 self
.db_vim_accounts
= yaml
.load(
207 descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
211 self
.msg
= asynctest
.Mock(MsgKafka())
214 if not getenv("OSMLCMTEST_FS_NOMOCK"):
215 self
.fs
= asynctest
.Mock(
216 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
218 self
.fs
.get_params
.return_value
= {
219 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
221 self
.fs
.file_open
= asynctest
.mock_open()
222 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
223 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
226 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
227 self
.lcm_tasks
.lock_HA
.return_value
= True
228 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
229 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
232 if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
233 ns
.K8sJujuConnector
= asynctest
.MagicMock(ns
.K8sJujuConnector
)
234 ns
.K8sHelmConnector
= asynctest
.MagicMock(ns
.K8sHelmConnector
)
235 ns
.K8sHelm3Connector
= asynctest
.MagicMock(ns
.K8sHelm3Connector
)
237 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
238 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
239 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
242 self
.my_ns
= ns
.NsLcm(self
.msg
, self
.lcm_tasks
, lcm_config
, self
.loop
)
243 self
.my_ns
.fs
= self
.fs
244 self
.my_ns
.db
= self
.db
245 self
.my_ns
._wait
_dependent
_n
2vc
= asynctest
.CoroutineMock()
248 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
249 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
252 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
253 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
254 # self.my_ns.n2vc = asynctest.Mock(N2VC())
255 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv(
256 "OSMLCM_VCA_PUBKEY", "public_key"
258 # allow several versions of n2vc
259 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(
260 side_effect
=self
._n
2vc
_FormatApplicationName
()
262 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(
263 side_effect
=self
._n
2vc
_DeployCharms
265 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
266 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
268 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(
271 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(
274 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
275 side_effect
=self
._return
_uuid
277 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
278 side_effect
=self
._return
_uuid
280 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(
281 return_value
="completed"
283 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(
284 return_value
={"result": "ok", "pubkey": pub_key
}
286 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(
289 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
290 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key")
292 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(
297 if not getenv("OSMLCMTEST_RO_NOMOCK"):
298 self
.my_ns
.RO
= asynctest
.Mock(
299 NgRoClient(self
.loop
, **lcm_config
["ro_config"])
301 # TODO first time should be empty list, following should return a dict
302 # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
303 self
.my_ns
.RO
.deploy
= asynctest
.CoroutineMock(
304 self
.my_ns
.RO
.deploy
, side_effect
=self
._ro
_deploy
306 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status)
307 # self.my_ns.RO.create_action = asynctest.CoroutineMock(self.my_ns.RO.create_action,
308 # return_value={"vm-id": {"vim_result": 200,
309 # "description": "done"}})
310 self
.my_ns
.RO
.delete
= asynctest
.CoroutineMock(self
.my_ns
.RO
.delete
)
312 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
313 # async def test_instantiate(self):
314 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
315 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
316 # # print("Test instantiate started")
318 # # delete deployed information of database
319 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
320 # if self.db.get_list("nsrs")[0]["_admin"].get("deployed"):
321 # del self.db.get_list("nsrs")[0]["_admin"]["deployed"]
322 # for db_vnfr in self.db.get_list("vnfrs"):
323 # db_vnfr.pop("ip_address", None)
324 # for db_vdur in db_vnfr["vdur"]:
325 # db_vdur.pop("ip_address", None)
326 # db_vdur.pop("mac_address", None)
327 # if getenv("OSMLCMTEST_RO_VIMID"):
328 # self.db.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
329 # if getenv("OSMLCMTEST_RO_VIMID"):
330 # self.db.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
332 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
334 # self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
335 # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
336 # "operationState": "COMPLETED"},
338 # self.lcm_tasks.lock_HA.assert_called_once_with('ns', 'nslcmops', nslcmop_id)
339 # if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
340 # self.assertTrue(self.my_ns.logger.debug.called, "Debug method not called")
341 # self.my_ns.logger.error.assert_not_called()
342 # self.my_ns.logger.exception().assert_not_called()
344 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
345 # self.assertTrue(self.db.set_one.called, "db.set_one not called")
346 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
347 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
348 # self.assertEqual(db_nsr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
349 # for vnfr in db_vnfrs_list:
350 # self.assertEqual(vnfr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
352 # if not getenv("OSMLCMTEST_VCA_NOMOCK"):
353 # # check intial-primitives called
354 # self.assertTrue(self.my_ns.n2vc.exec_primitive.called,
355 # "Exec primitive not called for initial config primitive")
356 # for _call in self.my_ns.n2vc.exec_primitive.call_args_list:
357 # self.assertIn(_call[1]["primitive_name"], ("config", "touch"),
358 # "called exec primitive with a primitive different than config or touch")
360 # # TODO add more checks of called methods
361 # # TODO add a terminate
363 # async def test_instantiate_ee_list(self):
364 # # Using modern IM where configuration is in the new format of execution_environment_list
365 # ee_descriptor_id = "charm_simple"
366 # non_used_initial_primitive = {
367 # "name": "not_to_be_called",
369 # "execution-environment-ref": "not_used_ee"
373 # "id": ee_descriptor_id,
374 # "juju": {"charm": "simple"},
381 # q_filter={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"},
382 # update_dict={"vnf-configuration.0.execution-environment-list": ee_list,
383 # "vnf-configuration.0.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id,
384 # "vnf-configuration.0.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id,
385 # "vnf-configuration.0.initial-config-primitive.2": non_used_initial_primitive,
386 # "vnf-configuration.0.config-primitive.0.execution-environment-ref": ee_descriptor_id,
387 # "vnf-configuration.0.config-primitive.0.execution-environment-primitive": "touch_charm",
389 # unset={"vnf-configuration.juju": None})
390 # await self.test_instantiate()
391 # # this will check that the initial-congig-primitive 'not_to_be_called' is not called
393 @asynctest.fail_on(active_handles
=True)
394 async def test_start_stop_rebuild_pass(self
):
395 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
396 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops"]
397 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
398 additional_param
= {"count-index": "0"}
399 operation_type
= "start"
400 await self
.my_ns
.rebuild_start_stop(
401 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
403 expected_value
= "COMPLETED"
404 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
407 self
.assertEqual(return_value
, expected_value
)
409 @asynctest.fail_on(active_handles
=True)
410 async def test_start_stop_rebuild_fail(self
):
411 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
412 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops1"]
413 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
414 additional_param
= {"count-index": "0"}
415 operation_type
= "stop"
416 await self
.my_ns
.rebuild_start_stop(
417 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
419 expected_value
= "Error"
420 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
423 self
.assertEqual(return_value
, expected_value
)
425 # Test scale() and related methods
426 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
427 async def test_scale(self
):
428 # print("Test scale started")
430 # TODO: Add more higher-lever tests here, for example:
431 # scale-out/scale-in operations with success/error result
433 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
434 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
435 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
436 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
437 expected_value
= "FAILED"
438 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
441 self
.assertEqual(return_value
, expected_value
)
442 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
444 # Test scale() for native kdu
445 # this also includes testing _scale_kdu()
446 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["ns"]
447 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["instantiate"]
449 self
.my_ns
.k8sclusterjuju
.scale
= asynctest
.mock
.CoroutineMock()
450 self
.my_ns
.k8sclusterjuju
.exec_primitive
= asynctest
.mock
.CoroutineMock()
451 self
.my_ns
.k8sclusterjuju
.get_scale_count
= asynctest
.mock
.CoroutineMock(
454 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
455 expected_value
= "COMPLETED"
456 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
459 self
.assertEqual(return_value
, expected_value
)
460 self
.my_ns
.k8sclusterjuju
.scale
.assert_called_once()
462 # Test scale() for native kdu with 2 resource
463 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["ns"]
464 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["instantiate"]
466 self
.my_ns
.k8sclusterjuju
.get_scale_count
.return_value
= 2
467 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
468 expected_value
= "COMPLETED"
469 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
472 self
.assertEqual(return_value
, expected_value
)
473 self
.my_ns
.k8sclusterjuju
.scale
.assert_called()
475 async def test_vca_status_refresh(self
):
476 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
477 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
478 await self
.my_ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
479 expected_value
= dict()
480 return_value
= dict()
481 vnf_descriptors
= self
.db
.get_list("vnfds")
482 for i
, _
in enumerate(vnf_descriptors
):
483 for j
, value
in enumerate(vnf_descriptors
[i
]["df"]):
484 if "lcm-operations-configuration" in vnf_descriptors
[i
]["df"][j
]:
487 in value
["lcm-operations-configuration"][
488 "operate-vnf-op-config"
491 for k
, v
in enumerate(
492 value
["lcm-operations-configuration"][
493 "operate-vnf-op-config"
497 v
.get("execution-environment-list")
498 and "juju" in v
["execution-environment-list"][k
]
500 expected_value
= self
.db
.get_list("nsrs")[i
][
503 await self
.my_ns
._on
_update
_n
2vc
_db
(
504 "nsrs", {"_id": nsr_id
}, "_admin.deployed.VCA.0", {}
506 return_value
= self
.db
.get_list("nsrs")[i
]["vcaStatus"]
507 self
.assertEqual(return_value
, expected_value
)
509 # Test _retry_or_skip_suboperation()
511 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
512 # - if marked as anything but 'COMPLETED', the suboperation index is expected
513 def test_scale_retry_or_skip_suboperation(self
):
514 # Load an alternative 'nslcmops' YAML for this test
515 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
516 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
518 # Test when 'operationState' is 'COMPLETED'
519 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = "COMPLETED"
520 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
521 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
522 self
.assertEqual(return_value
, expected_value
)
523 # Test when 'operationState' is not 'COMPLETED'
524 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = None
525 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
526 expected_value
= op_index
527 self
.assertEqual(return_value
, expected_value
)
529 # Test _find_suboperation()
530 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
531 def test_scale_find_suboperation(self
):
532 # Load an alternative 'nslcmops' YAML for this test
533 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
534 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
535 # Find this sub-operation
537 vnf_index
= db_nslcmop
["_admin"]["operations"][op_index
]["member_vnf_index"]
538 primitive
= db_nslcmop
["_admin"]["operations"][op_index
]["primitive"]
539 primitive_params
= db_nslcmop
["_admin"]["operations"][op_index
][
543 "member_vnf_index": vnf_index
,
544 "primitive": primitive
,
545 "primitive_params": primitive_params
,
547 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
548 self
.assertEqual(found_op_index
, op_index
)
549 # Test with not-matching params
551 "member_vnf_index": vnf_index
,
553 "primitive_params": primitive_params
,
555 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
556 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
559 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
560 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
562 # Test _update_suboperation_status()
563 def test_scale_update_suboperation_status(self
):
564 self
.db
.set_one
= asynctest
.Mock()
565 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
566 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
568 # Force the initial values to be distinct from the updated ones
569 q_filter
= {"_id": db_nslcmop
["_id"]}
570 # Test to change 'operationState' and 'detailed-status'
571 operationState
= "COMPLETED"
572 detailed_status
= "Done"
573 expected_update_dict
= {
574 "_admin.operations.0.operationState": operationState
,
575 "_admin.operations.0.detailed-status": detailed_status
,
577 self
.my_ns
._update
_suboperation
_status
(
578 db_nslcmop
, op_index
, operationState
, detailed_status
580 self
.db
.set_one
.assert_called_once_with(
583 update_dict
=expected_update_dict
,
587 def test_scale_add_suboperation(self
):
588 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
589 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
591 num_ops_before
= len(db_nslcmop
.get("_admin", {}).get("operations", [])) - 1
593 vdu_count_index
= None
596 mapped_primitive_params
= {
599 "data-type": "STRING",
601 "default-value": "<touch_filename2>",
606 operationState
= "PROCESSING"
607 detailed_status
= "In progress"
608 operationType
= "PRE-SCALE"
609 # Add a 'pre-scale' suboperation
610 op_index_after
= self
.my_ns
._add
_suboperation
(
617 mapped_primitive_params
,
622 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
624 # Delete all suboperations and add the same operation again
625 del db_nslcmop
["_admin"]["operations"]
626 op_index_zero
= self
.my_ns
._add
_suboperation
(
633 mapped_primitive_params
,
638 self
.assertEqual(op_index_zero
, 0)
640 # Add a 'RO' suboperation
641 RO_nsr_id
= "1234567890"
646 "member-vnf-index": "1",
647 "osm_vdu_id": "dataVM",
650 op_index
= self
.my_ns
._add
_suboperation
(
657 mapped_primitive_params
,
664 db_RO_nsr_id
= db_nslcmop
["_admin"]["operations"][op_index
]["RO_nsr_id"]
665 self
.assertEqual(op_index
, 1)
666 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
668 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
669 op_index_invalid
= self
.my_ns
._add
_suboperation
(
670 None, None, None, None, None, None, None, None, None, None, None
672 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
674 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
675 # check the possible return values:
676 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
677 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
678 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
679 def test_scale_check_or_add_scale_suboperation(self
):
680 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
681 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
682 operationType
= "PRE-SCALE"
688 "data-type": "STRING",
690 "default-value": "<touch_filename2>",
696 # Delete all sub-operations to be sure this is a new sub-operation
697 del db_nslcmop
["_admin"]["operations"]
699 # Add a new sub-operation
700 # For new sub-operations, operationState is set to 'PROCESSING' by default
701 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
702 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
704 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
706 # Use the same parameters again to match the already added sub-operation
707 # which has status 'PROCESSING' (!= 'COMPLETED') by default
708 # The expected return value is a non-negative number
709 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
710 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
712 self
.assertTrue(op_index_existing
>= 0)
714 # Change operationState 'manually' for this sub-operation
715 db_nslcmop
["_admin"]["operations"][op_index_existing
][
718 # Then use the same parameters again to match the already added sub-operation,
719 # which now has status 'COMPLETED'
720 # The expected return value is SUBOPERATION_STATUS_SKIP
721 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
722 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
724 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
726 # RO sub-operation test:
727 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
728 RO_nsr_id
= "1234567890"
733 "member-vnf-index": "1",
734 "osm_vdu_id": "dataVM",
737 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
738 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
740 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
742 # Use the same parameters again to match the already added RO sub-operation
743 op_index_existing_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
.assertTrue(op_index_existing_RO
>= 0)
748 # Change operationState 'manually' for this RO sub-operation
749 db_nslcmop
["_admin"]["operations"][op_index_existing_RO
][
752 # Then use the same parameters again to match the already added sub-operation,
753 # which now has status 'COMPLETED'
754 # The expected return value is SUBOPERATION_STATUS_SKIP
755 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
756 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
758 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
760 async def test_deploy_kdus(self
):
761 nsr_id
= descriptors
.test_ids
["TEST-KDU"]["ns"]
762 nslcmop_id
= descriptors
.test_ids
["TEST-KDU"]["instantiate"]
763 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
764 db_vnfr
= self
.db
.get_one(
765 "vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "multikdu"}
767 db_vnfrs
= {"multikdu": db_vnfr
}
768 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
772 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
= asynctest
.mock
.Mock()
773 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
.return_value
= "k8s_id"
774 self
.my_ns
.k8sclusterhelm3
.install
= asynctest
.CoroutineMock()
775 self
.my_ns
.k8sclusterhelm3
.synchronize_repos
= asynctest
.CoroutineMock(
776 return_value
=("", "")
778 self
.my_ns
.k8sclusterhelm3
.get_services
= asynctest
.CoroutineMock(
781 await self
.my_ns
.deploy_kdus(
782 logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
784 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
785 db_nsr
= self
.db
.get_list("nsrs")[1]
788 db_nsr
["_admin"]["deployed"],
789 "K8s entry not created at '_admin.deployed'",
791 self
.assertIsInstance(
792 db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
795 len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
798 "kdu-instance": "k8s_id",
799 "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
800 "k8scluster-type": "helm-chart-v3",
802 "member-vnf-index": "multikdu",
804 "kdu-deployment-name": None,
807 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][0])
808 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
809 expected_kdu_model
= "stable/openldap:1.2.1"
810 self
.assertEqual(nsr_result
, k8s_instace_info
)
812 nsr_kdu_model_result
in expected_kdu_model
813 or expected_kdu_model
in nsr_kdu_model_result
815 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][1])
816 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
817 k8s_instace_info
["kdu-name"] = "mongo"
818 expected_kdu_model
= "stable/mongodb"
819 self
.assertEqual(nsr_result
, k8s_instace_info
)
821 nsr_kdu_model_result
in expected_kdu_model
822 or expected_kdu_model
in nsr_kdu_model_result
825 # Test remove_vnf() and related methods
826 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
827 async def test_remove_vnf(self
):
829 nsr_id
= descriptors
.test_ids
["TEST-UPDATE"]["ns"]
830 nslcmop_id
= descriptors
.test_ids
["TEST-UPDATE"]["removeVnf"]
831 vnf_instance_id
= descriptors
.test_ids
["TEST-UPDATE"]["vnf"]
832 mock_wait_ng_ro
= asynctest
.CoroutineMock()
833 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
834 await self
.my_ns
.update(nsr_id
, nslcmop_id
)
835 expected_value
= "COMPLETED"
836 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
839 self
.assertEqual(return_value
, expected_value
)
840 with self
.assertRaises(Exception) as context
:
841 self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
842 self
.assertTrue("database exception Not found entry with filter" in str(context
.exception
))
844 # test vertical scale executes sucessfully
845 # @patch("osm_lcm.ng_ro.status.response")
846 @asynctest.fail_on(active_handles
=True)
847 async def test_vertical_scaling(self
):
848 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
849 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate"]
851 # calling the vertical scale fucntion
852 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status("update"))
853 mock_wait_ng_ro
= asynctest
.CoroutineMock()
854 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
855 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
856 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
859 expected_value
= "COMPLETED"
860 self
.assertEqual(return_value
, expected_value
)
862 # test vertical scale executes fail
863 @asynctest.fail_on(active_handles
=True)
864 async def test_vertical_scaling_fail(self
):
865 # get th nsr nad nslcmops id from descriptors
866 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
867 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate-1"]
869 # calling the vertical scale fucntion
870 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
871 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
874 expected_value
= "FAILED"
875 self
.assertEqual(return_value
, expected_value
)
877 # async def test_instantiate_pdu(self):
878 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
879 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
880 # # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
881 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
882 # update_dict={"ip-address": "10.205.1.46",
883 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
884 # "vdur.0.pdu-type": "PDU-TYPE-1",
885 # "vdur.0.ip-address": "10.205.1.46",
887 # unset={"vdur.status": None})
888 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
889 # update_dict={"ip-address": "10.205.1.47",
890 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
891 # "vdur.0.pdu-type": "PDU-TYPE-1",
892 # "vdur.0.ip-address": "10.205.1.47",
894 # unset={"vdur.status": None})
896 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
897 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
898 # self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
899 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
900 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
901 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
902 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
904 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
905 # async def test_terminate_without_configuration(self):
906 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
907 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
908 # # set instantiation task as completed
909 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
910 # update_dict={"operationState": "COMPLETED"})
911 # self.db.set_one("nsrs", {"_id": nsr_id},
912 # update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
914 # await self.my_ns.terminate(nsr_id, nslcmop_id)
915 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
916 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
917 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
918 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
919 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
920 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
921 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
922 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
923 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
924 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
925 # for vnfr in db_vnfrs_list:
926 # self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
928 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
929 # async def test_terminate_primitive(self):
930 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
931 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
932 # # set instantiation task as completed
933 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
934 # update_dict={"operationState": "COMPLETED"})
936 # # modify vnfd descriptor to include terminate_primitive
937 # terminate_primitive = [{
939 # "parameter": [{"name": "filename", "value": "terminate_filename"}],
942 # db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
943 # self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
944 # {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
946 # await self.my_ns.terminate(nsr_id, nslcmop_id)
947 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
948 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
949 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
950 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
951 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
952 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
953 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
954 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
955 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
959 async def test_update(self
):
961 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
962 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["update"]
963 vnfr_id
= "6421c7c9-d865-4fb4-9a13-d4275d243e01"
964 vnfd_id
= "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
967 mock_charm_hash
.reset_mock()
968 mock_juju_bundle
.reset_mock()
970 mock_charm_upgrade
.reset_mock()
971 mock_software_version
.reset_mock()
975 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
976 "Charm package changed, sw-version is not changed.",
981 q_filter
={"_id": vnfd_id
},
982 update_dict
={"_admin.revision": 3, "kdu": []},
987 q_filter
={"_id": vnfd_id
+ ":1"},
988 update_dict
={"_admin.revision": 1, "kdu": []},
992 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
995 mock_charm_hash
= Mock(autospec
=True)
996 mock_charm_hash
.return_value
= True
998 mock_juju_bundle
= Mock(return_value
=None)
1000 mock_software_version
= Mock(autospec
=True)
1001 mock_software_version
.side_effect
= ["1.0", "1.0"]
1003 mock_charm_upgrade
= asynctest
.Mock(autospec
=True)
1004 task
= asyncio
.Future()
1005 task
.set_result(("COMPLETED", "some_output"))
1006 mock_charm_upgrade
.return_value
= task
1008 fs
= Mock(autospec
=True)
1009 fs
.path
.__add
__ = Mock()
1010 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1011 fs
.sync
.side_effect
= [None, None]
1013 instance
= self
.my_ns
1015 expected_operation_state
= "COMPLETED"
1016 expected_operation_error
= ""
1017 expected_vnfr_revision
= 3
1018 expected_ns_state
= "INSTANTIATED"
1019 expected_ns_operational_state
= "running"
1021 with patch
.object(instance
, "fs", fs
), patch(
1022 "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
1023 ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade
), patch(
1024 "osm_lcm.data_utils.vnfd.find_software_version", mock_software_version
1026 "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
1029 await instance
.update(nsr_id
, nslcmop_id
)
1030 return_operation_state
= self
.db
.get_one(
1031 "nslcmops", {"_id": nslcmop_id
}
1032 ).get("operationState")
1033 return_operation_error
= self
.db
.get_one(
1034 "nslcmops", {"_id": nslcmop_id
}
1035 ).get("errorMessage")
1036 return_ns_operational_state
= self
.db
.get_one(
1037 "nsrs", {"_id": nsr_id
}
1038 ).get("operational-status")
1040 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1044 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1048 mock_charm_hash
.assert_called_with(
1049 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
1050 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:3/hackfest_3charmed_vnfd/charms/simple",
1053 self
.assertEqual(fs
.sync
.call_count
, 2)
1054 self
.assertEqual(return_ns_state
, expected_ns_state
)
1055 self
.assertEqual(return_operation_state
, expected_operation_state
)
1056 self
.assertEqual(return_operation_error
, expected_operation_error
)
1058 return_ns_operational_state
, expected_ns_operational_state
1060 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1065 i
=2, t
="Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"
1069 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 1}
1073 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1076 mock_charm_hash
= Mock(autospec
=True)
1077 mock_charm_hash
.return_value
= True
1079 mock_juju_bundle
= Mock(return_value
=None)
1080 mock_software_version
= Mock(autospec
=True)
1082 mock_charm_upgrade
= asynctest
.Mock(autospec
=True)
1083 task
= asyncio
.Future()
1084 task
.set_result(("COMPLETED", "some_output"))
1085 mock_charm_upgrade
.return_value
= task
1087 fs
= Mock(autospec
=True)
1088 fs
.path
.__add
__ = Mock()
1089 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1090 fs
.sync
.side_effect
= [None, None]
1092 instance
= self
.my_ns
1094 expected_operation_state
= "COMPLETED"
1095 expected_operation_error
= ""
1096 expected_vnfr_revision
= 1
1097 expected_ns_state
= "INSTANTIATED"
1098 expected_ns_operational_state
= "running"
1100 with patch
.object(instance
, "fs", fs
), patch(
1101 "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
1102 ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade
), patch(
1103 "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
1106 await instance
.update(nsr_id
, nslcmop_id
)
1108 return_operation_state
= self
.db
.get_one(
1109 "nslcmops", {"_id": nslcmop_id
}
1110 ).get("operationState")
1112 return_operation_error
= self
.db
.get_one(
1113 "nslcmops", {"_id": nslcmop_id
}
1114 ).get("errorMessage")
1116 return_ns_operational_state
= self
.db
.get_one(
1117 "nsrs", {"_id": nsr_id
}
1118 ).get("operational-status")
1120 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1124 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1128 mock_charm_hash
.assert_not_called()
1129 mock_software_version
.assert_not_called()
1130 mock_juju_bundle
.assert_not_called()
1131 mock_charm_upgrade
.assert_not_called()
1132 fs
.sync
.assert_not_called()
1134 self
.assertEqual(return_ns_state
, expected_ns_state
)
1135 self
.assertEqual(return_operation_state
, expected_operation_state
)
1136 self
.assertEqual(return_operation_error
, expected_operation_error
)
1138 return_ns_operational_state
, expected_ns_operational_state
1140 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1146 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed, "
1147 "Charm package is not changed, sw-version is not changed.",
1151 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 3}
1156 q_filter
={"_id": vnfd_id
+ ":1"},
1157 update_dict
={"_admin.revision": 1},
1161 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1164 mock_charm_hash
= Mock(autospec
=True)
1165 mock_charm_hash
.return_value
= False
1167 mock_juju_bundle
= Mock(return_value
=None)
1169 mock_software_version
= Mock(autospec
=True)
1171 mock_charm_upgrade
= asynctest
.Mock(autospec
=True)
1172 task
= asyncio
.Future()
1173 task
.set_result(("COMPLETED", "some_output"))
1174 mock_charm_upgrade
.return_value
= task
1175 mock_software_version
.side_effect
= ["1.0", "1.0"]
1177 fs
= Mock(autospec
=True)
1178 fs
.path
.__add
__ = Mock()
1179 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1180 fs
.sync
.side_effect
= [None, None]
1182 instance
= self
.my_ns
1184 expected_operation_state
= "COMPLETED"
1185 expected_operation_error
= ""
1186 expected_vnfr_revision
= 3
1187 expected_ns_state
= "INSTANTIATED"
1188 expected_ns_operational_state
= "running"
1190 with patch
.object(instance
, "fs", fs
), patch(
1191 "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
1192 ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade
), patch(
1193 "osm_lcm.lcm_utils.check_juju_bundle_existence", mock_juju_bundle
1196 await instance
.update(nsr_id
, nslcmop_id
)
1198 return_operation_state
= self
.db
.get_one(
1199 "nslcmops", {"_id": nslcmop_id
}
1200 ).get("operationState")
1202 return_operation_error
= self
.db
.get_one(
1203 "nslcmops", {"_id": nslcmop_id
}
1204 ).get("errorMessage")
1206 return_ns_operational_state
= self
.db
.get_one(
1207 "nsrs", {"_id": nsr_id
}
1208 ).get("operational-status")
1210 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1214 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1218 mock_charm_hash
.assert_called_with(
1219 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
1220 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:3/hackfest_3charmed_vnfd/charms/simple",
1223 self
.assertEqual(fs
.sync
.call_count
, 2)
1224 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1226 mock_juju_bundle
.assert_not_called()
1227 mock_charm_upgrade
.assert_not_called()
1229 self
.assertEqual(return_ns_state
, expected_ns_state
)
1230 self
.assertEqual(return_operation_state
, expected_operation_state
)
1231 self
.assertEqual(return_operation_error
, expected_operation_error
)
1233 return_ns_operational_state
, expected_ns_operational_state
1235 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1241 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed, "
1242 "Charm package exists, sw-version changed.",
1247 q_filter
={"_id": vnfd_id
},
1248 update_dict
={"_admin.revision": 3, "software-version": "3.0"},
1253 q_filter
={"_id": vnfd_id
+ ":1"},
1254 update_dict
={"_admin.revision": 1},
1259 q_filter
={"_id": vnfr_id
},
1260 update_dict
={"revision": 1},
1263 mock_charm_hash
= Mock(autospec
=True)
1264 mock_charm_hash
.return_value
= False
1266 mock_juju_bundle
= Mock(return_value
=None)
1268 mock_charm_upgrade
= asynctest
.Mock(autospec
=True)
1269 task
= asyncio
.Future()
1270 task
.set_result(("COMPLETED", "some_output"))
1271 mock_charm_upgrade
.return_value
= task
1273 mock_charm_artifact
= Mock(autospec
=True)
1274 mock_charm_artifact
.side_effect
= [
1275 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
1276 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/hackfest_3charmed_vnfd/charms/simple",
1279 fs
= Mock(autospec
=True)
1280 fs
.path
.__add
__ = Mock()
1281 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1282 fs
.sync
.side_effect
= [None, None]
1284 instance
= self
.my_ns
1286 expected_operation_state
= "FAILED"
1287 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."
1288 expected_vnfr_revision
= 1
1289 expected_ns_state
= "INSTANTIATED"
1290 expected_ns_operational_state
= "running"
1292 with patch
.object(instance
, "fs", fs
), patch(
1293 "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
1294 ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade
), patch(
1295 "osm_lcm.lcm_utils.get_charm_artifact_path", mock_charm_artifact
1298 await instance
.update(nsr_id
, nslcmop_id
)
1300 return_operation_state
= self
.db
.get_one(
1301 "nslcmops", {"_id": nslcmop_id
}
1302 ).get("operationState")
1304 return_operation_error
= self
.db
.get_one(
1305 "nslcmops", {"_id": nslcmop_id
}
1306 ).get("errorMessage")
1308 return_ns_operational_state
= self
.db
.get_one(
1309 "nsrs", {"_id": nsr_id
}
1310 ).get("operational-status")
1312 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1316 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1320 self
.assertEqual(fs
.sync
.call_count
, 2)
1321 mock_charm_hash
.assert_not_called()
1323 mock_juju_bundle
.assert_not_called()
1324 mock_charm_upgrade
.assert_not_called()
1326 self
.assertEqual(return_ns_state
, expected_ns_state
)
1327 self
.assertEqual(return_operation_state
, expected_operation_state
)
1328 self
.assertEqual(return_operation_error
, expected_operation_error
)
1330 return_ns_operational_state
, expected_ns_operational_state
1332 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1338 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
1339 "Charm package exists, sw-version not changed, juju-bundle exists",
1344 q_filter
={"_id": vnfd_id
},
1346 "_admin.revision": 3,
1347 "software-version": "1.0",
1348 "kdu.0.juju-bundle": "stable/native-kdu",
1354 q_filter
={"_id": vnfd_id
+ ":1"},
1356 "_admin.revision": 1,
1357 "software-version": "1.0",
1358 "kdu.0.juju-bundle": "stable/native-kdu",
1363 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1366 mock_charm_hash
= Mock(autospec
=True)
1367 mock_charm_hash
.return_value
= True
1369 mock_charm_artifact
= Mock(autospec
=True)
1370 mock_charm_artifact
.side_effect
= [
1371 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
1372 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/hackfest_3charmed_vnfd/charms/simple",
1375 fs
= Mock(autospec
=True)
1376 fs
.path
.__add
__ = Mock()
1377 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1378 fs
.sync
.side_effect
= [None, None]
1380 instance
= self
.my_ns
1382 expected_operation_state
= "FAILED"
1383 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"
1384 expected_vnfr_revision
= 1
1385 expected_ns_state
= "INSTANTIATED"
1386 expected_ns_operational_state
= "running"
1388 with patch
.object(instance
, "fs", fs
), patch(
1389 "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
1390 ), patch("osm_lcm.lcm_utils.get_charm_artifact_path", mock_charm_artifact
):
1392 await instance
.update(nsr_id
, nslcmop_id
)
1394 return_operation_state
= self
.db
.get_one(
1395 "nslcmops", {"_id": nslcmop_id
}
1396 ).get("operationState")
1398 return_operation_error
= self
.db
.get_one(
1399 "nslcmops", {"_id": nslcmop_id
}
1400 ).get("errorMessage")
1402 return_ns_operational_state
= self
.db
.get_one(
1403 "nsrs", {"_id": nsr_id
}
1404 ).get("operational-status")
1406 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1410 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1414 self
.assertEqual(fs
.sync
.call_count
, 2)
1415 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1416 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1418 mock_charm_upgrade
.assert_not_called()
1420 self
.assertEqual(return_ns_state
, expected_ns_state
)
1421 self
.assertEqual(return_operation_state
, expected_operation_state
)
1422 self
.assertEqual(return_operation_error
, expected_operation_error
)
1424 return_ns_operational_state
, expected_ns_operational_state
1426 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1432 t
="Update type: CHANGE_VNFPKG, latest_vnfd revision changed,"
1433 "Charm package exists, sw-version not changed, charm-upgrade failed",
1438 q_filter
={"_id": vnfd_id
},
1440 "_admin.revision": 3,
1441 "software-version": "1.0",
1448 q_filter
={"_id": vnfd_id
+ ":1"},
1450 "_admin.revision": 1,
1451 "software-version": "1.0",
1457 "vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1}
1460 mock_charm_hash
= Mock(autospec
=True)
1461 mock_charm_hash
.return_value
= True
1463 mock_charm_upgrade
= asynctest
.Mock(autospec
=True)
1464 task
= asyncio
.Future()
1465 task
.set_result(("FAILED", "some_error"))
1466 mock_charm_upgrade
.return_value
= task
1468 mock_charm_artifact
= Mock(autospec
=True)
1469 mock_charm_artifact
.side_effect
= [
1470 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:1/hackfest_3charmed_vnfd/charms/simple",
1471 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/hackfest_3charmed_vnfd/charms/simple",
1474 fs
= Mock(autospec
=True)
1475 fs
.path
.__add
__ = Mock()
1476 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1477 fs
.sync
.side_effect
= [None, None]
1479 instance
= self
.my_ns
1481 expected_operation_state
= "FAILED"
1482 expected_operation_error
= "some_error"
1483 expected_vnfr_revision
= 1
1484 expected_ns_state
= "INSTANTIATED"
1485 expected_ns_operational_state
= "running"
1487 with patch
.object(instance
, "fs", fs
), patch(
1488 "osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed", mock_charm_hash
1489 ), patch("osm_lcm.ns.NsLcm._ns_charm_upgrade", mock_charm_upgrade
):
1491 await instance
.update(nsr_id
, nslcmop_id
)
1493 return_operation_state
= self
.db
.get_one(
1494 "nslcmops", {"_id": nslcmop_id
}
1495 ).get("operationState")
1497 return_operation_error
= self
.db
.get_one(
1498 "nslcmops", {"_id": nslcmop_id
}
1499 ).get("errorMessage")
1501 return_ns_operational_state
= self
.db
.get_one(
1502 "nsrs", {"_id": nsr_id
}
1503 ).get("operational-status")
1505 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1509 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1513 self
.assertEqual(fs
.sync
.call_count
, 2)
1514 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1515 self
.assertEqual(mock_charm_upgrade
.call_count
, 1)
1517 self
.assertEqual(return_ns_state
, expected_ns_state
)
1518 self
.assertEqual(return_operation_state
, expected_operation_state
)
1519 self
.assertEqual(return_operation_error
, expected_operation_error
)
1521 return_ns_operational_state
, expected_ns_operational_state
1523 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1527 def test_ns_update_helper_methods(self
):
1530 mock_path
.mock_reset()
1531 mock_checksumdir
.mock_reset()
1534 i
=1, t
="Find software version, VNFD does not have have software version"
1536 # Testing method find_software_version
1538 db_vnfd
= self
.db
.get_one(
1539 "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
1541 expected_result
= "1.0"
1542 result
= find_software_version(db_vnfd
)
1544 result
, expected_result
, "Default sw version should be 1.0"
1548 i
=2, t
="Find software version, VNFD includes software version"
1550 # Testing method find_software_version
1552 db_vnfd
= self
.db
.get_one(
1553 "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
1555 db_vnfd
["software-version"] = "3.1"
1556 expected_result
= "3.1"
1557 result
= find_software_version(db_vnfd
)
1558 self
.assertEqual(result
, expected_result
, "VNFD software version is wrong")
1560 with self
.subTest(i
=3, t
="Check charm hash, Hash has did not change"):
1561 # Testing method check_charm_hash_changed
1563 current_path
, target_path
= "/tmp/charm1", "/tmp/charm1"
1564 fs
= Mock(autospec
=True)
1565 fs
.path
.__add
__ = Mock()
1566 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1568 mock_path
= Mock(autospec
=True)
1569 mock_path
.exists
.side_effect
= [True, True]
1571 mock_checksumdir
= Mock(autospec
=True)
1572 mock_checksumdir
.dirhash
.side_effect
= ["hash_value", "hash_value"]
1574 instance
= self
.my_ns
1575 expected_result
= False
1577 with patch
.object(instance
, "fs", fs
), patch(
1578 "checksumdir.dirhash", mock_checksumdir
.dirhash
1579 ), patch("os.path.exists", mock_path
.exists
):
1581 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1583 result
, expected_result
, "Wrong charm hash control value"
1585 self
.assertEqual(mock_path
.exists
.call_count
, 2)
1586 self
.assertEqual(mock_checksumdir
.dirhash
.call_count
, 2)
1590 with self
.subTest(i
=4, t
="Check charm hash, Hash has changed"):
1591 # Testing method check_charm_hash_changed
1593 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1594 fs
= Mock(autospec
=True)
1595 fs
.path
.__add
__ = Mock()
1596 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1598 mock_path
= Mock(autospec
=True)
1599 mock_path
.exists
.side_effect
= [True, True]
1601 mock_checksumdir
= Mock(autospec
=True)
1602 mock_checksumdir
.dirhash
.side_effect
= ["hash_value", "another_hash_value"]
1604 instance
= self
.my_ns
1605 expected_result
= True
1607 with patch
.object(instance
, "fs", fs
), patch(
1608 "checksumdir.dirhash", mock_checksumdir
.dirhash
1609 ), patch("os.path.exists", mock_path
.exists
):
1611 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1613 result
, expected_result
, "Wrong charm hash control value"
1615 self
.assertEqual(mock_path
.exists
.call_count
, 2)
1616 self
.assertEqual(mock_checksumdir
.dirhash
.call_count
, 2)
1620 with self
.subTest(i
=5, t
="Check charm hash, Charm path does not exists"):
1621 # Testing method check_charm_hash_changed
1623 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1624 fs
= Mock(autospec
=True)
1625 fs
.path
.__add
__ = Mock()
1626 fs
.path
.side_effect
= ["/", "/", "/", "/"]
1628 mock_path
= Mock(autospec
=True)
1629 mock_path
.exists
.side_effect
= [True, False]
1631 mock_checksumdir
= Mock(autospec
=True)
1632 mock_checksumdir
.dirhash
.side_effect
= ["hash_value", "hash_value"]
1634 instance
= self
.my_ns
1636 with patch
.object(instance
, "fs", fs
), patch(
1637 "checksumdir.dirhash", mock_checksumdir
.dirhash
1638 ), patch("os.path.exists", mock_path
.exists
):
1640 with self
.assertRaises(LcmException
):
1642 instance
.check_charm_hash_changed(current_path
, target_path
)
1643 self
.assertEqual(mock_path
.exists
.call_count
, 2)
1644 self
.assertEqual(mock_checksumdir
.dirhash
.call_count
, 0)
1648 with self
.subTest(i
=6, t
="Check juju bundle existence"):
1649 # Testing method check_juju_bundle_existence
1651 test_vnfd1
= self
.db
.get_one(
1652 "vnfds", {"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}
1654 test_vnfd2
= self
.db
.get_one(
1655 "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
1658 expected_result
= None
1659 result
= check_juju_bundle_existence(test_vnfd1
)
1660 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1662 expected_result
= "stable/native-kdu"
1663 result
= check_juju_bundle_existence(test_vnfd2
)
1664 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1666 with self
.subTest(i
=7, t
="Check charm artifacts"):
1667 # Testing method check_juju_bundle_existence
1670 "folder": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
1671 "pkg-dir": "hackfest_3charmed_vnfd",
1673 charm_name
= "simple"
1674 charm_type
= "lxc_proxy_charm"
1677 expected_result
= "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77:3/hackfest_3charmed_vnfd/charms/simple"
1678 result
= get_charm_artifact_path(
1679 base_folder
, charm_name
, charm_type
, revision
1681 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1685 "folder": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
1687 charm_name
= "basic"
1692 "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77/Scripts/helm-charts/basic"
1694 result
= get_charm_artifact_path(
1695 base_folder
, charm_name
, charm_type
, revision
1697 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1700 if __name__
== "__main__":