2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
3 # not use this file except in compliance with the License. You may obtain
4 # a copy of the License at
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11 # License for the specific language governing permissions and limitations
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: alfonso.tiernosepulveda@telefonica.com
19 import asynctest
# pip3 install asynctest --user
24 from osm_lcm
.ns
import NsLcm
25 from osm_common
.dbmemory
import DbMemory
26 from osm_common
.msgkafka
import MsgKafka
27 from osm_common
.fslocal
import FsLocal
28 from osm_lcm
.lcm_utils
import TaskRegistry
29 from n2vc
.vnf
import N2VC
30 # from n2vc.k8s_helm_conn import K8sHelmConnector
31 from uuid
import uuid4
32 from asynctest
.mock
import patch
34 from osm_lcm
.tests
import test_db_descriptors
as descriptors
36 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
38 """ Perform unittests using asynctest of osm_lcm.ns module
39 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
40 OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
41 OSMLCMTEST_NS_NAME: change name of NS
42 OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
43 OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
44 OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
45 OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
46 OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
47 OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
48 OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
49 OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
50 OSMLCM_VCA_XXX: configuration of N2VC
51 OSMLCM_RO_XXX: configuration of RO
56 "VCA": { # TODO replace with os.get_env to get other configurations
57 "host": getenv("OSMLCM_VCA_HOST", "vca"),
58 "port": getenv("OSMLCM_VCA_PORT", 17070),
59 "user": getenv("OSMLCM_VCA_USER", "admin"),
60 "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
61 "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
62 'ca_cert': getenv("OSMLCM_VCA_CACERT", None)
65 "endpoint_url": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"),
66 getenv("OSMLCM_RO_PORT", "9090")),
67 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
68 "logger_name": "lcm.ROclient",
74 class TestMyNS(asynctest
.TestCase
):
76 async def _n2vc_DeployCharms(self
, model_name
, application_name
, vnfd
, charm_path
, params
={}, machine_spec
={},
77 callback
=None, *callback_args
):
79 for status
, message
in (("maintenance", "installing sofwware"), ("active", "Ready!")):
80 # call callback after some time
81 asyncio
.sleep(5, loop
=self
.loop
)
82 callback(model_name
, application_name
, status
, message
, *callback_args
)
85 def _n2vc_FormatApplicationName(*args
):
88 yield "app_name-{}".format(num_calls
)
91 def _n2vc_CreateExecutionEnvironment(self
, namespace
, reuse_ee_id
, db_dict
):
92 k_list
= namespace
.split(".")
93 ee_id
= k_list
[1] + "."
101 def _ro_show(self
, *args
, **kwargs
):
102 ro_ns_desc
= yaml
.load(descriptors
.db_ro_ns_text
, Loader
=yaml
.Loader
)
104 # if ip address provided, replace descriptor
105 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
107 ip_addresses_list
= ip_addresses
.split(",")
108 for vnf
in ro_ns_desc
["vnfs"]:
109 if not ip_addresses_list
:
111 vnf
["ip_address"] = ip_addresses_list
[0]
112 for vm
in vnf
["vms"]:
113 if not ip_addresses_list
:
115 vm
["ip_address"] = ip_addresses_list
.pop(0)
119 for net
in ro_ns_desc
["nets"]:
120 if net
["status"] != "ACTIVE":
121 net
["status"] = "ACTIVE"
124 for vnf
in ro_ns_desc
["vnfs"]:
125 for vm
in vnf
["vms"]:
126 if vm
["status"] != "ACTIVE":
127 vm
["status"] = "ACTIVE"
130 def _ro_create(self
, *args
, **kwargs
):
132 yield {"uuid": str(uuid4())}
134 def _return_uuid(self
, *args
, **kwargs
):
137 @patch("osm_lcm.ns.N2VCJujuConnector")
138 @patch("osm_lcm.ns.K8sHelmConnector")
139 async def setUp(self
, k8s_mock
, n2vc_mock
):
141 if not getenv("OSMLCMTEST_DB_NOMOCK"):
143 self
.db
.create_list("vnfds", yaml
.load(descriptors
.db_vnfds_text
, Loader
=yaml
.Loader
))
144 self
.db
.create_list("nsds", yaml
.load(descriptors
.db_nsds_text
, Loader
=yaml
.Loader
))
145 self
.db
.create_list("nsrs", yaml
.load(descriptors
.db_nsrs_text
, Loader
=yaml
.Loader
))
146 self
.db
.create_list("vim_accounts", yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
))
147 self
.db
.create_list("k8sclusters", yaml
.load(descriptors
.db_k8sclusters_text
, Loader
=yaml
.Loader
))
148 self
.db
.create_list("nslcmops", yaml
.load(descriptors
.db_nslcmops_text
, Loader
=yaml
.Loader
))
149 self
.db
.create_list("vnfrs", yaml
.load(descriptors
.db_vnfrs_text
, Loader
=yaml
.Loader
))
151 self
.db_vim_accounts
= yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
)
154 self
.msg
= asynctest
.Mock(MsgKafka())
157 if not getenv("OSMLCMTEST_FS_NOMOCK"):
158 self
.fs
= asynctest
.Mock(FsLocal())
159 self
.fs
.get_params
.return_value
= {"path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")}
160 self
.fs
.file_open
= asynctest
.mock_open()
161 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
162 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
165 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
166 self
.lcm_tasks
.lock_HA
.return_value
= True
167 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
168 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
171 self
.my_ns
= NsLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, lcm_config
, self
.loop
)
172 self
.my_ns
._wait
_dependent
_n
2vc
= asynctest
.CoroutineMock()
175 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
176 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
179 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
180 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
181 self
.my_ns
.n2vc
= asynctest
.Mock(N2VC())
182 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv("OSMLCM_VCA_PUBKEY", "public_key")
183 # allow several versions of n2vc
184 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(side_effect
=self
._n
2vc
_FormatApplicationName
())
185 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(side_effect
=self
._n
2vc
_DeployCharms
)
186 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
187 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
)
188 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(return_value
=pub_key
)
189 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(return_value
=pub_key
)
190 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(side_effect
=self
._return
_uuid
)
191 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(return_value
="completed")
192 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(return_value
={"result": "ok",
194 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
195 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key"))
198 # if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
199 # pub_key = getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
200 # self.my_ns.k8sclusterhelm = asynctest.Mock(K8sHelmConnector())
203 if not getenv("OSMLCMTEST_RO_NOMOCK"):
204 # self.my_ns.RO = asynctest.Mock(ROclient.ROClient(self.loop, **lcm_config["ro_config"]))
205 # TODO first time should be empty list, following should return a dict
206 self
.my_ns
.RO
.get_list
= asynctest
.CoroutineMock(self
.my_ns
.RO
.get_list
, return_value
=[])
207 self
.my_ns
.RO
.create
= asynctest
.CoroutineMock(self
.my_ns
.RO
.create
, side_effect
=self
._ro
_create
())
208 self
.my_ns
.RO
.show
= asynctest
.CoroutineMock(self
.my_ns
.RO
.show
, side_effect
=self
._ro
_show
())
209 self
.my_ns
.RO
.create_action
= asynctest
.CoroutineMock(self
.my_ns
.RO
.create_action
,
210 return_value
={"vm-id": {"vim_result": 200,
211 "description": "done"}})
212 # self.my_ns.wait_vm_up_insert_key_ro = asynctest.CoroutineMock(return_value="ip-address")
214 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
215 async def test_instantiate(self
):
216 self
.db
.set_one
= asynctest
.Mock()
217 nsr_id
= self
.db
.get_list("nsrs")[0]["_id"]
218 nslcmop_id
= self
.db
.get_list("nslcmops")[0]["_id"]
219 # print("Test instantiate started")
221 # delete deployed information of database
222 if not getenv("OSMLCMTEST_DB_NOMOCK"):
223 if self
.db
.get_list("nsrs")[0]["_admin"].get("deployed"):
224 del self
.db
.get_list("nsrs")[0]["_admin"]["deployed"]
225 for db_vnfr
in self
.db
.get_list("vnfrs"):
226 db_vnfr
.pop("ip_address", None)
227 for db_vdur
in db_vnfr
["vdur"]:
228 db_vdur
.pop("ip_address", None)
229 db_vdur
.pop("mac_address", None)
230 if getenv("OSMLCMTEST_RO_VIMID"):
231 self
.db
.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
232 if getenv("OSMLCMTEST_RO_VIMID"):
233 self
.db
.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
235 await self
.my_ns
.instantiate(nsr_id
, nslcmop_id
)
237 # print("instantiate_result: {}".format(self.db.get_one("nslcmops",
238 # {"_id": nslcmop_id}).get("detailed-status")))
240 self
.msg
.aiowrite
.assert_called_once_with("ns", "instantiated",
241 {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
242 "operationState": "COMPLETED"},
244 self
.lcm_tasks
.lock_HA
.assert_called_once_with('ns', 'nslcmops', nslcmop_id
)
245 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
246 self
.assertTrue(self
.my_ns
.logger
.debug
.called
, "Debug method not called")
247 self
.my_ns
.logger
.error
.assert_not_called()
248 self
.my_ns
.logger
.exception().assert_not_called()
250 if not getenv("OSMLCMTEST_DB_NOMOCK"):
251 self
.assertTrue(self
.db
.set_one
.called
, "db.set_one not called")
253 # TODO add more checks of called methods
254 # TODO add a terminate
256 def test_ns_params_2_RO(self
):
257 vim
= self
.db
.get_list("vim_accounts")[0]
259 ro_vim_id
= vim
["_admin"]["deployed"]["RO"]
260 ns_params
= {"vimAccountId": vim_id
}
261 mgmt_interface
= {"cp": "cp"}
262 vdu
= [{"id": "vdu_id", "interface": [{"external-connection-point-ref": "cp"}]}]
264 "1": {"vdu": vdu
, "mgmt-interface": mgmt_interface
},
265 "2": {"vdu": vdu
, "mgmt-interface": mgmt_interface
, "vnf-configuration": None},
266 "3": {"vdu": vdu
, "mgmt-interface": mgmt_interface
, "vnf-configuration": {"config-access": None}},
267 "4": {"vdu": vdu
, "mgmt-interface": mgmt_interface
,
268 "vnf-configuration": {"config-access": {"ssh-access": None}}},
269 "5": {"vdu": vdu
, "mgmt-interface": mgmt_interface
,
270 "vnf-configuration": {"config-access": {"ssh-access": {"required": True, "default_user": "U"}}}},
272 nsd
= {"constituent-vnfd": []}
273 for k
in vnfd_dict
.keys():
274 nsd
["constituent-vnfd"].append({"vnfd-id-ref": k
, "member-vnf-index": k
})
276 n2vc_key_list
= ["key"]
277 ro_ns_params
= self
.my_ns
.ns_params_2_RO(ns_params
, nsd
, vnfd_dict
, n2vc_key_list
)
278 ro_params_expected
= {'wim_account': None, "datacenter": ro_vim_id
,
279 "vnfs": {"5": {"vdus": {"vdu_id": {"mgmt_keys": n2vc_key_list
}}}}}
280 self
.assertEqual(ro_ns_params
, ro_params_expected
)
282 # Test scale() and related methods
283 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
284 async def test_scale(self
):
285 # print("Test scale started")
287 # TODO: Add more higher-lever tests here, for example:
288 # scale-out/scale-in operations with success/error result
290 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
291 nsr_id
= self
.db
.get_list("nsrs")[0]["_id"]
292 nslcmop_id
= self
.db
.get_list("nslcmops")[0]["_id"]
293 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
294 expected_value
= 'FAILED'
295 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get("operationState")
296 self
.assertEqual(return_value
, expected_value
)
297 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
299 # Test _reintent_or_skip_suboperation()
301 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
302 # - if marked as anything but 'COMPLETED', the suboperation index is expected
303 def test_scale_reintent_or_skip_suboperation(self
):
304 # Load an alternative 'nslcmops' YAML for this test
305 db_nslcmop
= self
.db
.get_list('nslcmops')[0]
307 # Test when 'operationState' is 'COMPLETED'
308 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = 'COMPLETED'
309 return_value
= self
.my_ns
._reintent
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
310 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
311 self
.assertEqual(return_value
, expected_value
)
312 # Test when 'operationState' is not 'COMPLETED'
313 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = None
314 return_value
= self
.my_ns
._reintent
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
315 expected_value
= op_index
316 self
.assertEqual(return_value
, expected_value
)
318 # Test _find_suboperation()
319 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
320 def test_scale_find_suboperation(self
):
321 # Load an alternative 'nslcmops' YAML for this test
322 db_nslcmop
= self
.db
.get_list('nslcmops')[0]
323 # Find this sub-operation
325 vnf_index
= db_nslcmop
['_admin']['operations'][op_index
]['member_vnf_index']
326 primitive
= db_nslcmop
['_admin']['operations'][op_index
]['primitive']
327 primitive_params
= db_nslcmop
['_admin']['operations'][op_index
]['primitive_params']
329 'member_vnf_index': vnf_index
,
330 'primitive': primitive
,
331 'primitive_params': primitive_params
,
333 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
334 self
.assertEqual(found_op_index
, op_index
)
335 # Test with not-matching params
337 'member_vnf_index': vnf_index
,
339 'primitive_params': primitive_params
,
341 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
342 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
345 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
346 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
348 # Test _update_suboperation_status()
349 def test_scale_update_suboperation_status(self
):
350 self
.db
.set_one
= asynctest
.Mock()
351 db_nslcmop
= self
.db
.get_list('nslcmops')[0]
353 # Force the initial values to be distinct from the updated ones
354 q_filter
= {"_id": db_nslcmop
["_id"]}
355 # Test to change 'operationState' and 'detailed-status'
356 operationState
= 'COMPLETED'
357 detailed_status
= 'Done'
358 expected_update_dict
= {'_admin.operations.0.operationState': operationState
,
359 '_admin.operations.0.detailed-status': detailed_status
,
361 self
.my_ns
._update
_suboperation
_status
(db_nslcmop
, op_index
, operationState
, detailed_status
)
362 self
.db
.set_one
.assert_called_once_with("nslcmops", q_filter
=q_filter
, update_dict
=expected_update_dict
,
365 def test_scale_add_suboperation(self
):
366 db_nslcmop
= self
.db
.get_list('nslcmops')[0]
368 num_ops_before
= len(db_nslcmop
.get('_admin', {}).get('operations', [])) - 1
370 vdu_count_index
= None
373 mapped_primitive_params
= {'parameter':
374 [{'data-type': 'STRING',
376 'default-value': '<touch_filename2>'}],
378 operationState
= 'PROCESSING'
379 detailed_status
= 'In progress'
380 operationType
= 'PRE-SCALE'
381 # Add a 'pre-scale' suboperation
382 op_index_after
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
383 vdu_name
, primitive
, mapped_primitive_params
,
384 operationState
, detailed_status
, operationType
)
385 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
387 # Delete all suboperations and add the same operation again
388 del db_nslcmop
['_admin']['operations']
389 op_index_zero
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
390 vdu_name
, primitive
, mapped_primitive_params
,
391 operationState
, detailed_status
, operationType
)
392 self
.assertEqual(op_index_zero
, 0)
394 # Add a 'RO' suboperation
395 RO_nsr_id
= '1234567890'
396 RO_scaling_info
= [{'type': 'create', 'count': 1, 'member-vnf-index': '1', 'osm_vdu_id': 'dataVM'}]
397 op_index
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
398 vdu_name
, primitive
, mapped_primitive_params
,
399 operationState
, detailed_status
, operationType
,
400 RO_nsr_id
, RO_scaling_info
)
401 db_RO_nsr_id
= db_nslcmop
['_admin']['operations'][op_index
]['RO_nsr_id']
402 self
.assertEqual(op_index
, 1)
403 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
405 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
406 op_index_invalid
= self
.my_ns
._add
_suboperation
(None, None, None, None, None,
409 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
411 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
412 # check the possible return values:
413 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
414 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
415 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
416 def test_scale_check_or_add_scale_suboperation(self
):
417 db_nslcmop
= self
.db
.get_list('nslcmops')[0]
418 operationType
= 'PRE-SCALE'
421 primitive_params
= {'parameter':
422 [{'data-type': 'STRING',
424 'default-value': '<touch_filename2>'}],
427 # Delete all sub-operations to be sure this is a new sub-operation
428 del db_nslcmop
['_admin']['operations']
430 # Add a new sub-operation
431 # For new sub-operations, operationState is set to 'PROCESSING' by default
432 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
433 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
434 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
436 # Use the same parameters again to match the already added sub-operation
437 # which has status 'PROCESSING' (!= 'COMPLETED') by default
438 # The expected return value is a non-negative number
439 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
440 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
441 self
.assertTrue(op_index_existing
>= 0)
443 # Change operationState 'manually' for this sub-operation
444 db_nslcmop
['_admin']['operations'][op_index_existing
]['operationState'] = 'COMPLETED'
445 # Then use the same parameters again to match the already added sub-operation,
446 # which now has status 'COMPLETED'
447 # The expected return value is SUBOPERATION_STATUS_SKIP
448 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
449 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
450 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
452 # RO sub-operation test:
453 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
454 RO_nsr_id
= '1234567890'
455 RO_scaling_info
= [{'type': 'create', 'count': 1, 'member-vnf-index': '1', 'osm_vdu_id': 'dataVM'}]
456 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
457 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
458 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
460 # Use the same parameters again to match the already added RO sub-operation
461 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
462 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
463 self
.assertTrue(op_index_existing_RO
>= 0)
465 # Change operationState 'manually' for this RO sub-operation
466 db_nslcmop
['_admin']['operations'][op_index_existing_RO
]['operationState'] = 'COMPLETED'
467 # Then use the same parameters again to match the already added sub-operation,
468 # which now has status 'COMPLETED'
469 # The expected return value is SUBOPERATION_STATUS_SKIP
470 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
471 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
472 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
474 async def test_deploy_kdus(self
):
475 db_nsr
= self
.db
.get_list("nsrs")[1]
476 db_vnfr
= self
.db
.get_list("vnfrs")[2]
477 db_vnfrs
= {"multikdu": db_vnfr
}
478 db_vnfd
= self
.db
.get_list("vnfds")[1]
479 db_vnfds
= {db_vnfd
["_id"]: db_vnfd
}
480 nsr_id
= db_nsr
["_id"]
481 # nslcmop_id = self.db.get_list("nslcmops")[1]["_id"]
483 self
.my_ns
.k8sclusterhelm
.install
= asynctest
.CoroutineMock(return_value
="k8s_id")
484 await self
.my_ns
.deploy_kdus(logging_text
, nsr_id
, db_nsr
, db_vnfrs
, db_vnfds
)
485 db_nsr
= self
.db
.get_list("nsrs")[1]
486 self
.assertIn("K8s", db_nsr
["_admin"]["deployed"], "K8s entry not created at '_admin.deployed'")
487 self
.assertIsInstance(db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list")
488 self
.assertEqual(len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list")
489 k8s_instace_info
= {"kdu-instance": "k8s_id", "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
490 "k8scluster-type": "chart",
491 "kdu-name": "ldap", "kdu-model": "stable/openldap:1.2.1"}
493 self
.assertEqual(db_nsr
["_admin"]["deployed"]["K8s"][0], k8s_instace_info
)
494 k8s_instace_info
["kdu-name"] = "mongo"
495 k8s_instace_info
["kdu-model"] = "stable/mongodb"
496 self
.assertEqual(db_nsr
["_admin"]["deployed"]["K8s"][1], k8s_instace_info
)
499 if __name__
== '__main__':