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
24 from n2vc
.exceptions
import N2VCException
26 from osm_lcm
import ns
27 from osm_common
.msgkafka
import MsgKafka
29 from osm_lcm
.data_utils
.lcm_config
import LcmCfg
30 from osm_lcm
.lcm_utils
import TaskRegistry
31 from osm_lcm
.ng_ro
import NgRoClient
32 from osm_lcm
.data_utils
.database
.database
import Database
33 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
34 from osm_lcm
.data_utils
.vca
import Relation
, EERelation
, DeployedVCA
35 from osm_lcm
.data_utils
.vnfd
import find_software_version
36 from osm_lcm
.lcm_utils
import check_juju_bundle_existence
, get_charm_artifact_path
37 from osm_lcm
.lcm_utils
import LcmException
38 from uuid
import uuid4
39 from unittest
.mock
import Mock
, patch
41 from osm_lcm
.tests
import test_db_descriptors
as descriptors
43 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
45 """ Perform unittests using asynctest of osm_lcm.ns module
46 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
47 OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
48 OSMLCMTEST_NS_NAME: change name of NS
49 OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
50 OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
51 OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
52 OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
53 OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
54 OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
55 OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
56 OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
57 OSMLCM_VCA_XXX: configuration of N2VC
58 OSMLCM_RO_XXX: configuration of RO
62 "global": {"loglevel": "DEBUG"},
64 "VCA": { # TODO replace with os.get_env to get other configurations
65 "host": getenv("OSMLCM_VCA_HOST", "vca"),
66 "port": getenv("OSMLCM_VCA_PORT", 17070),
67 "user": getenv("OSMLCM_VCA_USER", "admin"),
68 "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
69 "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
70 "ca_cert": getenv("OSMLCM_VCA_CACERT", None),
71 "apiproxy": getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1"),
74 "uri": "http://{}:{}/openmano".format(
75 getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")
77 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
78 "logger_name": "lcm.ROclient",
85 lcm_config
.set_from_dict(lcm_config_dict
)
86 lcm_config
.transform()
88 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
89 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["update"]
90 vnfr_id
= "6421c7c9-d865-4fb4-9a13-d4275d243e01"
91 vnfd_id
= "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"
92 update_fs
= Mock(autospec
=True)
93 update_fs
.path
.__add
__ = Mock()
94 update_fs
.path
.side_effect
= ["/", "/", "/", "/"]
95 update_fs
.sync
.side_effect
= [None, None]
102 class TestBaseNS(asynctest
.TestCase
):
103 async def _n2vc_DeployCharms(
115 for status
, message
in (
116 ("maintenance", "installing sofwware"),
117 ("active", "Ready!"),
119 # call callback after some time
120 asyncio
.sleep(5, loop
=self
.loop
)
121 callback(model_name
, application_name
, status
, message
, *callback_args
)
124 def _n2vc_FormatApplicationName(*args
):
127 yield "app_name-{}".format(num_calls
)
130 def _n2vc_CreateExecutionEnvironment(
131 self
, namespace
, reuse_ee_id
, db_dict
, *args
, **kwargs
133 k_list
= namespace
.split(".")
134 ee_id
= k_list
[1] + "."
136 for k
in k_list
[2:4]:
142 def _ro_status(self
, *args
, **kwargs
):
143 print("Args > {}".format(args
))
144 print("kwargs > {}".format(kwargs
))
147 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_update_action_text
)
150 if kwargs
.get("delete"):
151 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_delete_action_text
)
155 ro_ns_desc
= yaml
.safe_load(descriptors
.ro_ns_text
)
157 # if ip address provided, replace descriptor
158 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
160 ip_addresses_list
= ip_addresses
.split(",")
161 for vnf
in ro_ns_desc
["vnfs"]:
162 if not ip_addresses_list
:
164 vnf
["ip_address"] = ip_addresses_list
[0]
165 for vm
in vnf
["vms"]:
166 if not ip_addresses_list
:
168 vm
["ip_address"] = ip_addresses_list
.pop(0)
172 for net
in ro_ns_desc
["nets"]:
173 if net
["status"] != "ACTIVE":
174 net
["status"] = "ACTIVE"
177 for vnf
in ro_ns_desc
["vnfs"]:
178 for vm
in vnf
["vms"]:
179 if vm
["status"] != "ACTIVE":
180 vm
["status"] = "ACTIVE"
183 def _ro_deploy(self
, *args
, **kwargs
):
184 return {"action_id": args
[1]["action_id"], "nsr_id": args
[0], "status": "ok"}
186 def _return_uuid(self
, *args
, **kwargs
):
189 async def setUp(self
):
192 self
.mock_filesystem()
193 self
.mock_task_registry()
195 self
.create_nslcm_class()
201 if not getenv("OSMLCMTEST_DB_NOMOCK"):
202 # Cleanup singleton Database instance
203 Database
.instance
= None
205 self
.db
= Database({"database": {"driver": "memory"}}).instance
.db
206 self
.db
.create_list("vnfds", yaml
.safe_load(descriptors
.db_vnfds_text
))
208 "vnfds_revisions", yaml
.safe_load(descriptors
.db_vnfds_revisions_text
)
210 self
.db
.create_list("nsds", yaml
.safe_load(descriptors
.db_nsds_text
))
211 self
.db
.create_list("nsrs", yaml
.safe_load(descriptors
.db_nsrs_text
))
213 "vim_accounts", yaml
.safe_load(descriptors
.db_vim_accounts_text
)
216 "k8sclusters", yaml
.safe_load(descriptors
.db_k8sclusters_text
)
219 "nslcmops", yaml
.safe_load(descriptors
.db_nslcmops_text
)
221 self
.db
.create_list("vnfrs", yaml
.safe_load(descriptors
.db_vnfrs_text
))
222 self
.db_vim_accounts
= yaml
.safe_load(descriptors
.db_vim_accounts_text
)
224 def mock_kafka(self
):
225 self
.msg
= asynctest
.Mock(MsgKafka())
227 def mock_filesystem(self
):
228 if not getenv("OSMLCMTEST_FS_NOMOCK"):
229 self
.fs
= asynctest
.Mock(
230 Filesystem({"storage": {"driver": "local", "path": "/"}}).instance
.fs
232 self
.fs
.get_params
.return_value
= {
233 "path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")
235 self
.fs
.file_open
= asynctest
.mock_open()
236 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
237 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
239 def mock_task_registry(self
):
240 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
241 self
.lcm_tasks
.lock_HA
.return_value
= True
242 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
243 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
245 def mock_vca_k8s(self
):
246 if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
247 ns
.K8sJujuConnector
= asynctest
.MagicMock(ns
.K8sJujuConnector
)
248 ns
.K8sHelmConnector
= asynctest
.MagicMock(ns
.K8sHelmConnector
)
249 ns
.K8sHelm3Connector
= asynctest
.MagicMock(ns
.K8sHelm3Connector
)
251 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
252 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
253 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
255 def create_nslcm_class(self
):
256 self
.my_ns
= ns
.NsLcm(self
.msg
, self
.lcm_tasks
, lcm_config
, self
.loop
)
257 self
.my_ns
.fs
= self
.fs
258 self
.my_ns
.db
= self
.db
259 self
.my_ns
._wait
_dependent
_n
2vc
= asynctest
.CoroutineMock()
261 def mock_logging(self
):
262 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
263 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
265 def mock_vca_n2vc(self
):
266 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
267 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
268 # self.my_ns.n2vc = asynctest.Mock(N2VC())
269 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv(
270 "OSMLCM_VCA_PUBKEY", "public_key"
272 # allow several versions of n2vc
273 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(
274 side_effect
=self
._n
2vc
_FormatApplicationName
()
276 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(
277 side_effect
=self
._n
2vc
_DeployCharms
279 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
280 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
282 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(
285 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(
288 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
289 side_effect
=self
._return
_uuid
291 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(
292 side_effect
=self
._return
_uuid
294 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(
295 return_value
="completed"
297 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(
298 return_value
={"result": "ok", "pubkey": pub_key
}
300 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(
303 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
304 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key")
306 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(
309 self
.my_ns
.n2vc
.register_execution_environment
= asynctest
.CoroutineMock(
310 return_value
="model-name.application-name.k8s"
314 if not getenv("OSMLCMTEST_RO_NOMOCK"):
315 self
.my_ns
.RO
= asynctest
.Mock(
316 NgRoClient(self
.loop
, **lcm_config
.RO
.to_dict())
318 # TODO first time should be empty list, following should return a dict
319 # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
320 self
.my_ns
.RO
.deploy
= asynctest
.CoroutineMock(
321 self
.my_ns
.RO
.deploy
, side_effect
=self
._ro
_deploy
323 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status)
324 # self.my_ns.RO.create_action = asynctest.CoroutineMock(self.my_ns.RO.create_action,
325 # return_value={"vm-id": {"vim_result": 200,
326 # "description": "done"}})
327 self
.my_ns
.RO
.delete
= asynctest
.CoroutineMock(self
.my_ns
.RO
.delete
)
329 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
330 # async def test_instantiate(self):
331 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
332 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
333 # # print("Test instantiate started")
335 # # delete deployed information of database
336 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
337 # if self.db.get_list("nsrs")[0]["_admin"].get("deployed"):
338 # del self.db.get_list("nsrs")[0]["_admin"]["deployed"]
339 # for db_vnfr in self.db.get_list("vnfrs"):
340 # db_vnfr.pop("ip_address", None)
341 # for db_vdur in db_vnfr["vdur"]:
342 # db_vdur.pop("ip_address", None)
343 # db_vdur.pop("mac_address", None)
344 # if getenv("OSMLCMTEST_RO_VIMID"):
345 # self.db.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
346 # if getenv("OSMLCMTEST_RO_VIMID"):
347 # self.db.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
349 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
351 # self.msg.aiowrite.assert_called_once_with("ns", "instantiated",
352 # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id,
353 # "operationState": "COMPLETED"},
355 # self.lcm_tasks.lock_HA.assert_called_once_with('ns', 'nslcmops', nslcmop_id)
356 # if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
357 # self.assertTrue(self.my_ns.logger.debug.called, "Debug method not called")
358 # self.my_ns.logger.error.assert_not_called()
359 # self.my_ns.logger.exception().assert_not_called()
361 # if not getenv("OSMLCMTEST_DB_NOMOCK"):
362 # self.assertTrue(self.db.set_one.called, "db.set_one not called")
363 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
364 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
365 # self.assertEqual(db_nsr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
366 # for vnfr in db_vnfrs_list:
367 # self.assertEqual(vnfr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
369 # if not getenv("OSMLCMTEST_VCA_NOMOCK"):
370 # # check intial-primitives called
371 # self.assertTrue(self.my_ns.n2vc.exec_primitive.called,
372 # "Exec primitive not called for initial config primitive")
373 # for _call in self.my_ns.n2vc.exec_primitive.call_args_list:
374 # self.assertIn(_call[1]["primitive_name"], ("config", "touch"),
375 # "called exec primitive with a primitive different than config or touch")
377 # # TODO add more checks of called methods
378 # # TODO add a terminate
380 # async def test_instantiate_ee_list(self):
381 # # Using modern IM where configuration is in the new format of execution_environment_list
382 # ee_descriptor_id = "charm_simple"
383 # non_used_initial_primitive = {
384 # "name": "not_to_be_called",
386 # "execution-environment-ref": "not_used_ee"
390 # "id": ee_descriptor_id,
391 # "juju": {"charm": "simple"},
398 # q_filter={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"},
399 # update_dict={"vnf-configuration.0.execution-environment-list": ee_list,
400 # "vnf-configuration.0.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id,
401 # "vnf-configuration.0.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id,
402 # "vnf-configuration.0.initial-config-primitive.2": non_used_initial_primitive,
403 # "vnf-configuration.0.config-primitive.0.execution-environment-ref": ee_descriptor_id,
404 # "vnf-configuration.0.config-primitive.0.execution-environment-primitive": "touch_charm",
406 # unset={"vnf-configuration.juju": None})
407 # await self.test_instantiate()
408 # # this will check that the initial-congig-primitive 'not_to_be_called' is not called
411 class TestMyNS(TestBaseNS
):
412 @asynctest.fail_on(active_handles
=True)
413 async def test_start_stop_rebuild_pass(self
):
414 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
415 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops"]
416 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
417 additional_param
= {"count-index": "0"}
418 operation_type
= "start"
419 await self
.my_ns
.rebuild_start_stop(
420 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
422 expected_value
= "COMPLETED"
423 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
426 self
.assertEqual(return_value
, expected_value
)
428 @asynctest.fail_on(active_handles
=True)
429 async def test_start_stop_rebuild_fail(self
):
430 nsr_id
= descriptors
.test_ids
["TEST-OP-VNF"]["ns"]
431 nslcmop_id
= descriptors
.test_ids
["TEST-OP-VNF"]["nslcmops1"]
432 vnf_id
= descriptors
.test_ids
["TEST-OP-VNF"]["vnfrs"]
433 additional_param
= {"count-index": "0"}
434 operation_type
= "stop"
435 await self
.my_ns
.rebuild_start_stop(
436 nsr_id
, nslcmop_id
, vnf_id
, additional_param
, operation_type
438 expected_value
= "Error"
439 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
442 self
.assertEqual(return_value
, expected_value
)
444 # Test scale() and related methods
445 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
446 async def test_scale(self
):
447 # print("Test scale started")
449 # TODO: Add more higher-lever tests here, for example:
450 # scale-out/scale-in operations with success/error result
452 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
453 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
454 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
455 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
456 expected_value
= "FAILED"
457 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
460 self
.assertEqual(return_value
, expected_value
)
461 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
463 # Test scale() for native kdu
464 # this also includes testing _scale_kdu()
465 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["ns"]
466 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU"]["instantiate"]
468 self
.my_ns
.k8sclusterjuju
.scale
= asynctest
.mock
.CoroutineMock()
469 self
.my_ns
.k8sclusterjuju
.exec_primitive
= asynctest
.mock
.CoroutineMock()
470 self
.my_ns
.k8sclusterjuju
.get_scale_count
= asynctest
.mock
.CoroutineMock(
473 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
474 expected_value
= "COMPLETED"
475 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
478 self
.assertEqual(return_value
, expected_value
)
479 self
.my_ns
.k8sclusterjuju
.scale
.assert_called_once()
481 # Test scale() for native kdu with 2 resource
482 nsr_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["ns"]
483 nslcmop_id
= descriptors
.test_ids
["TEST-NATIVE-KDU-2"]["instantiate"]
485 self
.my_ns
.k8sclusterjuju
.get_scale_count
.return_value
= 2
486 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
487 expected_value
= "COMPLETED"
488 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
491 self
.assertEqual(return_value
, expected_value
)
492 self
.my_ns
.k8sclusterjuju
.scale
.assert_called()
494 async def test_vca_status_refresh(self
):
495 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
496 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
497 await self
.my_ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
498 expected_value
= dict()
499 return_value
= dict()
500 vnf_descriptors
= self
.db
.get_list("vnfds")
501 for i
, _
in enumerate(vnf_descriptors
):
502 for j
, value
in enumerate(vnf_descriptors
[i
]["df"]):
503 if "lcm-operations-configuration" in vnf_descriptors
[i
]["df"][j
]:
506 in value
["lcm-operations-configuration"][
507 "operate-vnf-op-config"
510 for k
, v
in enumerate(
511 value
["lcm-operations-configuration"][
512 "operate-vnf-op-config"
516 v
.get("execution-environment-list")
517 and "juju" in v
["execution-environment-list"][0]
519 expected_value
= self
.db
.get_list("nsrs")[i
][
522 await self
.my_ns
._on
_update
_n
2vc
_db
(
525 "_admin.deployed.VCA.{}".format(k
),
528 return_value
= self
.db
.get_list("nsrs")[i
]["vcaStatus"]
529 self
.assertEqual(return_value
, expected_value
)
531 # Test _retry_or_skip_suboperation()
533 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
534 # - if marked as anything but 'COMPLETED', the suboperation index is expected
535 def test_scale_retry_or_skip_suboperation(self
):
536 # Load an alternative 'nslcmops' YAML for this test
537 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
538 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
540 # Test when 'operationState' is 'COMPLETED'
541 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = "COMPLETED"
542 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
543 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
544 self
.assertEqual(return_value
, expected_value
)
545 # Test when 'operationState' is not 'COMPLETED'
546 db_nslcmop
["_admin"]["operations"][op_index
]["operationState"] = None
547 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
548 expected_value
= op_index
549 self
.assertEqual(return_value
, expected_value
)
551 # Test _find_suboperation()
552 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
553 def test_scale_find_suboperation(self
):
554 # Load an alternative 'nslcmops' YAML for this test
555 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
556 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
557 # Find this sub-operation
559 vnf_index
= db_nslcmop
["_admin"]["operations"][op_index
]["member_vnf_index"]
560 primitive
= db_nslcmop
["_admin"]["operations"][op_index
]["primitive"]
561 primitive_params
= db_nslcmop
["_admin"]["operations"][op_index
][
565 "member_vnf_index": vnf_index
,
566 "primitive": primitive
,
567 "primitive_params": primitive_params
,
569 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
570 self
.assertEqual(found_op_index
, op_index
)
571 # Test with not-matching params
573 "member_vnf_index": vnf_index
,
575 "primitive_params": primitive_params
,
577 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
578 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
581 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
582 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
584 # Test _update_suboperation_status()
585 def test_scale_update_suboperation_status(self
):
586 self
.db
.set_one
= asynctest
.Mock()
587 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
588 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
590 # Force the initial values to be distinct from the updated ones
591 q_filter
= {"_id": db_nslcmop
["_id"]}
592 # Test to change 'operationState' and 'detailed-status'
593 operationState
= "COMPLETED"
594 detailed_status
= "Done"
595 expected_update_dict
= {
596 "_admin.operations.0.operationState": operationState
,
597 "_admin.operations.0.detailed-status": detailed_status
,
599 self
.my_ns
._update
_suboperation
_status
(
600 db_nslcmop
, op_index
, operationState
, detailed_status
602 self
.db
.set_one
.assert_called_once_with(
605 update_dict
=expected_update_dict
,
609 def test_scale_add_suboperation(self
):
610 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
611 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
613 num_ops_before
= len(db_nslcmop
.get("_admin", {}).get("operations", [])) - 1
615 vdu_count_index
= None
618 mapped_primitive_params
= {
621 "data-type": "STRING",
623 "default-value": "<touch_filename2>",
628 operationState
= "PROCESSING"
629 detailed_status
= "In progress"
630 operationType
= "PRE-SCALE"
631 # Add a 'pre-scale' suboperation
632 op_index_after
= self
.my_ns
._add
_suboperation
(
639 mapped_primitive_params
,
644 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
646 # Delete all suboperations and add the same operation again
647 del db_nslcmop
["_admin"]["operations"]
648 op_index_zero
= self
.my_ns
._add
_suboperation
(
655 mapped_primitive_params
,
660 self
.assertEqual(op_index_zero
, 0)
662 # Add a 'RO' suboperation
663 RO_nsr_id
= "1234567890"
668 "member-vnf-index": "1",
669 "osm_vdu_id": "dataVM",
672 op_index
= self
.my_ns
._add
_suboperation
(
679 mapped_primitive_params
,
686 db_RO_nsr_id
= db_nslcmop
["_admin"]["operations"][op_index
]["RO_nsr_id"]
687 self
.assertEqual(op_index
, 1)
688 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
690 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
691 op_index_invalid
= self
.my_ns
._add
_suboperation
(
692 None, None, None, None, None, None, None, None, None, None, None
694 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
696 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
697 # check the possible return values:
698 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
699 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
700 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
701 def test_scale_check_or_add_scale_suboperation(self
):
702 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
703 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
704 operationType
= "PRE-SCALE"
710 "data-type": "STRING",
712 "default-value": "<touch_filename2>",
718 # Delete all sub-operations to be sure this is a new sub-operation
719 del db_nslcmop
["_admin"]["operations"]
721 # Add a new sub-operation
722 # For new sub-operations, operationState is set to 'PROCESSING' by default
723 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
724 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
726 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
728 # Use the same parameters again to match the already added sub-operation
729 # which has status 'PROCESSING' (!= 'COMPLETED') by default
730 # The expected return value is a non-negative number
731 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
732 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
734 self
.assertTrue(op_index_existing
>= 0)
736 # Change operationState 'manually' for this sub-operation
737 db_nslcmop
["_admin"]["operations"][op_index_existing
][
740 # Then use the same parameters again to match the already added sub-operation,
741 # which now has status 'COMPLETED'
742 # The expected return value is SUBOPERATION_STATUS_SKIP
743 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
744 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
746 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
748 # RO sub-operation test:
749 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
750 RO_nsr_id
= "1234567890"
755 "member-vnf-index": "1",
756 "osm_vdu_id": "dataVM",
759 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
760 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
762 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
764 # Use the same parameters again to match the already added RO sub-operation
765 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
766 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
768 self
.assertTrue(op_index_existing_RO
>= 0)
770 # Change operationState 'manually' for this RO sub-operation
771 db_nslcmop
["_admin"]["operations"][op_index_existing_RO
][
774 # Then use the same parameters again to match the already added sub-operation,
775 # which now has status 'COMPLETED'
776 # The expected return value is SUBOPERATION_STATUS_SKIP
777 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
778 db_nslcmop
, vnf_index
, None, None, "SCALE-RO", RO_nsr_id
, RO_scaling_info
780 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
782 async def test_deploy_kdus(self
):
783 nsr_id
= descriptors
.test_ids
["TEST-KDU"]["ns"]
784 nslcmop_id
= descriptors
.test_ids
["TEST-KDU"]["instantiate"]
785 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
786 db_vnfr
= self
.db
.get_one(
787 "vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "multikdu"}
789 db_vnfrs
= {"multikdu": db_vnfr
}
790 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
794 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
= asynctest
.mock
.Mock()
795 self
.my_ns
.k8sclusterhelm3
.generate_kdu_instance_name
.return_value
= "k8s_id"
796 self
.my_ns
.k8sclusterhelm3
.install
= asynctest
.CoroutineMock()
797 self
.my_ns
.k8sclusterhelm3
.synchronize_repos
= asynctest
.CoroutineMock(
798 return_value
=("", "")
800 self
.my_ns
.k8sclusterhelm3
.get_services
= asynctest
.CoroutineMock(
803 await self
.my_ns
.deploy_kdus(
804 logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
806 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
807 db_nsr
= self
.db
.get_list("nsrs")[1]
810 db_nsr
["_admin"]["deployed"],
811 "K8s entry not created at '_admin.deployed'",
813 self
.assertIsInstance(
814 db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list"
817 len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list"
820 "kdu-instance": "k8s_id",
821 "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
822 "k8scluster-type": "helm-chart-v3",
824 "member-vnf-index": "multikdu",
826 "kdu-deployment-name": None,
829 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][0])
830 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
831 expected_kdu_model
= "stable/openldap:1.2.1"
832 self
.assertEqual(nsr_result
, k8s_instace_info
)
834 nsr_kdu_model_result
in expected_kdu_model
835 or expected_kdu_model
in nsr_kdu_model_result
837 nsr_result
= copy
.deepcopy(db_nsr
["_admin"]["deployed"]["K8s"][1])
838 nsr_kdu_model_result
= nsr_result
.pop("kdu-model")
839 k8s_instace_info
["kdu-name"] = "mongo"
840 expected_kdu_model
= "stable/mongodb"
841 self
.assertEqual(nsr_result
, k8s_instace_info
)
843 nsr_kdu_model_result
in expected_kdu_model
844 or expected_kdu_model
in nsr_kdu_model_result
847 # Test remove_vnf() and related methods
848 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
849 async def test_remove_vnf(self
):
851 nsr_id
= descriptors
.test_ids
["TEST-UPDATE"]["ns"]
852 nslcmop_id
= descriptors
.test_ids
["TEST-UPDATE"]["removeVnf"]
853 vnf_instance_id
= descriptors
.test_ids
["TEST-UPDATE"]["vnf"]
854 mock_wait_ng_ro
= asynctest
.CoroutineMock()
855 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
856 await self
.my_ns
.update(nsr_id
, nslcmop_id
)
857 expected_value
= "COMPLETED"
858 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
861 self
.assertEqual(return_value
, expected_value
)
862 with self
.assertRaises(Exception) as context
:
863 self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
865 "database exception Not found entry with filter"
866 in str(context
.exception
)
869 # test vertical scale executes sucessfully
870 # @patch("osm_lcm.ng_ro.status.response")
871 @asynctest.fail_on(active_handles
=True)
872 async def test_vertical_scaling(self
):
873 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
874 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate"]
876 # calling the vertical scale fucntion
877 # self.my_ns.RO.status = asynctest.CoroutineMock(self.my_ns.RO.status, side_effect=self._ro_status("update"))
878 mock_wait_ng_ro
= asynctest
.CoroutineMock()
879 with
patch("osm_lcm.ns.NsLcm._wait_ng_ro", mock_wait_ng_ro
):
880 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
881 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
884 expected_value
= "COMPLETED"
885 self
.assertEqual(return_value
, expected_value
)
887 # test vertical scale executes fail
888 @asynctest.fail_on(active_handles
=True)
889 async def test_vertical_scaling_fail(self
):
890 # get th nsr nad nslcmops id from descriptors
891 nsr_id
= descriptors
.test_ids
["TEST-V-SCALE"]["ns"]
892 nslcmop_id
= descriptors
.test_ids
["TEST-V-SCALE"]["instantiate-1"]
894 # calling the vertical scale fucntion
895 await self
.my_ns
.vertical_scale(nsr_id
, nslcmop_id
)
896 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
899 expected_value
= "FAILED"
900 self
.assertEqual(return_value
, expected_value
)
902 # async def test_instantiate_pdu(self):
903 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
904 # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
905 # # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
906 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
907 # update_dict={"ip-address": "10.205.1.46",
908 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
909 # "vdur.0.pdu-type": "PDU-TYPE-1",
910 # "vdur.0.ip-address": "10.205.1.46",
912 # unset={"vdur.status": None})
913 # self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
914 # update_dict={"ip-address": "10.205.1.47",
915 # "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
916 # "vdur.0.pdu-type": "PDU-TYPE-1",
917 # "vdur.0.ip-address": "10.205.1.47",
919 # unset={"vdur.status": None})
921 # await self.my_ns.instantiate(nsr_id, nslcmop_id)
922 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
923 # self.assertEqual(db_nsr.get("nsState"), "READY", str(db_nsr.get("errorDescription ")))
924 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
925 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
926 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
927 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
929 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
930 # async def test_terminate_without_configuration(self):
931 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
932 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
933 # # set instantiation task as completed
934 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
935 # update_dict={"operationState": "COMPLETED"})
936 # self.db.set_one("nsrs", {"_id": nsr_id},
937 # update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
939 # await self.my_ns.terminate(nsr_id, nslcmop_id)
940 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
941 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
942 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
943 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
944 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
945 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
946 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
947 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
948 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
949 # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
950 # for vnfr in db_vnfrs_list:
951 # self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
953 # @asynctest.fail_on(active_handles=True) # all async tasks must be completed
954 # async def test_terminate_primitive(self):
955 # nsr_id = descriptors.test_ids["TEST-A"]["ns"]
956 # nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
957 # # set instantiation task as completed
958 # self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
959 # update_dict={"operationState": "COMPLETED"})
961 # # modify vnfd descriptor to include terminate_primitive
962 # terminate_primitive = [{
964 # "parameter": [{"name": "filename", "value": "terminate_filename"}],
967 # db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
968 # self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
969 # {"vnf-configuration.0.terminate-config-primitive": terminate_primitive})
971 # await self.my_ns.terminate(nsr_id, nslcmop_id)
972 # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
973 # self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
974 # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
975 # self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
976 # self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
977 # self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
978 # self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
979 # self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
980 # self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
982 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
984 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
986 @patch("osm_lcm.data_utils.vnfd.find_software_version")
987 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
988 async def test_update_change_vnfpkg_sw_version_not_changed(
991 mock_software_version
,
995 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed,
996 Charm package changed, sw-version is not changed"""
999 q_filter
={"_id": vnfd_id
},
1000 update_dict
={"_admin.revision": 3, "kdu": []},
1005 q_filter
={"_id": vnfd_id
+ ":1"},
1006 update_dict
={"_admin.revision": 1, "kdu": []},
1009 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1011 mock_charm_hash
.return_value
= True
1012 mock_software_version
.side_effect
= ["1.0", "1.0"]
1014 task
= asyncio
.Future()
1015 task
.set_result(("COMPLETED", "some_output"))
1016 mock_charm_upgrade
.return_value
= task
1018 instance
= self
.my_ns
1019 fs
= deepcopy(update_fs
)
1022 expected_operation_state
= "COMPLETED"
1023 expected_operation_error
= ""
1024 expected_vnfr_revision
= 3
1025 expected_ns_state
= "INSTANTIATED"
1026 expected_ns_operational_state
= "running"
1028 await instance
.update(nsr_id
, nslcmop_id
)
1029 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1032 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1035 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1036 "operational-status"
1038 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1041 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1042 mock_charm_hash
.assert_called_with(
1043 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1044 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1046 self
.assertEqual(fs
.sync
.call_count
, 2)
1047 self
.assertEqual(return_ns_state
, expected_ns_state
)
1048 self
.assertEqual(return_operation_state
, expected_operation_state
)
1049 self
.assertEqual(return_operation_error
, expected_operation_error
)
1050 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1051 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1053 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1055 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1057 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1058 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1059 async def test_update_change_vnfpkg_vnfd_revision_not_changed(
1062 mock_software_version
,
1066 """Update type: CHANGE_VNFPKG, latest_vnfd revision not changed"""
1068 "vnfds", q_filter
={"_id": vnfd_id
}, update_dict
={"_admin.revision": 1}
1070 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1072 mock_charm_hash
.return_value
= True
1074 task
= asyncio
.Future()
1075 task
.set_result(("COMPLETED", "some_output"))
1076 mock_charm_upgrade
.return_value
= task
1078 instance
= self
.my_ns
1080 expected_operation_state
= "COMPLETED"
1081 expected_operation_error
= ""
1082 expected_vnfr_revision
= 1
1083 expected_ns_state
= "INSTANTIATED"
1084 expected_ns_operational_state
= "running"
1086 await instance
.update(nsr_id
, nslcmop_id
)
1087 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1090 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1093 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1094 "operational-status"
1096 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1097 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1100 mock_charm_hash
.assert_not_called()
1101 mock_software_version
.assert_not_called()
1102 mock_juju_bundle
.assert_not_called()
1103 mock_charm_upgrade
.assert_not_called()
1104 update_fs
.sync
.assert_not_called()
1105 self
.assertEqual(return_ns_state
, expected_ns_state
)
1106 self
.assertEqual(return_operation_state
, expected_operation_state
)
1107 self
.assertEqual(return_operation_error
, expected_operation_error
)
1108 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1109 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1111 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1113 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1115 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1116 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1117 async def test_update_change_vnfpkg_charm_is_not_changed(
1120 mock_software_version
,
1124 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1125 Charm package is not changed, sw-version is not changed"""
1128 q_filter
={"_id": vnfd_id
},
1129 update_dict
={"_admin.revision": 3, "kdu": []},
1133 q_filter
={"_id": vnfd_id
+ ":1"},
1134 update_dict
={"_admin.revision": 1, "kdu": []},
1136 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1138 mock_charm_hash
.return_value
= False
1139 mock_software_version
.side_effect
= ["1.0", "1.0"]
1141 instance
= self
.my_ns
1142 fs
= deepcopy(update_fs
)
1144 expected_operation_state
= "COMPLETED"
1145 expected_operation_error
= ""
1146 expected_vnfr_revision
= 3
1147 expected_ns_state
= "INSTANTIATED"
1148 expected_ns_operational_state
= "running"
1150 await instance
.update(nsr_id
, nslcmop_id
)
1151 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1154 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1157 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1158 "operational-status"
1160 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1163 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1164 mock_charm_hash
.assert_called_with(
1165 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1166 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1168 self
.assertEqual(fs
.sync
.call_count
, 2)
1169 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1170 mock_juju_bundle
.assert_not_called()
1171 mock_charm_upgrade
.assert_not_called()
1172 self
.assertEqual(return_ns_state
, expected_ns_state
)
1173 self
.assertEqual(return_operation_state
, expected_operation_state
)
1174 self
.assertEqual(return_operation_error
, expected_operation_error
)
1175 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1176 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1178 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1179 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1181 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1183 @patch("osm_lcm.lcm_utils.get_charm_artifact_path")
1184 async def test_update_change_vnfpkg_sw_version_changed(
1185 self
, mock_charm_artifact
, mock_charm_upgrade
, mock_charm_hash
, mock_juju_bundle
1187 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1188 Charm package exists, sw-version changed."""
1191 q_filter
={"_id": vnfd_id
},
1192 update_dict
={"_admin.revision": 3, "software-version": "3.0", "kdu": []},
1196 q_filter
={"_id": vnfd_id
+ ":1"},
1197 update_dict
={"_admin.revision": 1, "kdu": []},
1199 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1200 mock_charm_hash
.return_value
= False
1202 mock_charm_artifact
.side_effect
= [
1203 f
"{vnfd_id}:1/hackfest_3charmed_vnfd/charms/simple",
1204 f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple",
1207 instance
= self
.my_ns
1208 fs
= deepcopy(update_fs
)
1210 expected_operation_state
= "FAILED"
1211 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."
1212 expected_vnfr_revision
= 1
1213 expected_ns_state
= "INSTANTIATED"
1214 expected_ns_operational_state
= "running"
1216 await instance
.update(nsr_id
, nslcmop_id
)
1217 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1220 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1223 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1224 "operational-status"
1226 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1229 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1230 self
.assertEqual(fs
.sync
.call_count
, 2)
1231 mock_charm_hash
.assert_not_called()
1232 mock_juju_bundle
.assert_not_called()
1233 mock_charm_upgrade
.assert_not_called()
1234 self
.assertEqual(return_ns_state
, expected_ns_state
)
1235 self
.assertEqual(return_operation_state
, expected_operation_state
)
1236 self
.assertEqual(return_operation_error
, expected_operation_error
)
1237 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1238 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1240 @patch("osm_lcm.lcm_utils.check_juju_bundle_existence")
1241 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1243 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1245 @patch("osm_lcm.data_utils.vnfd.find_software_version")
1246 async def test_update_change_vnfpkg_juju_bundle_exists(
1248 mock_software_version
,
1253 """Update type: CHANGE_VNFPKG, latest_vnfd revision changed
1254 Charm package exists, sw-version not changed, juju-bundle exists"""
1255 # Upgrade is not allowed with juju bundles, this will cause TypeError
1258 q_filter
={"_id": vnfd_id
},
1260 "_admin.revision": 5,
1261 "software-version": "1.0",
1262 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1267 q_filter
={"_id": vnfd_id
+ ":1"},
1269 "_admin.revision": 1,
1270 "software-version": "1.0",
1271 "kdu": [{"kdu_name": "native-kdu", "juju-bundle": "stable/native-kdu"}],
1276 q_filter
={"_id": nsr_id
},
1277 update_dict
={"_admin.deployed.VCA.0.kdu_name": "native-kdu"},
1279 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1281 mock_charm_hash
.side_effect
= [True]
1282 mock_software_version
.side_effect
= ["1.0", "1.0"]
1283 mock_juju_bundle
.return_value
= True
1284 instance
= self
.my_ns
1285 fs
= deepcopy(update_fs
)
1288 expected_vnfr_revision
= 1
1289 expected_ns_state
= "INSTANTIATED"
1290 expected_ns_operational_state
= "running"
1292 await instance
.update(nsr_id
, nslcmop_id
)
1293 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1294 "operational-status"
1296 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1299 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1300 self
.assertEqual(fs
.sync
.call_count
, 2)
1301 mock_charm_upgrade
.assert_not_called()
1302 mock_charm_hash
.assert_not_called()
1303 self
.assertEqual(return_ns_state
, expected_ns_state
)
1304 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1305 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1307 @patch("osm_lcm.lcm_utils.LcmBase.check_charm_hash_changed")
1309 "osm_lcm.ns.NsLcm._ns_charm_upgrade", new_callable
=asynctest
.Mock(autospec
=True)
1311 async def test_update_change_vnfpkg_charm_upgrade_failed(
1312 self
, mock_charm_upgrade
, mock_charm_hash
1314 """ "Update type: CHANGE_VNFPKG, latest_vnfd revision changed"
1315 Charm package exists, sw-version not changed, charm-upgrade failed"""
1318 q_filter
={"_id": vnfd_id
},
1319 update_dict
={"_admin.revision": 3, "software-version": "1.0", "kdu": []},
1323 q_filter
={"_id": vnfd_id
+ ":1"},
1324 update_dict
={"_admin.revision": 1, "software-version": "1.0", "kdu": []},
1326 self
.db
.set_one("vnfrs", q_filter
={"_id": vnfr_id
}, update_dict
={"revision": 1})
1328 mock_charm_hash
.return_value
= True
1330 task
= asyncio
.Future()
1331 task
.set_result(("FAILED", "some_error"))
1332 mock_charm_upgrade
.return_value
= task
1334 instance
= self
.my_ns
1335 fs
= deepcopy(update_fs
)
1337 expected_operation_state
= "FAILED"
1338 expected_operation_error
= "some_error"
1339 expected_vnfr_revision
= 1
1340 expected_ns_state
= "INSTANTIATED"
1341 expected_ns_operational_state
= "running"
1343 await instance
.update(nsr_id
, nslcmop_id
)
1344 return_operation_state
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1347 return_operation_error
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get(
1350 return_ns_operational_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get(
1351 "operational-status"
1353 return_vnfr_revision
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
}).get(
1356 return_ns_state
= self
.db
.get_one("nsrs", {"_id": nsr_id
}).get("nsState")
1357 self
.assertEqual(fs
.sync
.call_count
, 2)
1358 self
.assertEqual(mock_charm_hash
.call_count
, 1)
1359 self
.assertEqual(mock_charm_upgrade
.call_count
, 1)
1360 self
.assertEqual(return_ns_state
, expected_ns_state
)
1361 self
.assertEqual(return_operation_state
, expected_operation_state
)
1362 self
.assertEqual(return_operation_error
, expected_operation_error
)
1363 self
.assertEqual(return_ns_operational_state
, expected_ns_operational_state
)
1364 self
.assertEqual(return_vnfr_revision
, expected_vnfr_revision
)
1366 def test_ns_update_find_sw_version_vnfd_not_includes(self
):
1367 """Find software version, VNFD does not have software version"""
1369 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1370 expected_result
= "1.0"
1371 result
= find_software_version(db_vnfd
)
1372 self
.assertEqual(result
, expected_result
, "Default sw version should be 1.0")
1374 def test_ns_update_find_sw_version_vnfd_includes(self
):
1375 """Find software version, VNFD includes software version"""
1377 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1378 db_vnfd
["software-version"] = "3.1"
1379 expected_result
= "3.1"
1380 result
= find_software_version(db_vnfd
)
1381 self
.assertEqual(result
, expected_result
, "VNFD software version is wrong")
1383 @patch("os.path.exists")
1384 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1385 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1386 def test_ns_update_check_charm_hash_not_changed(
1387 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1389 """Check charm hash, Hash did not change"""
1391 current_path
, target_path
= "/tmp/charm1", "/tmp/charm1"
1394 fs
.path
.__add
__ = Mock()
1395 fs
.path
.side_effect
= [current_path
, target_path
]
1396 fs
.path
.__add
__.side_effect
= [current_path
, target_path
]
1398 mock_path_exists
.side_effect
= [True, True]
1400 mock_compare_charmdir_hash
.return_value
= callable(False)
1401 mock_compare_charm_hash
.return_value
= callable(False)
1403 instance
= self
.my_ns
1405 expected_result
= False
1407 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1408 self
.assertEqual(result
, expected_result
, "Wrong charm hash control value")
1409 self
.assertEqual(mock_path_exists
.call_count
, 2)
1410 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 1)
1411 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1413 @patch("os.path.exists")
1414 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1415 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1416 def test_ns_update_check_charm_hash_changed(
1417 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1419 """Check charm hash, Hash has changed"""
1421 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1424 fs
.path
.__add
__ = Mock()
1425 fs
.path
.side_effect
= [current_path
, target_path
, current_path
, target_path
]
1426 fs
.path
.__add
__.side_effect
= [
1433 mock_path_exists
.side_effect
= [True, True]
1434 mock_compare_charmdir_hash
.return_value
= callable(True)
1435 mock_compare_charm_hash
.return_value
= callable(True)
1437 instance
= self
.my_ns
1439 expected_result
= True
1441 result
= instance
.check_charm_hash_changed(current_path
, target_path
)
1442 self
.assertEqual(result
, expected_result
, "Wrong charm hash control value")
1443 self
.assertEqual(mock_path_exists
.call_count
, 2)
1444 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 1)
1445 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1447 @patch("os.path.exists")
1448 @patch("osm_lcm.lcm_utils.LcmBase.compare_charmdir_hash")
1449 @patch("osm_lcm.lcm_utils.LcmBase.compare_charm_hash")
1450 def test_ns_update_check_no_charm_path(
1451 self
, mock_compare_charm_hash
, mock_compare_charmdir_hash
, mock_path_exists
1453 """Check charm hash, Charm path does not exist"""
1455 current_path
, target_path
= "/tmp/charm1", "/tmp/charm2"
1458 fs
.path
.__add
__ = Mock()
1459 fs
.path
.side_effect
= [current_path
, target_path
, current_path
, target_path
]
1460 fs
.path
.__add
__.side_effect
= [
1467 mock_path_exists
.side_effect
= [True, False]
1469 mock_compare_charmdir_hash
.return_value
= callable(False)
1470 mock_compare_charm_hash
.return_value
= callable(False)
1471 instance
= self
.my_ns
1474 with self
.assertRaises(LcmException
):
1475 instance
.check_charm_hash_changed(current_path
, target_path
)
1476 self
.assertEqual(mock_path_exists
.call_count
, 2)
1477 self
.assertEqual(mock_compare_charmdir_hash
.call_count
, 0)
1478 self
.assertEqual(mock_compare_charm_hash
.call_count
, 0)
1480 def test_ns_update_check_juju_bundle_existence_bundle_exists(self
):
1481 """Check juju bundle existence"""
1482 test_vnfd2
= self
.db
.get_one(
1483 "vnfds", {"_id": "d96b1cdf-5ad6-49f7-bf65-907ada989293"}
1485 expected_result
= "stable/native-kdu"
1486 result
= check_juju_bundle_existence(test_vnfd2
)
1487 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1489 def test_ns_update_check_juju_bundle_existence_bundle_does_not_exist(self
):
1490 """Check juju bundle existence"""
1491 test_vnfd1
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1492 expected_result
= None
1493 result
= check_juju_bundle_existence(test_vnfd1
)
1494 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1496 def test_ns_update_check_juju_bundle_existence_empty_vnfd(self
):
1497 """Check juju bundle existence"""
1499 expected_result
= None
1500 result
= check_juju_bundle_existence(test_vnfd1
)
1501 self
.assertEqual(result
, expected_result
, "Wrong juju bundle name")
1503 def test_ns_update_check_juju_bundle_existence_invalid_vnfd(self
):
1504 """Check juju bundle existence"""
1505 test_vnfd1
= [{"_id": vnfd_id
}]
1506 with self
.assertRaises(AttributeError):
1507 check_juju_bundle_existence(test_vnfd1
)
1509 def test_ns_update_check_juju_charm_artifacts_base_folder_wth_pkgdir(self
):
1510 """Check charm artifacts"""
1511 base_folder
= {"folder": vnfd_id
, "pkg-dir": "hackfest_3charmed_vnfd"}
1512 charm_name
= "simple"
1513 charm_type
= "lxc_proxy_charm"
1515 expected_result
= f
"{vnfd_id}:3/hackfest_3charmed_vnfd/charms/simple"
1516 result
= get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
)
1517 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1519 def test_ns_update_check_juju_charm_artifacts_base_folder_wthout_pkgdir(self
):
1520 """Check charm artifacts, SOL004 packages"""
1521 base_folder
= {"folder": vnfd_id
}
1522 charm_name
= "basic"
1523 charm_type
, revision
= "", ""
1524 expected_result
= f
"{vnfd_id}/Scripts/helm-charts/basic"
1525 result
= get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
)
1526 self
.assertEqual(result
, expected_result
, "Wrong charm artifact path")
1529 class TestInstantiateN2VC(TestBaseNS
):
1530 async def setUp(self
):
1531 await super().setUp()
1532 self
.db_nsr
= yaml
.safe_load(descriptors
.db_nsrs_text
)[0]
1533 self
.db_vnfr
= yaml
.safe_load(descriptors
.db_vnfrs_text
)[0]
1535 self
.my_ns
._write
_configuration
_status
= Mock()
1537 async def call_instantiate_N2VC(self
):
1538 logging_text
= "N2VC Instantiation"
1539 config_descriptor
= {"config-access": {"ssh-access": {"default-user": "admin"}}}
1540 base_folder
= {"pkg-dir": "", "folder": "~"}
1541 stage
= ["Stage", "Message"]
1543 await self
.my_ns
.instantiate_N2VC(
1544 logging_text
=logging_text
,
1545 vca_index
=self
.vca_index
,
1548 db_vnfr
=self
.db_vnfr
,
1553 config_descriptor
=config_descriptor
,
1555 base_folder
=base_folder
,
1556 nslcmop_id
="nslcmop_id",
1558 vca_type
="native_charm",
1559 vca_name
="vca_name",
1560 ee_config_descriptor
={},
1563 def check_config_status(self
, expected_status
):
1564 self
.my_ns
._write
_configuration
_status
.assert_called_with(
1565 nsr_id
=self
.db_nsr
["_id"], vca_index
=self
.vca_index
, status
=expected_status
1568 async def call_ns_add_relation(self
):
1569 ee_relation
= EERelation(
1571 "nsr-id": self
.db_nsr
["_id"],
1572 "vdu-profile-id": None,
1573 "kdu-resource-profile-id": None,
1574 "vnf-profile-id": "hackfest_vnf1",
1575 "execution-environment-ref": "f48163a6-c807-47bc-9682-f72caef5af85.alf-c-ab",
1576 "endpoint": "127.0.0.1",
1580 relation
= Relation("relation-name", ee_relation
, ee_relation
)
1581 cached_vnfrs
= {"hackfest_vnf1": self
.db_vnfr
}
1583 return await self
.my_ns
._add
_relation
(
1585 vca_type
="native_charm",
1588 cached_vnfrs
=cached_vnfrs
,
1591 async def test_add_relation_ok(self
):
1592 await self
.call_instantiate_N2VC()
1593 self
.check_config_status(expected_status
="READY")
1595 async def test_add_relation_returns_false_raises_exception(self
):
1596 self
.my_ns
._add
_vca
_relations
= asynctest
.CoroutineMock(return_value
=False)
1598 with self
.assertRaises(LcmException
) as exception
:
1599 await self
.call_instantiate_N2VC()
1601 exception_msg
= "Relations could not be added to VCA."
1602 self
.assertTrue(exception_msg
in str(exception
.exception
))
1603 self
.check_config_status(expected_status
="BROKEN")
1605 async def test_add_relation_raises_lcm_exception(self
):
1606 exception_msg
= "Relations FAILED"
1607 self
.my_ns
._add
_vca
_relations
= asynctest
.CoroutineMock(
1608 side_effect
=LcmException(exception_msg
)
1611 with self
.assertRaises(LcmException
) as exception
:
1612 await self
.call_instantiate_N2VC()
1614 self
.assertTrue(exception_msg
in str(exception
.exception
))
1615 self
.check_config_status(expected_status
="BROKEN")
1617 async def test_n2vc_add_relation_fails_raises_exception(self
):
1618 exception_msg
= "N2VC failed to add relations"
1619 self
.my_ns
.n2vc
.add_relation
= asynctest
.CoroutineMock(
1620 side_effect
=N2VCException(exception_msg
)
1622 with self
.assertRaises(LcmException
) as exception
:
1623 await self
.call_ns_add_relation()
1624 self
.assertTrue(exception_msg
in str(exception
.exception
))
1626 async def test_n2vc_add_relation_ok_returns_true(self
):
1627 self
.my_ns
.n2vc
.add_relation
= asynctest
.CoroutineMock(return_value
=None)
1628 self
.assertTrue(await self
.call_ns_add_relation())
1631 class TestGetVNFRelations(TestBaseNS
):
1632 async def setUp(self
):
1633 await super().setUp()
1634 self
.db_nsd
= yaml
.safe_load(descriptors
.db_nsds_text
)[0]
1636 def test_ns_charm_vca_returns_empty_relations(self
):
1637 ns_charm_vca
= {"member-vnf-index": None, "target_element": "ns"}
1638 nsr_id
= self
.db_nsd
["id"]
1639 deployed_vca
= DeployedVCA(nsr_id
, ns_charm_vca
)
1641 expected_relations
= []
1644 self
.my_ns
._get
_vnf
_relations
(
1645 nsr_id
=nsr_id
, nsd
=self
.db_nsd
, vca
=deployed_vca
, cached_vnfds
={}
1649 def test_vnf_returns_relation(self
):
1651 "member-vnf-index": "1",
1652 "target_element": "vnf/0",
1653 "ee_descriptor_id": "simple-ee",
1656 nsr_id
= self
.db_nsd
["id"]
1657 deployed_vca
= DeployedVCA(nsr_id
, vnf_vca
)
1661 "vnf-profile-id": "1",
1662 "vdu-profile-id": "mgmtVM",
1663 "kdu-resource-profile-id": None,
1664 "execution-environment-ref": "simple-ee",
1665 "endpoint": "interface",
1670 "vnf-profile-id": "1",
1671 "vdu-profile-id": "dataVM",
1672 "kdu-resource-profile-id": None,
1673 "execution-environment-ref": "simple-ee",
1674 "endpoint": "interface",
1677 provider
= EERelation(provider_dict
)
1678 requirer
= EERelation(requirer_dict
)
1679 relation
= Relation("relation", provider
, requirer
)
1681 relations_found
= self
.my_ns
._get
_vnf
_relations
(
1682 nsr_id
=nsr_id
, nsd
=self
.db_nsd
, vca
=deployed_vca
, cached_vnfds
={}
1685 self
.assertEqual(1, len(relations_found
))
1686 self
.assertEqual(relation
, relations_found
[0])
1689 if __name__
== "__main__":