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
27 from osm_lcm
.lcm_utils
import TaskRegistry
28 from osm_lcm
.ng_ro
import NgRoClient
29 from osm_lcm
.data_utils
.database
.database
import Database
30 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
31 from osm_lcm
.data_utils
.vnfd
import find_software_version
32 from osm_lcm
.lcm_utils
import check_juju_bundle_existence
, get_charm_artifact_path
33 from osm_lcm
.lcm_utils
import LcmException
34 from uuid
import uuid4
35 from unittest
.mock
import Mock
, patch
37 from osm_lcm
.tests
import test_db_descriptors
as descriptors
39 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
41 """ Perform unittests using asynctest of osm_lcm.ns module
42 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
43 OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
44 OSMLCMTEST_NS_NAME: change name of NS
45 OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
46 OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
47 OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
48 OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
49 OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
50 OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
51 OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
52 OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
53 OSMLCM_VCA_XXX: configuration of N2VC
54 OSMLCM_RO_XXX: configuration of RO
58 "global": {"loglevel": "DEBUG"},
60 "VCA": { # TODO replace with os.get_env to get other configurations
61 "host": getenv("OSMLCM_VCA_HOST", "vca"),
62 "port": getenv("OSMLCM_VCA_PORT", 17070),
63 "user": getenv("OSMLCM_VCA_USER", "admin"),
64 "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
65 "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
66 "ca_cert": getenv("OSMLCM_VCA_CACERT", None),
67 "apiproxy": getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1"),
70 "uri": "http://{}:{}/openmano".format(
71 getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
73 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
74 "logger_name": "lcm.ROclient",
79 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
80 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["update"]
81 vnfr_id
= "6421c7c9-d865-4fb4-9a13-d4275d243e01"
82 vnfd_id
= "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
83 update_fs
= Mock(autospec
=True)
84 update_fs
.path
.__add
__ = Mock()
85 update_fs
.path
.side_effect
= ["/", "/", "/", "/"]
86 update_fs
.sync
.side_effect
= [None, None]
93 class TestMyNS(asynctest
.TestCase
):
94 async def _n2vc_DeployCharms(
106 for status
, message
in (
107 ("maintenance", "installing sofwware"),
108 ("active", "Ready!"),
110 # call callback after some time
111 asyncio
.sleep(5, loop
=self
.loop
)
112 callback(model_name
, application_name
, status
, message
, *callback_args
)
115 def _n2vc_FormatApplicationName(*args
):
118 yield "app_name-{}".format(num_calls
)
121 def _n2vc_CreateExecutionEnvironment(
122 self
, namespace
, reuse_ee_id
, db_dict
, *args
, **kwargs
124 k_list
= namespace
.split(".")
125 ee_id
= k_list
[1] + "."
127 for k
in k_list
[2:4]:
133 def _ro_status(self
, *args
, **kwargs
):
134 print("Args > {}".format(args
))
135 print("kwargs > {}".format(kwargs
))
138 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_update_action_text
)
141 if kwargs
.get("delete"):
142 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_delete_action_text
)
146 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_ns_text
)
148 # if ip address provided, replace descriptor
149 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
151 ip_addresses_list
= ip_addresses
.split(",")
152 for vnf
in ro_ns_desc
["vnfs"]:
153 if not ip_addresses_list
:
155 vnf
["ip_address"] = ip_addresses_list
[0]
156 for vm
in vnf
["vms"]:
157 if not ip_addresses_list
:
159 vm
["ip_address"] = ip_addresses_list
.pop(0)
163 for net
in ro_ns_desc
["nets"]:
164 if net
["status"] != "ACTIVE":
165 net
["status"] = "ACTIVE"
168 for vnf
in ro_ns_desc
["vnfs"]:
169 for vm
in vnf
["vms"]:
170 if vm
["status"] != "ACTIVE":
171 vm
["status"] = "ACTIVE"
174 def _ro_deploy(self
, *args
, **kwargs
):
175 return {"action_id": args
[1]["action_id"], "nsr_id": args
[0], "status": "ok"}
177 def _return_uuid(self
, *args
, **kwargs
):
180 async def setUp(self
):
183 if not getenv("OSMLCMTEST_DB_NOMOCK"):
184 # Cleanup singleton Database instance
185 Database
.instance
= None
187 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
188 self
.db
.create_list("vnfds", yaml
.safe_load(descriptors
.db_vnfds_text
))
191 yaml
.safe_load(descriptors
.db_vnfds_revisions_text
),
193 self
.db
.create_list("nsds", yaml
.safe_load(descriptors
.db_nsds_text
))
194 self
.db
.create_list("nsrs", yaml
.safe_load(descriptors
.db_nsrs_text
))
197 yaml
.safe_load(descriptors
.db_vim_accounts_text
),
201 yaml
.safe_load(descriptors
.db_k8sclusters_text
),
204 "nslcmops", yaml
.safe_load(descriptors
.db_nslcmops_text
)
206 self
.db
.create_list("vnfrs", yaml
.safe_load(descriptors
.db_vnfrs_text
))
207 self
.db_vim_accounts
= yaml
.safe_load(descriptors
.db_vim_accounts_text
)
210 self
.msg
= asynctest
.Mock(MsgKafka())
213 if not getenv("OSMLCMTEST_FS_NOMOCK"):
214 self
.fs
= asynctest
.Mock(
215 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
217 self
.fs
.get_params
.return_value
= {
218 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
220 self
.fs
.file_open
= asynctest
.mock_open()
221 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
222 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
225 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
226 self
.lcm_tasks
.lock_HA
.return_value
= True
227 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
228 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
231 if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
232 ns
.K8sJujuConnector
= asynctest
.MagicMock(ns
.K8sJujuConnector
)
233 ns
.K8sHelmConnector
= asynctest
.MagicMock(ns
.K8sHelmConnector
)
234 ns
.K8sHelm3Connector
= asynctest
.MagicMock(ns
.K8sHelm3Connector
)
236 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
237 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
238 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
241 self
.my_ns
= ns
.NsLcm(self
.msg
, self
.lcm_tasks
, lcm_config
, self
.loop
)
242 self
.my_ns
.fs
= self
.fs
243 self
.my_ns
.db
= self
.db
244 self
.my_ns
._wait
_dependent
_n
2vc
= asynctest
.CoroutineMock()
247 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
248 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
251 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
252 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
253 # self.my_ns.n2vc = asynctest.Mock(N2VC())
254 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv(
255 "OSMLCM_VCA_PUBKEY", "public_key"
257 # allow several versions of n2vc
258 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(
259 side_effect
=self
._n
2vc
_FormatApplicationName
()
261 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(
262 side_effect
=self
._n
2vc
_DeployCharms
264 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
265 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
267 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(
270 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(
273 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
274 side_effect
=self
._return
_uuid
276 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
277 side_effect
=self
._return
_uuid
279 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(
280 return_value
="completed"
282 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(
283 return_value
={"result": "ok", "pubkey": pub_key
}
285 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(
288 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
289 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key")
291 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(
296 if not getenv("OSMLCMTEST_RO_NOMOCK"):
297 self
.my_ns
.RO
= asynctest
.Mock(
298 NgRoClient(self
.loop
, **lcm_config
["ro_config"])
300 # TODO first time should be empty list, following should return a dict
301 # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
302 self
.my_ns
.RO
.deploy
= asynctest
.CoroutineMock(
303 self
.my_ns
.RO
.deploy
, side_effect
=self
._ro
_deploy
305 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status)
306 # self.my_ns.RO.create_action = asynctest.CoroutineMock(self.my_ns.RO.create_action,
307 # return_value={"vm-id": {"vim_result": 200,
308 # "description": "done"}})
309 self
.my_ns
.RO
.delete
= asynctest
.CoroutineMock(self
.my_ns
.RO
.delete
)
311 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
312 # async def test_instantiate(self):
313 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
314 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
315 # # print("Test instantiate started")
317 # # delete deployed information of database
318 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
319 # if self.db.get_list("nsrs")[0]["_admin"].get("deployed"):
320 # del self.db.get_list("nsrs")[0]["_admin"]["deployed"]
321 # for db_vnfr in self.db.get_list("vnfrs"):
322 # db_vnfr.pop("ip_address", None)
323 # for db_vdur in db_vnfr["vdur"]:
324 # db_vdur.pop("ip_address", None)
325 # db_vdur.pop("mac_address", None)
326 # if getenv("OSMLCMTEST_RO_VIMID"):
327 # self.db.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
328 # if getenv("OSMLCMTEST_RO_VIMID"):
329 # self.db.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
331 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
333 # self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
334 # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
335 # "operationState": "COMPLETED"},
337 # self.lcm_tasks.lock_HA.assert_called_once_with('ns', 'nslcmops', nslcmop_id)
338 # if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
339 # self.assertTrue(self.my_ns.logger.debug.called, "Debug method not called")
340 # self.my_ns.logger.error.assert_not_called()
341 # self.my_ns.logger.exception().assert_not_called()
343 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
344 # self.assertTrue(self.db.set_one.called, "db.set_one not called")
345 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
346 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
347 # self.assertEqual(db_nsr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
348 # for vnfr in db_vnfrs_list:
349 # self.assertEqual(vnfr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
351 # if not getenv("OSMLCMTEST_VCA_NOMOCK"):
352 # # check intial-primitives called
353 # self.assertTrue(self.my_ns.n2vc.exec_primitive.called,
354 # "Exec primitive not called for initial config primitive")
355 # for _call in self.my_ns.n2vc.exec_primitive.call_args_list:
356 # self.assertIn(_call[1]["primitive_name"], ("config", "touch"),
357 # "called exec primitive with a primitive different than config or touch")
359 # # TODO add more checks of called methods
360 # # TODO add a terminate
362 # async def test_instantiate_ee_list(self):
363 # # Using modern IM where configuration is in the new format of execution_environment_list
364 # ee_descriptor_id = "charm_simple"
365 # non_used_initial_primitive = {
366 # "name": "not_to_be_called",
368 # "execution-environment-ref": "not_used_ee"
372 # "id": ee_descriptor_id,
373 # "juju": {"charm": "simple"},
380 # q_filter={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"},
381 # update_dict={"vnf-configuration.0.execution-environment-list": ee_list,
382 # "vnf-configuration.0.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id,
383 # "vnf-configuration.0.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id,
384 # "vnf-configuration.0.initial-config-primitive.2": non_used_initial_primitive,
385 # "vnf-configuration.0.config-primitive.0.execution-environment-ref": ee_descriptor_id,
386 # "vnf-configuration.0.config-primitive.0.execution-environment-primitive": "touch_charm",
388 # unset={"vnf-configuration.juju": None})
389 # await self.test_instantiate()
390 # # this will check that the initial-congig-primitive 'not_to_be_called' is not called
392 @asynctest.fail_on(active_handles
=True)
393 async def test_start_stop_rebuild_pass(self
):
394 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
395 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops"]
396 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
397 additional_param
= {"count-index": "0"}
398 operation_type
= "start"
399 await self
.my_ns
.rebuild_start_stop(
400 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
402 expected_value
= "COMPLETED"
403 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
406 self
.assertEqual(return_value
, expected_value
)
408 @asynctest.fail_on(active_handles
=True)
409 async def test_start_stop_rebuild_fail(self
):
410 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
411 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops1"]
412 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
413 additional_param
= {"count-index": "0"}
414 operation_type
= "stop"
415 await self
.my_ns
.rebuild_start_stop(
416 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
418 expected_value
= "Error"
419 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
422 self
.assertEqual(return_value
, expected_value
)
424 # Test scale() and related methods
425 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
426 async def test_scale(self
):
427 # print("Test scale started")
429 # TODO: Add more higher-lever tests here, for example:
430 # scale-out/scale-in operations with success/error result
432 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
433 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
434 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
435 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
436 expected_value
= "FAILED"
437 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
440 self
.assertEqual(return_value
, expected_value
)
441 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
443 # Test scale() for native kdu
444 # this also includes testing _scale_kdu()
445 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["ns"]
446 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["instantiate"]
448 self
.my_ns
.k8sclusterjuju
.scale
= asynctest
.mock
.CoroutineMock()
449 self
.my_ns
.k8sclusterjuju
.exec_primitive
= asynctest
.mock
.CoroutineMock()
450 self
.my_ns
.k8sclusterjuju
.get_scale_count
= asynctest
.mock
.CoroutineMock(
453 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
454 expected_value
= "COMPLETED"
455 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
458 self
.assertEqual(return_value
, expected_value
)
459 self
.my_ns
.k8sclusterjuju
.scale
.assert_called_once()
461 # Test scale() for native kdu with 2 resource
462 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["ns"]
463 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["instantiate"]
465 self
.my_ns
.k8sclusterjuju
.get_scale_count
.return_value
= 2
466 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
467 expected_value
= "COMPLETED"
468 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
471 self
.assertEqual(return_value
, expected_value
)
472 self
.my_ns
.k8sclusterjuju
.scale
.assert_called()
474 async def test_vca_status_refresh(self
):
475 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
476 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
477 await self
.my_ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
478 expected_value
= dict()
479 return_value
= dict()
480 vnf_descriptors
= self
.db
.get_list("vnfds")
481 for i
, _
in enumerate(vnf_descriptors
):
482 for j
, value
in enumerate(vnf_descriptors
[i
]["df"]):
483 if "lcm-operations-configuration" in vnf_descriptors
[i
]["df"][j
]:
486 in value
["lcm-operations-configuration"][
487 "operate-vnf-op-config"
490 for k
, v
in enumerate(
491 value
["lcm-operations-configuration"][
492 "operate-vnf-op-config"
496 v
.get("execution-environment-list")
497 and "juju" in v
["execution-environment-list"][k
]
499 expected_value
= self
.db
.get_list("nsrs")[i
][
502 await self
.my_ns
._on
_update
_n
2vc
_db
(
503 "nsrs", {"_id": nsr_id
}, "_admin.deployed.VCA.0", {}
505 return_value
= self
.db
.get_list("nsrs")[i
]["vcaStatus"]
506 self
.assertEqual(return_value
, expected_value
)
508 # Test _retry_or_skip_suboperation()
510 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
511 # - if marked as anything but 'COMPLETED', the suboperation index is expected
512 def test_scale_retry_or_skip_suboperation(self
):
513 # Load an alternative 'nslcmops' YAML for this test
514 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
515 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
517 # Test when 'operationState' is 'COMPLETED'
518 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = "COMPLETED"
519 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
520 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
521 self
.assertEqual(return_value
, expected_value
)
522 # Test when 'operationState' is not 'COMPLETED'
523 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = None
524 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
525 expected_value
= op_index
526 self
.assertEqual(return_value
, expected_value
)
528 # Test _find_suboperation()
529 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
530 def test_scale_find_suboperation(self
):
531 # Load an alternative 'nslcmops' YAML for this test
532 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
533 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
534 # Find this sub-operation
536 vnf_index
= db_nslcmop
["_admin"]["operations"][op_index
]["member_vnf_index"]
537 primitive
= db_nslcmop
["_admin"]["operations"][op_index
]["primitive"]
538 primitive_params
= db_nslcmop
["_admin"]["operations"][op_index
][
542 "member_vnf_index": vnf_index
,
543 "primitive": primitive
,
544 "primitive_params": primitive_params
,
546 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
547 self
.assertEqual(found_op_index
, op_index
)
548 # Test with not-matching params
550 "member_vnf_index": vnf_index
,
552 "primitive_params": primitive_params
,
554 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
555 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
558 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
559 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
561 # Test _update_suboperation_status()
562 def test_scale_update_suboperation_status(self
):
563 self
.db
.set_one
= asynctest
.Mock()
564 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
565 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
567 # Force the initial values to be distinct from the updated ones
568 q_filter
= {"_id": db_nslcmop
["_id"]}
569 # Test to change 'operationState' and 'detailed-status'
570 operationState
= "COMPLETED"
571 detailed_status
= "Done"
572 expected_update_dict
= {
573 "_admin.operations.0.operationState": operationState
,
574 "_admin.operations.0.detailed-status": detailed_status
,
576 self
.my_ns
._update
_suboperation
_status
(
577 db_nslcmop
, op_index
, operationState
, detailed_status
579 self
.db
.set_one
.assert_called_once_with(
582 update_dict
=expected_update_dict
,
586 def test_scale_add_suboperation(self
):
587 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
588 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
590 num_ops_before
= len(db_nslcmop
.get("_admin", {}).get("operations", [])) - 1
592 vdu_count_index
= None
595 mapped_primitive_params
= {
598 "data-type": "STRING",
600 "default-value": "<touch_filename2>",
605 operationState
= "PROCESSING"
606 detailed_status
= "In progress"
607 operationType
= "PRE-SCALE"
608 # Add a 'pre-scale' suboperation
609 op_index_after
= self
.my_ns
._add
_suboperation
(
616 mapped_primitive_params
,
621 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
623 # Delete all suboperations and add the same operation again
624 del db_nslcmop
["_admin"]["operations"]
625 op_index_zero
= self
.my_ns
._add
_suboperation
(
632 mapped_primitive_params
,
637 self
.assertEqual(op_index_zero
, 0)
639 # Add a 'RO' suboperation
640 RO_nsr_id
= "1234567890"
645 "member-vnf-index": "1",
646 "osm_vdu_id": "dataVM",
649 op_index
= self
.my_ns
._add
_suboperation
(
656 mapped_primitive_params
,
663 db_RO_nsr_id
= db_nslcmop
["_admin"]["operations"][op_index
]["RO_nsr_id"]
664 self
.assertEqual(op_index
, 1)
665 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
667 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
668 op_index_invalid
= self
.my_ns
._add
_suboperation
(
669 None, None, None, None, None, None, None, None, None, None, None
671 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
673 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
674 # check the possible return values:
675 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
676 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
677 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
678 def test_scale_check_or_add_scale_suboperation(self
):
679 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
680 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
681 operationType
= "PRE-SCALE"
687 "data-type": "STRING",
689 "default-value": "<touch_filename2>",
695 # Delete all sub-operations to be sure this is a new sub-operation
696 del db_nslcmop
["_admin"]["operations"]
698 # Add a new sub-operation
699 # For new sub-operations, operationState is set to 'PROCESSING' by default
700 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
701 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
703 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
705 # Use the same parameters again to match the already added sub-operation
706 # which has status 'PROCESSING' (!= 'COMPLETED') by default
707 # The expected return value is a non-negative number
708 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
709 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
711 self
.assertTrue(op_index_existing
>= 0)
713 # Change operationState 'manually' for this sub-operation
714 db_nslcmop
["_admin"]["operations"][op_index_existing
][
717 # Then use the same parameters again to match the already added sub-operation,
718 # which now has status 'COMPLETED'
719 # The expected return value is SUBOPERATION_STATUS_SKIP
720 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
721 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
723 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
725 # RO sub-operation test:
726 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
727 RO_nsr_id
= "1234567890"
732 "member-vnf-index": "1",
733 "osm_vdu_id": "dataVM",
736 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
737 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
739 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
741 # Use the same parameters again to match the already added RO sub-operation
742 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
743 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
745 self
.assertTrue(op_index_existing_RO
>= 0)
747 # Change operationState 'manually' for this RO sub-operation
748 db_nslcmop
["_admin"]["operations"][op_index_existing_RO
][
751 # Then use the same parameters again to match the already added sub-operation,
752 # which now has status 'COMPLETED'
753 # The expected return value is SUBOPERATION_STATUS_SKIP
754 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
755 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
757 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
759 async def test_deploy_kdus(self
):
760 nsr_id
= descriptors
.test_ids
["TEST-KDU"]["ns"]
761 nslcmop_id
= descriptors
.test_ids
["TEST-KDU"]["instantiate"]
762 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
763 db_vnfr
= self
.db
.get_one(
764 "vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "multikdu"}
766 db_vnfrs
= {"multikdu": db_vnfr
}
767 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
771 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
= asynctest
.mock
.Mock()
772 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
.return_value
= "k8s_id"
773 self
.my_ns
.k8sclusterhelm3
.install
= asynctest
.CoroutineMock()
774 self
.my_ns
.k8sclusterhelm3
.synchronize_repos
= asynctest
.CoroutineMock(
775 return_value
=("", "")
777 self
.my_ns
.k8sclusterhelm3
.get_services
= asynctest
.CoroutineMock(
780 await self
.my_ns
.deploy_kdus(
781 logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
783 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
784 db_nsr
= self
.db
.get_list("nsrs")[1]
787 db_nsr
["_admin"]["deployed"],
788 "K8s entry not created at '_admin.deployed'",
790 self
.assertIsInstance(
791 db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
794 len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
797 "kdu-instance": "k8s_id",
798 "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
799 "k8scluster-type": "helm-chart-v3",
801 "member-vnf-index": "multikdu",
803 "kdu-deployment-name": None,
806 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][0])
807 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
808 expected_kdu_model
= "stable/openldap:1.2.1"
809 self
.assertEqual(nsr_result
, k8s_instace_info
)
811 nsr_kdu_model_result
in expected_kdu_model
812 or expected_kdu_model
in nsr_kdu_model_result
814 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][1])
815 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
816 k8s_instace_info
["kdu-name"] = "mongo"
817 expected_kdu_model
= "stable/mongodb"
818 self
.assertEqual(nsr_result
, k8s_instace_info
)
820 nsr_kdu_model_result
in expected_kdu_model
821 or expected_kdu_model
in nsr_kdu_model_result
824 # Test remove_vnf() and related methods
825 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
826 async def test_remove_vnf(self
):
828 nsr_id
= descriptors
.test_ids
["TEST-UPDATE"]["ns"]
829 nslcmop_id
= descriptors
.test_ids
["TEST-UPDATE"]["removeVnf"]
830 vnf_instance_id
= descriptors
.test_ids
["TEST-UPDATE"]["vnf"]
831 mock_wait_ng_ro
= asynctest
.CoroutineMock()
832 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
833 await self
.my_ns
.update(nsr_id
, nslcmop_id
)
834 expected_value
= "COMPLETED"
835 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
838 self
.assertEqual(return_value
, expected_value
)
839 with self
.assertRaises(Exception) as context
:
840 self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
842 "database exception Not found entry with filter"
843 in str(context
.exception
)
846 # test vertical scale executes sucessfully
847 # @patch("osm_lcm.ng_ro.status.response")
848 @asynctest.fail_on(active_handles
=True)
849 async def test_vertical_scaling(self
):
850 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
851 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate"]
853 # calling the vertical scale fucntion
854 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status("update"))
855 mock_wait_ng_ro
= asynctest
.CoroutineMock()
856 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
857 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
858 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
861 expected_value
= "COMPLETED"
862 self
.assertEqual(return_value
, expected_value
)
864 # test vertical scale executes fail
865 @asynctest.fail_on(active_handles
=True)
866 async def test_vertical_scaling_fail(self
):
867 # get th nsr nad nslcmops id from descriptors
868 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
869 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate-1"]
871 # calling the vertical scale fucntion
872 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
873 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
876 expected_value
= "FAILED"
877 self
.assertEqual(return_value
, expected_value
)
879 # async def test_instantiate_pdu(self):
880 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
881 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
882 # # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
883 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
884 # update_dict={"ip-address": "10.205.1.46",
885 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
886 # "vdur.0.pdu-type": "PDU-TYPE-1",
887 # "vdur.0.ip-address": "10.205.1.46",
889 # unset={"vdur.status": None})
890 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
891 # update_dict={"ip-address": "10.205.1.47",
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.47",
896 # unset={"vdur.status": None})
898 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
899 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
900 # self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
901 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
902 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
903 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
904 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
906 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
907 # async def test_terminate_without_configuration(self):
908 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
909 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
910 # # set instantiation task as completed
911 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
912 # update_dict={"operationState": "COMPLETED"})
913 # self.db.set_one("nsrs", {"_id": nsr_id},
914 # update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
916 # await self.my_ns.terminate(nsr_id, nslcmop_id)
917 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
918 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
919 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
920 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
921 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
922 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
923 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
924 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
925 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
926 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
927 # for vnfr in db_vnfrs_list:
928 # self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
930 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
931 # async def test_terminate_primitive(self):
932 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
933 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
934 # # set instantiation task as completed
935 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
936 # update_dict={"operationState": "COMPLETED"})
938 # # modify vnfd descriptor to include terminate_primitive
939 # terminate_primitive = [{
941 # "parameter": [{"name": "filename", "value": "terminate_filename"}],
944 # db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
945 # self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
946 # {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
948 # await self.my_ns.terminate(nsr_id, nslcmop_id)
949 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
950 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
951 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
952 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
953 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
954 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
955 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
956 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
957 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
959 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
961 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
963 @patch("osm_lcm.data_utils.vnfd.find_software_version")
964 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
965 async def test_update_change_vnfpkg_sw_version_not_changed(
968 mock_software_version
,
972 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed,
973 Charm package changed, sw-version is not changed"""
976 q_filter
={"_id": vnfd_id
},
977 update_dict
={"_admin.revision": 3, "kdu": []},
982 q_filter
={"_id": vnfd_id
+ ":1"},
983 update_dict
={"_admin.revision": 1, "kdu": []},
986 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
988 mock_charm_hash
.return_value
= True
989 mock_software_version
.side_effect
= ["1.0", "1.0"]
991 task
= asyncio
.Future()
992 task
.set_result(("COMPLETED", "some_output"))
993 mock_charm_upgrade
.return_value
= task
995 instance
= self
.my_ns
996 fs
= deepcopy(update_fs
)
999 expected_operation_state
= "COMPLETED"
1000 expected_operation_error
= ""
1001 expected_vnfr_revision
= 3
1002 expected_ns_state
= "INSTANTIATED"
1003 expected_ns_operational_state
= "running"
1005 await instance
.update(nsr_id
, nslcmop_id
)
1006 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1009 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1012 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1013 "operational-status"
1015 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1018 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1019 mock_charm_hash
.assert_called_with(
1020 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1021 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1023 self
.assertEqual(fs
.sync
.call_count
, 2)
1024 self
.assertEqual(return_ns_state
, expected_ns_state
)
1025 self
.assertEqual(return_operation_state
, expected_operation_state
)
1026 self
.assertEqual(return_operation_error
, expected_operation_error
)
1027 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1028 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1030 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1032 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1034 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1035 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1036 async def test_update_change_vnfpkg_vnfd_revision_not_changed(
1039 mock_software_version
,
1043 """Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"""
1045 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 1}
1047 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1049 mock_charm_hash
.return_value
= True
1051 task
= asyncio
.Future()
1052 task
.set_result(("COMPLETED", "some_output"))
1053 mock_charm_upgrade
.return_value
= task
1055 instance
= self
.my_ns
1057 expected_operation_state
= "COMPLETED"
1058 expected_operation_error
= ""
1059 expected_vnfr_revision
= 1
1060 expected_ns_state
= "INSTANTIATED"
1061 expected_ns_operational_state
= "running"
1063 await instance
.update(nsr_id
, nslcmop_id
)
1064 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1067 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1070 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1071 "operational-status"
1073 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1074 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1077 mock_charm_hash
.assert_not_called()
1078 mock_software_version
.assert_not_called()
1079 mock_juju_bundle
.assert_not_called()
1080 mock_charm_upgrade
.assert_not_called()
1081 update_fs
.sync
.assert_not_called()
1082 self
.assertEqual(return_ns_state
, expected_ns_state
)
1083 self
.assertEqual(return_operation_state
, expected_operation_state
)
1084 self
.assertEqual(return_operation_error
, expected_operation_error
)
1085 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1086 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1088 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1090 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1092 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1093 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1094 async def test_update_change_vnfpkg_charm_is_not_changed(
1097 mock_software_version
,
1101 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1102 Charm package is not changed, sw-version is not changed"""
1105 q_filter
={"_id": vnfd_id
},
1106 update_dict
={"_admin.revision": 3, "kdu": []},
1110 q_filter
={"_id": vnfd_id
+ ":1"},
1111 update_dict
={"_admin.revision": 1, "kdu": []},
1113 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1115 mock_charm_hash
.return_value
= False
1116 mock_software_version
.side_effect
= ["1.0", "1.0"]
1118 instance
= self
.my_ns
1119 fs
= deepcopy(update_fs
)
1121 expected_operation_state
= "COMPLETED"
1122 expected_operation_error
= ""
1123 expected_vnfr_revision
= 3
1124 expected_ns_state
= "INSTANTIATED"
1125 expected_ns_operational_state
= "running"
1127 await instance
.update(nsr_id
, nslcmop_id
)
1128 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1131 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1134 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1135 "operational-status"
1137 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1140 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1141 mock_charm_hash
.assert_called_with(
1142 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1143 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1145 self
.assertEqual(fs
.sync
.call_count
, 2)
1146 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1147 mock_juju_bundle
.assert_not_called()
1148 mock_charm_upgrade
.assert_not_called()
1149 self
.assertEqual(return_ns_state
, expected_ns_state
)
1150 self
.assertEqual(return_operation_state
, expected_operation_state
)
1151 self
.assertEqual(return_operation_error
, expected_operation_error
)
1152 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1153 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1155 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1156 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1158 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1160 @patch("osm_lcm.lcm_utils.get_charm_artifact_path")
1161 async def test_update_change_vnfpkg_sw_version_changed(
1162 self
, mock_charm_artifact
, mock_charm_upgrade
, mock_charm_hash
, mock_juju_bundle
1164 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1165 Charm package exists, sw-version changed."""
1168 q_filter
={"_id": vnfd_id
},
1169 update_dict
={"_admin.revision": 3, "software-version": "3.0", "kdu": []},
1173 q_filter
={"_id": vnfd_id
+ ":1"},
1174 update_dict
={"_admin.revision": 1, "kdu": []},
1178 q_filter
={"_id": vnfr_id
},
1179 update_dict
={"revision": 1},
1181 mock_charm_hash
.return_value
= False
1183 mock_charm_artifact
.side_effect
= [
1184 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1185 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1188 instance
= self
.my_ns
1189 fs
= deepcopy(update_fs
)
1191 expected_operation_state
= "FAILED"
1192 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."
1193 expected_vnfr_revision
= 1
1194 expected_ns_state
= "INSTANTIATED"
1195 expected_ns_operational_state
= "running"
1197 await instance
.update(nsr_id
, nslcmop_id
)
1198 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1201 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1204 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1205 "operational-status"
1207 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1210 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1211 self
.assertEqual(fs
.sync
.call_count
, 2)
1212 mock_charm_hash
.assert_not_called()
1213 mock_juju_bundle
.assert_not_called()
1214 mock_charm_upgrade
.assert_not_called()
1215 self
.assertEqual(return_ns_state
, expected_ns_state
)
1216 self
.assertEqual(return_operation_state
, expected_operation_state
)
1217 self
.assertEqual(return_operation_error
, expected_operation_error
)
1218 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1219 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1221 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1222 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1224 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1226 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1227 async def test_update_change_vnfpkg_juju_bundle_exists(
1229 mock_software_version
,
1234 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1235 Charm package exists, sw-version not changed, juju-bundle exists"""
1236 # Upgrade is not allowed with juju bundles, this will cause TypeError
1239 q_filter
={"_id": vnfd_id
},
1241 "_admin.revision": 5,
1242 "software-version": "1.0",
1243 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1248 q_filter
={"_id": vnfd_id
+ ":1"},
1250 "_admin.revision": 1,
1251 "software-version": "1.0",
1252 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1257 q_filter
={"_id": nsr_id
},
1259 "_admin.deployed.VCA.0.kdu_name": "native-kdu",
1262 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1264 mock_charm_hash
.side_effect
= [True]
1265 mock_software_version
.side_effect
= ["1.0", "1.0"]
1266 mock_juju_bundle
.return_value
= True
1267 instance
= self
.my_ns
1268 fs
= deepcopy(update_fs
)
1271 expected_vnfr_revision
= 1
1272 expected_ns_state
= "INSTANTIATED"
1273 expected_ns_operational_state
= "running"
1275 await instance
.update(nsr_id
, nslcmop_id
)
1276 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1277 "operational-status"
1279 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1282 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1283 self
.assertEqual(fs
.sync
.call_count
, 2)
1284 mock_charm_upgrade
.assert_not_called()
1285 mock_charm_hash
.assert_not_called()
1286 self
.assertEqual(return_ns_state
, expected_ns_state
)
1287 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1288 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1290 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1292 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1294 async def test_update_change_vnfpkg_charm_upgrade_failed(
1295 self
, mock_charm_upgrade
, mock_charm_hash
1297 """ "Update type: CHANGE_VNFPKG, latest_vnfd revision changed"
1298 Charm package exists, sw-version not changed, charm-upgrade failed"""
1301 q_filter
={"_id": vnfd_id
},
1303 "_admin.revision": 3,
1304 "software-version": "1.0",
1310 q_filter
={"_id": vnfd_id
+ ":1"},
1312 "_admin.revision": 1,
1313 "software-version": "1.0",
1317 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1319 mock_charm_hash
.return_value
= True
1321 task
= asyncio
.Future()
1322 task
.set_result(("FAILED", "some_error"))
1323 mock_charm_upgrade
.return_value
= task
1325 instance
= self
.my_ns
1326 fs
= deepcopy(update_fs
)
1328 expected_operation_state
= "FAILED"
1329 expected_operation_error
= "some_error"
1330 expected_vnfr_revision
= 1
1331 expected_ns_state
= "INSTANTIATED"
1332 expected_ns_operational_state
= "running"
1334 await instance
.update(nsr_id
, nslcmop_id
)
1335 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1338 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1341 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1342 "operational-status"
1344 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1347 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1348 self
.assertEqual(fs
.sync
.call_count
, 2)
1349 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1350 self
.assertEqual(mock_charm_upgrade
.call_count
, 1)
1351 self
.assertEqual(return_ns_state
, expected_ns_state
)
1352 self
.assertEqual(return_operation_state
, expected_operation_state
)
1353 self
.assertEqual(return_operation_error
, expected_operation_error
)
1354 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1355 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1357 def test_ns_update_find_sw_version_vnfd_not_includes(self
):
1358 """Find software version, VNFD does not have software version"""
1360 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1361 expected_result
= "1.0"
1362 result
= find_software_version(db_vnfd
)
1363 self
.assertEqual(result
, expected_result
, "Default sw version should be 1.0")
1365 def test_ns_update_find_sw_version_vnfd_includes(self
):
1366 """Find software version, VNFD includes software version"""
1368 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1369 db_vnfd
["software-version"] = "3.1"
1370 expected_result
= "3.1"
1371 result
= find_software_version(db_vnfd
)
1372 self
.assertEqual(result
, expected_result
, "VNFD software version is wrong")
1374 @patch("os.path.exists")
1375 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1376 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1377 def test_ns_update_check_charm_hash_not_changed(
1378 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1380 """Check charm hash, Hash did not change"""
1382 current_path
, target_path
= "/tmp/charm1", "/tmp/charm1"
1385 fs
.path
.__add
__ = Mock()
1386 fs
.path
.side_effect
= [current_path
, target_path
]
1387 fs
.path
.__add
__.side_effect
= [current_path
, target_path
]
1389 mock_path_exists
.side_effect
= [True, True]
1391 mock_compare_charmdir_hash
.return_value
= callable(False)
1392 mock_compare_charm_hash
.return_value
= callable(False)
1394 instance
= self
.my_ns
1396 expected_result
= False
1398 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1399 self
.assertEqual(result
, expected_result
, "Wrong charm hash control value")
1400 self
.assertEqual(mock_path_exists
.call_count
, 2)
1401 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 1)
1402 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1404 @patch("os.path.exists")
1405 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1406 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1407 def test_ns_update_check_charm_hash_changed(
1408 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1410 """Check charm hash, Hash has changed"""
1412 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1415 fs
.path
.__add
__ = Mock()
1416 fs
.path
.side_effect
= [current_path
, target_path
, current_path
, target_path
]
1417 fs
.path
.__add
__.side_effect
= [
1424 mock_path_exists
.side_effect
= [True, True]
1425 mock_compare_charmdir_hash
.return_value
= callable(True)
1426 mock_compare_charm_hash
.return_value
= callable(True)
1428 instance
= self
.my_ns
1430 expected_result
= True
1432 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1433 self
.assertEqual(result
, expected_result
, "Wrong charm hash control value")
1434 self
.assertEqual(mock_path_exists
.call_count
, 2)
1435 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 1)
1436 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1438 @patch("os.path.exists")
1439 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1440 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1441 def test_ns_update_check_no_charm_path(
1442 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1444 """Check charm hash, Charm path does not exist"""
1446 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1449 fs
.path
.__add
__ = Mock()
1450 fs
.path
.side_effect
= [current_path
, target_path
, current_path
, target_path
]
1451 fs
.path
.__add
__.side_effect
= [
1458 mock_path_exists
.side_effect
= [True, False]
1460 mock_compare_charmdir_hash
.return_value
= callable(False)
1461 mock_compare_charm_hash
.return_value
= callable(False)
1462 instance
= self
.my_ns
1465 with self
.assertRaises(LcmException
):
1466 instance
.check_charm_hash_changed(current_path
, target_path
)
1467 self
.assertEqual(mock_path_exists
.call_count
, 2)
1468 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 0)
1469 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1471 def test_ns_update_check_juju_bundle_existence_bundle_exists(self
):
1472 """Check juju bundle existence"""
1473 test_vnfd2
= self
.db
.get_one(
1474 "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
1476 expected_result
= "stable/native-kdu"
1477 result
= check_juju_bundle_existence(test_vnfd2
)
1478 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1480 def test_ns_update_check_juju_bundle_existence_bundle_does_not_exist(self
):
1481 """Check juju bundle existence"""
1482 test_vnfd1
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1483 expected_result
= None
1484 result
= check_juju_bundle_existence(test_vnfd1
)
1485 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1487 def test_ns_update_check_juju_bundle_existence_empty_vnfd(self
):
1488 """Check juju bundle existence"""
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_invalid_vnfd(self
):
1495 """Check juju bundle existence"""
1496 test_vnfd1
= [{"_id": vnfd_id
}]
1497 with self
.assertRaises(AttributeError):
1498 check_juju_bundle_existence(test_vnfd1
)
1500 def test_ns_update_check_juju_charm_artifacts_base_folder_wth_pkgdir(self
):
1501 """Check charm artifacts"""
1504 "pkg-dir": "hackfest_3charmed_vnfd",
1506 charm_name
= "simple"
1507 charm_type
= "lxc_proxy_charm"
1509 expected_result
= f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple"
1510 result
= get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
)
1511 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1513 def test_ns_update_check_juju_charm_artifacts_base_folder_wthout_pkgdir(self
):
1514 """Check charm artifacts, SOL004 packages"""
1518 charm_name
= "basic"
1519 charm_type
, revision
= "", ""
1520 expected_result
= f
"{vnfd_id}/Scripts/helm-charts/basic"
1521 result
= get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
)
1522 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1525 if __name__
== "__main__":