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: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
19 import asynctest
# pip3 install asynctest --user
24 from osm_lcm
.ns
import NsLcm
25 from osm_common
.dbmongo
import DbMongo
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 uuid
import uuid4
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
53 vca_config
= { # TODO replace with os.get_env to get other configurations
54 "host": getenv("OSMLCM_VCA_HOST", "vca"),
55 "port": getenv("OSMLCM_VCA_PORT", 17070),
56 "user": getenv("OSMLCM_VCA_USER", "admin"),
57 "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
58 "pubkey": getenv("OSMLCM_VCA_PUBKEY", None),
59 'cacert': getenv("OSMLCM_VCA_CACERT", None)
63 "endpoint_url": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")),
64 "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
65 "logger_name": "lcm.ROclient",
70 class TestMyNS(asynctest
.TestCase
):
72 def _db_get_one(self
, table
, q_filter
=None, fail_on_empty
=True, fail_on_more
=True):
73 if table
not in self
.db_content
:
74 self
.assertTrue(False, "db.get_one called with table={}".format(table
))
75 for db_item
in self
.db_content
[table
]:
76 if db_item
["_id"] == q_filter
["_id"]:
79 self
.assertTrue(False, "db.get_one, table={}, not found _id={}".format(table
, q_filter
["_id"]))
81 def _db_get_list(self
, table
, q_filter
=None):
82 if table
not in self
.db_content
:
83 self
.assertTrue(False, "db.get_list called with table={} not found".format(table
))
84 return self
.db_content
[table
]
86 def _db_set_one(self
, table
, q_filter
, update_dict
, fail_on_empty
=True, unset
=None, pull
=None, push
=None):
87 db_item
= self
._db
_get
_one
(table
, q_filter
, fail_on_empty
=fail_on_empty
)
88 for k
, v
in update_dict
.items():
91 for k_nested
in k_list
[0:-1]:
92 if isinstance(db_nested
, list):
93 db_nested
= db_nested
[int(k_nested
)]
95 if k_nested
not in db_nested
:
96 db_nested
[k_nested
] = {}
97 db_nested
= db_nested
[k_nested
]
99 if isinstance(db_nested
, list):
100 if int(k_nested
) < len(db_nested
):
101 db_nested
[int(k_nested
)] = v
103 db_nested
.insert(int(k_nested
), v
)
105 db_nested
[k_nested
] = v
107 async def _n2vc_DeployCharms(self
, model_name
, application_name
, vnfd
, charm_path
, params
={}, machine_spec
={},
108 callback
=None, *callback_args
):
110 for status
, message
in (("maintenance", "installing sofwware"), ("active", "Ready!")):
111 # call callback after some time
112 asyncio
.sleep(5, loop
=self
.loop
)
113 callback(model_name
, application_name
, status
, message
, *callback_args
)
116 def _n2vc_FormatApplicationName(*args
):
119 yield "app_name-{}".format(num_calls
)
122 def _n2vc_CreateExecutionEnvironment(self
, namespace
):
123 k_list
= namespace
.split(".")
124 ee_id
= k_list
[1] + "."
126 for k
in k_list
[2:4]:
132 def _ro_show(self
, *args
, **kwargs
):
133 ro_ns_desc
= yaml
.load(descriptors
.db_ro_ns_text
, Loader
=yaml
.Loader
)
135 # if ip address provided, replace descriptor
136 ip_addresses
= getenv("OSMLCMTEST_NS_IPADDRESS", "")
138 ip_addresses_list
= ip_addresses
.split(",")
139 for vnf
in ro_ns_desc
["vnfs"]:
140 if not ip_addresses_list
:
142 vnf
["ip_address"] = ip_addresses_list
[0]
143 for vm
in vnf
["vms"]:
144 if not ip_addresses_list
:
146 vm
["ip_address"] = ip_addresses_list
.pop(0)
150 for net
in ro_ns_desc
["nets"]:
151 if net
["status"] != "ACTIVE":
152 net
["status"] = "ACTIVE"
155 for vnf
in ro_ns_desc
["vnfs"]:
156 for vm
in vnf
["vms"]:
157 if vm
["status"] != "ACTIVE":
158 vm
["status"] = "ACTIVE"
161 def _ro_create(self
, *args
, **kwargs
):
163 yield {"uuid": str(uuid4())}
165 def _return_uuid(self
, *args
, **kwargs
):
168 async def setUp(self
):
170 if not getenv("OSMLCMTEST_DB_NOMOCK"):
171 self
.db
= asynctest
.Mock(DbMongo())
172 self
.db
.get_one
.side_effect
= self
._db
_get
_one
173 self
.db
.get_list
.side_effect
= self
._db
_get
_list
174 self
.db
.set_one
.side_effect
= self
._db
_set
_one
176 "nsrs": yaml
.load(descriptors
.db_nsrs_text
, Loader
=yaml
.Loader
),
177 "nslcmops": yaml
.load(descriptors
.db_nslcmops_text
, Loader
=yaml
.Loader
),
178 "vnfrs": yaml
.load(descriptors
.db_vnfrs_text
, Loader
=yaml
.Loader
),
179 "vnfds": yaml
.load(descriptors
.db_vnfds_text
, Loader
=yaml
.Loader
),
180 "vim_accounts": yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
),
182 self
.db_vim_accounts
= yaml
.load(descriptors
.db_vim_accounts_text
, Loader
=yaml
.Loader
)
185 self
.msg
= asynctest
.Mock(MsgKafka())
188 if not getenv("OSMLCMTEST_FS_NOMOCK"):
189 self
.fs
= asynctest
.Mock(FsLocal())
190 self
.fs
.get_params
.return_value
= {"path": getenv("OSMLCMTEST_PACKAGES_PATH", "./test/temp/packages")}
191 self
.fs
.file_open
= asynctest
.mock_open()
192 # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with"
193 # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file
196 self
.lcm_tasks
= asynctest
.Mock(TaskRegistry())
197 self
.lcm_tasks
.lock_HA
.return_value
= True
198 self
.lcm_tasks
.waitfor_related_HA
.return_value
= None
199 self
.lcm_tasks
.lookfor_related
.return_value
= ("", [])
202 self
.my_ns
= NsLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, ro_config
, vca_config
, self
.loop
)
205 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
206 self
.my_ns
.logger
= asynctest
.Mock(self
.my_ns
.logger
)
209 if not getenv("OSMLCMTEST_VCA_NOMOCK"):
210 pub_key
= getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com")
211 self
.my_ns
.n2vc
= asynctest
.Mock(N2VC())
212 self
.my_ns
.n2vc
.GetPublicKey
.return_value
= getenv("OSMLCM_VCA_PUBKEY", "public_key")
213 # allow several versions of n2vc
214 self
.my_ns
.n2vc
.FormatApplicationName
= asynctest
.Mock(side_effect
=self
._n
2vc
_FormatApplicationName
())
215 self
.my_ns
.n2vc
.DeployCharms
= asynctest
.CoroutineMock(side_effect
=self
._n
2vc
_DeployCharms
)
216 self
.my_ns
.n2vc
.CreateExecutionEnvironment
= asynctest
.CoroutineMock(
217 side_effect
=self
._n
2vc
_CreateExecutionEnvironment
)
218 self
.my_ns
.n2vc
.InstallConfigurationSW
= asynctest
.CoroutineMock(return_value
=pub_key
)
219 self
.my_ns
.n2vc
.ExecutePrimitive
= asynctest
.CoroutineMock(side_effect
=self
._return
_uuid
)
220 self
.my_ns
.n2vc
.GetPrimitiveStatus
= asynctest
.CoroutineMock(return_value
="completed")
221 self
.my_ns
.n2vc
.GetPrimitiveOutput
= asynctest
.CoroutineMock(return_value
={"result": "ok",
225 if not getenv("OSMLCMTEST_RO_NOMOCK"):
226 # self.my_ns.RO = asynctest.Mock(ROclient.ROClient(self.loop, **ro_config))
227 # TODO first time should be empty list, following should return a dict
228 self
.my_ns
.RO
.get_list
= asynctest
.CoroutineMock(self
.my_ns
.RO
.get_list
, return_value
=[])
229 self
.my_ns
.RO
.create
= asynctest
.CoroutineMock(self
.my_ns
.RO
.create
, side_effect
=self
._ro
_create
())
230 self
.my_ns
.RO
.show
= asynctest
.CoroutineMock(self
.my_ns
.RO
.show
, side_effect
=self
._ro
_show
())
231 self
.my_ns
.RO
.create_action
= asynctest
.CoroutineMock(self
.my_ns
.RO
.create_action
)
233 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
234 async def test_instantiate(self
):
235 nsr_id
= self
.db_content
["nsrs"][0]["_id"]
236 nslcmop_id
= self
.db_content
["nslcmops"][0]["_id"]
237 print("Test instantiate started")
239 # delete deployed information of database
240 if not getenv("OSMLCMTEST_DB_NOMOCK"):
241 if self
.db_content
["nsrs"][0]["_admin"].get("deployed"):
242 del self
.db_content
["nsrs"][0]["_admin"]["deployed"]
243 for db_vnfr
in self
.db_content
["vnfrs"]:
244 db_vnfr
.pop("ip_address", None)
245 for db_vdur
in db_vnfr
["vdur"]:
246 db_vdur
.pop("ip_address", None)
247 db_vdur
.pop("mac_address", None)
248 if getenv("OSMLCMTEST_RO_VIMID"):
249 self
.db_content
["vim_accounts"][0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
250 if getenv("OSMLCMTEST_RO_VIMID"):
251 self
.db_content
["nsrs"][0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID")
253 await self
.my_ns
.instantiate(nsr_id
, nslcmop_id
)
255 print("instantiate_result: {}".format(self
._db
_get
_one
("nslcmops", {"_id": nslcmop_id
}).get("detailed-status")))
257 self
.msg
.aiowrite
.assert_called_once_with("ns", "instantiated",
258 {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
259 "operationState": "COMPLETED"},
261 self
.lcm_tasks
.lock_HA
.assert_called_once_with('ns', 'nslcmops', nslcmop_id
)
262 if not getenv("OSMLCMTEST_LOGGING_NOMOCK"):
263 self
.assertTrue(self
.my_ns
.logger
.debug
.called
, "Debug method not called")
264 self
.my_ns
.logger
.error
.assert_not_called()
265 self
.my_ns
.logger
.exception().assert_not_called()
267 if not getenv("OSMLCMTEST_DB_NOMOCK"):
268 self
.assertTrue(self
.db
.set_one
.called
, "db.set_one not called")
270 # TODO add more checks of called methods
271 # TODO add a terminate
273 def test_ns_params_2_RO(self
):
274 vim
= self
._db
_get
_list
("vim_accounts")[0]
276 ro_vim_id
= vim
["_admin"]["deployed"]["RO"]
277 ns_params
= {"vimAccountId": vim_id
}
278 mgmt_interface
= {"cp": "cp"}
279 vdu
= [{"id": "vdu_id", "interface": [{"external-connection-point-ref": "cp"}]}]
281 "1": {"vdu": vdu
, "mgmt-interface": mgmt_interface
},
282 "2": {"vdu": vdu
, "mgmt-interface": mgmt_interface
, "vnf-configuration": None},
283 "3": {"vdu": vdu
, "mgmt-interface": mgmt_interface
, "vnf-configuration": {"config-access": None}},
284 "4": {"vdu": vdu
, "mgmt-interface": mgmt_interface
,
285 "vnf-configuration": {"config-access": {"ssh-access": None}}},
286 "5": {"vdu": vdu
, "mgmt-interface": mgmt_interface
,
287 "vnf-configuration": {"config-access": {"ssh-access": {"required": True, "default_user": "U"}}}},
289 nsd
= {"constituent-vnfd": []}
290 for k
in vnfd_dict
.keys():
291 nsd
["constituent-vnfd"].append({"vnfd-id-ref": k
, "member-vnf-index": k
})
293 n2vc_key_list
= ["key"]
294 ro_ns_params
= self
.my_ns
.ns_params_2_RO(ns_params
, nsd
, vnfd_dict
, n2vc_key_list
)
295 ro_params_expected
= {'wim_account': None, "datacenter": ro_vim_id
,
296 "vnfs": {"5": {"vdus": {"vdu_id": {"mgmt_keys": n2vc_key_list
}}}}}
297 self
.assertEqual(ro_ns_params
, ro_params_expected
)
299 # Test scale() and related methods
300 @asynctest.fail_on(active_handles
=True) # all async tasks must be completed
301 async def test_scale(self
):
302 # print("Test scale started")
304 # TODO: Add more higher-lever tests here, for example:
305 # scale-out/scale-in operations with success/error result
307 # Test scale() with missing 'scaleVnfData', should return operationState = 'FAILED'
308 nsr_id
= self
.db_content
["nsrs"][0]["_id"]
309 nslcmop_id
= self
.db_content
["nslcmops"][0]["_id"]
310 await self
.my_ns
.scale(nsr_id
, nslcmop_id
)
311 expected_value
= 'FAILED'
312 return_value
= self
._db
_get
_one
("nslcmops", {"_id": nslcmop_id
}).get("operationState")
313 self
.assertEqual(return_value
, expected_value
)
314 # print("scale_result: {}".format(self._db_get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
316 # Test _reintent_or_skip_suboperation()
318 # - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
319 # - if marked as anything but 'COMPLETED', the suboperation index is expected
320 def test_scale_reintent_or_skip_suboperation(self
):
321 # Load an alternative 'nslcmops' YAML for this test
322 self
.db_content
['nslcmops'] = yaml
.load(descriptors
.db_nslcmops_scale_text
, Loader
=yaml
.Loader
)
323 db_nslcmop
= self
.db_content
['nslcmops'][0]
325 # Test when 'operationState' is 'COMPLETED'
326 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = 'COMPLETED'
327 return_value
= self
.my_ns
._reintent
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
328 expected_value
= self
.my_ns
.SUBOPERATION_STATUS_SKIP
329 self
.assertEqual(return_value
, expected_value
)
330 # Test when 'operationState' is not 'COMPLETED'
331 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = None
332 return_value
= self
.my_ns
._reintent
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
333 expected_value
= op_index
334 self
.assertEqual(return_value
, expected_value
)
336 # Test _find_suboperation()
337 # Expected result: index of the found sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if not found
338 def test_scale_find_suboperation(self
):
339 # Load an alternative 'nslcmops' YAML for this test
340 self
.db_content
['nslcmops'] = yaml
.load(descriptors
.db_nslcmops_scale_text
, Loader
=yaml
.Loader
)
341 db_nslcmop
= self
.db_content
['nslcmops'][0]
342 # Find this sub-operation
344 vnf_index
= db_nslcmop
['_admin']['operations'][op_index
]['member_vnf_index']
345 primitive
= db_nslcmop
['_admin']['operations'][op_index
]['primitive']
346 primitive_params
= db_nslcmop
['_admin']['operations'][op_index
]['primitive_params']
348 'member_vnf_index': vnf_index
,
349 'primitive': primitive
,
350 'primitive_params': primitive_params
,
352 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
353 self
.assertEqual(found_op_index
, op_index
)
354 # Test with not-matching params
356 'member_vnf_index': vnf_index
,
358 'primitive_params': primitive_params
,
360 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
361 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
364 found_op_index
= self
.my_ns
._find
_suboperation
(db_nslcmop
, match
)
365 self
.assertEqual(found_op_index
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
367 # Test _update_suboperation_status()
368 def test_scale_update_suboperation_status(self
):
369 db_nslcmop
= self
.db_content
['nslcmops'][0]
371 # Force the initial values to be distinct from the updated ones
372 db_nslcmop
['_admin']['operations'][op_index
]['operationState'] = 'PROCESSING'
373 db_nslcmop
['_admin']['operations'][op_index
]['detailed-status'] = 'In progress'
374 # Test to change 'operationState' and 'detailed-status'
375 operationState
= 'COMPLETED'
376 detailed_status
= 'Done'
377 self
.my_ns
._update
_suboperation
_status
(
378 db_nslcmop
, op_index
, operationState
, detailed_status
)
379 operationState_new
= db_nslcmop
['_admin']['operations'][op_index
]['operationState']
380 detailed_status_new
= db_nslcmop
['_admin']['operations'][op_index
]['detailed-status']
381 # print("DEBUG: operationState_new={}, detailed_status_new={}".format(operationState_new, detailed_status_new))
382 self
.assertEqual(operationState
, operationState_new
)
383 self
.assertEqual(detailed_status
, detailed_status_new
)
385 # Test _add_suboperation()
386 def test_scale_add_suboperation(self
):
387 db_nslcmop
= self
.db_content
['nslcmops'][0]
389 num_ops_before
= len(db_nslcmop
.get('_admin', {}).get('operations', [])) - 1
391 vdu_count_index
= None
394 mapped_primitive_params
= {'parameter':
395 [{'data-type': 'STRING',
397 'default-value': '<touch_filename2>'}],
399 operationState
= 'PROCESSING'
400 detailed_status
= 'In progress'
401 operationType
= 'PRE-SCALE'
402 # Add a 'pre-scale' suboperation
403 op_index_after
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
404 vdu_name
, primitive
, mapped_primitive_params
,
405 operationState
, detailed_status
, operationType
)
406 self
.assertEqual(op_index_after
, num_ops_before
+ 1)
408 # Delete all suboperations and add the same operation again
409 del db_nslcmop
['_admin']['operations']
410 op_index_zero
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
411 vdu_name
, primitive
, mapped_primitive_params
,
412 operationState
, detailed_status
, operationType
)
413 self
.assertEqual(op_index_zero
, 0)
415 # Add a 'RO' suboperation
416 RO_nsr_id
= '1234567890'
417 RO_scaling_info
= [{'type': 'create', 'count': 1, 'member-vnf-index': '1', 'osm_vdu_id': 'dataVM'}]
418 op_index
= self
.my_ns
._add
_suboperation
(db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
,
419 vdu_name
, primitive
, mapped_primitive_params
,
420 operationState
, detailed_status
, operationType
,
421 RO_nsr_id
, RO_scaling_info
)
422 db_RO_nsr_id
= db_nslcmop
['_admin']['operations'][op_index
]['RO_nsr_id']
423 self
.assertEqual(op_index
, 1)
424 self
.assertEqual(RO_nsr_id
, db_RO_nsr_id
)
426 # Try to add an invalid suboperation, should return SUBOPERATION_STATUS_NOT_FOUND
427 op_index_invalid
= self
.my_ns
._add
_suboperation
(None, None, None, None, None,
430 self
.assertEqual(op_index_invalid
, self
.my_ns
.SUBOPERATION_STATUS_NOT_FOUND
)
432 # Test _check_or_add_scale_suboperation() and _check_or_add_scale_suboperation_RO()
433 # check the possible return values:
434 # - SUBOPERATION_STATUS_NEW: This is a new sub-operation
435 # - op_index (non-negative number): This is an existing sub-operation, operationState != 'COMPLETED'
436 # - SUBOPERATION_STATUS_SKIP: This is an existing sub-operation, operationState == 'COMPLETED'
437 def test_scale_check_or_add_scale_suboperation(self
):
438 db_nslcmop
= self
.db_content
['nslcmops'][0]
439 operationType
= 'PRE-SCALE'
442 primitive_params
= {'parameter':
443 [{'data-type': 'STRING',
445 'default-value': '<touch_filename2>'}],
448 # Delete all sub-operations to be sure this is a new sub-operation
449 del db_nslcmop
['_admin']['operations']
451 # Add a new sub-operation
452 # For new sub-operations, operationState is set to 'PROCESSING' by default
453 op_index_new
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
454 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
455 self
.assertEqual(op_index_new
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
457 # Use the same parameters again to match the already added sub-operation
458 # which has status 'PROCESSING' (!= 'COMPLETED') by default
459 # The expected return value is a non-negative number
460 op_index_existing
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
461 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
462 self
.assertTrue(op_index_existing
>= 0)
464 # Change operationState 'manually' for this sub-operation
465 db_nslcmop
['_admin']['operations'][op_index_existing
]['operationState'] = 'COMPLETED'
466 # Then use the same parameters again to match the already added sub-operation,
467 # which now has status 'COMPLETED'
468 # The expected return value is SUBOPERATION_STATUS_SKIP
469 op_index_skip
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
470 db_nslcmop
, vnf_index
, primitive
, primitive_params
, operationType
)
471 self
.assertEqual(op_index_skip
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
473 # RO sub-operation test:
474 # Repeat tests for the very similar _check_or_add_scale_suboperation_RO(),
475 RO_nsr_id
= '1234567890'
476 RO_scaling_info
= [{'type': 'create', 'count': 1, 'member-vnf-index': '1', 'osm_vdu_id': 'dataVM'}]
477 op_index_new_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
478 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
479 self
.assertEqual(op_index_new_RO
, self
.my_ns
.SUBOPERATION_STATUS_NEW
)
481 # Use the same parameters again to match the already added RO sub-operation
482 op_index_existing_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
483 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
484 self
.assertTrue(op_index_existing_RO
>= 0)
486 # Change operationState 'manually' for this RO sub-operation
487 db_nslcmop
['_admin']['operations'][op_index_existing_RO
]['operationState'] = 'COMPLETED'
488 # Then use the same parameters again to match the already added sub-operation,
489 # which now has status 'COMPLETED'
490 # The expected return value is SUBOPERATION_STATUS_SKIP
491 op_index_skip_RO
= self
.my_ns
._check
_or
_add
_scale
_suboperation
(
492 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
493 self
.assertEqual(op_index_skip_RO
, self
.my_ns
.SUBOPERATION_STATUS_SKIP
)
496 if __name__
== '__main__':