1 #######################################################################################
2 # Copyright ETSI Contributors and Others.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #######################################################################################
17 from copy
import deepcopy
19 from unittest
.mock
import MagicMock
, Mock
, patch
29 from osm_ng_ro
.ns
import Ns
, NsException
32 __author__
= "Eduardo Sousa"
33 __date__
= "$19-NOV-2021 00:00:00$"
36 # Variables used in Tests
37 vnfd_wth_persistent_storage
= {
38 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
42 "vdu-profile": [{"id": "several_volumes-VM", "min-number-of-instances": 1}],
45 "id": "several_volumes-vnf",
46 "product-name": "several_volumes-vnf",
49 "id": "several_volumes-VM",
50 "name": "several_volumes-VM",
51 "sw-image-desc": "ubuntu20.04",
52 "alternative-sw-image-desc": [
56 "virtual-storage-desc": [
57 "persistent-root-volume",
64 "virtual-storage-desc": [
66 "id": "persistent-volume2",
67 "type-of-storage": "persistent-storage:persistent-storage",
68 "size-of-storage": "10",
71 "id": "persistent-root-volume",
72 "type-of-storage": "persistent-storage:persistent-storage",
73 "size-of-storage": "10",
74 "vdu-storage-requirements": [
75 {"key": "keep-volume", "value": "true"},
79 "id": "ephemeral-volume",
80 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
81 "size-of-storage": "1",
87 "path": "/app/storage/",
92 vim_volume_id
= "ru937f49-3870-4169-b758-9732e1ff40f3"
93 task_by_target_record_id
= {
94 "nsrs:th47f48-9870-4169-b758-9732e1ff40f3": {
95 "extra_dict": {"params": {"net_type": "SR-IOV"}}
98 interfaces_wthout_positions
= [
109 "ns-vld-id": "mgmtnet",
112 interfaces_wth_all_positions
= [
125 "ns-vld-id": "mgmtnet",
129 target_vdu_wth_persistent_storage
= {
130 "_id": "09a0baa7-b7cb-4924-bd63-9f04a1c23960",
133 "vdu-name": "several_volumes-VM",
137 "ns-vld-id": "mgmtnet",
140 "virtual-storages": [
142 "id": "persistent-volume2",
143 "size-of-storage": "10",
144 "type-of-storage": "persistent-storage:persistent-storage",
147 "id": "persistent-root-volume",
148 "size-of-storage": "10",
149 "type-of-storage": "persistent-storage:persistent-storage",
150 "vdu-storage-requirements": [
151 {"key": "keep-volume", "value": "true"},
155 "id": "ephemeral-volume",
156 "size-of-storage": "1",
157 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
161 db
= MagicMock(name
="database mock")
162 fs
= MagicMock(name
="database mock")
163 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
164 vnf_preffix
= "vnfrs:wh47f48-y870-4169-b758-5732e1ff40f5"
165 vnfr_id
= "wh47f48-y870-4169-b758-5732e1ff40f5"
166 nsr_id
= "th47f48-9870-4169-b758-9732e1ff40f3"
168 "name": "sample_name",
170 expected_extra_dict
= {
172 f
"{ns_preffix}:image.0",
173 f
"{ns_preffix}:flavor.0",
176 "affinity_group_list": [],
177 "availability_zone_index": None,
178 "availability_zone_list": None,
179 "cloud_config": None,
180 "description": "several_volumes-VM",
182 "flavor_id": f
"TASK-{ns_preffix}:flavor.0",
183 "image_id": f
"TASK-{ns_preffix}:image.0",
184 "name": "sample_name-vnf-several-volu-several_volumes-VM-0",
190 expected_extra_dict2
= {
192 f
"{ns_preffix}:image.0",
193 f
"{ns_preffix}:flavor.0",
196 "affinity_group_list": [],
197 "availability_zone_index": None,
198 "availability_zone_list": None,
199 "cloud_config": None,
200 "description": "without_volumes-VM",
202 "flavor_id": f
"TASK-{ns_preffix}:flavor.0",
203 "image_id": f
"TASK-{ns_preffix}:image.0",
204 "name": "sample_name-vnf-several-volu-without_volumes-VM-0",
210 tasks_by_target_record_id
= {
211 "nsrs:th47f48-9870-4169-b758-9732e1ff40f3": {
214 "net_type": "SR-IOV",
221 "vdu2cloud_init": {},
223 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
224 "member-vnf-index-ref": "vnf-several-volumes",
227 vnfd_wthout_persistent_storage
= {
228 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
232 "vdu-profile": [{"id": "without_volumes-VM", "min-number-of-instances": 1}],
235 "id": "without_volumes-vnf",
236 "product-name": "without_volumes-vnf",
239 "id": "without_volumes-VM",
240 "name": "without_volumes-VM",
241 "sw-image-desc": "ubuntu20.04",
242 "alternative-sw-image-desc": [
246 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
250 "virtual-storage-desc": [
251 {"id": "root-volume", "size-of-storage": "10"},
253 "id": "ephemeral-volume",
254 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
255 "size-of-storage": "1",
261 "path": "/app/storage/",
267 target_vdu_wthout_persistent_storage
= {
268 "_id": "09a0baa7-b7cb-4924-bd63-9f04a1c23960",
271 "vdu-name": "without_volumes-VM",
275 "ns-vld-id": "mgmtnet",
278 "virtual-storages": [
281 "size-of-storage": "10",
284 "id": "ephemeral-volume",
285 "size-of-storage": "1",
286 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
290 cloud_init_content
= """
295 overwrite: {{is_override}}
298 - [ sh, -xc, "echo $(date) '{{command}}'" ]
309 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
311 vdu_id
= "bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
312 vnf_id
= "665b4165-ce24-4320-bf19-b9a45bade49f"
313 target_vim
= "vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
314 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
315 nsr_id_2
= "993166fe-723e-4680-ac4b-b1af2541ae31"
316 target_record_1
= "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
318 "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
319 "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
321 expected_result_vertical_scale
= {
322 "target_id": target_vim
,
323 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
324 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
325 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
326 "status": "SCHEDULED",
328 "item": "verticalscale",
329 "target_record": target_record_1
,
330 "target_record_id": target_record_id
,
332 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
333 "flavor_dict": "flavor_dict",
334 "flavor_id": "TASK-nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0",
336 "depends_on": ["nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0"],
340 "vim_info": {target_vim
: {"interfaces": []}},
343 vnf
= {"_id": vnf_id
}
344 extra_dict_vertical_scale
= {
346 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
347 "flavor_dict": "flavor_dict",
350 extra_dict_migrate
= {
352 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
353 "migrate_host": "migrateToHost",
356 expected_result_migrate
= {
357 "target_id": target_vim
,
358 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
359 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
360 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
361 "status": "SCHEDULED",
364 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
365 "target_record_id": target_record_id
,
367 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
368 "migrate_host": "migrateToHost",
371 expected_result_rebuild_start_stop
= {
372 "target_id": target_vim
,
373 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
374 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
375 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:0",
376 "status": "SCHEDULED",
379 "target_record_id": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
383 class TestException(Exception):
387 class CopyingMock(MagicMock
):
388 def __call__(self
, *args
, **kwargs
):
389 args
= deepcopy(args
)
390 kwargs
= deepcopy(kwargs
)
391 return super(CopyingMock
, self
).__call
__(*args
, **kwargs
)
394 class TestNs(unittest
.TestCase
):
398 def test__create_task_without_extra_dict(self
):
400 "target_id": "vim_openstack_1",
401 "action_id": "123456",
403 "task_id": "123456:1",
404 "status": "SCHEDULED",
407 "target_record": "test_target_record",
408 "target_record_id": "test_target_record_id",
411 "action_id": "123456",
416 task
= Ns
._create
_task
(
417 deployment_info
=deployment_info
,
418 target_id
="vim_openstack_1",
421 target_record
="test_target_record",
422 target_record_id
="test_target_record_id",
425 self
.assertEqual(deployment_info
.get("task_index"), 2)
426 self
.assertDictEqual(task
, expected_result
)
428 def test__create_task(self
):
430 "target_id": "vim_openstack_1",
431 "action_id": "123456",
433 "task_id": "123456:1",
434 "status": "SCHEDULED",
437 "target_record": "test_target_record",
438 "target_record_id": "test_target_record_id",
439 # values coming from extra_dict
440 "params": "test_params",
441 "find_params": "test_find_params",
442 "depends_on": "test_depends_on",
445 "action_id": "123456",
450 task
= Ns
._create
_task
(
451 deployment_info
=deployment_info
,
452 target_id
="vim_openstack_1",
455 target_record
="test_target_record",
456 target_record_id
="test_target_record_id",
458 "params": "test_params",
459 "find_params": "test_find_params",
460 "depends_on": "test_depends_on",
464 self
.assertEqual(deployment_info
.get("task_index"), 2)
465 self
.assertDictEqual(task
, expected_result
)
467 @patch("osm_ng_ro.ns.time")
468 def test__create_ro_task(self
, mock_time
: Mock
):
469 now
= 1637324838.994551
470 mock_time
.return_value
= now
472 "target_id": "vim_openstack_1",
473 "action_id": "123456",
475 "task_id": "123456:1",
476 "status": "SCHEDULED",
479 "target_record": "test_target_record",
480 "target_record_id": "test_target_record_id",
481 # values coming from extra_dict
482 "params": "test_params",
483 "find_params": "test_find_params",
484 "depends_on": "test_depends_on",
490 "target_id": "vim_openstack_1",
493 "created_items": None,
507 ro_task
= Ns
._create
_ro
_task
(
508 target_id
="vim_openstack_1",
512 self
.assertDictEqual(ro_task
, expected_result
)
514 def test__process_image_params_with_empty_target_image(self
):
520 result
= Ns
._process
_image
_params
(
521 target_image
=target_image
,
524 target_record_id
=None,
527 self
.assertDictEqual(expected_result
, result
)
529 def test__process_image_params_with_wrong_target_image(self
):
534 "no_image": "to_see_here",
537 result
= Ns
._process
_image
_params
(
538 target_image
=target_image
,
541 target_record_id
=None,
544 self
.assertDictEqual(expected_result
, result
)
546 def test__process_image_params_with_image(self
):
558 result
= Ns
._process
_image
_params
(
559 target_image
=target_image
,
562 target_record_id
=None,
565 self
.assertDictEqual(expected_result
, result
)
567 def test__process_image_params_with_vim_image_id(self
):
576 "vim_image_id": "123456",
579 result
= Ns
._process
_image
_params
(
580 target_image
=target_image
,
583 target_record_id
=None,
586 self
.assertDictEqual(expected_result
, result
)
588 def test__process_image_params_with_image_checksum(self
):
592 "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
597 "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
600 result
= Ns
._process
_image
_params
(
601 target_image
=target_image
,
604 target_record_id
=None,
607 self
.assertDictEqual(expected_result
, result
)
609 def test__get_resource_allocation_params_with_empty_target_image(self
):
611 quota_descriptor
= {}
613 result
= Ns
._get
_resource
_allocation
_params
(
614 quota_descriptor
=quota_descriptor
,
617 self
.assertDictEqual(expected_result
, result
)
619 def test__get_resource_allocation_params_with_wrong_target_image(self
):
622 "no_quota": "present_here",
625 result
= Ns
._get
_resource
_allocation
_params
(
626 quota_descriptor
=quota_descriptor
,
629 self
.assertDictEqual(expected_result
, result
)
631 def test__get_resource_allocation_params_with_limit(self
):
639 result
= Ns
._get
_resource
_allocation
_params
(
640 quota_descriptor
=quota_descriptor
,
643 self
.assertDictEqual(expected_result
, result
)
645 def test__get_resource_allocation_params_with_reserve(self
):
653 result
= Ns
._get
_resource
_allocation
_params
(
654 quota_descriptor
=quota_descriptor
,
657 self
.assertDictEqual(expected_result
, result
)
659 def test__get_resource_allocation_params_with_shares(self
):
667 result
= Ns
._get
_resource
_allocation
_params
(
668 quota_descriptor
=quota_descriptor
,
671 self
.assertDictEqual(expected_result
, result
)
673 def test__get_resource_allocation_params(self
):
685 result
= Ns
._get
_resource
_allocation
_params
(
686 quota_descriptor
=quota_descriptor
,
689 self
.assertDictEqual(expected_result
, result
)
691 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
692 def test__process_guest_epa_quota_params_with_empty_quota_epa_cpu(
700 result
= Ns
._process
_guest
_epa
_quota
_params
(
701 guest_epa_quota
=guest_epa_quota
,
702 epa_vcpu_set
=epa_vcpu_set
,
705 self
.assertDictEqual(expected_result
, result
)
706 self
.assertFalse(resource_allocation
.called
)
708 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
709 def test__process_guest_epa_quota_params_with_empty_quota_false_epa_cpu(
717 result
= Ns
._process
_guest
_epa
_quota
_params
(
718 guest_epa_quota
=guest_epa_quota
,
719 epa_vcpu_set
=epa_vcpu_set
,
722 self
.assertDictEqual(expected_result
, result
)
723 self
.assertFalse(resource_allocation
.called
)
725 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
726 def test__process_guest_epa_quota_params_with_wrong_quota_epa_cpu(
732 "no-quota": "nothing",
736 result
= Ns
._process
_guest
_epa
_quota
_params
(
737 guest_epa_quota
=guest_epa_quota
,
738 epa_vcpu_set
=epa_vcpu_set
,
741 self
.assertDictEqual(expected_result
, result
)
742 self
.assertFalse(resource_allocation
.called
)
744 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
745 def test__process_guest_epa_quota_params_with_wrong_quota_false_epa_cpu(
751 "no-quota": "nothing",
755 result
= Ns
._process
_guest
_epa
_quota
_params
(
756 guest_epa_quota
=guest_epa_quota
,
757 epa_vcpu_set
=epa_vcpu_set
,
760 self
.assertDictEqual(expected_result
, result
)
761 self
.assertFalse(resource_allocation
.called
)
763 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
764 def test__process_guest_epa_quota_params_with_cpu_quota_epa_cpu(
778 result
= Ns
._process
_guest
_epa
_quota
_params
(
779 guest_epa_quota
=guest_epa_quota
,
780 epa_vcpu_set
=epa_vcpu_set
,
783 self
.assertDictEqual(expected_result
, result
)
784 self
.assertFalse(resource_allocation
.called
)
786 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
787 def test__process_guest_epa_quota_params_with_cpu_quota_false_epa_cpu(
807 resource_allocation_param
= {
812 resource_allocation
.return_value
= {
818 result
= Ns
._process
_guest
_epa
_quota
_params
(
819 guest_epa_quota
=guest_epa_quota
,
820 epa_vcpu_set
=epa_vcpu_set
,
823 resource_allocation
.assert_called_once_with(resource_allocation_param
)
824 self
.assertDictEqual(expected_result
, result
)
826 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
827 def test__process_guest_epa_quota_params_with_mem_quota_epa_cpu(
847 resource_allocation_param
= {
852 resource_allocation
.return_value
= {
858 result
= Ns
._process
_guest
_epa
_quota
_params
(
859 guest_epa_quota
=guest_epa_quota
,
860 epa_vcpu_set
=epa_vcpu_set
,
863 resource_allocation
.assert_called_once_with(resource_allocation_param
)
864 self
.assertDictEqual(expected_result
, result
)
866 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
867 def test__process_guest_epa_quota_params_with_mem_quota_false_epa_cpu(
887 resource_allocation_param
= {
892 resource_allocation
.return_value
= {
898 result
= Ns
._process
_guest
_epa
_quota
_params
(
899 guest_epa_quota
=guest_epa_quota
,
900 epa_vcpu_set
=epa_vcpu_set
,
903 resource_allocation
.assert_called_once_with(resource_allocation_param
)
904 self
.assertDictEqual(expected_result
, result
)
906 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
907 def test__process_guest_epa_quota_params_with_disk_io_quota_epa_cpu(
927 resource_allocation_param
= {
932 resource_allocation
.return_value
= {
938 result
= Ns
._process
_guest
_epa
_quota
_params
(
939 guest_epa_quota
=guest_epa_quota
,
940 epa_vcpu_set
=epa_vcpu_set
,
943 resource_allocation
.assert_called_once_with(resource_allocation_param
)
944 self
.assertDictEqual(expected_result
, result
)
946 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
947 def test__process_guest_epa_quota_params_with_disk_io_quota_false_epa_cpu(
967 resource_allocation_param
= {
972 resource_allocation
.return_value
= {
978 result
= Ns
._process
_guest
_epa
_quota
_params
(
979 guest_epa_quota
=guest_epa_quota
,
980 epa_vcpu_set
=epa_vcpu_set
,
983 resource_allocation
.assert_called_once_with(resource_allocation_param
)
984 self
.assertDictEqual(expected_result
, result
)
986 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
987 def test__process_guest_epa_quota_params_with_vif_quota_epa_cpu(
1007 resource_allocation_param
= {
1012 resource_allocation
.return_value
= {
1018 result
= Ns
._process
_guest
_epa
_quota
_params
(
1019 guest_epa_quota
=guest_epa_quota
,
1020 epa_vcpu_set
=epa_vcpu_set
,
1023 resource_allocation
.assert_called_once_with(resource_allocation_param
)
1024 self
.assertDictEqual(expected_result
, result
)
1026 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1027 def test__process_guest_epa_quota_params_with_vif_quota_false_epa_cpu(
1029 resource_allocation
,
1045 epa_vcpu_set
= False
1047 resource_allocation_param
= {
1052 resource_allocation
.return_value
= {
1058 result
= Ns
._process
_guest
_epa
_quota
_params
(
1059 guest_epa_quota
=guest_epa_quota
,
1060 epa_vcpu_set
=epa_vcpu_set
,
1063 resource_allocation
.assert_called_once_with(resource_allocation_param
)
1064 self
.assertDictEqual(expected_result
, result
)
1066 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1067 def test__process_guest_epa_quota_params_with_quota_epa_cpu(
1069 resource_allocation
,
1112 resource_allocation
.return_value
= {
1118 result
= Ns
._process
_guest
_epa
_quota
_params
(
1119 guest_epa_quota
=guest_epa_quota
,
1120 epa_vcpu_set
=epa_vcpu_set
,
1123 self
.assertTrue(resource_allocation
.called
)
1124 self
.assertDictEqual(expected_result
, result
)
1126 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1127 def test__process_guest_epa_quota_params_with_quota_epa_cpu_no_set(
1129 resource_allocation
,
1175 epa_vcpu_set
= False
1177 resource_allocation
.return_value
= {
1183 result
= Ns
._process
_guest
_epa
_quota
_params
(
1184 guest_epa_quota
=guest_epa_quota
,
1185 epa_vcpu_set
=epa_vcpu_set
,
1188 self
.assertTrue(resource_allocation
.called
)
1189 self
.assertDictEqual(expected_result
, result
)
1191 def test__process_guest_epa_numa_params_with_empty_numa_params(self
):
1192 expected_numa_result
= []
1193 expected_epa_vcpu_set_result
= False
1194 guest_epa_quota
= {}
1196 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1197 guest_epa_quota
=guest_epa_quota
,
1199 self
.assertEqual(expected_numa_result
, numa_result
)
1200 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1202 def test__process_guest_epa_numa_params_with_wrong_numa_params(self
):
1203 expected_numa_result
= []
1204 expected_epa_vcpu_set_result
= False
1205 guest_epa_quota
= {"no_nume": "here"}
1207 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1208 guest_epa_quota
=guest_epa_quota
,
1211 self
.assertEqual(expected_numa_result
, numa_result
)
1212 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1214 def test__process_guest_epa_numa_params_with_numa_node_policy(self
):
1215 expected_numa_result
= []
1216 expected_epa_vcpu_set_result
= False
1217 guest_epa_quota
= {"numa-node-policy": {}}
1219 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1220 guest_epa_quota
=guest_epa_quota
,
1223 self
.assertEqual(expected_numa_result
, numa_result
)
1224 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1226 def test__process_guest_epa_numa_params_with_no_node(self
):
1227 expected_numa_result
= []
1228 expected_epa_vcpu_set_result
= False
1230 "numa-node-policy": {
1235 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1236 guest_epa_quota
=guest_epa_quota
,
1239 self
.assertEqual(expected_numa_result
, numa_result
)
1240 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1242 def test__process_guest_epa_numa_params_with_1_node_num_cores(self
):
1243 expected_numa_result
= [{"cores": 3}]
1244 expected_epa_vcpu_set_result
= True
1246 "numa-node-policy": {
1255 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1256 guest_epa_quota
=guest_epa_quota
,
1259 self
.assertEqual(expected_numa_result
, numa_result
)
1260 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1262 def test__process_guest_epa_numa_params_with_1_node_paired_threads(self
):
1263 expected_numa_result
= [{"paired_threads": 3}]
1264 expected_epa_vcpu_set_result
= True
1266 "numa-node-policy": {
1269 "paired-threads": {"num-paired-threads": "3"},
1275 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1276 guest_epa_quota
=guest_epa_quota
,
1279 self
.assertEqual(expected_numa_result
, numa_result
)
1280 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1282 def test__process_guest_epa_numa_params_with_1_node_paired_threads_ids(self
):
1283 expected_numa_result
= [
1285 "paired-threads-id": [("0", "1"), ("4", "5")],
1288 expected_epa_vcpu_set_result
= False
1290 "numa-node-policy": {
1294 "paired-thread-ids": [
1310 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1311 guest_epa_quota
=guest_epa_quota
,
1314 self
.assertEqual(expected_numa_result
, numa_result
)
1315 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1317 def test__process_guest_epa_numa_params_with_1_node_num_threads(self
):
1318 expected_numa_result
= [{"threads": 3}]
1319 expected_epa_vcpu_set_result
= True
1321 "numa-node-policy": {
1330 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1331 guest_epa_quota
=guest_epa_quota
,
1334 self
.assertEqual(expected_numa_result
, numa_result
)
1335 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1337 def test__process_guest_epa_numa_params_with_1_node_memory_mb(self
):
1338 expected_numa_result
= [{"memory": 2}]
1339 expected_epa_vcpu_set_result
= False
1341 "numa-node-policy": {
1350 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1351 guest_epa_quota
=guest_epa_quota
,
1354 self
.assertEqual(expected_numa_result
, numa_result
)
1355 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1357 def test__process_guest_epa_numa_params_with_1_node_vcpu(self
):
1358 expected_numa_result
= [
1364 expected_epa_vcpu_set_result
= False
1366 "numa-node-policy": {
1367 "node": [{"id": "0", "vcpu": [{"id": "0"}, {"id": "1"}]}],
1371 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1372 guest_epa_quota
=guest_epa_quota
,
1375 self
.assertEqual(expected_numa_result
, numa_result
)
1376 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1378 def test__process_guest_epa_numa_params_with_2_node_vcpu(self
):
1379 expected_numa_result
= [
1390 expected_epa_vcpu_set_result
= False
1392 "numa-node-policy": {
1394 {"id": "0", "vcpu": [{"id": "0"}, {"id": "1"}]},
1395 {"id": "1", "vcpu": [{"id": "2"}, {"id": "3"}]},
1400 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1401 guest_epa_quota
=guest_epa_quota
,
1404 self
.assertEqual(expected_numa_result
, numa_result
)
1405 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1407 def test__process_guest_epa_numa_params_with_1_node(self
):
1408 expected_numa_result
= [
1413 "paired_threads": 3,
1414 "paired-threads-id": [("0", "1"), ("4", "5")],
1419 expected_epa_vcpu_set_result
= True
1421 "numa-node-policy": {
1426 "num-paired-threads": "3",
1427 "paired-thread-ids": [
1445 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1446 guest_epa_quota
=guest_epa_quota
,
1449 self
.assertEqual(expected_numa_result
, numa_result
)
1450 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1452 def test__process_guest_epa_numa_params_with_2_nodes(self
):
1453 expected_numa_result
= [
1456 "paired_threads": 3,
1457 "paired-threads-id": [("0", "1"), ("4", "5")],
1463 "paired_threads": 7,
1464 "paired-threads-id": [("2", "3"), ("5", "6")],
1469 expected_epa_vcpu_set_result
= True
1471 "numa-node-policy": {
1476 "num-paired-threads": "3",
1477 "paired-thread-ids": [
1494 "num-paired-threads": "7",
1495 "paired-thread-ids": [
1513 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1514 guest_epa_quota
=guest_epa_quota
,
1517 self
.assertEqual(expected_numa_result
, numa_result
)
1518 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1520 def test__process_guest_epa_cpu_pinning_params_with_empty_params(self
):
1521 expected_numa_result
= {}
1522 expected_epa_vcpu_set_result
= False
1523 guest_epa_quota
= {}
1525 epa_vcpu_set
= False
1527 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1528 guest_epa_quota
=guest_epa_quota
,
1529 vcpu_count
=vcpu_count
,
1530 epa_vcpu_set
=epa_vcpu_set
,
1533 self
.assertDictEqual(expected_numa_result
, numa_result
)
1534 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1536 def test__process_guest_epa_cpu_pinning_params_with_wrong_params(self
):
1537 expected_numa_result
= {}
1538 expected_epa_vcpu_set_result
= False
1540 "no-cpu-pinning-policy": "here",
1543 epa_vcpu_set
= False
1545 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1546 guest_epa_quota
=guest_epa_quota
,
1547 vcpu_count
=vcpu_count
,
1548 epa_vcpu_set
=epa_vcpu_set
,
1551 self
.assertDictEqual(expected_numa_result
, numa_result
)
1552 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1554 def test__process_guest_epa_cpu_pinning_params_with_epa_vcpu_set(self
):
1555 expected_numa_result
= {}
1556 expected_epa_vcpu_set_result
= True
1558 "cpu-pinning-policy": "DEDICATED",
1563 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1564 guest_epa_quota
=guest_epa_quota
,
1565 vcpu_count
=vcpu_count
,
1566 epa_vcpu_set
=epa_vcpu_set
,
1569 self
.assertDictEqual(expected_numa_result
, numa_result
)
1570 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1572 def test__process_guest_epa_cpu_pinning_params_with_policy_prefer(self
):
1573 expected_numa_result
= {"threads": 3}
1574 expected_epa_vcpu_set_result
= True
1576 "cpu-pinning-policy": "DEDICATED",
1577 "cpu-thread-pinning-policy": "PREFER",
1580 epa_vcpu_set
= False
1582 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1583 guest_epa_quota
=guest_epa_quota
,
1584 vcpu_count
=vcpu_count
,
1585 epa_vcpu_set
=epa_vcpu_set
,
1588 self
.assertDictEqual(expected_numa_result
, numa_result
)
1589 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1591 def test__process_guest_epa_cpu_pinning_params_with_policy_isolate(self
):
1592 expected_numa_result
= {"cores": 3}
1593 expected_epa_vcpu_set_result
= True
1595 "cpu-pinning-policy": "DEDICATED",
1596 "cpu-thread-pinning-policy": "ISOLATE",
1599 epa_vcpu_set
= False
1601 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1602 guest_epa_quota
=guest_epa_quota
,
1603 vcpu_count
=vcpu_count
,
1604 epa_vcpu_set
=epa_vcpu_set
,
1607 self
.assertDictEqual(expected_numa_result
, numa_result
)
1608 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1610 def test__process_guest_epa_cpu_pinning_params_with_policy_require(self
):
1611 expected_numa_result
= {"threads": 3}
1612 expected_epa_vcpu_set_result
= True
1614 "cpu-pinning-policy": "DEDICATED",
1615 "cpu-thread-pinning-policy": "REQUIRE",
1618 epa_vcpu_set
= False
1620 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1621 guest_epa_quota
=guest_epa_quota
,
1622 vcpu_count
=vcpu_count
,
1623 epa_vcpu_set
=epa_vcpu_set
,
1626 self
.assertDictEqual(expected_numa_result
, numa_result
)
1627 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1629 def test__process_guest_epa_cpu_pinning_params(self
):
1630 expected_numa_result
= {"threads": 3}
1631 expected_epa_vcpu_set_result
= True
1633 "cpu-pinning-policy": "DEDICATED",
1636 epa_vcpu_set
= False
1638 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1639 guest_epa_quota
=guest_epa_quota
,
1640 vcpu_count
=vcpu_count
,
1641 epa_vcpu_set
=epa_vcpu_set
,
1644 self
.assertDictEqual(expected_numa_result
, numa_result
)
1645 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1647 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1648 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1649 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1650 def test__process_guest_epa_params_with_empty_params(
1652 guest_epa_numa_params
,
1653 guest_epa_cpu_pinning_params
,
1654 guest_epa_quota_params
,
1656 expected_result
= {}
1659 result
= Ns
._process
_epa
_params
(
1660 target_flavor
=target_flavor
,
1663 self
.assertDictEqual(expected_result
, result
)
1664 self
.assertFalse(guest_epa_numa_params
.called
)
1665 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1666 self
.assertFalse(guest_epa_quota_params
.called
)
1668 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1669 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1670 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1671 def test__process_guest_epa_params_with_wrong_params(
1673 guest_epa_numa_params
,
1674 guest_epa_cpu_pinning_params
,
1675 guest_epa_quota_params
,
1677 expected_result
= {}
1679 "no-guest-epa": "here",
1682 result
= Ns
._process
_epa
_params
(
1683 target_flavor
=target_flavor
,
1686 self
.assertDictEqual(expected_result
, result
)
1687 self
.assertFalse(guest_epa_numa_params
.called
)
1688 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1689 self
.assertFalse(guest_epa_quota_params
.called
)
1691 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1692 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1693 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1694 def test__process_guest_epa_params(
1696 guest_epa_numa_params
,
1697 guest_epa_cpu_pinning_params
,
1698 guest_epa_quota_params
,
1701 "mem-policy": "STRICT",
1706 "numa-node-policy": {
1707 "mem-policy": "STRICT",
1712 guest_epa_numa_params
.return_value
= ({}, False)
1713 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1714 guest_epa_quota_params
.return_value
= {}
1716 result
= Ns
._process
_epa
_params
(
1717 target_flavor
=target_flavor
,
1720 self
.assertDictEqual(expected_result
, result
)
1721 self
.assertTrue(guest_epa_numa_params
.called
)
1722 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1723 self
.assertTrue(guest_epa_quota_params
.called
)
1725 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1726 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1727 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1728 def test__process_guest_epa_params_with_mempage_size(
1730 guest_epa_numa_params
,
1731 guest_epa_cpu_pinning_params
,
1732 guest_epa_quota_params
,
1735 "mempage-size": "1G",
1736 "mem-policy": "STRICT",
1741 "mempage-size": "1G",
1742 "numa-node-policy": {
1743 "mem-policy": "STRICT",
1748 guest_epa_numa_params
.return_value
= ({}, False)
1749 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1750 guest_epa_quota_params
.return_value
= {}
1752 result
= Ns
._process
_epa
_params
(
1753 target_flavor
=target_flavor
,
1756 self
.assertDictEqual(expected_result
, result
)
1757 self
.assertTrue(guest_epa_numa_params
.called
)
1758 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1759 self
.assertTrue(guest_epa_quota_params
.called
)
1761 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1762 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1763 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1764 def test__process_guest_epa_params_with_numa(
1766 guest_epa_numa_params
,
1767 guest_epa_cpu_pinning_params
,
1768 guest_epa_quota_params
,
1771 "mempage-size": "1G",
1772 "cpu-pinning-policy": "DEDICATED",
1773 "cpu-thread-pinning-policy": "PREFER",
1778 "paired-threads": 3,
1779 "paired-threads-id": [("0", "1"), ("4", "5")],
1783 "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1784 "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1785 "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1786 "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1791 "mempage-size": "1G",
1792 "cpu-pinning-policy": "DEDICATED",
1793 "cpu-thread-pinning-policy": "PREFER",
1794 "numa-node-policy": {
1799 "num-paired-threads": "3",
1800 "paired-thread-ids": [
1839 guest_epa_numa_params
.return_value
= (
1843 "paired-threads": 3,
1844 "paired-threads-id": [("0", "1"), ("4", "5")],
1851 guest_epa_cpu_pinning_params
.return_value
= (
1857 guest_epa_quota_params
.return_value
= {
1880 result
= Ns
._process
_epa
_params
(
1881 target_flavor
=target_flavor
,
1883 self
.assertEqual(expected_result
, result
)
1884 self
.assertTrue(guest_epa_numa_params
.called
)
1885 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1886 self
.assertTrue(guest_epa_quota_params
.called
)
1888 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1889 def test__process_flavor_params_with_empty_target_flavor(
1897 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1902 target_record_id
= ""
1904 with self
.assertRaises(KeyError):
1905 Ns
._process
_flavor
_params
(
1906 target_flavor
=target_flavor
,
1909 target_record_id
=target_record_id
,
1912 self
.assertFalse(epa_params
.called
)
1914 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1915 def test__process_flavor_params_with_wrong_target_flavor(
1920 "no-target-flavor": "here",
1924 target_record_id
= ""
1926 with self
.assertRaises(KeyError):
1927 Ns
._process
_flavor
_params
(
1928 target_flavor
=target_flavor
,
1931 target_record_id
=target_record_id
,
1934 self
.assertFalse(epa_params
.called
)
1936 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1937 def test__process_flavor_params_with_empty_indata(
1961 "memory-mb": "1024",
1966 target_record_id
= ""
1968 epa_params
.return_value
= {}
1970 result
= Ns
._process
_flavor
_params
(
1971 target_flavor
=target_flavor
,
1974 target_record_id
=target_record_id
,
1977 self
.assertTrue(epa_params
.called
)
1978 self
.assertDictEqual(result
, expected_result
)
1980 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1981 def test__process_flavor_params_with_wrong_indata(
2005 "memory-mb": "1024",
2012 target_record_id
= ""
2014 epa_params
.return_value
= {}
2016 result
= Ns
._process
_flavor
_params
(
2017 target_flavor
=target_flavor
,
2020 target_record_id
=target_record_id
,
2023 self
.assertTrue(epa_params
.called
)
2024 self
.assertDictEqual(result
, expected_result
)
2026 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2027 def test__process_flavor_params_with_ephemeral_disk(
2035 db
.get_one
.return_value
= {
2036 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2041 {"id": "without_volumes-VM", "min-number-of-instances": 1}
2045 "id": "without_volumes-vnf",
2046 "product-name": "without_volumes-vnf",
2049 "id": "without_volumes-VM",
2050 "name": "without_volumes-VM",
2051 "sw-image-desc": "ubuntu20.04",
2052 "alternative-sw-image-desc": [
2054 "ubuntu20.04-azure",
2056 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2060 "virtual-storage-desc": [
2061 {"id": "root-volume", "size-of-storage": "10"},
2063 "id": "ephemeral-volume",
2064 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2065 "size-of-storage": "1",
2071 "path": "/app/storage/",
2099 "memory-mb": "1024",
2107 "ns-flavor-id": "test_id",
2108 "virtual-storages": [
2110 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2111 "size-of-storage": "10",
2116 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2121 target_record_id
= ""
2123 epa_params
.return_value
= {}
2125 result
= Ns
._process
_flavor
_params
(
2126 target_flavor
=target_flavor
,
2129 target_record_id
=target_record_id
,
2133 self
.assertTrue(epa_params
.called
)
2134 self
.assertDictEqual(result
, expected_result
)
2136 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2137 def test__process_flavor_params_with_swap_disk(
2164 "memory-mb": "1024",
2172 "ns-flavor-id": "test_id",
2173 "virtual-storages": [
2175 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2176 "size-of-storage": "20",
2181 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2186 target_record_id
= ""
2188 epa_params
.return_value
= {}
2190 result
= Ns
._process
_flavor
_params
(
2191 target_flavor
=target_flavor
,
2194 target_record_id
=target_record_id
,
2197 self
.assertTrue(epa_params
.called
)
2198 self
.assertDictEqual(result
, expected_result
)
2200 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2201 def test__process_flavor_params_with_persistent_root_disk(
2209 db
.get_one
.return_value
= {
2210 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2215 {"id": "several_volumes-VM", "min-number-of-instances": 1}
2219 "id": "several_volumes-vnf",
2220 "product-name": "several_volumes-vnf",
2223 "id": "several_volumes-VM",
2224 "name": "several_volumes-VM",
2225 "sw-image-desc": "ubuntu20.04",
2226 "alternative-sw-image-desc": [
2228 "ubuntu20.04-azure",
2230 "virtual-storage-desc": [
2231 "persistent-root-volume",
2236 "virtual-storage-desc": [
2238 "id": "persistent-root-volume",
2239 "type-of-storage": "persistent-storage:persistent-storage",
2240 "size-of-storage": "10",
2246 "path": "/app/storage/",
2272 "memory-mb": "1024",
2280 "vdu-name": "several_volumes-VM",
2281 "ns-flavor-id": "test_id",
2282 "virtual-storages": [
2284 "type-of-storage": "persistent-storage:persistent-storage",
2285 "size-of-storage": "10",
2290 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2295 target_record_id
= ""
2297 epa_params
.return_value
= {}
2299 result
= Ns
._process
_flavor
_params
(
2300 target_flavor
=target_flavor
,
2303 target_record_id
=target_record_id
,
2307 self
.assertTrue(epa_params
.called
)
2308 self
.assertDictEqual(result
, expected_result
)
2310 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2311 def test__process_flavor_params_with_epa_params(
2322 "numa": "there-is-numa-here",
2333 "numa": "there-is-numa-here",
2342 "memory-mb": "1024",
2350 "ns-flavor-id": "test_id",
2353 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2358 target_record_id
= ""
2360 epa_params
.return_value
= {
2361 "numa": "there-is-numa-here",
2364 result
= Ns
._process
_flavor
_params
(
2365 target_flavor
=target_flavor
,
2368 target_record_id
=target_record_id
,
2371 self
.assertTrue(epa_params
.called
)
2372 self
.assertDictEqual(result
, expected_result
)
2374 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2375 def test__process_flavor_params_with_vim_flavor_id(
2381 "vim_flavor_id": "test.flavor",
2388 "memory-mb": "1024",
2396 "ns-flavor-id": "test_id",
2397 "additionalParams": {
2398 "OSM": {"vim_flavor_id": "test.flavor"}
2402 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2407 target_record_id
= ""
2409 epa_params
.return_value
= {}
2411 result
= Ns
._process
_flavor
_params
(
2412 target_flavor
=target_flavor
,
2415 target_record_id
=target_record_id
,
2418 self
.assertFalse(epa_params
.called
)
2419 self
.assertDictEqual(result
, expected_result
)
2421 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2422 def test__process_flavor_params(
2430 db
.get_one
.return_value
= {
2431 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2436 {"id": "without_volumes-VM", "min-number-of-instances": 1}
2440 "id": "without_volumes-vnf",
2441 "product-name": "without_volumes-vnf",
2444 "id": "without_volumes-VM",
2445 "name": "without_volumes-VM",
2446 "sw-image-desc": "ubuntu20.04",
2447 "alternative-sw-image-desc": [
2449 "ubuntu20.04-azure",
2451 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2455 "virtual-storage-desc": [
2456 {"id": "root-volume", "size-of-storage": "10"},
2458 "id": "ephemeral-volume",
2459 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2460 "size-of-storage": "1",
2466 "path": "/app/storage/",
2481 "numa": "there-is-numa-here",
2494 "numa": "there-is-numa-here",
2503 "memory-mb": "1024",
2511 "ns-flavor-id": "test_id",
2512 "virtual-storages": [
2514 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2515 "size-of-storage": "10",
2518 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2519 "size-of-storage": "20",
2524 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2529 target_record_id
= ""
2531 epa_params
.return_value
= {
2532 "numa": "there-is-numa-here",
2535 result
= Ns
._process
_flavor
_params
(
2536 target_flavor
=target_flavor
,
2539 target_record_id
=target_record_id
,
2543 self
.assertTrue(epa_params
.called
)
2544 self
.assertDictEqual(result
, expected_result
)
2546 def test__process_net_params_with_empty_params(
2556 "provider_network": "some-profile-here",
2558 "some_ip_profile": "here",
2561 target_record_id
= ""
2564 "net_name": "ns-name-vld-name",
2565 "net_type": "bridge",
2567 "some_ip_profile": "here",
2569 "provider_network_profile": "some-profile-here",
2573 result
= Ns
._process
_net
_params
(
2574 target_vld
=target_vld
,
2577 target_record_id
=target_record_id
,
2580 self
.assertDictEqual(expected_result
, result
)
2582 def test__process_net_params_with_vim_info_sdn(
2593 "sdn-ports": ["some", "ports", "here"],
2594 "vlds": ["some", "vlds", "here"],
2597 target_record_id
= "vld.sdn.something"
2600 "sdn-ports": ["some", "ports", "here"],
2601 "vlds": ["some", "vlds", "here"],
2606 result
= Ns
._process
_net
_params
(
2607 target_vld
=target_vld
,
2610 target_record_id
=target_record_id
,
2613 self
.assertDictEqual(expected_result
, result
)
2615 def test__process_net_params_with_vim_info_sdn_target_vim(
2626 "sdn-ports": ["some", "ports", "here"],
2627 "vlds": ["some", "vlds", "here"],
2628 "target_vim": "some-vim",
2631 target_record_id
= "vld.sdn.something"
2633 "depends_on": ["some-vim vld.sdn"],
2635 "sdn-ports": ["some", "ports", "here"],
2636 "vlds": ["some", "vlds", "here"],
2637 "target_vim": "some-vim",
2642 result
= Ns
._process
_net
_params
(
2643 target_vld
=target_vld
,
2646 target_record_id
=target_record_id
,
2649 self
.assertDictEqual(expected_result
, result
)
2651 def test__process_net_params_with_vim_network_name(
2661 "vim_network_name": "some-network-name",
2663 target_record_id
= "vld.sdn.something"
2667 "name": "some-network-name",
2672 result
= Ns
._process
_net
_params
(
2673 target_vld
=target_vld
,
2676 target_record_id
=target_record_id
,
2679 self
.assertDictEqual(expected_result
, result
)
2681 def test__process_net_params_with_vim_network_id(
2691 "vim_network_id": "some-network-id",
2693 target_record_id
= "vld.sdn.something"
2697 "id": "some-network-id",
2702 result
= Ns
._process
_net
_params
(
2703 target_vld
=target_vld
,
2706 target_record_id
=target_record_id
,
2709 self
.assertDictEqual(expected_result
, result
)
2711 def test__process_net_params_with_mgmt_network(
2717 "mgmt-network": "some-mgmt-network",
2723 target_record_id
= "vld.sdn.something"
2731 result
= Ns
._process
_net
_params
(
2732 target_vld
=target_vld
,
2735 target_record_id
=target_record_id
,
2738 self
.assertDictEqual(expected_result
, result
)
2740 def test__process_net_params_with_underlay_eline(
2745 "underlay": "some-underlay-here",
2752 "provider_network": "some-profile-here",
2754 "some_ip_profile": "here",
2757 target_record_id
= ""
2761 "some_ip_profile": "here",
2763 "net_name": "ns-name-vld-name",
2765 "provider_network_profile": "some-profile-here",
2769 result
= Ns
._process
_net
_params
(
2770 target_vld
=target_vld
,
2773 target_record_id
=target_record_id
,
2776 self
.assertDictEqual(expected_result
, result
)
2778 def test__process_net_params_with_underlay_elan(
2783 "underlay": "some-underlay-here",
2790 "provider_network": "some-profile-here",
2792 "some_ip_profile": "here",
2795 target_record_id
= ""
2799 "some_ip_profile": "here",
2801 "net_name": "ns-name-vld-name",
2803 "provider_network_profile": "some-profile-here",
2807 result
= Ns
._process
_net
_params
(
2808 target_vld
=target_vld
,
2811 target_record_id
=target_record_id
,
2814 self
.assertDictEqual(expected_result
, result
)
2816 def test__get_cloud_init_exception(self
):
2817 db_mock
= MagicMock(name
="database mock")
2822 with self
.assertRaises(NsException
):
2823 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2825 def test__get_cloud_init_file_fs_exception(self
):
2826 db_mock
= MagicMock(name
="database mock")
2829 location
= "vnfr_id_123456:file:test_file"
2830 db_mock
.get_one
.return_value
= {
2833 "folder": "/home/osm",
2834 "pkg-dir": "vnfr_test_dir",
2839 with self
.assertRaises(NsException
):
2840 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2842 def test__get_cloud_init_file(self
):
2843 db_mock
= MagicMock(name
="database mock")
2844 fs_mock
= MagicMock(name
="filesystem mock")
2845 file_mock
= MagicMock(name
="file mock")
2847 location
= "vnfr_id_123456:file:test_file"
2848 cloud_init_content
= "this is a cloud init file content"
2850 db_mock
.get_one
.return_value
= {
2853 "folder": "/home/osm",
2854 "pkg-dir": "vnfr_test_dir",
2858 fs_mock
.file_open
.return_value
= file_mock
2859 file_mock
.__enter
__.return_value
.read
.return_value
= cloud_init_content
2861 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2863 self
.assertEqual(cloud_init_content
, result
)
2865 def test__get_cloud_init_vdu(self
):
2866 db_mock
= MagicMock(name
="database mock")
2869 location
= "vnfr_id_123456:vdu:0"
2870 cloud_init_content
= "this is a cloud init file content"
2872 db_mock
.get_one
.return_value
= {
2875 "cloud-init": cloud_init_content
,
2880 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2882 self
.assertEqual(cloud_init_content
, result
)
2884 @patch("jinja2.Environment.__init__")
2885 def test__parse_jinja2_undefined_error(self
, env_mock
: Mock
):
2886 cloud_init_content
= None
2890 env_mock
.side_effect
= UndefinedError("UndefinedError occurred.")
2892 with self
.assertRaises(NsException
):
2894 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2897 @patch("jinja2.Environment.__init__")
2898 def test__parse_jinja2_template_error(self
, env_mock
: Mock
):
2899 cloud_init_content
= None
2903 env_mock
.side_effect
= TemplateError("TemplateError occurred.")
2905 with self
.assertRaises(NsException
):
2907 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2910 @patch("jinja2.Environment.__init__")
2911 def test__parse_jinja2_template_not_found(self
, env_mock
: Mock
):
2912 cloud_init_content
= None
2916 env_mock
.side_effect
= TemplateNotFound("TemplateNotFound occurred.")
2918 with self
.assertRaises(NsException
):
2920 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2923 def test_rendering_jinja2_temp_without_special_characters(self
):
2924 cloud_init_content
= """
2927 table_type: {{type}}
2929 overwrite: {{is_override}}
2932 - [ sh, -xc, "echo $(date) '{{command}}'" ]
2936 "is_override": "False",
2937 "command": "; mkdir abc",
2939 context
= "cloud-init for VM"
2940 expected_result
= """
2948 - [ sh, -xc, "echo $(date) '; mkdir abc'" ]
2950 result
= Ns
._parse
_jinja
2(
2951 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2953 self
.assertEqual(result
, expected_result
)
2955 def test_rendering_jinja2_temp_with_special_characters(self
):
2956 cloud_init_content
= """
2959 table_type: {{type}}
2961 overwrite: {{is_override}}
2964 - [ sh, -xc, "echo $(date) '{{command}}'" ]
2968 "is_override": "False",
2969 "command": "& rm -rf",
2971 context
= "cloud-init for VM"
2972 expected_result
= """
2980 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
2982 result
= Ns
._parse
_jinja
2(
2983 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2985 self
.assertNotEqual(result
, expected_result
)
2987 def test_rendering_jinja2_temp_with_special_characters_autoescape_is_false(self
):
2988 with
patch("osm_ng_ro.ns.Environment") as mock_environment
:
2989 mock_environment
.return_value
= Environment(
2990 undefined
=StrictUndefined
,
2991 autoescape
=select_autoescape(default_for_string
=False, default
=False),
2993 cloud_init_content
= """
2996 table_type: {{type}}
2998 overwrite: {{is_override}}
3001 - [ sh, -xc, "echo $(date) '{{command}}'" ]
3005 "is_override": "False",
3006 "command": "& rm -rf /",
3008 context
= "cloud-init for VM"
3009 expected_result
= """
3017 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
3019 result
= Ns
._parse
_jinja
2(
3020 cloud_init_content
=cloud_init_content
,
3024 self
.assertEqual(result
, expected_result
)
3026 @patch("osm_ng_ro.ns.Ns._assign_vim")
3027 def test__rebuild_start_stop_task__successful(self
, assign_vim
):
3030 actions
= ["start", "stop", "rebuild"]
3033 for action
in actions
:
3035 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3038 extra_dict
["params"] = params
3039 expected_result
= deepcopy(expected_result_rebuild_start_stop
)
3042 ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3043 expected_result
["params"] = params
3044 task
= self
.ns
.rebuild_start_stop_task(
3054 self
.assertDictEqual(task
, expected_result
)
3056 @patch("osm_ng_ro.ns.Ns._assign_vim")
3057 def test__rebuild_start_stop_task__empty_extra_dict__task_without_params(
3062 actions
= ["start", "stop", "rebuild"]
3065 expected_result
= deepcopy(expected_result_rebuild_start_stop
)
3068 ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3070 task
= self
.ns
.rebuild_start_stop_task(
3080 self
.assertDictEqual(task
, expected_result
)
3082 @patch("osm_ng_ro.ns.Ns._assign_vim")
3083 def test__rebuild_start_stop_task__different_vdu_index__target_record_changes(
3088 actions
= ["start", "stop", "rebuild"]
3091 for action
in actions
:
3093 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3096 extra_dict
["params"] = params
3097 expected_result
= deepcopy(expected_result_rebuild_start_stop
)
3100 ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3101 expected_result
["params"] = params
3102 task
= self
.ns
.rebuild_start_stop_task(
3112 self
.assertDictEqual(task
, expected_result
)
3114 @patch("osm_ng_ro.ns.Ns._assign_vim")
3115 def test__rebuild_start_stop_task__different_task_index__task_id_changes(
3120 actions
= ["start", "stop", "rebuild"]
3123 for action
in actions
:
3125 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3128 extra_dict
["params"] = params
3129 expected_result
= deepcopy(expected_result_rebuild_start_stop
)
3132 ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3133 expected_result
["params"] = params
3134 expected_result
["task_id"] = "bb937f49-3870-4169-b758-9732e1ff40f3:3"
3135 task
= self
.ns
.rebuild_start_stop_task(
3145 self
.assertDictEqual(task
, expected_result
)
3147 @patch("osm_ng_ro.ns.Ns._assign_vim")
3148 def test__rebuild_start_stop_task__assign_vim_raises__task_is_not_created(
3153 actions
= ["start", "stop", "rebuild"]
3156 for action
in actions
:
3158 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3161 extra_dict
["params"] = params
3162 assign_vim
.side_effect
= TestException("Can not connect to VIM.")
3163 with self
.assertRaises(TestException
) as err
:
3164 task
= self
.ns
.rebuild_start_stop_task(
3174 self
.assertEqual(task
, None)
3175 self
.assertEqual(str(err
.exception
), "Can not connect to VIM.")
3177 @patch("osm_ng_ro.ns.Ns._assign_vim")
3178 def test_verticalscale_task__successful(self
, assign_vim
):
3182 task
= self
.ns
.verticalscale_task(
3189 extra_dict_vertical_scale
,
3191 self
.assertDictEqual(task
, expected_result_vertical_scale
)
3193 @patch("osm_ng_ro.ns.Ns._assign_vim")
3194 def test_verticalscale_task__task_index_changes__task_id_changes(self
, assign_vim
):
3198 expected_result
= deepcopy(expected_result_vertical_scale
)
3199 expected_result
["task_id"] = "bb937f49-3870-4169-b758-9732e1ff40f3:2"
3200 task
= self
.ns
.verticalscale_task(
3207 extra_dict_vertical_scale
,
3209 self
.assertDictEqual(task
, expected_result
)
3211 @patch("osm_ng_ro.ns.Ns._assign_vim")
3212 def test_verticalscale_task__empty_extra_dict__expected_result_without_params(
3216 extra_dict
= {"params": {}}
3220 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3221 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3222 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3223 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
3224 "status": "SCHEDULED",
3226 "item": "verticalscale",
3227 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3228 "target_record_id": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
3230 "flavor_id": "TASK-nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0"
3232 "depends_on": ["nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0"],
3235 task
= self
.ns
.verticalscale_task(
3236 vdu
, vnf
, vdu_index
, action_id
, nsr_id_2
, task_index
, extra_dict
3238 self
.assertDictEqual(task
, expected_result
)
3240 @patch("osm_ng_ro.ns.Ns._assign_vim")
3241 def test_verticalscale_task__assign_vim_raises__task_is_not_created(
3247 assign_vim
.side_effect
= TestException("Can not connect to VIM.")
3248 with self
.assertRaises(TestException
) as err
:
3249 task
= self
.ns
.verticalscale_task(
3256 extra_dict_vertical_scale
,
3258 self
.assertEqual(task
, {})
3259 self
.assertEqual(str(err
.exception
), "Can not connect to VIM.")
3261 @patch("osm_ng_ro.ns.Ns._assign_vim")
3262 def test_migrate_task__successful(self
, assign_vim
):
3266 task
= self
.ns
.migrate_task(
3267 vdu
, vnf
, vdu_index
, action_id
, nsr_id_2
, task_index
, extra_dict_migrate
3269 self
.assertDictEqual(task
, expected_result_migrate
)
3271 @patch("osm_ng_ro.ns.Ns._assign_vim")
3272 def test_migrate_task__empty_extra_dict__task_without_params(self
, assign_vim
):
3277 expected_result
= deepcopy(expected_result_migrate
)
3278 expected_result
.pop("params")
3279 task
= self
.ns
.migrate_task(
3280 vdu
, vnf
, vdu_index
, action_id
, nsr_id_2
, task_index
, extra_dict
3282 self
.assertDictEqual(task
, expected_result
)
3284 @patch("osm_ng_ro.ns.Ns._assign_vim")
3285 def test_migrate_task__different_vdu_index__target_record_with_different_vdu_index(
3291 expected_result
= deepcopy(expected_result_migrate
)
3294 ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3295 task
= self
.ns
.migrate_task(
3296 vdu
, vnf
, vdu_index
, action_id
, nsr_id_2
, task_index
, extra_dict_migrate
3298 self
.assertDictEqual(task
, expected_result
)
3300 @patch("osm_ng_ro.ns.Ns._assign_vim")
3301 def test_migrate_task__assign_vim_raises__task_is_not_created(self
, assign_vim
):
3305 assign_vim
.side_effect
= TestException("Can not connect to VIM.")
3306 with self
.assertRaises(TestException
) as err
:
3307 task
= self
.ns
.migrate_task(
3308 vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict_migrate
3310 self
.assertDictEqual(task
, {})
3311 self
.assertEqual(str(err
.exception
), "Can not connect to VIM.")
3314 class TestProcessVduParams(unittest
.TestCase
):
3317 self
.logger
= CopyingMock(autospec
=True)
3319 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3320 def test_find_persistent_root_volumes_empty_instantiation_vol_list(
3321 self
, mock_volume_keeping_required
3323 """Find persistent root volume, instantiation_vol_list is empty."""
3324 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3325 target_vdu
= target_vdu_wth_persistent_storage
3326 vdu_instantiation_volumes_list
= []
3328 mock_volume_keeping_required
.return_value
= True
3329 expected_root_disk
= {
3330 "id": "persistent-root-volume",
3331 "type-of-storage": "persistent-storage:persistent-storage",
3332 "size-of-storage": "10",
3333 "vdu-storage-requirements": [{"key": "keep-volume", "value": "true"}],
3335 expected_persist_root_disk
= {
3336 "persistent-root-volume": {
3337 "image_id": "ubuntu20.04",
3342 expected_disk_list
= [
3344 "image_id": "ubuntu20.04",
3349 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3350 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3352 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3353 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3354 self
.assertEqual(disk_list
, expected_disk_list
)
3355 self
.assertEqual(len(disk_list
), 1)
3357 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3358 def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(
3359 self
, mock_volume_keeping_required
3361 """Find persistent root volume, always selects the first vsd as root volume."""
3362 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3363 vnfd
["vdu"][0]["virtual-storage-desc"] = [
3364 "persistent-volume2",
3365 "persistent-root-volume",
3368 target_vdu
= target_vdu_wth_persistent_storage
3369 vdu_instantiation_volumes_list
= []
3371 mock_volume_keeping_required
.return_value
= True
3372 expected_root_disk
= {
3373 "id": "persistent-volume2",
3374 "type-of-storage": "persistent-storage:persistent-storage",
3375 "size-of-storage": "10",
3377 expected_persist_root_disk
= {
3378 "persistent-volume2": {
3379 "image_id": "ubuntu20.04",
3384 expected_disk_list
= [
3386 "image_id": "ubuntu20.04",
3391 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3392 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3394 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3395 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3396 self
.assertEqual(disk_list
, expected_disk_list
)
3397 self
.assertEqual(len(disk_list
), 1)
3399 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3400 def test_find_persistent_root_volumes_empty_size_of_storage(
3401 self
, mock_volume_keeping_required
3403 """Find persistent root volume, size of storage is empty."""
3404 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3405 vnfd
["virtual-storage-desc"][0]["size-of-storage"] = ""
3406 vnfd
["vdu"][0]["virtual-storage-desc"] = [
3407 "persistent-volume2",
3408 "persistent-root-volume",
3411 target_vdu
= target_vdu_wth_persistent_storage
3412 vdu_instantiation_volumes_list
= []
3414 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3415 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3417 self
.assertEqual(persist_root_disk
, {})
3418 mock_volume_keeping_required
.assert_not_called()
3419 self
.assertEqual(disk_list
, [])
3421 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3422 def test_find_persistent_root_volumes_keeping_is_not_required(
3423 self
, mock_volume_keeping_required
3425 """Find persistent root volume, volume keeping is not required."""
3426 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3427 vnfd
["virtual-storage-desc"][1]["vdu-storage-requirements"] = [
3428 {"key": "keep-volume", "value": "false"},
3430 target_vdu
= target_vdu_wth_persistent_storage
3431 vdu_instantiation_volumes_list
= []
3433 mock_volume_keeping_required
.return_value
= False
3434 expected_root_disk
= {
3435 "id": "persistent-root-volume",
3436 "type-of-storage": "persistent-storage:persistent-storage",
3437 "size-of-storage": "10",
3438 "vdu-storage-requirements": [{"key": "keep-volume", "value": "false"}],
3440 expected_persist_root_disk
= {
3441 "persistent-root-volume": {
3442 "image_id": "ubuntu20.04",
3447 expected_disk_list
= [
3449 "image_id": "ubuntu20.04",
3454 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3455 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3457 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3458 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3459 self
.assertEqual(disk_list
, expected_disk_list
)
3460 self
.assertEqual(len(disk_list
), 1)
3462 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3463 def test_find_persistent_root_volumes_target_vdu_mismatch(
3464 self
, mock_volume_keeping_required
3466 """Find persistent root volume, target vdu name is not matching."""
3467 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3468 vnfd
["vdu"][0]["name"] = "Several_Volumes-VM"
3469 target_vdu
= target_vdu_wth_persistent_storage
3470 vdu_instantiation_volumes_list
= []
3472 result
= self
.ns
.find_persistent_root_volumes(
3473 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3475 self
.assertEqual(result
, None)
3476 mock_volume_keeping_required
.assert_not_called()
3477 self
.assertEqual(disk_list
, [])
3478 self
.assertEqual(len(disk_list
), 0)
3480 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3481 def test_find_persistent_root_volumes_with_instantiation_vol_list(
3482 self
, mock_volume_keeping_required
3484 """Find persistent root volume, existing volume needs to be used."""
3485 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3486 target_vdu
= target_vdu_wth_persistent_storage
3487 vdu_instantiation_volumes_list
= [
3489 "vim-volume-id": vim_volume_id
,
3490 "name": "persistent-root-volume",
3494 expected_persist_root_disk
= {
3495 "persistent-root-volume": {
3496 "vim_volume_id": vim_volume_id
,
3497 "image_id": "ubuntu20.04",
3500 expected_disk_list
= [
3502 "vim_volume_id": vim_volume_id
,
3503 "image_id": "ubuntu20.04",
3506 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3507 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3509 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3510 mock_volume_keeping_required
.assert_not_called()
3511 self
.assertEqual(disk_list
, expected_disk_list
)
3512 self
.assertEqual(len(disk_list
), 1)
3514 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3515 def test_find_persistent_root_volumes_invalid_instantiation_params(
3516 self
, mock_volume_keeping_required
3518 """Find persistent root volume, existing volume id keyword is invalid."""
3519 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3520 target_vdu
= target_vdu_wth_persistent_storage
3521 vdu_instantiation_volumes_list
= [
3523 "volume-id": vim_volume_id
,
3524 "name": "persistent-root-volume",
3528 with self
.assertRaises(KeyError):
3529 self
.ns
.find_persistent_root_volumes(
3530 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3532 mock_volume_keeping_required
.assert_not_called()
3533 self
.assertEqual(disk_list
, [])
3534 self
.assertEqual(len(disk_list
), 0)
3536 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3537 def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list(
3538 self
, mock_volume_keeping_required
3540 """Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty."""
3541 persistent_root_disk
= {
3542 "persistent-root-volume": {
3543 "image_id": "ubuntu20.04",
3548 mock_volume_keeping_required
.return_value
= False
3549 target_vdu
= target_vdu_wth_persistent_storage
3550 vdu_instantiation_volumes_list
= []
3553 "image_id": "ubuntu20.04",
3559 "id": "persistent-volume2",
3560 "size-of-storage": "10",
3561 "type-of-storage": "persistent-storage:persistent-storage",
3563 expected_disk_list
= [
3565 "image_id": "ubuntu20.04",
3574 self
.ns
.find_persistent_volumes(
3575 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3577 self
.assertEqual(disk_list
, expected_disk_list
)
3578 mock_volume_keeping_required
.assert_called_once_with(expected_disk
)
3580 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3581 def test_find_persistent_volumes_vdu_wth_inst_vol_list(
3582 self
, mock_volume_keeping_required
3584 """Find persistent ordinary volume, vim-volume-id is given as instantiation parameter."""
3585 persistent_root_disk
= {
3586 "persistent-root-volume": {
3587 "image_id": "ubuntu20.04",
3592 vdu_instantiation_volumes_list
= [
3594 "vim-volume-id": vim_volume_id
,
3595 "name": "persistent-volume2",
3598 target_vdu
= target_vdu_wth_persistent_storage
3601 "image_id": "ubuntu20.04",
3606 expected_disk_list
= [
3608 "image_id": "ubuntu20.04",
3613 "vim_volume_id": vim_volume_id
,
3616 self
.ns
.find_persistent_volumes(
3617 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3619 self
.assertEqual(disk_list
, expected_disk_list
)
3620 mock_volume_keeping_required
.assert_not_called()
3622 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3623 def test_find_persistent_volumes_vdu_wthout_persistent_storage(
3624 self
, mock_volume_keeping_required
3626 """Find persistent ordinary volume, there is not any persistent disk."""
3627 persistent_root_disk
= {}
3628 vdu_instantiation_volumes_list
= []
3629 mock_volume_keeping_required
.return_value
= False
3630 target_vdu
= target_vdu_wthout_persistent_storage
3632 self
.ns
.find_persistent_volumes(
3633 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3635 self
.assertEqual(disk_list
, disk_list
)
3636 mock_volume_keeping_required
.assert_not_called()
3638 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3639 def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk(
3640 self
, mock_volume_keeping_required
3642 """There is persistent root disk, but there is not ordinary persistent disk."""
3643 persistent_root_disk
= {
3644 "persistent-root-volume": {
3645 "image_id": "ubuntu20.04",
3650 vdu_instantiation_volumes_list
= []
3651 mock_volume_keeping_required
.return_value
= False
3652 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3653 target_vdu
["virtual-storages"] = [
3655 "id": "persistent-root-volume",
3656 "size-of-storage": "10",
3657 "type-of-storage": "persistent-storage:persistent-storage",
3658 "vdu-storage-requirements": [
3659 {"key": "keep-volume", "value": "true"},
3663 "id": "ephemeral-volume",
3664 "size-of-storage": "1",
3665 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
3670 "image_id": "ubuntu20.04",
3675 self
.ns
.find_persistent_volumes(
3676 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3678 self
.assertEqual(disk_list
, disk_list
)
3679 mock_volume_keeping_required
.assert_not_called()
3681 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3682 def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(
3683 self
, mock_volume_keeping_required
3685 """Find persistent ordinary volume, volume id is not persistent_root_disk dict,
3686 vim-volume-id is given as instantiation parameter but disk id is not matching.
3688 mock_volume_keeping_required
.return_value
= True
3689 vdu_instantiation_volumes_list
= [
3691 "vim-volume-id": vim_volume_id
,
3692 "name": "persistent-volume3",
3695 persistent_root_disk
= {
3696 "persistent-root-volume": {
3697 "image_id": "ubuntu20.04",
3704 "image_id": "ubuntu20.04",
3709 expected_disk_list
= [
3711 "image_id": "ubuntu20.04",
3721 "id": "persistent-volume2",
3722 "size-of-storage": "10",
3723 "type-of-storage": "persistent-storage:persistent-storage",
3725 target_vdu
= target_vdu_wth_persistent_storage
3726 self
.ns
.find_persistent_volumes(
3727 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3729 self
.assertEqual(disk_list
, expected_disk_list
)
3730 mock_volume_keeping_required
.assert_called_once_with(expected_disk
)
3732 def test_is_volume_keeping_required_true(self
):
3733 """Volume keeping is required."""
3734 virtual_storage_descriptor
= {
3735 "id": "persistent-root-volume",
3736 "type-of-storage": "persistent-storage:persistent-storage",
3737 "size-of-storage": "10",
3738 "vdu-storage-requirements": [
3739 {"key": "keep-volume", "value": "true"},
3742 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3743 self
.assertEqual(result
, True)
3745 def test_is_volume_keeping_required_false(self
):
3746 """Volume keeping is not required."""
3747 virtual_storage_descriptor
= {
3748 "id": "persistent-root-volume",
3749 "type-of-storage": "persistent-storage:persistent-storage",
3750 "size-of-storage": "10",
3751 "vdu-storage-requirements": [
3752 {"key": "keep-volume", "value": "false"},
3755 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3756 self
.assertEqual(result
, False)
3758 def test_is_volume_keeping_required_wthout_vdu_storage_reqirement(self
):
3759 """Volume keeping is not required, vdu-storage-requirements key does not exist."""
3760 virtual_storage_descriptor
= {
3761 "id": "persistent-root-volume",
3762 "type-of-storage": "persistent-storage:persistent-storage",
3763 "size-of-storage": "10",
3765 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3766 self
.assertEqual(result
, False)
3768 def test_is_volume_keeping_required_wrong_keyword(self
):
3769 """vdu-storage-requirements key to indicate keeping-volume is wrong."""
3770 virtual_storage_descriptor
= {
3771 "id": "persistent-root-volume",
3772 "type-of-storage": "persistent-storage:persistent-storage",
3773 "size-of-storage": "10",
3774 "vdu-storage-requirements": [
3775 {"key": "hold-volume", "value": "true"},
3778 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3779 self
.assertEqual(result
, False)
3781 def test_sort_vdu_interfaces_position_all_wth_positions(self
):
3782 """Interfaces are sorted according to position, all have positions."""
3783 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3784 target_vdu
["interfaces"] = [
3787 "ns-vld-id": "datanet",
3792 "ns-vld-id": "mgmtnet",
3796 sorted_interfaces
= [
3799 "ns-vld-id": "mgmtnet",
3804 "ns-vld-id": "datanet",
3808 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3809 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3811 def test_sort_vdu_interfaces_position_some_wth_position(self
):
3812 """Interfaces are sorted according to position, some of them have positions."""
3813 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3814 target_vdu
["interfaces"] = [
3817 "ns-vld-id": "mgmtnet",
3821 "ns-vld-id": "datanet",
3825 sorted_interfaces
= [
3828 "ns-vld-id": "datanet",
3833 "ns-vld-id": "mgmtnet",
3836 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3837 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3839 def test_sort_vdu_interfaces_position_empty_interface_list(self
):
3840 """Interface list is empty."""
3841 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3842 target_vdu
["interfaces"] = []
3843 sorted_interfaces
= []
3844 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3845 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3847 def test_partially_locate_vdu_interfaces(self
):
3848 """Some interfaces have positions."""
3849 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3850 target_vdu
["interfaces"] = [
3853 "ns-vld-id": "net1",
3855 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3858 "ns-vld-id": "mgmtnet",
3862 "ns-vld-id": "datanet",
3866 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3867 self
.assertDictEqual(
3868 target_vdu
["interfaces"][0],
3871 "ns-vld-id": "datanet",
3875 self
.assertDictEqual(
3876 target_vdu
["interfaces"][2],
3877 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3880 def test_partially_locate_vdu_interfaces_position_start_from_0(self
):
3881 """Some interfaces have positions, position start from 0."""
3882 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3883 target_vdu
["interfaces"] = [
3886 "ns-vld-id": "net1",
3888 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3891 "ns-vld-id": "mgmtnet",
3895 "ns-vld-id": "datanet",
3899 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3900 self
.assertDictEqual(
3901 target_vdu
["interfaces"][0],
3904 "ns-vld-id": "datanet",
3908 self
.assertDictEqual(
3909 target_vdu
["interfaces"][3],
3910 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3913 def test_partially_locate_vdu_interfaces_wthout_position(self
):
3914 """Interfaces do not have positions."""
3915 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3916 target_vdu
["interfaces"] = interfaces_wthout_positions
3917 expected_result
= deepcopy(target_vdu
["interfaces"])
3918 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3919 self
.assertEqual(target_vdu
["interfaces"], expected_result
)
3921 def test_partially_locate_vdu_interfaces_all_has_position(self
):
3922 """All interfaces have position."""
3923 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3924 target_vdu
["interfaces"] = interfaces_wth_all_positions
3925 expected_interfaces
= [
3928 "ns-vld-id": "net2",
3933 "ns-vld-id": "mgmtnet",
3938 "ns-vld-id": "net1",
3942 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3943 self
.assertEqual(target_vdu
["interfaces"], expected_interfaces
)
3945 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3946 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3947 def test_prepare_vdu_cloud_init(self
, mock_parse_jinja2
, mock_get_cloud_init
):
3948 """Target_vdu has cloud-init and boot-data-drive."""
3949 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3950 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3951 target_vdu
["boot-data-drive"] = "vda"
3953 mock_get_cloud_init
.return_value
= cloud_init_content
3954 mock_parse_jinja2
.return_value
= user_data
3956 "user-data": user_data
,
3957 "boot-data-drive": "vda",
3959 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3960 self
.assertDictEqual(result
, expected_result
)
3961 mock_get_cloud_init
.assert_called_once_with(
3962 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3964 mock_parse_jinja2
.assert_called_once_with(
3965 cloud_init_content
=cloud_init_content
,
3967 context
="sample-cloud-init-path",
3970 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3971 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3972 def test_prepare_vdu_cloud_init_get_cloud_init_raise_exception(
3973 self
, mock_parse_jinja2
, mock_get_cloud_init
3975 """Target_vdu has cloud-init and boot-data-drive, get_cloud_init method raises exception."""
3976 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3977 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3978 target_vdu
["boot-data-drive"] = "vda"
3980 mock_get_cloud_init
.side_effect
= NsException(
3981 "Mismatch descriptor for cloud init."
3984 with self
.assertRaises(NsException
) as err
:
3985 self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3986 self
.assertEqual(str(err
.exception
), "Mismatch descriptor for cloud init.")
3988 mock_get_cloud_init
.assert_called_once_with(
3989 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3991 mock_parse_jinja2
.assert_not_called()
3993 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3994 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3995 def test_prepare_vdu_cloud_init_parse_jinja2_raise_exception(
3996 self
, mock_parse_jinja2
, mock_get_cloud_init
3998 """Target_vdu has cloud-init and boot-data-drive, parse_jinja2 method raises exception."""
3999 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4000 target_vdu
["cloud-init"] = "sample-cloud-init-path"
4001 target_vdu
["boot-data-drive"] = "vda"
4003 mock_get_cloud_init
.return_value
= cloud_init_content
4004 mock_parse_jinja2
.side_effect
= NsException("Error parsing cloud-init content.")
4006 with self
.assertRaises(NsException
) as err
:
4007 self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
4008 self
.assertEqual(str(err
.exception
), "Error parsing cloud-init content.")
4009 mock_get_cloud_init
.assert_called_once_with(
4010 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
4012 mock_parse_jinja2
.assert_called_once_with(
4013 cloud_init_content
=cloud_init_content
,
4015 context
="sample-cloud-init-path",
4018 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
4019 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
4020 def test_prepare_vdu_cloud_init_vdu_wthout_boot_data_drive(
4021 self
, mock_parse_jinja2
, mock_get_cloud_init
4023 """Target_vdu has cloud-init but do not have boot-data-drive."""
4024 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4025 target_vdu
["cloud-init"] = "sample-cloud-init-path"
4027 mock_get_cloud_init
.return_value
= cloud_init_content
4028 mock_parse_jinja2
.return_value
= user_data
4030 "user-data": user_data
,
4032 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
4033 self
.assertDictEqual(result
, expected_result
)
4034 mock_get_cloud_init
.assert_called_once_with(
4035 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
4037 mock_parse_jinja2
.assert_called_once_with(
4038 cloud_init_content
=cloud_init_content
,
4040 context
="sample-cloud-init-path",
4043 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
4044 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
4045 def test_prepare_vdu_cloud_init_exists_in_vdu2cloud_init(
4046 self
, mock_parse_jinja2
, mock_get_cloud_init
4048 """Target_vdu has cloud-init, vdu2cloud_init dict has cloud-init_content."""
4049 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4050 target_vdu
["cloud-init"] = "sample-cloud-init-path"
4051 target_vdu
["boot-data-drive"] = "vda"
4052 vdu2cloud_init
= {"sample-cloud-init-path": cloud_init_content
}
4053 mock_parse_jinja2
.return_value
= user_data
4055 "user-data": user_data
,
4056 "boot-data-drive": "vda",
4058 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
4059 self
.assertDictEqual(result
, expected_result
)
4060 mock_get_cloud_init
.assert_not_called()
4061 mock_parse_jinja2
.assert_called_once_with(
4062 cloud_init_content
=cloud_init_content
,
4064 context
="sample-cloud-init-path",
4067 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
4068 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
4069 def test_prepare_vdu_cloud_init_no_cloud_init(
4070 self
, mock_parse_jinja2
, mock_get_cloud_init
4072 """Target_vdu do not have cloud-init."""
4073 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4074 target_vdu
["boot-data-drive"] = "vda"
4077 "boot-data-drive": "vda",
4079 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
4080 self
.assertDictEqual(result
, expected_result
)
4081 mock_get_cloud_init
.assert_not_called()
4082 mock_parse_jinja2
.assert_not_called()
4084 def test_check_vld_information_of_interfaces_ns_vld_vnf_vld_both_exist(self
):
4085 """ns_vld and vnf_vld both exist."""
4088 "ns-vld-id": "mgmtnet",
4089 "vnf-vld-id": "mgmt_cp_int",
4091 expected_result
= f
"{ns_preffix}:vld.mgmtnet"
4092 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
4093 interface
, ns_preffix
, vnf_preffix
4095 self
.assertEqual(result
, expected_result
)
4097 def test_check_vld_information_of_interfaces_empty_interfaces(self
):
4098 """Interface dict is empty."""
4100 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
4101 interface
, ns_preffix
, vnf_preffix
4103 self
.assertEqual(result
, "")
4105 def test_check_vld_information_of_interfaces_has_only_vnf_vld(self
):
4106 """Interface dict has only vnf_vld."""
4109 "vnf-vld-id": "mgmt_cp_int",
4111 expected_result
= f
"{vnf_preffix}:vld.mgmt_cp_int"
4112 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
4113 interface
, ns_preffix
, vnf_preffix
4115 self
.assertEqual(result
, expected_result
)
4117 def test_check_vld_information_of_interfaces_has_vnf_vld_wthout_vnf_prefix(
4120 """Interface dict has only vnf_vld but vnf_preffix does not exist."""
4123 "vnf-vld-id": "mgmt_cp_int",
4126 with self
.assertRaises(Exception) as err
:
4127 self
.ns
._check
_vld
_information
_of
_interfaces
(
4128 interface
, ns_preffix
, vnf_preffix
4130 self
.assertEqual(type(err
), TypeError)
4132 def test_prepare_interface_port_security_has_security_details(self
):
4133 """Interface dict has port security details."""
4136 "ns-vld-id": "mgmtnet",
4137 "vnf-vld-id": "mgmt_cp_int",
4138 "port-security-enabled": True,
4139 "port-security-disable-strategy": "allow-address-pairs",
4141 expected_interface
= {
4143 "ns-vld-id": "mgmtnet",
4144 "vnf-vld-id": "mgmt_cp_int",
4145 "port_security": True,
4146 "port_security_disable_strategy": "allow-address-pairs",
4148 self
.ns
._prepare
_interface
_port
_security
(interface
)
4149 self
.assertDictEqual(interface
, expected_interface
)
4151 def test_prepare_interface_port_security_empty_interfaces(self
):
4152 """Interface dict is empty."""
4154 expected_interface
= {}
4155 self
.ns
._prepare
_interface
_port
_security
(interface
)
4156 self
.assertDictEqual(interface
, expected_interface
)
4158 def test_prepare_interface_port_security_wthout_port_security(self
):
4159 """Interface dict does not have port security details."""
4162 "ns-vld-id": "mgmtnet",
4163 "vnf-vld-id": "mgmt_cp_int",
4165 expected_interface
= {
4167 "ns-vld-id": "mgmtnet",
4168 "vnf-vld-id": "mgmt_cp_int",
4170 self
.ns
._prepare
_interface
_port
_security
(interface
)
4171 self
.assertDictEqual(interface
, expected_interface
)
4173 def test_create_net_item_of_interface_floating_ip_port_security(self
):
4174 """Interface dict has floating ip, port-security details."""
4177 "vcpi": "sample_vcpi",
4178 "port_security": True,
4179 "port_security_disable_strategy": "allow-address-pairs",
4180 "floating_ip": "10.1.1.12",
4181 "ns-vld-id": "mgmtnet",
4182 "vnf-vld-id": "mgmt_cp_int",
4184 net_text
= f
"{ns_preffix}"
4185 expected_net_item
= {
4187 "port_security": True,
4188 "port_security_disable_strategy": "allow-address-pairs",
4189 "floating_ip": "10.1.1.12",
4190 "net_id": f
"TASK-{ns_preffix}",
4193 result
= self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
4194 self
.assertDictEqual(result
, expected_net_item
)
4196 def test_create_net_item_of_interface_invalid_net_text(self
):
4197 """net-text is invalid."""
4200 "vcpi": "sample_vcpi",
4201 "port_security": True,
4202 "port_security_disable_strategy": "allow-address-pairs",
4203 "floating_ip": "10.1.1.12",
4204 "ns-vld-id": "mgmtnet",
4205 "vnf-vld-id": "mgmt_cp_int",
4208 with self
.assertRaises(TypeError):
4209 self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
4211 def test_create_net_item_of_interface_empty_interface(self
):
4212 """Interface dict is empty."""
4214 net_text
= ns_preffix
4215 expected_net_item
= {
4216 "net_id": f
"TASK-{ns_preffix}",
4219 result
= self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
4220 self
.assertDictEqual(result
, expected_net_item
)
4222 @patch("osm_ng_ro.ns.deep_get")
4223 def test_prepare_type_of_interface_type_sriov(self
, mock_deep_get
):
4224 """Interface type is SR-IOV."""
4227 "vcpi": "sample_vcpi",
4228 "port_security": True,
4229 "port_security_disable_strategy": "allow-address-pairs",
4230 "floating_ip": "10.1.1.12",
4231 "ns-vld-id": "mgmtnet",
4232 "vnf-vld-id": "mgmt_cp_int",
4235 mock_deep_get
.return_value
= "SR-IOV"
4236 net_text
= ns_preffix
4238 expected_net_item
= {
4243 self
.ns
._prepare
_type
_of
_interface
(
4244 interface
, tasks_by_target_record_id
, net_text
, net_item
4246 self
.assertDictEqual(net_item
, expected_net_item
)
4249 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"]["net_type"],
4251 mock_deep_get
.assert_called_once_with(
4252 tasks_by_target_record_id
, net_text
, "extra_dict", "params", "net_type"
4255 @patch("osm_ng_ro.ns.deep_get")
4256 def test_prepare_type_of_interface_type_pic_passthrough_deep_get_return_empty_dict(
4259 """Interface type is PCI-PASSTHROUGH, deep_get method return empty dict."""
4262 "vcpi": "sample_vcpi",
4263 "port_security": True,
4264 "port_security_disable_strategy": "allow-address-pairs",
4265 "floating_ip": "10.1.1.12",
4266 "ns-vld-id": "mgmtnet",
4267 "vnf-vld-id": "mgmt_cp_int",
4268 "type": "PCI-PASSTHROUGH",
4270 mock_deep_get
.return_value
= {}
4271 tasks_by_target_record_id
= {}
4272 net_text
= ns_preffix
4274 expected_net_item
= {
4276 "model": "PCI-PASSTHROUGH",
4277 "type": "PCI-PASSTHROUGH",
4279 self
.ns
._prepare
_type
_of
_interface
(
4280 interface
, tasks_by_target_record_id
, net_text
, net_item
4282 self
.assertDictEqual(net_item
, expected_net_item
)
4283 mock_deep_get
.assert_called_once_with(
4284 tasks_by_target_record_id
, net_text
, "extra_dict", "params", "net_type"
4287 @patch("osm_ng_ro.ns.deep_get")
4288 def test_prepare_type_of_interface_type_mgmt(self
, mock_deep_get
):
4289 """Interface type is mgmt."""
4292 "vcpi": "sample_vcpi",
4293 "port_security": True,
4294 "port_security_disable_strategy": "allow-address-pairs",
4295 "floating_ip": "10.1.1.12",
4296 "ns-vld-id": "mgmtnet",
4297 "vnf-vld-id": "mgmt_cp_int",
4300 tasks_by_target_record_id
= {}
4301 net_text
= ns_preffix
4303 expected_net_item
= {
4306 self
.ns
._prepare
_type
_of
_interface
(
4307 interface
, tasks_by_target_record_id
, net_text
, net_item
4309 self
.assertDictEqual(net_item
, expected_net_item
)
4310 mock_deep_get
.assert_not_called()
4312 @patch("osm_ng_ro.ns.deep_get")
4313 def test_prepare_type_of_interface_type_bridge(self
, mock_deep_get
):
4314 """Interface type is bridge."""
4317 "vcpi": "sample_vcpi",
4318 "port_security": True,
4319 "port_security_disable_strategy": "allow-address-pairs",
4320 "floating_ip": "10.1.1.12",
4321 "ns-vld-id": "mgmtnet",
4322 "vnf-vld-id": "mgmt_cp_int",
4324 tasks_by_target_record_id
= {}
4325 net_text
= ns_preffix
4327 expected_net_item
= {
4331 self
.ns
._prepare
_type
_of
_interface
(
4332 interface
, tasks_by_target_record_id
, net_text
, net_item
4334 self
.assertDictEqual(net_item
, expected_net_item
)
4335 mock_deep_get
.assert_not_called()
4337 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4338 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4339 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4340 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4341 def test_prepare_vdu_interfaces(
4343 mock_type_of_interface
,
4344 mock_item_of_interface
,
4346 mock_vld_information_of_interface
,
4348 """Prepare vdu interfaces successfully."""
4349 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4352 "ns-vld-id": "net1",
4353 "ip-address": "13.2.12.31",
4354 "mgmt-interface": True,
4358 "vnf-vld-id": "net2",
4359 "mac-address": "d0:94:66:ed:fc:e2",
4363 "ns-vld-id": "mgmtnet",
4365 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4367 "params": "test_params",
4368 "find_params": "test_find_params",
4372 net_text_1
= f
"{ns_preffix}:net1"
4373 net_text_2
= f
"{vnf_preffix}:net2"
4374 net_text_3
= f
"{ns_preffix}:mgmtnet"
4377 "net_id": f
"TASK-{ns_preffix}",
4382 "net_id": f
"TASK-{ns_preffix}",
4387 "net_id": f
"TASK-{ns_preffix}",
4390 mock_item_of_interface
.side_effect
= [net_item_1
, net_item_2
, net_item_3
]
4391 mock_vld_information_of_interface
.side_effect
= [
4397 expected_extra_dict
= {
4398 "params": "test_params",
4399 "find_params": "test_find_params",
4400 "depends_on": [net_text_1
, net_text_2
, net_text_3
],
4401 "mgmt_vdu_interface": 0,
4403 updated_net_item1
= deepcopy(net_item_1
)
4404 updated_net_item1
.update({"ip_address": "13.2.12.31"})
4405 updated_net_item2
= deepcopy(net_item_2
)
4406 updated_net_item2
.update({"mac_address": "d0:94:66:ed:fc:e2"})
4407 expected_net_list
= [updated_net_item1
, updated_net_item2
, net_item_3
]
4408 self
.ns
._prepare
_vdu
_interfaces
(
4414 tasks_by_target_record_id
,
4417 _call_mock_vld_information_of_interface
= (
4418 mock_vld_information_of_interface
.call_args_list
4421 _call_mock_vld_information_of_interface
[0][0],
4422 (interface_1
, ns_preffix
, vnf_preffix
),
4425 _call_mock_vld_information_of_interface
[1][0],
4426 (interface_2
, ns_preffix
, vnf_preffix
),
4429 _call_mock_vld_information_of_interface
[2][0],
4430 (interface_3
, ns_preffix
, vnf_preffix
),
4433 _call_mock_port_security
= mock_port_security
.call_args_list
4434 self
.assertEqual(_call_mock_port_security
[0].args
[0], interface_1
)
4435 self
.assertEqual(_call_mock_port_security
[1].args
[0], interface_2
)
4436 self
.assertEqual(_call_mock_port_security
[2].args
[0], interface_3
)
4438 _call_mock_item_of_interface
= mock_item_of_interface
.call_args_list
4439 self
.assertEqual(_call_mock_item_of_interface
[0][0], (interface_1
, net_text_1
))
4440 self
.assertEqual(_call_mock_item_of_interface
[1][0], (interface_2
, net_text_2
))
4441 self
.assertEqual(_call_mock_item_of_interface
[2][0], (interface_3
, net_text_3
))
4443 _call_mock_type_of_interface
= mock_type_of_interface
.call_args_list
4445 _call_mock_type_of_interface
[0][0],
4446 (interface_1
, tasks_by_target_record_id
, net_text_1
, net_item_1
),
4449 _call_mock_type_of_interface
[1][0],
4450 (interface_2
, tasks_by_target_record_id
, net_text_2
, net_item_2
),
4453 _call_mock_type_of_interface
[2][0],
4454 (interface_3
, tasks_by_target_record_id
, net_text_3
, net_item_3
),
4456 self
.assertEqual(net_list
, expected_net_list
)
4457 self
.assertEqual(extra_dict
, expected_extra_dict
)
4458 self
.logger
.error
.assert_not_called()
4460 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4461 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4462 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4463 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4464 def test_prepare_vdu_interfaces_create_net_item_raise_exception(
4466 mock_type_of_interface
,
4467 mock_item_of_interface
,
4469 mock_vld_information_of_interface
,
4471 """Prepare vdu interfaces, create_net_item_of_interface method raise exception."""
4472 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4475 "ns-vld-id": "net1",
4476 "ip-address": "13.2.12.31",
4477 "mgmt-interface": True,
4481 "vnf-vld-id": "net2",
4482 "mac-address": "d0:94:66:ed:fc:e2",
4486 "ns-vld-id": "mgmtnet",
4488 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4490 "params": "test_params",
4491 "find_params": "test_find_params",
4494 net_text_1
= f
"{ns_preffix}:net1"
4495 mock_item_of_interface
.side_effect
= [TypeError, TypeError, TypeError]
4497 mock_vld_information_of_interface
.side_effect
= [net_text_1
]
4499 expected_extra_dict
= {
4500 "params": "test_params",
4501 "find_params": "test_find_params",
4502 "depends_on": [net_text_1
],
4504 with self
.assertRaises(TypeError):
4505 self
.ns
._prepare
_vdu
_interfaces
(
4511 tasks_by_target_record_id
,
4515 _call_mock_vld_information_of_interface
= (
4516 mock_vld_information_of_interface
.call_args_list
4519 _call_mock_vld_information_of_interface
[0][0],
4520 (interface_1
, ns_preffix
, vnf_preffix
),
4523 _call_mock_port_security
= mock_port_security
.call_args_list
4524 self
.assertEqual(_call_mock_port_security
[0].args
[0], interface_1
)
4526 _call_mock_item_of_interface
= mock_item_of_interface
.call_args_list
4527 self
.assertEqual(_call_mock_item_of_interface
[0][0], (interface_1
, net_text_1
))
4529 mock_type_of_interface
.assert_not_called()
4530 self
.logger
.error
.assert_not_called()
4531 self
.assertEqual(net_list
, [])
4532 self
.assertEqual(extra_dict
, expected_extra_dict
)
4534 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4535 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4536 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4537 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4538 def test_prepare_vdu_interfaces_vld_information_is_empty(
4540 mock_type_of_interface
,
4541 mock_item_of_interface
,
4543 mock_vld_information_of_interface
,
4545 """Prepare vdu interfaces, check_vld_information_of_interface method returns empty result."""
4546 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4549 "ns-vld-id": "net1",
4550 "ip-address": "13.2.12.31",
4551 "mgmt-interface": True,
4555 "vnf-vld-id": "net2",
4556 "mac-address": "d0:94:66:ed:fc:e2",
4560 "ns-vld-id": "mgmtnet",
4562 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4564 "params": "test_params",
4565 "find_params": "test_find_params",
4568 mock_vld_information_of_interface
.side_effect
= ["", "", ""]
4570 self
.ns
._prepare
_vdu
_interfaces
(
4576 tasks_by_target_record_id
,
4580 _call_mock_vld_information_of_interface
= (
4581 mock_vld_information_of_interface
.call_args_list
4584 _call_mock_vld_information_of_interface
[0][0],
4585 (interface_1
, ns_preffix
, vnf_preffix
),
4588 _call_mock_vld_information_of_interface
[1][0],
4589 (interface_2
, ns_preffix
, vnf_preffix
),
4592 _call_mock_vld_information_of_interface
[2][0],
4593 (interface_3
, ns_preffix
, vnf_preffix
),
4596 _call_logger
= self
.logger
.error
.call_args_list
4599 ("Interface 0 from vdu several_volumes-VM not connected to any vld",),
4603 ("Interface 1 from vdu several_volumes-VM not connected to any vld",),
4607 ("Interface 2 from vdu several_volumes-VM not connected to any vld",),
4609 self
.assertEqual(net_list
, [])
4613 "params": "test_params",
4614 "find_params": "test_find_params",
4619 mock_item_of_interface
.assert_not_called()
4620 mock_port_security
.assert_not_called()
4621 mock_type_of_interface
.assert_not_called()
4623 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4624 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4625 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4626 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4627 def test_prepare_vdu_interfaces_empty_interface_list(
4629 mock_type_of_interface
,
4630 mock_item_of_interface
,
4632 mock_vld_information_of_interface
,
4634 """Prepare vdu interfaces, interface list is empty."""
4635 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4636 target_vdu
["interfaces"] = []
4639 self
.ns
._prepare
_vdu
_interfaces
(
4645 tasks_by_target_record_id
,
4648 mock_type_of_interface
.assert_not_called()
4649 mock_vld_information_of_interface
.assert_not_called()
4650 mock_item_of_interface
.assert_not_called()
4651 mock_port_security
.assert_not_called()
4653 def test_prepare_vdu_ssh_keys(self
):
4654 """Target_vdu has ssh-keys and ro_nsr_public_key exists."""
4655 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4656 target_vdu
["ssh-keys"] = ["sample-ssh-key"]
4657 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4658 target_vdu
["ssh-access-required"] = True
4660 expected_cloud_config
= {
4661 "key-pairs": ["sample-ssh-key", {"public_key": "path_of_public_key"}]
4663 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4664 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4666 def test_prepare_vdu_ssh_keys_target_vdu_wthout_ssh_keys(self
):
4667 """Target_vdu does not have ssh-keys."""
4668 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4669 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4670 target_vdu
["ssh-access-required"] = True
4672 expected_cloud_config
= {"key-pairs": [{"public_key": "path_of_public_key"}]}
4673 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4674 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4676 def test_prepare_vdu_ssh_keys_ssh_access_is_not_required(self
):
4677 """Target_vdu has ssh-keys, ssh-access is not required."""
4678 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4679 target_vdu
["ssh-keys"] = ["sample-ssh-key"]
4680 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4681 target_vdu
["ssh-access-required"] = False
4683 expected_cloud_config
= {"key-pairs": ["sample-ssh-key"]}
4684 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4685 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4687 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4688 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4689 def test_add_persistent_root_disk_to_disk_list_keep_false(
4690 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4692 """Add persistent root disk to disk_list, keep volume set to False."""
4694 "id": "persistent-root-volume",
4695 "type-of-storage": "persistent-storage:persistent-storage",
4696 "size-of-storage": "10",
4698 mock_select_persistent_root_disk
.return_value
= root_disk
4699 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4700 vnfd
["virtual-storage-desc"][1] = root_disk
4701 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4702 persistent_root_disk
= {}
4704 mock_volume_keeping_required
.return_value
= False
4705 expected_disk_list
= [
4707 "image_id": "ubuntu20.04",
4712 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4713 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4715 self
.assertEqual(disk_list
, expected_disk_list
)
4716 mock_select_persistent_root_disk
.assert_called_once()
4717 mock_volume_keeping_required
.assert_called_once()
4719 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4720 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4721 def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises(
4722 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4724 """Add persistent root disk to disk_list"""
4726 "id": "persistent-root-volume",
4727 "type-of-storage": "persistent-storage:persistent-storage",
4728 "size-of-storage": "10",
4730 mock_select_persistent_root_disk
.side_effect
= AttributeError
4731 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4732 vnfd
["virtual-storage-desc"][1] = root_disk
4733 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4734 persistent_root_disk
= {}
4736 with self
.assertRaises(AttributeError):
4737 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4738 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4740 self
.assertEqual(disk_list
, [])
4741 mock_select_persistent_root_disk
.assert_called_once()
4742 mock_volume_keeping_required
.assert_not_called()
4744 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4745 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4746 def test_add_persistent_root_disk_to_disk_list_keep_true(
4747 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4749 """Add persistent root disk, keeo volume set to True."""
4750 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4751 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4752 mock_volume_keeping_required
.return_value
= True
4754 "id": "persistent-root-volume",
4755 "type-of-storage": "persistent-storage:persistent-storage",
4756 "size-of-storage": "10",
4757 "vdu-storage-requirements": [
4758 {"key": "keep-volume", "value": "true"},
4761 mock_select_persistent_root_disk
.return_value
= root_disk
4762 persistent_root_disk
= {}
4764 expected_disk_list
= [
4766 "image_id": "ubuntu20.04",
4771 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4772 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4774 self
.assertEqual(disk_list
, expected_disk_list
)
4775 mock_volume_keeping_required
.assert_called_once_with(root_disk
)
4777 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4778 def test_add_persistent_ordinary_disk_to_disk_list(
4779 self
, mock_volume_keeping_required
4781 """Add persistent ordinary disk, keeo volume set to True."""
4782 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4783 mock_volume_keeping_required
.return_value
= False
4784 persistent_root_disk
= {
4785 "persistent-root-volume": {
4786 "image_id": "ubuntu20.04",
4792 "id": "persistent-volume2",
4793 "type-of-storage": "persistent-storage:persistent-storage",
4794 "size-of-storage": "10",
4796 persistent_ordinary_disk
= {}
4799 expected_disk_list
= [
4803 "multiattach": False,
4804 "name": "persistent-volume2",
4807 self
.ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
4809 persistent_root_disk
,
4810 persistent_ordinary_disk
,
4814 self
.assertEqual(disk_list
, expected_disk_list
)
4815 mock_volume_keeping_required
.assert_called_once_with(ordinary_disk
)
4817 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4818 def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(
4819 self
, mock_volume_keeping_required
4821 """Add persistent ordinary disk, vsd id is in root_disk dict."""
4822 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4823 mock_volume_keeping_required
.return_value
= False
4824 persistent_root_disk
= {
4825 "persistent-root-volume": {
4826 "image_id": "ubuntu20.04",
4830 "persistent-volume2": {
4834 persistent_ordinary_disk
= {}
4838 self
.ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
4840 persistent_root_disk
,
4841 persistent_ordinary_disk
,
4845 self
.assertEqual(disk_list
, [])
4846 mock_volume_keeping_required
.assert_not_called()
4848 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4849 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4850 def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage(
4851 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4853 """VNFD does not have persistent storage."""
4854 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4855 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4856 mock_select_persistent_root_disk
.return_value
= None
4857 persistent_root_disk
= {}
4859 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4860 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4862 self
.assertEqual(disk_list
, [])
4863 self
.assertEqual(mock_select_persistent_root_disk
.call_count
, 2)
4864 mock_volume_keeping_required
.assert_not_called()
4866 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4867 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4868 def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk(
4869 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4871 """Persistent_root_disk dict is empty."""
4872 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4873 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4874 mock_select_persistent_root_disk
.return_value
= None
4875 persistent_root_disk
= {}
4877 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4878 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4880 self
.assertEqual(disk_list
, [])
4881 self
.assertEqual(mock_select_persistent_root_disk
.call_count
, 2)
4882 mock_volume_keeping_required
.assert_not_called()
4884 def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self
):
4885 """Invalid extra dict."""
4886 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4887 target_vdu
["affinity-or-anti-affinity-group-id"] = "sample_affinity-group-id"
4889 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4890 with self
.assertRaises(NsException
) as err
:
4891 self
.ns
._prepare
_vdu
_affinity
_group
_list
(target_vdu
, extra_dict
, ns_preffix
)
4892 self
.assertEqual(str(err
.exception
), "Invalid extra_dict format.")
4894 def test_prepare_vdu_affinity_group_list_one_affinity_group(self
):
4895 """There is one affinity-group."""
4896 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4897 target_vdu
["affinity-or-anti-affinity-group-id"] = ["sample_affinity-group-id"]
4898 extra_dict
= {"depends_on": []}
4899 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4900 affinity_group_txt
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.sample_affinity-group-id"
4901 expected_result
= [{"affinity_group_id": "TASK-" + affinity_group_txt
}]
4902 expected_extra_dict
= {"depends_on": [affinity_group_txt
]}
4903 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4904 target_vdu
, extra_dict
, ns_preffix
4906 self
.assertDictEqual(extra_dict
, expected_extra_dict
)
4907 self
.assertEqual(result
, expected_result
)
4909 def test_prepare_vdu_affinity_group_list_several_affinity_groups(self
):
4910 """There are two affinity-groups."""
4911 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4912 target_vdu
["affinity-or-anti-affinity-group-id"] = [
4913 "affinity-group-id1",
4914 "affinity-group-id2",
4916 extra_dict
= {"depends_on": []}
4917 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4918 affinity_group_txt1
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id1"
4919 affinity_group_txt2
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id2"
4921 {"affinity_group_id": "TASK-" + affinity_group_txt1
},
4922 {"affinity_group_id": "TASK-" + affinity_group_txt2
},
4924 expected_extra_dict
= {"depends_on": [affinity_group_txt1
, affinity_group_txt2
]}
4925 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4926 target_vdu
, extra_dict
, ns_preffix
4928 self
.assertDictEqual(extra_dict
, expected_extra_dict
)
4929 self
.assertEqual(result
, expected_result
)
4931 def test_prepare_vdu_affinity_group_list_no_affinity_group(self
):
4932 """There is not any affinity-group."""
4933 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4934 extra_dict
= {"depends_on": []}
4935 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4936 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4937 target_vdu
, extra_dict
, ns_preffix
4939 self
.assertDictEqual(extra_dict
, {"depends_on": []})
4940 self
.assertEqual(result
, [])
4942 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4943 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4944 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4945 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4946 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4947 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4948 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4949 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4950 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4951 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4952 def test_process_vdu_params_with_inst_vol_list(
4954 mock_prepare_vdu_affinity_group_list
,
4955 mock_add_persistent_ordinary_disks_to_disk_list
,
4956 mock_add_persistent_root_disk_to_disk_list
,
4957 mock_find_persistent_volumes
,
4958 mock_find_persistent_root_volumes
,
4959 mock_prepare_vdu_ssh_keys
,
4960 mock_prepare_vdu_cloud_init
,
4961 mock_prepare_vdu_interfaces
,
4962 mock_locate_vdu_interfaces
,
4963 mock_sort_vdu_interfaces
,
4965 """Instantiation volume list is empty."""
4966 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4968 target_vdu
["interfaces"] = interfaces_wth_all_positions
4970 vdu_instantiation_vol_list
= [
4972 "vim-volume-id": vim_volume_id
,
4973 "name": "persistent-volume2",
4976 target_vdu
["additionalParams"] = {
4977 "OSM": {"vdu_volumes": vdu_instantiation_vol_list
}
4979 mock_prepare_vdu_cloud_init
.return_value
= {}
4980 mock_prepare_vdu_affinity_group_list
.return_value
= []
4981 persistent_root_disk
= {
4982 "persistent-root-volume": {
4983 "image_id": "ubuntu20.04",
4987 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
4989 new_kwargs
= deepcopy(kwargs
)
4994 "tasks_by_target_record_id": {},
4998 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
4999 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5000 db
.get_one
.return_value
= vnfd
5001 result
= Ns
._process
_vdu
_params
(
5002 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5004 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
5005 mock_locate_vdu_interfaces
.assert_not_called()
5006 mock_prepare_vdu_cloud_init
.assert_called_once()
5007 mock_add_persistent_root_disk_to_disk_list
.assert_not_called()
5008 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5009 mock_prepare_vdu_interfaces
.assert_called_once_with(
5011 expected_extra_dict_copy
,
5018 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5019 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5020 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5021 mock_find_persistent_volumes
.assert_called_once_with(
5022 persistent_root_disk
, target_vdu
, vdu_instantiation_vol_list
, []
5025 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5026 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5027 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5028 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5029 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5030 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5031 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5032 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5033 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5034 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5035 def test_process_vdu_params_wth_affinity_groups(
5037 mock_prepare_vdu_affinity_group_list
,
5038 mock_add_persistent_ordinary_disks_to_disk_list
,
5039 mock_add_persistent_root_disk_to_disk_list
,
5040 mock_find_persistent_volumes
,
5041 mock_find_persistent_root_volumes
,
5042 mock_prepare_vdu_ssh_keys
,
5043 mock_prepare_vdu_cloud_init
,
5044 mock_prepare_vdu_interfaces
,
5045 mock_locate_vdu_interfaces
,
5046 mock_sort_vdu_interfaces
,
5048 """There is cloud-config."""
5049 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
5052 target_vdu
["interfaces"] = interfaces_wth_all_positions
5053 mock_prepare_vdu_cloud_init
.return_value
= {}
5054 mock_prepare_vdu_affinity_group_list
.return_value
= [
5059 new_kwargs
= deepcopy(kwargs
)
5064 "tasks_by_target_record_id": {},
5068 expected_extra_dict3
= deepcopy(expected_extra_dict2
)
5069 expected_extra_dict3
["params"]["affinity_group_list"] = [
5073 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5074 db
.get_one
.return_value
= vnfd
5075 result
= Ns
._process
_vdu
_params
(
5076 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5078 self
.assertDictEqual(result
, expected_extra_dict3
)
5079 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
5080 mock_locate_vdu_interfaces
.assert_not_called()
5081 mock_prepare_vdu_cloud_init
.assert_called_once()
5082 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5083 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5084 mock_prepare_vdu_interfaces
.assert_called_once_with(
5086 expected_extra_dict3
,
5094 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5095 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5096 mock_find_persistent_volumes
.assert_not_called()
5098 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5099 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5100 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5101 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5102 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5103 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5104 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5105 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5106 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5107 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5108 def test_process_vdu_params_wth_cloud_config(
5110 mock_prepare_vdu_affinity_group_list
,
5111 mock_add_persistent_ordinary_disks_to_disk_list
,
5112 mock_add_persistent_root_disk_to_disk_list
,
5113 mock_find_persistent_volumes
,
5114 mock_find_persistent_root_volumes
,
5115 mock_prepare_vdu_ssh_keys
,
5116 mock_prepare_vdu_cloud_init
,
5117 mock_prepare_vdu_interfaces
,
5118 mock_locate_vdu_interfaces
,
5119 mock_sort_vdu_interfaces
,
5121 """There is cloud-config."""
5122 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
5125 target_vdu
["interfaces"] = interfaces_wth_all_positions
5126 mock_prepare_vdu_cloud_init
.return_value
= {
5127 "user-data": user_data
,
5128 "boot-data-drive": "vda",
5130 mock_prepare_vdu_affinity_group_list
.return_value
= []
5132 new_kwargs
= deepcopy(kwargs
)
5137 "tasks_by_target_record_id": {},
5141 expected_extra_dict3
= deepcopy(expected_extra_dict2
)
5142 expected_extra_dict3
["params"]["cloud_config"] = {
5143 "user-data": user_data
,
5144 "boot-data-drive": "vda",
5146 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5147 db
.get_one
.return_value
= vnfd
5148 result
= Ns
._process
_vdu
_params
(
5149 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5151 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
5152 mock_locate_vdu_interfaces
.assert_not_called()
5153 mock_prepare_vdu_cloud_init
.assert_called_once()
5154 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5155 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5156 mock_prepare_vdu_interfaces
.assert_called_once_with(
5158 expected_extra_dict3
,
5165 self
.assertDictEqual(result
, expected_extra_dict3
)
5166 mock_prepare_vdu_ssh_keys
.assert_called_once_with(
5167 target_vdu
, None, {"user-data": user_data
, "boot-data-drive": "vda"}
5169 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5170 mock_find_persistent_volumes
.assert_not_called()
5172 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5173 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5174 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5175 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5176 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5177 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5178 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5179 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5180 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5181 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5182 def test_process_vdu_params_wthout_persistent_storage(
5184 mock_prepare_vdu_affinity_group_list
,
5185 mock_add_persistent_ordinary_disks_to_disk_list
,
5186 mock_add_persistent_root_disk_to_disk_list
,
5187 mock_find_persistent_volumes
,
5188 mock_find_persistent_root_volumes
,
5189 mock_prepare_vdu_ssh_keys
,
5190 mock_prepare_vdu_cloud_init
,
5191 mock_prepare_vdu_interfaces
,
5192 mock_locate_vdu_interfaces
,
5193 mock_sort_vdu_interfaces
,
5195 """There is not any persistent storage."""
5196 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
5199 target_vdu
["interfaces"] = interfaces_wth_all_positions
5200 mock_prepare_vdu_cloud_init
.return_value
= {}
5201 mock_prepare_vdu_affinity_group_list
.return_value
= []
5203 new_kwargs
= deepcopy(kwargs
)
5208 "tasks_by_target_record_id": {},
5212 expected_extra_dict_copy
= deepcopy(expected_extra_dict2
)
5213 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
5214 db
.get_one
.return_value
= vnfd
5215 result
= Ns
._process
_vdu
_params
(
5216 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5218 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
5219 mock_locate_vdu_interfaces
.assert_not_called()
5220 mock_prepare_vdu_cloud_init
.assert_called_once()
5221 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5222 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5223 mock_prepare_vdu_interfaces
.assert_called_once_with(
5225 expected_extra_dict_copy
,
5232 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5233 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5234 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5235 mock_find_persistent_volumes
.assert_not_called()
5237 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5238 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5239 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5240 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5241 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5242 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5243 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5244 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5245 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5246 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5247 def test_process_vdu_params_interfaces_partially_located(
5249 mock_prepare_vdu_affinity_group_list
,
5250 mock_add_persistent_ordinary_disks_to_disk_list
,
5251 mock_add_persistent_root_disk_to_disk_list
,
5252 mock_find_persistent_volumes
,
5253 mock_find_persistent_root_volumes
,
5254 mock_prepare_vdu_ssh_keys
,
5255 mock_prepare_vdu_cloud_init
,
5256 mock_prepare_vdu_interfaces
,
5257 mock_locate_vdu_interfaces
,
5258 mock_sort_vdu_interfaces
,
5260 """Some interfaces have position."""
5261 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5264 target_vdu
["interfaces"] = [
5267 "ns-vld-id": "net1",
5269 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 2},
5272 "ns-vld-id": "mgmtnet",
5275 mock_prepare_vdu_cloud_init
.return_value
= {}
5276 mock_prepare_vdu_affinity_group_list
.return_value
= []
5277 persistent_root_disk
= {
5278 "persistent-root-volume": {
5279 "image_id": "ubuntu20.04",
5284 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5286 new_kwargs
= deepcopy(kwargs
)
5291 "tasks_by_target_record_id": {},
5296 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5297 db
.get_one
.return_value
= vnfd
5298 result
= Ns
._process
_vdu
_params
(
5299 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5301 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
5302 mock_sort_vdu_interfaces
.assert_not_called()
5303 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5304 mock_prepare_vdu_cloud_init
.assert_called_once()
5305 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5306 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5307 mock_prepare_vdu_interfaces
.assert_called_once_with(
5309 expected_extra_dict_copy
,
5316 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5317 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5318 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5319 mock_find_persistent_volumes
.assert_not_called()
5320 mock_find_persistent_root_volumes
.assert_not_called()
5322 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5323 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5324 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5325 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5326 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5327 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5328 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5329 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5330 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5331 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5332 def test_process_vdu_params_no_interface_position(
5334 mock_prepare_vdu_affinity_group_list
,
5335 mock_add_persistent_ordinary_disks_to_disk_list
,
5336 mock_add_persistent_root_disk_to_disk_list
,
5337 mock_find_persistent_volumes
,
5338 mock_find_persistent_root_volumes
,
5339 mock_prepare_vdu_ssh_keys
,
5340 mock_prepare_vdu_cloud_init
,
5341 mock_prepare_vdu_interfaces
,
5342 mock_locate_vdu_interfaces
,
5343 mock_sort_vdu_interfaces
,
5345 """Interfaces do not have position."""
5346 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5349 target_vdu
["interfaces"] = interfaces_wthout_positions
5350 mock_prepare_vdu_cloud_init
.return_value
= {}
5351 mock_prepare_vdu_affinity_group_list
.return_value
= []
5352 persistent_root_disk
= {
5353 "persistent-root-volume": {
5354 "image_id": "ubuntu20.04",
5359 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5360 new_kwargs
= deepcopy(kwargs
)
5365 "tasks_by_target_record_id": {},
5370 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5371 db
.get_one
.return_value
= vnfd
5372 result
= Ns
._process
_vdu
_params
(
5373 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5375 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
5376 mock_sort_vdu_interfaces
.assert_not_called()
5377 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5378 mock_prepare_vdu_cloud_init
.assert_called_once()
5379 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5380 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5381 mock_prepare_vdu_interfaces
.assert_called_once_with(
5383 expected_extra_dict_copy
,
5390 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5391 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5392 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5393 mock_find_persistent_volumes
.assert_not_called()
5394 mock_find_persistent_root_volumes
.assert_not_called()
5396 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5397 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5398 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5399 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5400 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5401 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5402 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5403 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5404 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5405 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5406 def test_process_vdu_params_prepare_vdu_interfaces_raises_exception(
5408 mock_prepare_vdu_affinity_group_list
,
5409 mock_add_persistent_ordinary_disks_to_disk_list
,
5410 mock_add_persistent_root_disk_to_disk_list
,
5411 mock_find_persistent_volumes
,
5412 mock_find_persistent_root_volumes
,
5413 mock_prepare_vdu_ssh_keys
,
5414 mock_prepare_vdu_cloud_init
,
5415 mock_prepare_vdu_interfaces
,
5416 mock_locate_vdu_interfaces
,
5417 mock_sort_vdu_interfaces
,
5419 """Prepare vdu interfaces method raises exception."""
5420 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5423 target_vdu
["interfaces"] = interfaces_wthout_positions
5424 mock_prepare_vdu_cloud_init
.return_value
= {}
5425 mock_prepare_vdu_affinity_group_list
.return_value
= []
5426 persistent_root_disk
= {
5427 "persistent-root-volume": {
5428 "image_id": "ubuntu20.04",
5433 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5434 new_kwargs
= deepcopy(kwargs
)
5439 "tasks_by_target_record_id": {},
5443 mock_prepare_vdu_interfaces
.side_effect
= TypeError
5445 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5446 db
.get_one
.return_value
= vnfd
5447 with self
.assertRaises(Exception) as err
:
5448 Ns
._process
_vdu
_params
(
5449 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5451 self
.assertEqual(type(err
), TypeError)
5452 mock_sort_vdu_interfaces
.assert_not_called()
5453 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5454 mock_prepare_vdu_cloud_init
.assert_not_called()
5455 mock_add_persistent_root_disk_to_disk_list
.assert_not_called()
5456 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5457 mock_prepare_vdu_interfaces
.assert_called_once()
5458 mock_prepare_vdu_ssh_keys
.assert_not_called()
5459 mock_prepare_vdu_affinity_group_list
.assert_not_called()
5460 mock_find_persistent_volumes
.assert_not_called()
5461 mock_find_persistent_root_volumes
.assert_not_called()
5463 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5464 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5465 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5466 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5467 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5468 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5469 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5470 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5471 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5472 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5473 def test_process_vdu_params_add_persistent_root_disk_raises_exception(
5475 mock_prepare_vdu_affinity_group_list
,
5476 mock_add_persistent_ordinary_disks_to_disk_list
,
5477 mock_add_persistent_root_disk_to_disk_list
,
5478 mock_find_persistent_volumes
,
5479 mock_find_persistent_root_volumes
,
5480 mock_prepare_vdu_ssh_keys
,
5481 mock_prepare_vdu_cloud_init
,
5482 mock_prepare_vdu_interfaces
,
5483 mock_locate_vdu_interfaces
,
5484 mock_sort_vdu_interfaces
,
5486 """Add persistent root disk method raises exception."""
5487 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5490 target_vdu
["interfaces"] = interfaces_wthout_positions
5491 mock_prepare_vdu_cloud_init
.return_value
= {}
5492 mock_prepare_vdu_affinity_group_list
.return_value
= []
5493 mock_add_persistent_root_disk_to_disk_list
.side_effect
= KeyError
5494 new_kwargs
= deepcopy(kwargs
)
5499 "tasks_by_target_record_id": {},
5504 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5505 db
.get_one
.return_value
= vnfd
5506 with self
.assertRaises(Exception) as err
:
5507 Ns
._process
_vdu
_params
(
5508 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5510 self
.assertEqual(type(err
), KeyError)
5511 mock_sort_vdu_interfaces
.assert_not_called()
5512 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5513 mock_prepare_vdu_cloud_init
.assert_called_once()
5514 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5515 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5516 mock_prepare_vdu_interfaces
.assert_called_once_with(
5520 f
"{ns_preffix}:image.0",
5521 f
"{ns_preffix}:flavor.0",
5531 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5532 mock_prepare_vdu_affinity_group_list
.assert_not_called()
5533 mock_find_persistent_volumes
.assert_not_called()
5534 mock_find_persistent_root_volumes
.assert_not_called()
5536 def test_select_persistent_root_disk(self
):
5537 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5538 vdu
["virtual-storage-desc"] = [
5539 "persistent-root-volume",
5540 "persistent-volume2",
5543 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5544 expected_result
= vsd
5545 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5546 self
.assertEqual(result
, expected_result
)
5548 def test_select_persistent_root_disk_first_vsd_is_different(self
):
5549 """VDU first virtual-storage-desc is different than vsd id."""
5550 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5551 vdu
["virtual-storage-desc"] = [
5552 "persistent-volume2",
5553 "persistent-root-volume",
5556 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5557 expected_result
= None
5558 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5559 self
.assertEqual(result
, expected_result
)
5561 def test_select_persistent_root_disk_vsd_is_not_persistent(self
):
5562 """vsd type is not persistent."""
5563 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5564 vdu
["virtual-storage-desc"] = [
5565 "persistent-volume2",
5566 "persistent-root-volume",
5569 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5570 vsd
["type-of-storage"] = "etsi-nfv-descriptors:ephemeral-storage"
5571 expected_result
= None
5572 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5573 self
.assertEqual(result
, expected_result
)
5575 def test_select_persistent_root_disk_vsd_does_not_have_size(self
):
5576 """vsd size is None."""
5577 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5578 vdu
["virtual-storage-desc"] = [
5579 "persistent-volume2",
5580 "persistent-root-volume",
5583 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5584 vsd
["size-of-storage"] = None
5585 expected_result
= None
5586 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5587 self
.assertEqual(result
, expected_result
)
5589 def test_select_persistent_root_disk_vdu_wthout_vsd(self
):
5590 """VDU does not have virtual-storage-desc."""
5591 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5592 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5593 expected_result
= None
5594 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5595 self
.assertEqual(result
, expected_result
)
5597 def test_select_persistent_root_disk_invalid_vsd_type(self
):
5598 """vsd is list, expected to be a dict."""
5599 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5600 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"]
5601 with self
.assertRaises(AttributeError):
5602 Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5605 class TestSFC(unittest
.TestCase
):
5608 self
.logger
= CopyingMock(autospec
=True)
5610 @patch("osm_ng_ro.ns.Ns._prefix_ip_address")
5611 @patch("osm_ng_ro.ns.Ns._process_ip_proto")
5612 @patch("osm_ng_ro.ns.Ns._get_vnfr_vdur_text")
5613 def test_process_classification_params(
5614 self
, mock_get_vnfr_vdur_text
, mock_process_ip_proto
, mock_prefix_ip_address
5617 mock_prefix_ip_address
.side_effect
= ["10.10.10.10/32", "20.20.20.20/32"]
5618 mock_process_ip_proto
.return_value
= "tcp"
5619 mock_get_vnfr_vdur_text
.return_value
= "vdur_text"
5620 vim_info
, indata
, target_record_id
= {}, {}, ""
5621 target_classification
= {
5623 "source-ip-address": "10.10.10.10",
5624 "destination-ip-address": "20.20.20.20",
5628 "destination-port": 5555,
5630 "ingress_port_index": 0,
5631 "vim_info": vim_info
,
5636 "depends_on": ["vdur_text"],
5638 "destination_ip_prefix": "20.20.20.20/32",
5639 "destination_port_range_max": 5555,
5640 "destination_port_range_min": 5555,
5641 "logical_source_port": "TASK-vdur_text",
5642 "logical_source_port_index": 0,
5645 "source_ip_prefix": "10.10.10.10/32",
5646 "source_port_range_max": "0",
5647 "source_port_range_min": "0",
5651 result
= self
.ns
._process
_classification
_params
(
5652 target_classification
, indata
, vim_info
, target_record_id
, **kwargs
5654 self
.assertEqual(expected_result
, result
)
5656 def test_process_sfp_params(self
):
5657 sf_text
= "nsrs:1234:sf.sf1"
5658 classi_text
= "nsrs:1234:classification.rule1"
5659 vim_info
, indata
, target_record_id
= {}, {}, ""
5663 "classifications": ["rule1"],
5664 "vim_info": vim_info
,
5667 kwargs
= {"nsr_id": "1234"}
5670 "depends_on": [sf_text
, classi_text
],
5673 "sfs": ["TASK-" + sf_text
],
5674 "classifications": ["TASK-" + classi_text
],
5678 result
= self
.ns
._process
_sfp
_params
(
5679 target_sfp
, indata
, vim_info
, target_record_id
, **kwargs
5681 self
.assertEqual(expected_result
, result
)
5683 def test_process_sf_params(self
):
5684 sfi_text
= "nsrs::sfi.sfi1"
5685 vim_info
, indata
, target_record_id
= {}, {}, ""
5686 target_sf
= {"id": "sf1", "sfis": ["sfi1"], "vim_info": vim_info
}
5688 kwargs
= {"ns_id": "1234"}
5691 "depends_on": [sfi_text
],
5694 "sfis": ["TASK-" + sfi_text
],
5698 result
= self
.ns
._process
_sf
_params
(
5699 target_sf
, indata
, vim_info
, target_record_id
, **kwargs
5701 self
.assertEqual(expected_result
, result
)
5703 @patch("osm_ng_ro.ns.Ns._get_vnfr_vdur_text")
5704 def test_process_sfi_params(self
, mock_get_vnfr_vdur_text
):
5706 mock_get_vnfr_vdur_text
.return_value
= "vdur_text"
5707 vim_info
, indata
, target_record_id
= {}, {}, ""
5710 "ingress_port": "vnf-cp0-ext",
5711 "egress_port": "vnf-cp0-ext",
5714 "ingress_port_index": 0,
5715 "egress_port_index": 0,
5721 "depends_on": ["vdur_text"],
5724 "ingress_port": "TASK-vdur_text",
5725 "egress_port": "TASK-vdur_text",
5726 "ingress_port_index": 0,
5727 "egress_port_index": 0,
5731 result
= self
.ns
._process
_sfi
_params
(
5732 target_sfi
, indata
, vim_info
, target_record_id
, **kwargs
5734 self
.assertEqual(expected_result
, result
)
5736 def test_process_vnfgd_sfp(self
):
5739 "position-desc-id": [
5742 "cp-profile-id": [{"id": "sf1"}],
5743 "match-attributes": [{"id": "rule1"}],
5747 expected_result
= {"id": "sfp1", "sfs": ["sf1"], "classifications": ["rule1"]}
5749 result
= self
.ns
._process
_vnfgd
_sfp
(sfp
)
5750 self
.assertEqual(expected_result
, result
)
5752 def test_process_vnfgd_sf(self
):
5753 sf
= {"id": "sf1", "constituent-profile-elements": [{"id": "sfi1", "order": 0}]}
5754 expected_result
= {"id": "sf1", "sfis": ["sfi1"]}
5756 result
= self
.ns
._process
_vnfgd
_sf
(sf
)
5757 self
.assertEqual(expected_result
, result
)
5759 def test_process_vnfgd_sfi(self
):
5762 "constituent-base-element-id": "vnf",
5764 "ingress-constituent-cpd-id": "vnf-cp0-ext",
5765 "egress-constituent-cpd-id": "vnf-cp0-ext",
5770 "member-vnf-index-ref": "vnf",
5771 "connection-point": [
5773 "name": "vnf-cp0-ext",
5774 "connection-point-id": "vdu-eth0-int",
5775 "connection-point-vdu-id": "5678",
5776 "id": "vnf-cp0-ext",
5783 "ingress_port": "vnf-cp0-ext",
5784 "egress_port": "vnf-cp0-ext",
5787 "ingress_port_index": 0,
5788 "egress_port_index": 0,
5791 result
= self
.ns
._process
_vnfgd
_sfi
(sfi
, db_vnfrs
)
5792 self
.assertEqual(expected_result
, result
)
5794 def test_process_vnfgd_classification(self
):
5798 "source-ip-address": "10.10.10.10",
5799 "destination-ip-address": "20.20.20.20",
5800 "constituent-base-element-id": "vnf",
5801 "constituent-cpd-id": "vnf-cp0-ext",
5802 "destination-port": 5555,
5807 "member-vnf-index-ref": "vnf",
5808 "connection-point": [
5810 "name": "vnf-cp0-ext",
5811 "connection-point-id": "vdu-eth0-int",
5812 "connection-point-vdu-id": "5678",
5813 "id": "vnf-cp0-ext",
5822 "source-ip-address": "10.10.10.10",
5823 "destination-ip-address": "20.20.20.20",
5824 "destination-port": 5555,
5827 "ingress_port_index": 0,
5828 "constituent-base-element-id": "vnf",
5829 "constituent-cpd-id": "vnf-cp0-ext",
5832 result
= self
.ns
._process
_vnfgd
_classification
(classification
, db_vnfrs
)
5833 self
.assertEqual(expected_result
, result
)