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
23 from osm_lcm
import ns
24 from osm_common
.dbmemory
import DbMemory
25 from osm_common
.msgkafka
import MsgKafka
26 from osm_common
.fslocal
import FsLocal
27 from osm_lcm
.lcm_utils
import TaskRegistry
28 from osm_lcm
.ROclient
import ROClient
29 from uuid
import uuid4
30 # from asynctest.mock import patch
32 from osm_lcm
.tests
import test_db_descriptors
as descriptors
34 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
36 """ Perform unittests using asynctest of osm_lcm.ns module
37 It allows, if some testing ENV are supplied, testing without mocking some external libraries for debugging:
38 OSMLCMTEST_NS_PUBKEY: public ssh-key returned by N2VC to inject to VMs
39 OSMLCMTEST_NS_NAME: change name of NS
40 OSMLCMTEST_PACKAGES_PATH: path where the vnf-packages are stored (de-compressed), each one on a 'vnfd_id' folder
41 OSMLCMTEST_NS_IPADDRESS: IP address where emulated VMs are reached. Comma separate list
42 OSMLCMTEST_RO_VIMID: VIM id of RO target vim IP. Obtain it with openmano datcenter-list on RO container
43 OSMLCMTEST_VCA_NOMOCK: Do no mock the VCA, N2VC library, for debugging it
44 OSMLCMTEST_RO_NOMOCK: Do no mock the ROClient library, for debugging it
45 OSMLCMTEST_DB_NOMOCK: Do no mock the database library, for debugging it
46 OSMLCMTEST_FS_NOMOCK: Do no mock the File Storage library, for debugging it
47 OSMLCMTEST_LOGGING_NOMOCK: Do no mock the logging
48 OSMLCM_VCA_XXX: configuration of N2VC
49 OSMLCM_RO_XXX: configuration of RO
54 "VCA": { # TODO replace with os.get_env to get other configurations
55 "host": getenv("OSMLCM_VCA_HOST", "vca"),
56 "port": getenv("OSMLCM_VCA_PORT", 17070),
57 "user": getenv("OSMLCM_VCA_USER", "admin"),
58 "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
59 "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
60 'ca_cert': getenv("OSMLCM_VCA_CACERT", None),
61 'apiproxy': getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1")
64 "uri": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"),
65 getenv("OSMLCM_RO_PORT", "9090")),
66 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
67 "logger_name": "lcm.ROclient",
73 class TestMyNS(asynctest
.TestCase
):
75 async def _n2vc_DeployCharms(self
, model_name
, application_name
, vnfd
, charm_path
, params
={}, machine_spec
={},
76 callback
=None, *callback_args
):
78 for status
, message
in (("maintenance", "installing sofwware"), ("active", "Ready!")):
79 # call callback after some time
80 asyncio
.sleep(5, loop
=self
.loop
)
81 callback(model_name
, application_name
, status
, message
, *callback_args
)
84 def _n2vc_FormatApplicationName(*args
):
87 yield "app_name-{}".format(num_calls
)
90 def _n2vc_CreateExecutionEnvironment(self
, namespace
, reuse_ee_id
, db_dict
, *args
, **kwargs
):
91 k_list
= namespace
.split(".")
92 ee_id
= k_list
[1] + "."
100 def _ro_show(self
, *args
, **kwargs
):
101 if kwargs
.get("delete"):
102 ro_ns_desc
= yaml
.load(descriptors
.ro_delete_action_text
, Loader
=yaml
.Loader
)
106 ro_ns_desc
= yaml
.load(descriptors
.ro_ns_text
, Loader
=yaml
.Loader
)
108 # if ip address provided, replace descriptor
109 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
111 ip_addresses_list
= ip_addresses
.split(",")
112 for vnf
in ro_ns_desc
["vnfs"]:
113 if not ip_addresses_list
:
115 vnf
["ip_address"] = ip_addresses_list
[0]
116 for vm
in vnf
["vms"]:
117 if not ip_addresses_list
:
119 vm
["ip_address"] = ip_addresses_list
.pop(0)
123 for net
in ro_ns_desc
["nets"]:
124 if net
["status"] != "ACTIVE":
125 net
["status"] = "ACTIVE"
128 for vnf
in ro_ns_desc
["vnfs"]:
129 for vm
in vnf
["vms"]:
130 if vm
["status"] != "ACTIVE":
131 vm
["status"] = "ACTIVE"
134 def _ro_create(self
, *args
, **kwargs
):
136 yield {"uuid": str(uuid4())}
138 def _return_uuid(self
, *args
, **kwargs
):
141 async def setUp(self
):
144 if not getenv("OSMLCMTEST_DB_NOMOCK"):
146 self
.db
.create_list("vnfds", yaml
.load(descriptors
.db_vnfds_text
, Loader
=yaml
.Loader
))
147 self
.db
.create_list("nsds", yaml
.load(descriptors
.db_nsds_text
, Loader
=yaml
.Loader
))
148 self
.db
.create_list("nsrs", yaml
.load(descriptors
.db_nsrs_text
, Loader
=yaml
.Loader
))
149 self
.db
.create_list("vim_accounts", yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
))
150 self
.db
.create_list("k8sclusters", yaml
.load(descriptors
.db_k8sclusters_text
, Loader
=yaml
.Loader
))
151 self
.db
.create_list("nslcmops", yaml
.load(descriptors
.db_nslcmops_text
, Loader
=yaml
.Loader
))
152 self
.db
.create_list("vnfrs", yaml
.load(descriptors
.db_vnfrs_text
, Loader
=yaml
.Loader
))
153 self
.db_vim_accounts
= yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
)
156 self
.msg
= asynctest
.Mock(MsgKafka())
159 if not getenv("OSMLCMTEST_FS_NOMOCK"):
160 self
.fs
= asynctest
.Mock(FsLocal())
161 self
.fs
.get_params
.return_value
= {"path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")}
162 self
.fs
.file_open
= asynctest
.mock_open()
163 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
164 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
167 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
168 self
.lcm_tasks
.lock_HA
.return_value
= True
169 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
170 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
173 if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
174 ns
.K8sJujuConnector
= asynctest
.MagicMock(ns
.K8sJujuConnector
)
175 ns
.K8sHelmConnector
= asynctest
.MagicMock(ns
.K8sHelmConnector
)
177 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
178 ns
.N2VCJujuConnector
= asynctest
.MagicMock(ns
.N2VCJujuConnector
)
179 ns
.N2VCJujuConnectorLCM
= asynctest
.MagicMock(ns
.N2VCJujuConnectorLCM
)
180 ns
.LCMHelmConn
= asynctest
.MagicMock(ns
.LCMHelmConn
)
183 self
.my_ns
= ns
.NsLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, lcm_config
, self
.loop
)
184 self
.my_ns
._wait
_dependent
_n
2vc
= asynctest
.CoroutineMock()
187 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
188 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
191 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
192 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
193 # self.my_ns.n2vc = asynctest.Mock(N2VC())
194 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv("OSMLCM_VCA_PUBKEY", "public_key")
195 # allow several versions of n2vc
196 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(side_effect
=self
._n
2vc
_FormatApplicationName
())
197 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(side_effect
=self
._n
2vc
_DeployCharms
)
198 self
.my_ns
.n2vc
.create_execution_environment
= asynctest
.CoroutineMock(
199 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
)
200 self
.my_ns
.n2vc
.install_configuration_sw
= asynctest
.CoroutineMock(return_value
=pub_key
)
201 self
.my_ns
.n2vc
.get_ee_ssh_public__key
= asynctest
.CoroutineMock(return_value
=pub_key
)
202 self
.my_ns
.n2vc
.exec_primitive
= asynctest
.CoroutineMock(side_effect
=self
._return
_uuid
)
203 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(return_value
="completed")
204 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(return_value
={"result": "ok",
206 self
.my_ns
.n2vc
.delete_execution_environment
= asynctest
.CoroutineMock(return_value
=None)
207 self
.my_ns
.n2vc
.get_public_key
= asynctest
.CoroutineMock(
208 return_value
=getenv("OSMLCM_VCA_PUBKEY", "public_key"))
209 self
.my_ns
.n2vc
.delete_namespace
= asynctest
.CoroutineMock(return_value
=None)
212 if not getenv("OSMLCMTEST_RO_NOMOCK"):
213 # self.my_ns.RO = asynctest.Mock(ROclient.ROClient(self.loop, **lcm_config["ro_config"]))
214 # TODO first time should be empty list, following should return a dict
215 self
.my_ns
.RO
.get_list
= asynctest
.CoroutineMock(self
.my_ns
.RO
.get_list
, return_value
=[])
216 self
.my_ns
.RO
.create
= asynctest
.CoroutineMock(self
.my_ns
.RO
.create
, side_effect
=self
._ro
_create
())
217 self
.my_ns
.RO
.show
= asynctest
.CoroutineMock(self
.my_ns
.RO
.show
, side_effect
=self
._ro
_show
())
218 self
.my_ns
.RO
.create_action
= asynctest
.CoroutineMock(self
.my_ns
.RO
.create_action
,
219 return_value
={"vm-id": {"vim_result": 200,
220 "description": "done"}})
221 self
.my_ns
.RO
.delete
= asynctest
.CoroutineMock(self
.my_ns
.RO
.delete
, return_value
={"action_id": "del"})
222 # self.my_ns.wait_vm_up_insert_key_ro = asynctest.CoroutineMock(return_value="ip-address")
224 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
225 async def test_instantiate(self
):
226 self
.db
.set_one
= asynctest
.Mock()
227 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
228 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
229 # print("Test instantiate started")
231 # delete deployed information of database
232 if not getenv("OSMLCMTEST_DB_NOMOCK"):
233 if self
.db
.get_list("nsrs")[0]["_admin"].get("deployed"):
234 del self
.db
.get_list("nsrs")[0]["_admin"]["deployed"]
235 for db_vnfr
in self
.db
.get_list("vnfrs"):
236 db_vnfr
.pop("ip_address", None)
237 for db_vdur
in db_vnfr
["vdur"]:
238 db_vdur
.pop("ip_address", None)
239 db_vdur
.pop("mac_address", None)
240 if getenv("OSMLCMTEST_RO_VIMID"):
241 self
.db
.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
242 if getenv("OSMLCMTEST_RO_VIMID"):
243 self
.db
.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
245 await self
.my_ns
.instantiate(nsr_id
, nslcmop_id
)
247 # print("instantiate_result: {}".format(self.db.get_one("nslcmops",
248 # {"_id": nslcmop_id}).get("detailed-status")))
250 self
.msg
.aiowrite
.assert_called_once_with("ns", "instantiated",
251 {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
252 "operationState": "COMPLETED"},
254 self
.lcm_tasks
.lock_HA
.assert_called_once_with('ns', 'nslcmops', nslcmop_id
)
255 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
256 self
.assertTrue(self
.my_ns
.logger
.debug
.called
, "Debug method not called")
257 self
.my_ns
.logger
.error
.assert_not_called()
258 self
.my_ns
.logger
.exception().assert_not_called()
260 if not getenv("OSMLCMTEST_DB_NOMOCK"):
261 self
.assertTrue(self
.db
.set_one
.called
, "db.set_one not called")
262 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
263 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
264 self
.assertEqual(db_nsr
["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
265 for vnfr
in db_vnfrs_list
:
266 self
.assertEqual(vnfr
["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated")
268 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
269 # check intial-primitives called
270 self
.assertTrue(self
.my_ns
.n2vc
.exec_primitive
.called
,
271 "Exec primitive not called for initial config primitive")
272 for _call
in self
.my_ns
.n2vc
.exec_primitive
.call_args_list
:
273 self
.assertIn(_call
[1]["primitive_name"], ("config", "touch"),
274 "called exec primitive with a primitive different than config or touch")
276 # TODO add more checks of called methods
277 # TODO add a terminate
279 async def test_instantiate_ee_list(self
):
280 # Using modern IM where configuration is in the new format of execution_environment_list
281 ee_descriptor_id
= "charm_simple"
282 non_used_initial_primitive
= {
283 "name": "not_to_be_called",
285 "execution-environment-ref": "not_used_ee"
289 "id": ee_descriptor_id
,
290 "juju": {"charm": "simple"},
297 q_filter
={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"},
298 update_dict
={"vnf-configuration.execution-environment-list": ee_list
,
299 "vnf-configuration.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id
,
300 "vnf-configuration.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id
,
301 "vnf-configuration.initial-config-primitive.2": non_used_initial_primitive
,
302 "vnf-configuration.config-primitive.0.execution-environment-ref": ee_descriptor_id
,
303 "vnf-configuration.config-primitive.0.execution-environment-primitive": "touch_charm",
305 unset
={"vnf-configuration.juju": None})
306 await self
.test_instantiate()
307 # this will check that the initial-congig-primitive 'not_to_be_called' is not called
309 def test_ns_params_2_RO(self
):
310 vims
= self
.db
.get_list("vim_accounts")
311 vim_id
= vims
[0]["_id"]
312 ro_vim_id
= vims
[0]["_admin"]["deployed"]["RO"]
313 vim_id2
= vims
[1]["_id"]
314 ro_vim_id2
= vims
[1]["_admin"]["deployed"]["RO"]
316 ns_params
= {"vimAccountId": vim_id
}
317 mgmt_interface
= {"cp": "cp"}
318 vdu
= [{"id": "vdu_id", "interface": [{"external-connection-point-ref": "cp"}]}]
320 "1": {"vdu": vdu
, "mgmt-interface": mgmt_interface
},
321 "2": {"vdu": vdu
, "mgmt-interface": mgmt_interface
, "vnf-configuration": None},
322 "3": {"vdu": vdu
, "mgmt-interface": mgmt_interface
, "vnf-configuration": {"config-access": None}},
323 "4": {"vdu": vdu
, "mgmt-interface": mgmt_interface
,
324 "vnf-configuration": {"config-access": {"ssh-access": None}}},
325 "5": {"vdu": vdu
, "mgmt-interface": mgmt_interface
,
326 "vnf-configuration": {"config-access": {"ssh-access": {"required": True, "default_user": "U"}}}},
328 nsd
= {"constituent-vnfd": []}
330 for k
in vnfd_dict
.keys():
331 nsd
["constituent-vnfd"].append({"vnfd-id-ref": k
, "member-vnf-index": "member " + k
})
332 db_vnfrs
["member " + k
] = {"vim-account-id": vim_id2
if k
== "1" else vim_id
}
334 n2vc_key_list
= ["key"]
335 ro_ns_params
= self
.my_ns
._ns
_params
_2_RO
(ns_params
, nsd
, vnfd_dict
, db_vnfrs
, n2vc_key_list
)
336 ro_params_expected
= {'wim_account': None, "datacenter": ro_vim_id
,
338 "member 5": {"vdus": {"vdu_id": {"mgmt_keys": n2vc_key_list
}}},
339 "member 1": {"datacenter": ro_vim_id2
}
341 self
.assertEqual(ro_ns_params
, ro_params_expected
)
343 # Test scale() and related methods
344 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
345 async def test_scale(self
):
346 # print("Test scale started")
348 # TODO: Add more higher-lever tests here, for example:
349 # scale-out/scale-in operations with success/error result
351 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
352 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
353 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
354 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
355 expected_value
= 'FAILED'
356 return_value
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
}).get("operationState")
357 self
.assertEqual(return_value
, expected_value
)
358 # print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
360 async def test_vca_status_refresh(self
):
361 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
362 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
363 await self
.my_ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
364 for vnf_descriptor
in self
.db
.get_list("vnfds"):
365 expected_value
= dict()
366 return_value
= dict()
367 if vnf_descriptor
.get("vnf-configuration"):
368 if "juju" in vnf_descriptor
["vnf-configuration"]:
369 expected_value
= self
.db
.get_list("nsrs")[0]["vcaStatus"]
370 await self
.my_ns
._on
_update
_n
2vc
_db
("nsrs", {"_id": nsr_id
}, "_admin.deployed.VCA.0", {})
371 return_value
= self
.db
.get_list("nsrs")[0]["vcaStatus"]
372 self
.assertEqual(return_value
, expected_value
)
374 # Test _retry_or_skip_suboperation()
376 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
377 # - if marked as anything but 'COMPLETED', the suboperation index is expected
378 def test_scale_retry_or_skip_suboperation(self
):
379 # Load an alternative 'nslcmops' YAML for this test
380 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
381 db_nslcmop
= self
.db
.get_one('nslcmops', {"_id": nslcmop_id
})
383 # Test when 'operationState' is 'COMPLETED'
384 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = 'COMPLETED'
385 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
386 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
387 self
.assertEqual(return_value
, expected_value
)
388 # Test when 'operationState' is not 'COMPLETED'
389 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = None
390 return_value
= self
.my_ns
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
391 expected_value
= op_index
392 self
.assertEqual(return_value
, expected_value
)
394 # Test _find_suboperation()
395 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
396 def test_scale_find_suboperation(self
):
397 # Load an alternative 'nslcmops' YAML for this test
398 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
399 db_nslcmop
= self
.db
.get_one('nslcmops', {"_id": nslcmop_id
})
400 # Find this sub-operation
402 vnf_index
= db_nslcmop
['_admin']['operations'][op_index
]['member_vnf_index']
403 primitive
= db_nslcmop
['_admin']['operations'][op_index
]['primitive']
404 primitive_params
= db_nslcmop
['_admin']['operations'][op_index
]['primitive_params']
406 'member_vnf_index': vnf_index
,
407 'primitive': primitive
,
408 'primitive_params': primitive_params
,
410 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
411 self
.assertEqual(found_op_index
, op_index
)
412 # Test with not-matching params
414 'member_vnf_index': vnf_index
,
416 'primitive_params': primitive_params
,
418 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
419 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
422 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
423 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
425 # Test _update_suboperation_status()
426 def test_scale_update_suboperation_status(self
):
427 self
.db
.set_one
= asynctest
.Mock()
428 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
429 db_nslcmop
= self
.db
.get_one('nslcmops', {"_id": nslcmop_id
})
431 # Force the initial values to be distinct from the updated ones
432 q_filter
= {"_id": db_nslcmop
["_id"]}
433 # Test to change 'operationState' and 'detailed-status'
434 operationState
= 'COMPLETED'
435 detailed_status
= 'Done'
436 expected_update_dict
= {'_admin.operations.0.operationState': operationState
,
437 '_admin.operations.0.detailed-status': detailed_status
,
439 self
.my_ns
._update
_suboperation
_status
(db_nslcmop
, op_index
, operationState
, detailed_status
)
440 self
.db
.set_one
.assert_called_once_with("nslcmops", q_filter
=q_filter
, update_dict
=expected_update_dict
,
443 def test_scale_add_suboperation(self
):
444 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
445 db_nslcmop
= self
.db
.get_one('nslcmops', {"_id": nslcmop_id
})
447 num_ops_before
= len(db_nslcmop
.get('_admin', {}).get('operations', [])) - 1
449 vdu_count_index
= None
452 mapped_primitive_params
= {'parameter':
453 [{'data-type': 'STRING',
455 'default-value': '<touch_filename2>'}],
457 operationState
= 'PROCESSING'
458 detailed_status
= 'In progress'
459 operationType
= 'PRE-SCALE'
460 # Add a 'pre-scale' suboperation
461 op_index_after
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
462 vdu_name
, primitive
, mapped_primitive_params
,
463 operationState
, detailed_status
, operationType
)
464 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
466 # Delete all suboperations and add the same operation again
467 del db_nslcmop
['_admin']['operations']
468 op_index_zero
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
469 vdu_name
, primitive
, mapped_primitive_params
,
470 operationState
, detailed_status
, operationType
)
471 self
.assertEqual(op_index_zero
, 0)
473 # Add a 'RO' suboperation
474 RO_nsr_id
= '1234567890'
475 RO_scaling_info
= [{'type': 'create', 'count': 1, 'member-vnf-index': '1', 'osm_vdu_id': 'dataVM'}]
476 op_index
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
477 vdu_name
, primitive
, mapped_primitive_params
,
478 operationState
, detailed_status
, operationType
,
479 RO_nsr_id
, RO_scaling_info
)
480 db_RO_nsr_id
= db_nslcmop
['_admin']['operations'][op_index
]['RO_nsr_id']
481 self
.assertEqual(op_index
, 1)
482 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
484 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
485 op_index_invalid
= self
.my_ns
._add
_suboperation
(None, None, None, None, None,
488 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
490 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
491 # check the possible return values:
492 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
493 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
494 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
495 def test_scale_check_or_add_scale_suboperation(self
):
496 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
497 db_nslcmop
= self
.db
.get_one('nslcmops', {"_id": nslcmop_id
})
498 operationType
= 'PRE-SCALE'
501 primitive_params
= {'parameter':
502 [{'data-type': 'STRING',
504 'default-value': '<touch_filename2>'}],
507 # Delete all sub-operations to be sure this is a new sub-operation
508 del db_nslcmop
['_admin']['operations']
510 # Add a new sub-operation
511 # For new sub-operations, operationState is set to 'PROCESSING' by default
512 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
513 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
514 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
516 # Use the same parameters again to match the already added sub-operation
517 # which has status 'PROCESSING' (!= 'COMPLETED') by default
518 # The expected return value is a non-negative number
519 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
520 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
521 self
.assertTrue(op_index_existing
>= 0)
523 # Change operationState 'manually' for this sub-operation
524 db_nslcmop
['_admin']['operations'][op_index_existing
]['operationState'] = 'COMPLETED'
525 # Then use the same parameters again to match the already added sub-operation,
526 # which now has status 'COMPLETED'
527 # The expected return value is SUBOPERATION_STATUS_SKIP
528 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
529 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
530 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
532 # RO sub-operation test:
533 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
534 RO_nsr_id
= '1234567890'
535 RO_scaling_info
= [{'type': 'create', 'count': 1, 'member-vnf-index': '1', 'osm_vdu_id': 'dataVM'}]
536 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
537 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
538 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
540 # Use the same parameters again to match the already added RO sub-operation
541 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
542 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
543 self
.assertTrue(op_index_existing_RO
>= 0)
545 # Change operationState 'manually' for this RO sub-operation
546 db_nslcmop
['_admin']['operations'][op_index_existing_RO
]['operationState'] = 'COMPLETED'
547 # Then use the same parameters again to match the already added sub-operation,
548 # which now has status 'COMPLETED'
549 # The expected return value is SUBOPERATION_STATUS_SKIP
550 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
551 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
552 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
554 async def test_deploy_kdus(self
):
555 nsr_id
= descriptors
.test_ids
["TEST-KDU"]["ns"]
556 nslcmop_id
= descriptors
.test_ids
["TEST-KDU"]["instantiate"]
557 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
558 db_vnfr
= self
.db
.get_one("vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "multikdu"})
559 db_vnfrs
= {"multikdu": db_vnfr
}
560 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
561 db_vnfds
= {db_vnfd
["_id"]: db_vnfd
}
564 self
.my_ns
.k8sclusterhelm
.install
= asynctest
.CoroutineMock(return_value
="k8s_id")
565 self
.my_ns
.k8sclusterhelm
.synchronize_repos
= asynctest
.CoroutineMock(return_value
=("", ""))
566 self
.my_ns
.k8sclusterhelm
.get_services
= asynctest
.CoroutineMock(return_value
=([]))
567 await self
.my_ns
.deploy_kdus(logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_register
)
568 await asyncio
.wait(list(task_register
.keys()), timeout
=100)
569 db_nsr
= self
.db
.get_list("nsrs")[1]
570 self
.assertIn("K8s", db_nsr
["_admin"]["deployed"], "K8s entry not created at '_admin.deployed'")
571 self
.assertIsInstance(db_nsr
["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list")
572 self
.assertEqual(len(db_nsr
["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list")
573 k8s_instace_info
= {"kdu-instance": "k8s_id", "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
574 "k8scluster-type": "helm-chart",
575 "kdu-name": "ldap", "kdu-model": "stable/openldap:1.2.1",
576 "member-vnf-index": "multikdu", "namespace": None}
578 self
.assertEqual(db_nsr
["_admin"]["deployed"]["K8s"][0], k8s_instace_info
)
579 k8s_instace_info
["kdu-name"] = "mongo"
580 k8s_instace_info
["kdu-model"] = "stable/mongodb"
581 self
.assertEqual(db_nsr
["_admin"]["deployed"]["K8s"][1], k8s_instace_info
)
583 async def test_instantiate_pdu(self
):
584 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
585 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["instantiate"]
586 # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
587 self
.db
.set_one("vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "1"},
588 update_dict
={"ip-address": "10.205.1.46",
589 "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
590 "vdur.0.pdu-type": "PDU-TYPE-1",
591 "vdur.0.ip-address": "10.205.1.46",
593 unset
={"vdur.status": None})
594 self
.db
.set_one("vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "2"},
595 update_dict
={"ip-address": "10.205.1.47",
596 "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
597 "vdur.0.pdu-type": "PDU-TYPE-1",
598 "vdur.0.ip-address": "10.205.1.47",
600 unset
={"vdur.status": None})
602 await self
.my_ns
.instantiate(nsr_id
, nslcmop_id
)
603 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
604 self
.assertEqual(db_nsr
.get("nsState"), "READY", str(db_nsr
.get("errorDescription ")))
605 self
.assertEqual(db_nsr
.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
606 self
.assertEqual(db_nsr
.get("currentOperationID"), None, "currentOperationID different than None")
607 self
.assertEqual(db_nsr
.get("errorDescription "), None, "errorDescription different than None")
608 self
.assertEqual(db_nsr
.get("errorDetail"), None, "errorDetail different than None")
610 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
611 async def test_terminate_without_configuration(self
):
612 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
613 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["terminate"]
614 # set instantiation task as completed
615 self
.db
.set_list("nslcmops", {"nsInstanceId": nsr_id
, "_id.ne": nslcmop_id
},
616 update_dict
={"operationState": "COMPLETED"})
617 self
.my_ns
.RO
.show
= asynctest
.CoroutineMock(ROClient
.show
, side_effect
=self
._ro
_show
(delete
=nslcmop_id
))
618 self
.db
.set_one("nsrs", {"_id": nsr_id
},
619 update_dict
={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
621 await self
.my_ns
.terminate(nsr_id
, nslcmop_id
)
622 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
623 self
.assertEqual(db_nslcmop
.get("operationState"), 'COMPLETED', db_nslcmop
.get("detailed-status"))
624 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
625 self
.assertEqual(db_nsr
.get("nsState"), "NOT_INSTANTIATED", str(db_nsr
.get("errorDescription ")))
626 self
.assertEqual(db_nsr
["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr
.get("errorDescription ")))
627 self
.assertEqual(db_nsr
.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
628 self
.assertEqual(db_nsr
.get("currentOperationID"), None, "currentOperationID different than None")
629 self
.assertEqual(db_nsr
.get("errorDescription "), None, "errorDescription different than None")
630 self
.assertEqual(db_nsr
.get("errorDetail"), None, "errorDetail different than None")
631 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
632 for vnfr
in db_vnfrs_list
:
633 self
.assertEqual(vnfr
["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
635 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
636 async def test_terminate_primitive(self
):
637 nsr_id
= descriptors
.test_ids
["TEST-A"]["ns"]
638 nslcmop_id
= descriptors
.test_ids
["TEST-A"]["terminate"]
639 self
.my_ns
.RO
.show
= asynctest
.CoroutineMock(ROClient
.show
, side_effect
=self
._ro
_show
(delete
=nslcmop_id
))
640 # set instantiation task as completed
641 self
.db
.set_list("nslcmops", {"nsInstanceId": nsr_id
, "_id.ne": nslcmop_id
},
642 update_dict
={"operationState": "COMPLETED"})
644 # modify vnfd descriptor to include terminate_primitive
645 terminate_primitive
= [{
647 "parameter": [{"name": "filename", "value": "terminate_filename"}],
650 db_vnfr
= self
.db
.get_one("vnfrs", {"nsr-id-ref": nsr_id
, "member-vnf-index-ref": "1"})
651 self
.db
.set_one("vnfds", {"_id": db_vnfr
["vnfd-id"]},
652 {"vnf-configuration.terminate-config-primitive": terminate_primitive
})
654 await self
.my_ns
.terminate(nsr_id
, nslcmop_id
)
655 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
656 self
.assertEqual(db_nslcmop
.get("operationState"), 'COMPLETED', db_nslcmop
.get("detailed-status"))
657 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
658 self
.assertEqual(db_nsr
.get("nsState"), "NOT_INSTANTIATED", str(db_nsr
.get("errorDescription ")))
659 self
.assertEqual(db_nsr
["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr
.get("errorDescription ")))
660 self
.assertEqual(db_nsr
.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
661 self
.assertEqual(db_nsr
.get("currentOperationID"), None, "currentOperationID different than None")
662 self
.assertEqual(db_nsr
.get("errorDescription "), None, "errorDescription different than None")
663 self
.assertEqual(db_nsr
.get("errorDetail"), None, "errorDetail different than None")
666 if __name__
== '__main__':