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",
209 tasks_by_target_record_id
= {
210 "nsrs:th47f48-9870-4169-b758-9732e1ff40f3": {
213 "net_type": "SR-IOV",
220 "vdu2cloud_init": {},
222 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
223 "member-vnf-index-ref": "vnf-several-volumes",
226 vnfd_wthout_persistent_storage
= {
227 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
231 "vdu-profile": [{"id": "without_volumes-VM", "min-number-of-instances": 1}],
234 "id": "without_volumes-vnf",
235 "product-name": "without_volumes-vnf",
238 "id": "without_volumes-VM",
239 "name": "without_volumes-VM",
240 "sw-image-desc": "ubuntu20.04",
241 "alternative-sw-image-desc": [
245 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
249 "virtual-storage-desc": [
250 {"id": "root-volume", "size-of-storage": "10"},
252 "id": "ephemeral-volume",
253 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
254 "size-of-storage": "1",
260 "path": "/app/storage/",
266 target_vdu_wthout_persistent_storage
= {
267 "_id": "09a0baa7-b7cb-4924-bd63-9f04a1c23960",
270 "vdu-name": "without_volumes-VM",
274 "ns-vld-id": "mgmtnet",
277 "virtual-storages": [
280 "size-of-storage": "10",
283 "id": "ephemeral-volume",
284 "size-of-storage": "1",
285 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
289 cloud_init_content
= """
294 overwrite: {{is_override}}
297 - [ sh, -xc, "echo $(date) '{{command}}'" ]
308 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
312 class CopyingMock(MagicMock
):
313 def __call__(self
, *args
, **kwargs
):
314 args
= deepcopy(args
)
315 kwargs
= deepcopy(kwargs
)
316 return super(CopyingMock
, self
).__call
__(*args
, **kwargs
)
319 class TestNs(unittest
.TestCase
):
323 def test__create_task_without_extra_dict(self
):
325 "target_id": "vim_openstack_1",
326 "action_id": "123456",
328 "task_id": "123456:1",
329 "status": "SCHEDULED",
332 "target_record": "test_target_record",
333 "target_record_id": "test_target_record_id",
336 "action_id": "123456",
341 task
= Ns
._create
_task
(
342 deployment_info
=deployment_info
,
343 target_id
="vim_openstack_1",
346 target_record
="test_target_record",
347 target_record_id
="test_target_record_id",
350 self
.assertEqual(deployment_info
.get("task_index"), 2)
351 self
.assertDictEqual(task
, expected_result
)
353 def test__create_task(self
):
355 "target_id": "vim_openstack_1",
356 "action_id": "123456",
358 "task_id": "123456:1",
359 "status": "SCHEDULED",
362 "target_record": "test_target_record",
363 "target_record_id": "test_target_record_id",
364 # values coming from extra_dict
365 "params": "test_params",
366 "find_params": "test_find_params",
367 "depends_on": "test_depends_on",
370 "action_id": "123456",
375 task
= Ns
._create
_task
(
376 deployment_info
=deployment_info
,
377 target_id
="vim_openstack_1",
380 target_record
="test_target_record",
381 target_record_id
="test_target_record_id",
383 "params": "test_params",
384 "find_params": "test_find_params",
385 "depends_on": "test_depends_on",
389 self
.assertEqual(deployment_info
.get("task_index"), 2)
390 self
.assertDictEqual(task
, expected_result
)
392 @patch("osm_ng_ro.ns.time")
393 def test__create_ro_task(self
, mock_time
: Mock
):
394 now
= 1637324838.994551
395 mock_time
.return_value
= now
397 "target_id": "vim_openstack_1",
398 "action_id": "123456",
400 "task_id": "123456:1",
401 "status": "SCHEDULED",
404 "target_record": "test_target_record",
405 "target_record_id": "test_target_record_id",
406 # values coming from extra_dict
407 "params": "test_params",
408 "find_params": "test_find_params",
409 "depends_on": "test_depends_on",
415 "target_id": "vim_openstack_1",
418 "created_items": None,
432 ro_task
= Ns
._create
_ro
_task
(
433 target_id
="vim_openstack_1",
437 self
.assertDictEqual(ro_task
, expected_result
)
439 def test__process_image_params_with_empty_target_image(self
):
445 result
= Ns
._process
_image
_params
(
446 target_image
=target_image
,
449 target_record_id
=None,
452 self
.assertDictEqual(expected_result
, result
)
454 def test__process_image_params_with_wrong_target_image(self
):
459 "no_image": "to_see_here",
462 result
= Ns
._process
_image
_params
(
463 target_image
=target_image
,
466 target_record_id
=None,
469 self
.assertDictEqual(expected_result
, result
)
471 def test__process_image_params_with_image(self
):
483 result
= Ns
._process
_image
_params
(
484 target_image
=target_image
,
487 target_record_id
=None,
490 self
.assertDictEqual(expected_result
, result
)
492 def test__process_image_params_with_vim_image_id(self
):
501 "vim_image_id": "123456",
504 result
= Ns
._process
_image
_params
(
505 target_image
=target_image
,
508 target_record_id
=None,
511 self
.assertDictEqual(expected_result
, result
)
513 def test__process_image_params_with_image_checksum(self
):
517 "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
522 "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
525 result
= Ns
._process
_image
_params
(
526 target_image
=target_image
,
529 target_record_id
=None,
532 self
.assertDictEqual(expected_result
, result
)
534 def test__get_resource_allocation_params_with_empty_target_image(self
):
536 quota_descriptor
= {}
538 result
= Ns
._get
_resource
_allocation
_params
(
539 quota_descriptor
=quota_descriptor
,
542 self
.assertDictEqual(expected_result
, result
)
544 def test__get_resource_allocation_params_with_wrong_target_image(self
):
547 "no_quota": "present_here",
550 result
= Ns
._get
_resource
_allocation
_params
(
551 quota_descriptor
=quota_descriptor
,
554 self
.assertDictEqual(expected_result
, result
)
556 def test__get_resource_allocation_params_with_limit(self
):
564 result
= Ns
._get
_resource
_allocation
_params
(
565 quota_descriptor
=quota_descriptor
,
568 self
.assertDictEqual(expected_result
, result
)
570 def test__get_resource_allocation_params_with_reserve(self
):
578 result
= Ns
._get
_resource
_allocation
_params
(
579 quota_descriptor
=quota_descriptor
,
582 self
.assertDictEqual(expected_result
, result
)
584 def test__get_resource_allocation_params_with_shares(self
):
592 result
= Ns
._get
_resource
_allocation
_params
(
593 quota_descriptor
=quota_descriptor
,
596 self
.assertDictEqual(expected_result
, result
)
598 def test__get_resource_allocation_params(self
):
610 result
= Ns
._get
_resource
_allocation
_params
(
611 quota_descriptor
=quota_descriptor
,
614 self
.assertDictEqual(expected_result
, result
)
616 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
617 def test__process_guest_epa_quota_params_with_empty_quota_epa_cpu(
625 result
= Ns
._process
_guest
_epa
_quota
_params
(
626 guest_epa_quota
=guest_epa_quota
,
627 epa_vcpu_set
=epa_vcpu_set
,
630 self
.assertDictEqual(expected_result
, result
)
631 self
.assertFalse(resource_allocation
.called
)
633 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
634 def test__process_guest_epa_quota_params_with_empty_quota_false_epa_cpu(
642 result
= Ns
._process
_guest
_epa
_quota
_params
(
643 guest_epa_quota
=guest_epa_quota
,
644 epa_vcpu_set
=epa_vcpu_set
,
647 self
.assertDictEqual(expected_result
, result
)
648 self
.assertFalse(resource_allocation
.called
)
650 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
651 def test__process_guest_epa_quota_params_with_wrong_quota_epa_cpu(
657 "no-quota": "nothing",
661 result
= Ns
._process
_guest
_epa
_quota
_params
(
662 guest_epa_quota
=guest_epa_quota
,
663 epa_vcpu_set
=epa_vcpu_set
,
666 self
.assertDictEqual(expected_result
, result
)
667 self
.assertFalse(resource_allocation
.called
)
669 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
670 def test__process_guest_epa_quota_params_with_wrong_quota_false_epa_cpu(
676 "no-quota": "nothing",
680 result
= Ns
._process
_guest
_epa
_quota
_params
(
681 guest_epa_quota
=guest_epa_quota
,
682 epa_vcpu_set
=epa_vcpu_set
,
685 self
.assertDictEqual(expected_result
, result
)
686 self
.assertFalse(resource_allocation
.called
)
688 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
689 def test__process_guest_epa_quota_params_with_cpu_quota_epa_cpu(
703 result
= Ns
._process
_guest
_epa
_quota
_params
(
704 guest_epa_quota
=guest_epa_quota
,
705 epa_vcpu_set
=epa_vcpu_set
,
708 self
.assertDictEqual(expected_result
, result
)
709 self
.assertFalse(resource_allocation
.called
)
711 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
712 def test__process_guest_epa_quota_params_with_cpu_quota_false_epa_cpu(
732 resource_allocation_param
= {
737 resource_allocation
.return_value
= {
743 result
= Ns
._process
_guest
_epa
_quota
_params
(
744 guest_epa_quota
=guest_epa_quota
,
745 epa_vcpu_set
=epa_vcpu_set
,
748 resource_allocation
.assert_called_once_with(resource_allocation_param
)
749 self
.assertDictEqual(expected_result
, result
)
751 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
752 def test__process_guest_epa_quota_params_with_mem_quota_epa_cpu(
772 resource_allocation_param
= {
777 resource_allocation
.return_value
= {
783 result
= Ns
._process
_guest
_epa
_quota
_params
(
784 guest_epa_quota
=guest_epa_quota
,
785 epa_vcpu_set
=epa_vcpu_set
,
788 resource_allocation
.assert_called_once_with(resource_allocation_param
)
789 self
.assertDictEqual(expected_result
, result
)
791 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
792 def test__process_guest_epa_quota_params_with_mem_quota_false_epa_cpu(
812 resource_allocation_param
= {
817 resource_allocation
.return_value
= {
823 result
= Ns
._process
_guest
_epa
_quota
_params
(
824 guest_epa_quota
=guest_epa_quota
,
825 epa_vcpu_set
=epa_vcpu_set
,
828 resource_allocation
.assert_called_once_with(resource_allocation_param
)
829 self
.assertDictEqual(expected_result
, result
)
831 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
832 def test__process_guest_epa_quota_params_with_disk_io_quota_epa_cpu(
852 resource_allocation_param
= {
857 resource_allocation
.return_value
= {
863 result
= Ns
._process
_guest
_epa
_quota
_params
(
864 guest_epa_quota
=guest_epa_quota
,
865 epa_vcpu_set
=epa_vcpu_set
,
868 resource_allocation
.assert_called_once_with(resource_allocation_param
)
869 self
.assertDictEqual(expected_result
, result
)
871 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
872 def test__process_guest_epa_quota_params_with_disk_io_quota_false_epa_cpu(
892 resource_allocation_param
= {
897 resource_allocation
.return_value
= {
903 result
= Ns
._process
_guest
_epa
_quota
_params
(
904 guest_epa_quota
=guest_epa_quota
,
905 epa_vcpu_set
=epa_vcpu_set
,
908 resource_allocation
.assert_called_once_with(resource_allocation_param
)
909 self
.assertDictEqual(expected_result
, result
)
911 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
912 def test__process_guest_epa_quota_params_with_vif_quota_epa_cpu(
932 resource_allocation_param
= {
937 resource_allocation
.return_value
= {
943 result
= Ns
._process
_guest
_epa
_quota
_params
(
944 guest_epa_quota
=guest_epa_quota
,
945 epa_vcpu_set
=epa_vcpu_set
,
948 resource_allocation
.assert_called_once_with(resource_allocation_param
)
949 self
.assertDictEqual(expected_result
, result
)
951 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
952 def test__process_guest_epa_quota_params_with_vif_quota_false_epa_cpu(
972 resource_allocation_param
= {
977 resource_allocation
.return_value
= {
983 result
= Ns
._process
_guest
_epa
_quota
_params
(
984 guest_epa_quota
=guest_epa_quota
,
985 epa_vcpu_set
=epa_vcpu_set
,
988 resource_allocation
.assert_called_once_with(resource_allocation_param
)
989 self
.assertDictEqual(expected_result
, result
)
991 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
992 def test__process_guest_epa_quota_params_with_quota_epa_cpu(
1037 resource_allocation
.return_value
= {
1043 result
= Ns
._process
_guest
_epa
_quota
_params
(
1044 guest_epa_quota
=guest_epa_quota
,
1045 epa_vcpu_set
=epa_vcpu_set
,
1048 self
.assertTrue(resource_allocation
.called
)
1049 self
.assertDictEqual(expected_result
, result
)
1051 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1052 def test__process_guest_epa_quota_params_with_quota_epa_cpu_no_set(
1054 resource_allocation
,
1100 epa_vcpu_set
= False
1102 resource_allocation
.return_value
= {
1108 result
= Ns
._process
_guest
_epa
_quota
_params
(
1109 guest_epa_quota
=guest_epa_quota
,
1110 epa_vcpu_set
=epa_vcpu_set
,
1113 self
.assertTrue(resource_allocation
.called
)
1114 self
.assertDictEqual(expected_result
, result
)
1116 def test__process_guest_epa_numa_params_with_empty_numa_params(self
):
1117 expected_numa_result
= []
1118 expected_epa_vcpu_set_result
= False
1119 guest_epa_quota
= {}
1121 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1122 guest_epa_quota
=guest_epa_quota
,
1124 self
.assertEqual(expected_numa_result
, numa_result
)
1125 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1127 def test__process_guest_epa_numa_params_with_wrong_numa_params(self
):
1128 expected_numa_result
= []
1129 expected_epa_vcpu_set_result
= False
1130 guest_epa_quota
= {"no_nume": "here"}
1132 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1133 guest_epa_quota
=guest_epa_quota
,
1136 self
.assertEqual(expected_numa_result
, numa_result
)
1137 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1139 def test__process_guest_epa_numa_params_with_numa_node_policy(self
):
1140 expected_numa_result
= []
1141 expected_epa_vcpu_set_result
= False
1142 guest_epa_quota
= {"numa-node-policy": {}}
1144 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1145 guest_epa_quota
=guest_epa_quota
,
1148 self
.assertEqual(expected_numa_result
, numa_result
)
1149 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1151 def test__process_guest_epa_numa_params_with_no_node(self
):
1152 expected_numa_result
= []
1153 expected_epa_vcpu_set_result
= False
1155 "numa-node-policy": {
1160 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1161 guest_epa_quota
=guest_epa_quota
,
1164 self
.assertEqual(expected_numa_result
, numa_result
)
1165 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1167 def test__process_guest_epa_numa_params_with_1_node_num_cores(self
):
1168 expected_numa_result
= [{"cores": 3}]
1169 expected_epa_vcpu_set_result
= True
1171 "numa-node-policy": {
1180 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1181 guest_epa_quota
=guest_epa_quota
,
1184 self
.assertEqual(expected_numa_result
, numa_result
)
1185 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1187 def test__process_guest_epa_numa_params_with_1_node_paired_threads(self
):
1188 expected_numa_result
= [{"paired_threads": 3}]
1189 expected_epa_vcpu_set_result
= True
1191 "numa-node-policy": {
1194 "paired-threads": {"num-paired-threads": "3"},
1200 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1201 guest_epa_quota
=guest_epa_quota
,
1204 self
.assertEqual(expected_numa_result
, numa_result
)
1205 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1207 def test__process_guest_epa_numa_params_with_1_node_paired_threads_ids(self
):
1208 expected_numa_result
= [
1210 "paired-threads-id": [("0", "1"), ("4", "5")],
1213 expected_epa_vcpu_set_result
= False
1215 "numa-node-policy": {
1219 "paired-thread-ids": [
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_threads(self
):
1243 expected_numa_result
= [{"threads": 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_memory_mb(self
):
1263 expected_numa_result
= [{"memory": 2}]
1264 expected_epa_vcpu_set_result
= False
1266 "numa-node-policy": {
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_vcpu(self
):
1283 expected_numa_result
= [
1289 expected_epa_vcpu_set_result
= False
1291 "numa-node-policy": {
1292 "node": [{"id": "0", "vcpu": [{"id": "0"}, {"id": "1"}]}],
1296 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1297 guest_epa_quota
=guest_epa_quota
,
1300 self
.assertEqual(expected_numa_result
, numa_result
)
1301 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1303 def test__process_guest_epa_numa_params_with_2_node_vcpu(self
):
1304 expected_numa_result
= [
1315 expected_epa_vcpu_set_result
= False
1317 "numa-node-policy": {
1319 {"id": "0", "vcpu": [{"id": "0"}, {"id": "1"}]},
1320 {"id": "1", "vcpu": [{"id": "2"}, {"id": "3"}]},
1325 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1326 guest_epa_quota
=guest_epa_quota
,
1329 self
.assertEqual(expected_numa_result
, numa_result
)
1330 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1332 def test__process_guest_epa_numa_params_with_1_node(self
):
1333 expected_numa_result
= [
1338 "paired_threads": 3,
1339 "paired-threads-id": [("0", "1"), ("4", "5")],
1344 expected_epa_vcpu_set_result
= True
1346 "numa-node-policy": {
1351 "num-paired-threads": "3",
1352 "paired-thread-ids": [
1370 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1371 guest_epa_quota
=guest_epa_quota
,
1374 self
.assertEqual(expected_numa_result
, numa_result
)
1375 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1377 def test__process_guest_epa_numa_params_with_2_nodes(self
):
1378 expected_numa_result
= [
1381 "paired_threads": 3,
1382 "paired-threads-id": [("0", "1"), ("4", "5")],
1388 "paired_threads": 7,
1389 "paired-threads-id": [("2", "3"), ("5", "6")],
1394 expected_epa_vcpu_set_result
= True
1396 "numa-node-policy": {
1401 "num-paired-threads": "3",
1402 "paired-thread-ids": [
1419 "num-paired-threads": "7",
1420 "paired-thread-ids": [
1438 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1439 guest_epa_quota
=guest_epa_quota
,
1442 self
.assertEqual(expected_numa_result
, numa_result
)
1443 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1445 def test__process_guest_epa_cpu_pinning_params_with_empty_params(self
):
1446 expected_numa_result
= {}
1447 expected_epa_vcpu_set_result
= False
1448 guest_epa_quota
= {}
1450 epa_vcpu_set
= False
1452 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1453 guest_epa_quota
=guest_epa_quota
,
1454 vcpu_count
=vcpu_count
,
1455 epa_vcpu_set
=epa_vcpu_set
,
1458 self
.assertDictEqual(expected_numa_result
, numa_result
)
1459 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1461 def test__process_guest_epa_cpu_pinning_params_with_wrong_params(self
):
1462 expected_numa_result
= {}
1463 expected_epa_vcpu_set_result
= False
1465 "no-cpu-pinning-policy": "here",
1468 epa_vcpu_set
= False
1470 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1471 guest_epa_quota
=guest_epa_quota
,
1472 vcpu_count
=vcpu_count
,
1473 epa_vcpu_set
=epa_vcpu_set
,
1476 self
.assertDictEqual(expected_numa_result
, numa_result
)
1477 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1479 def test__process_guest_epa_cpu_pinning_params_with_epa_vcpu_set(self
):
1480 expected_numa_result
= {}
1481 expected_epa_vcpu_set_result
= True
1483 "cpu-pinning-policy": "DEDICATED",
1488 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1489 guest_epa_quota
=guest_epa_quota
,
1490 vcpu_count
=vcpu_count
,
1491 epa_vcpu_set
=epa_vcpu_set
,
1494 self
.assertDictEqual(expected_numa_result
, numa_result
)
1495 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1497 def test__process_guest_epa_cpu_pinning_params_with_policy_prefer(self
):
1498 expected_numa_result
= {"threads": 3}
1499 expected_epa_vcpu_set_result
= True
1501 "cpu-pinning-policy": "DEDICATED",
1502 "cpu-thread-pinning-policy": "PREFER",
1505 epa_vcpu_set
= False
1507 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1508 guest_epa_quota
=guest_epa_quota
,
1509 vcpu_count
=vcpu_count
,
1510 epa_vcpu_set
=epa_vcpu_set
,
1513 self
.assertDictEqual(expected_numa_result
, numa_result
)
1514 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1516 def test__process_guest_epa_cpu_pinning_params_with_policy_isolate(self
):
1517 expected_numa_result
= {"cores": 3}
1518 expected_epa_vcpu_set_result
= True
1520 "cpu-pinning-policy": "DEDICATED",
1521 "cpu-thread-pinning-policy": "ISOLATE",
1524 epa_vcpu_set
= False
1526 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1527 guest_epa_quota
=guest_epa_quota
,
1528 vcpu_count
=vcpu_count
,
1529 epa_vcpu_set
=epa_vcpu_set
,
1532 self
.assertDictEqual(expected_numa_result
, numa_result
)
1533 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1535 def test__process_guest_epa_cpu_pinning_params_with_policy_require(self
):
1536 expected_numa_result
= {"threads": 3}
1537 expected_epa_vcpu_set_result
= True
1539 "cpu-pinning-policy": "DEDICATED",
1540 "cpu-thread-pinning-policy": "REQUIRE",
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(self
):
1555 expected_numa_result
= {"threads": 3}
1556 expected_epa_vcpu_set_result
= True
1558 "cpu-pinning-policy": "DEDICATED",
1561 epa_vcpu_set
= False
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 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1573 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1574 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1575 def test__process_guest_epa_params_with_empty_params(
1577 guest_epa_numa_params
,
1578 guest_epa_cpu_pinning_params
,
1579 guest_epa_quota_params
,
1581 expected_result
= {}
1584 result
= Ns
._process
_epa
_params
(
1585 target_flavor
=target_flavor
,
1588 self
.assertDictEqual(expected_result
, result
)
1589 self
.assertFalse(guest_epa_numa_params
.called
)
1590 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1591 self
.assertFalse(guest_epa_quota_params
.called
)
1593 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1594 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1595 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1596 def test__process_guest_epa_params_with_wrong_params(
1598 guest_epa_numa_params
,
1599 guest_epa_cpu_pinning_params
,
1600 guest_epa_quota_params
,
1602 expected_result
= {}
1604 "no-guest-epa": "here",
1607 result
= Ns
._process
_epa
_params
(
1608 target_flavor
=target_flavor
,
1611 self
.assertDictEqual(expected_result
, result
)
1612 self
.assertFalse(guest_epa_numa_params
.called
)
1613 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1614 self
.assertFalse(guest_epa_quota_params
.called
)
1616 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1617 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1618 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1619 def test__process_guest_epa_params(
1621 guest_epa_numa_params
,
1622 guest_epa_cpu_pinning_params
,
1623 guest_epa_quota_params
,
1626 "mem-policy": "STRICT",
1631 "numa-node-policy": {
1632 "mem-policy": "STRICT",
1637 guest_epa_numa_params
.return_value
= ({}, False)
1638 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1639 guest_epa_quota_params
.return_value
= {}
1641 result
= Ns
._process
_epa
_params
(
1642 target_flavor
=target_flavor
,
1645 self
.assertDictEqual(expected_result
, result
)
1646 self
.assertTrue(guest_epa_numa_params
.called
)
1647 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1648 self
.assertTrue(guest_epa_quota_params
.called
)
1650 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1651 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1652 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1653 def test__process_guest_epa_params_with_mempage_size(
1655 guest_epa_numa_params
,
1656 guest_epa_cpu_pinning_params
,
1657 guest_epa_quota_params
,
1660 "mempage-size": "1G",
1661 "mem-policy": "STRICT",
1666 "mempage-size": "1G",
1667 "numa-node-policy": {
1668 "mem-policy": "STRICT",
1673 guest_epa_numa_params
.return_value
= ({}, False)
1674 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1675 guest_epa_quota_params
.return_value
= {}
1677 result
= Ns
._process
_epa
_params
(
1678 target_flavor
=target_flavor
,
1681 self
.assertDictEqual(expected_result
, result
)
1682 self
.assertTrue(guest_epa_numa_params
.called
)
1683 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1684 self
.assertTrue(guest_epa_quota_params
.called
)
1686 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1687 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1688 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1689 def test__process_guest_epa_params_with_numa(
1691 guest_epa_numa_params
,
1692 guest_epa_cpu_pinning_params
,
1693 guest_epa_quota_params
,
1696 "mempage-size": "1G",
1697 "cpu-pinning-policy": "DEDICATED",
1698 "cpu-thread-pinning-policy": "PREFER",
1703 "paired-threads": 3,
1704 "paired-threads-id": [("0", "1"), ("4", "5")],
1708 "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1709 "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1710 "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1711 "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1716 "mempage-size": "1G",
1717 "cpu-pinning-policy": "DEDICATED",
1718 "cpu-thread-pinning-policy": "PREFER",
1719 "numa-node-policy": {
1724 "num-paired-threads": "3",
1725 "paired-thread-ids": [
1764 guest_epa_numa_params
.return_value
= (
1768 "paired-threads": 3,
1769 "paired-threads-id": [("0", "1"), ("4", "5")],
1776 guest_epa_cpu_pinning_params
.return_value
= (
1782 guest_epa_quota_params
.return_value
= {
1805 result
= Ns
._process
_epa
_params
(
1806 target_flavor
=target_flavor
,
1808 self
.assertEqual(expected_result
, result
)
1809 self
.assertTrue(guest_epa_numa_params
.called
)
1810 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1811 self
.assertTrue(guest_epa_quota_params
.called
)
1813 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1814 def test__process_flavor_params_with_empty_target_flavor(
1822 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1827 target_record_id
= ""
1829 with self
.assertRaises(KeyError):
1830 Ns
._process
_flavor
_params
(
1831 target_flavor
=target_flavor
,
1834 target_record_id
=target_record_id
,
1837 self
.assertFalse(epa_params
.called
)
1839 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1840 def test__process_flavor_params_with_wrong_target_flavor(
1845 "no-target-flavor": "here",
1849 target_record_id
= ""
1851 with self
.assertRaises(KeyError):
1852 Ns
._process
_flavor
_params
(
1853 target_flavor
=target_flavor
,
1856 target_record_id
=target_record_id
,
1859 self
.assertFalse(epa_params
.called
)
1861 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1862 def test__process_flavor_params_with_empty_indata(
1886 "memory-mb": "1024",
1891 target_record_id
= ""
1893 epa_params
.return_value
= {}
1895 result
= Ns
._process
_flavor
_params
(
1896 target_flavor
=target_flavor
,
1899 target_record_id
=target_record_id
,
1902 self
.assertTrue(epa_params
.called
)
1903 self
.assertDictEqual(result
, expected_result
)
1905 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1906 def test__process_flavor_params_with_wrong_indata(
1930 "memory-mb": "1024",
1937 target_record_id
= ""
1939 epa_params
.return_value
= {}
1941 result
= Ns
._process
_flavor
_params
(
1942 target_flavor
=target_flavor
,
1945 target_record_id
=target_record_id
,
1948 self
.assertTrue(epa_params
.called
)
1949 self
.assertDictEqual(result
, expected_result
)
1951 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1952 def test__process_flavor_params_with_ephemeral_disk(
1960 db
.get_one
.return_value
= {
1961 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1966 {"id": "without_volumes-VM", "min-number-of-instances": 1}
1970 "id": "without_volumes-vnf",
1971 "product-name": "without_volumes-vnf",
1974 "id": "without_volumes-VM",
1975 "name": "without_volumes-VM",
1976 "sw-image-desc": "ubuntu20.04",
1977 "alternative-sw-image-desc": [
1979 "ubuntu20.04-azure",
1981 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
1985 "virtual-storage-desc": [
1986 {"id": "root-volume", "size-of-storage": "10"},
1988 "id": "ephemeral-volume",
1989 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1990 "size-of-storage": "1",
1996 "path": "/app/storage/",
2024 "memory-mb": "1024",
2032 "ns-flavor-id": "test_id",
2033 "virtual-storages": [
2035 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2036 "size-of-storage": "10",
2041 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2046 target_record_id
= ""
2048 epa_params
.return_value
= {}
2050 result
= Ns
._process
_flavor
_params
(
2051 target_flavor
=target_flavor
,
2054 target_record_id
=target_record_id
,
2058 self
.assertTrue(epa_params
.called
)
2059 self
.assertDictEqual(result
, expected_result
)
2061 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2062 def test__process_flavor_params_with_swap_disk(
2089 "memory-mb": "1024",
2097 "ns-flavor-id": "test_id",
2098 "virtual-storages": [
2100 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2101 "size-of-storage": "20",
2106 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2111 target_record_id
= ""
2113 epa_params
.return_value
= {}
2115 result
= Ns
._process
_flavor
_params
(
2116 target_flavor
=target_flavor
,
2119 target_record_id
=target_record_id
,
2122 self
.assertTrue(epa_params
.called
)
2123 self
.assertDictEqual(result
, expected_result
)
2125 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2126 def test__process_flavor_params_with_persistent_root_disk(
2134 db
.get_one
.return_value
= {
2135 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2140 {"id": "several_volumes-VM", "min-number-of-instances": 1}
2144 "id": "several_volumes-vnf",
2145 "product-name": "several_volumes-vnf",
2148 "id": "several_volumes-VM",
2149 "name": "several_volumes-VM",
2150 "sw-image-desc": "ubuntu20.04",
2151 "alternative-sw-image-desc": [
2153 "ubuntu20.04-azure",
2155 "virtual-storage-desc": [
2156 "persistent-root-volume",
2161 "virtual-storage-desc": [
2163 "id": "persistent-root-volume",
2164 "type-of-storage": "persistent-storage:persistent-storage",
2165 "size-of-storage": "10",
2171 "path": "/app/storage/",
2197 "memory-mb": "1024",
2205 "vdu-name": "several_volumes-VM",
2206 "ns-flavor-id": "test_id",
2207 "virtual-storages": [
2209 "type-of-storage": "persistent-storage:persistent-storage",
2210 "size-of-storage": "10",
2215 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2220 target_record_id
= ""
2222 epa_params
.return_value
= {}
2224 result
= Ns
._process
_flavor
_params
(
2225 target_flavor
=target_flavor
,
2228 target_record_id
=target_record_id
,
2232 self
.assertTrue(epa_params
.called
)
2233 self
.assertDictEqual(result
, expected_result
)
2235 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2236 def test__process_flavor_params_with_epa_params(
2247 "numa": "there-is-numa-here",
2258 "numa": "there-is-numa-here",
2267 "memory-mb": "1024",
2275 "ns-flavor-id": "test_id",
2278 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2283 target_record_id
= ""
2285 epa_params
.return_value
= {
2286 "numa": "there-is-numa-here",
2289 result
= Ns
._process
_flavor
_params
(
2290 target_flavor
=target_flavor
,
2293 target_record_id
=target_record_id
,
2296 self
.assertTrue(epa_params
.called
)
2297 self
.assertDictEqual(result
, expected_result
)
2299 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2300 def test__process_flavor_params(
2308 db
.get_one
.return_value
= {
2309 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2314 {"id": "without_volumes-VM", "min-number-of-instances": 1}
2318 "id": "without_volumes-vnf",
2319 "product-name": "without_volumes-vnf",
2322 "id": "without_volumes-VM",
2323 "name": "without_volumes-VM",
2324 "sw-image-desc": "ubuntu20.04",
2325 "alternative-sw-image-desc": [
2327 "ubuntu20.04-azure",
2329 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2333 "virtual-storage-desc": [
2334 {"id": "root-volume", "size-of-storage": "10"},
2336 "id": "ephemeral-volume",
2337 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2338 "size-of-storage": "1",
2344 "path": "/app/storage/",
2359 "numa": "there-is-numa-here",
2372 "numa": "there-is-numa-here",
2381 "memory-mb": "1024",
2389 "ns-flavor-id": "test_id",
2390 "virtual-storages": [
2392 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2393 "size-of-storage": "10",
2396 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2397 "size-of-storage": "20",
2402 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2407 target_record_id
= ""
2409 epa_params
.return_value
= {
2410 "numa": "there-is-numa-here",
2413 result
= Ns
._process
_flavor
_params
(
2414 target_flavor
=target_flavor
,
2417 target_record_id
=target_record_id
,
2421 self
.assertTrue(epa_params
.called
)
2422 self
.assertDictEqual(result
, expected_result
)
2424 def test__process_net_params_with_empty_params(
2434 "provider_network": "some-profile-here",
2436 "some_ip_profile": "here",
2439 target_record_id
= ""
2442 "net_name": "ns-name-vld-name",
2443 "net_type": "bridge",
2445 "some_ip_profile": "here",
2447 "provider_network_profile": "some-profile-here",
2451 result
= Ns
._process
_net
_params
(
2452 target_vld
=target_vld
,
2455 target_record_id
=target_record_id
,
2458 self
.assertDictEqual(expected_result
, result
)
2460 def test__process_net_params_with_vim_info_sdn(
2471 "sdn-ports": ["some", "ports", "here"],
2472 "vlds": ["some", "vlds", "here"],
2475 target_record_id
= "vld.sdn.something"
2478 "sdn-ports": ["some", "ports", "here"],
2479 "vlds": ["some", "vlds", "here"],
2484 result
= Ns
._process
_net
_params
(
2485 target_vld
=target_vld
,
2488 target_record_id
=target_record_id
,
2491 self
.assertDictEqual(expected_result
, result
)
2493 def test__process_net_params_with_vim_info_sdn_target_vim(
2504 "sdn-ports": ["some", "ports", "here"],
2505 "vlds": ["some", "vlds", "here"],
2506 "target_vim": "some-vim",
2509 target_record_id
= "vld.sdn.something"
2511 "depends_on": ["some-vim vld.sdn"],
2513 "sdn-ports": ["some", "ports", "here"],
2514 "vlds": ["some", "vlds", "here"],
2515 "target_vim": "some-vim",
2520 result
= Ns
._process
_net
_params
(
2521 target_vld
=target_vld
,
2524 target_record_id
=target_record_id
,
2527 self
.assertDictEqual(expected_result
, result
)
2529 def test__process_net_params_with_vim_network_name(
2539 "vim_network_name": "some-network-name",
2541 target_record_id
= "vld.sdn.something"
2545 "name": "some-network-name",
2550 result
= Ns
._process
_net
_params
(
2551 target_vld
=target_vld
,
2554 target_record_id
=target_record_id
,
2557 self
.assertDictEqual(expected_result
, result
)
2559 def test__process_net_params_with_vim_network_id(
2569 "vim_network_id": "some-network-id",
2571 target_record_id
= "vld.sdn.something"
2575 "id": "some-network-id",
2580 result
= Ns
._process
_net
_params
(
2581 target_vld
=target_vld
,
2584 target_record_id
=target_record_id
,
2587 self
.assertDictEqual(expected_result
, result
)
2589 def test__process_net_params_with_mgmt_network(
2595 "mgmt-network": "some-mgmt-network",
2601 target_record_id
= "vld.sdn.something"
2609 result
= Ns
._process
_net
_params
(
2610 target_vld
=target_vld
,
2613 target_record_id
=target_record_id
,
2616 self
.assertDictEqual(expected_result
, result
)
2618 def test__process_net_params_with_underlay_eline(
2623 "underlay": "some-underlay-here",
2630 "provider_network": "some-profile-here",
2632 "some_ip_profile": "here",
2635 target_record_id
= ""
2639 "some_ip_profile": "here",
2641 "net_name": "ns-name-vld-name",
2643 "provider_network_profile": "some-profile-here",
2647 result
= Ns
._process
_net
_params
(
2648 target_vld
=target_vld
,
2651 target_record_id
=target_record_id
,
2654 self
.assertDictEqual(expected_result
, result
)
2656 def test__process_net_params_with_underlay_elan(
2661 "underlay": "some-underlay-here",
2668 "provider_network": "some-profile-here",
2670 "some_ip_profile": "here",
2673 target_record_id
= ""
2677 "some_ip_profile": "here",
2679 "net_name": "ns-name-vld-name",
2681 "provider_network_profile": "some-profile-here",
2685 result
= Ns
._process
_net
_params
(
2686 target_vld
=target_vld
,
2689 target_record_id
=target_record_id
,
2692 self
.assertDictEqual(expected_result
, result
)
2694 def test__get_cloud_init_exception(self
):
2695 db_mock
= MagicMock(name
="database mock")
2700 with self
.assertRaises(NsException
):
2701 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2703 def test__get_cloud_init_file_fs_exception(self
):
2704 db_mock
= MagicMock(name
="database mock")
2707 location
= "vnfr_id_123456:file:test_file"
2708 db_mock
.get_one
.return_value
= {
2711 "folder": "/home/osm",
2712 "pkg-dir": "vnfr_test_dir",
2717 with self
.assertRaises(NsException
):
2718 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2720 def test__get_cloud_init_file(self
):
2721 db_mock
= MagicMock(name
="database mock")
2722 fs_mock
= MagicMock(name
="filesystem mock")
2723 file_mock
= MagicMock(name
="file mock")
2725 location
= "vnfr_id_123456:file:test_file"
2726 cloud_init_content
= "this is a cloud init file content"
2728 db_mock
.get_one
.return_value
= {
2731 "folder": "/home/osm",
2732 "pkg-dir": "vnfr_test_dir",
2736 fs_mock
.file_open
.return_value
= file_mock
2737 file_mock
.__enter
__.return_value
.read
.return_value
= cloud_init_content
2739 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2741 self
.assertEqual(cloud_init_content
, result
)
2743 def test__get_cloud_init_vdu(self
):
2744 db_mock
= MagicMock(name
="database mock")
2747 location
= "vnfr_id_123456:vdu:0"
2748 cloud_init_content
= "this is a cloud init file content"
2750 db_mock
.get_one
.return_value
= {
2753 "cloud-init": cloud_init_content
,
2758 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2760 self
.assertEqual(cloud_init_content
, result
)
2762 @patch("jinja2.Environment.__init__")
2763 def test__parse_jinja2_undefined_error(self
, env_mock
: Mock
):
2764 cloud_init_content
= None
2768 env_mock
.side_effect
= UndefinedError("UndefinedError occurred.")
2770 with self
.assertRaises(NsException
):
2772 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2775 @patch("jinja2.Environment.__init__")
2776 def test__parse_jinja2_template_error(self
, env_mock
: Mock
):
2777 cloud_init_content
= None
2781 env_mock
.side_effect
= TemplateError("TemplateError occurred.")
2783 with self
.assertRaises(NsException
):
2785 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2788 @patch("jinja2.Environment.__init__")
2789 def test__parse_jinja2_template_not_found(self
, env_mock
: Mock
):
2790 cloud_init_content
= None
2794 env_mock
.side_effect
= TemplateNotFound("TemplateNotFound occurred.")
2796 with self
.assertRaises(NsException
):
2798 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2801 def test_rendering_jinja2_temp_without_special_characters(self
):
2802 cloud_init_content
= """
2805 table_type: {{type}}
2807 overwrite: {{is_override}}
2810 - [ sh, -xc, "echo $(date) '{{command}}'" ]
2814 "is_override": "False",
2815 "command": "; mkdir abc",
2817 context
= "cloud-init for VM"
2818 expected_result
= """
2826 - [ sh, -xc, "echo $(date) '; mkdir abc'" ]
2828 result
= Ns
._parse
_jinja
2(
2829 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2831 self
.assertEqual(result
, expected_result
)
2833 def test_rendering_jinja2_temp_with_special_characters(self
):
2834 cloud_init_content
= """
2837 table_type: {{type}}
2839 overwrite: {{is_override}}
2842 - [ sh, -xc, "echo $(date) '{{command}}'" ]
2846 "is_override": "False",
2847 "command": "& rm -rf",
2849 context
= "cloud-init for VM"
2850 expected_result
= """
2858 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
2860 result
= Ns
._parse
_jinja
2(
2861 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2863 self
.assertNotEqual(result
, expected_result
)
2865 def test_rendering_jinja2_temp_with_special_characters_autoescape_is_false(self
):
2866 with
patch("osm_ng_ro.ns.Environment") as mock_environment
:
2867 mock_environment
.return_value
= Environment(
2868 undefined
=StrictUndefined
,
2869 autoescape
=select_autoescape(default_for_string
=False, default
=False),
2871 cloud_init_content
= """
2874 table_type: {{type}}
2876 overwrite: {{is_override}}
2879 - [ sh, -xc, "echo $(date) '{{command}}'" ]
2883 "is_override": "False",
2884 "command": "& rm -rf /",
2886 context
= "cloud-init for VM"
2887 expected_result
= """
2895 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
2897 result
= Ns
._parse
_jinja
2(
2898 cloud_init_content
=cloud_init_content
,
2902 self
.assertEqual(result
, expected_result
)
2904 @patch("osm_ng_ro.ns.Ns._assign_vim")
2905 def test__rebuild_start_stop_task(self
, assign_vim
):
2908 actions
= ["start", "stop", "rebuild"]
2909 vdu_id
= "bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
2910 vnf_id
= "665b4165-ce24-4320-bf19-b9a45bade49f"
2912 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
2913 nsr_id
= "993166fe-723e-4680-ac4b-b1af2541ae31"
2915 target_vim
= "vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
2916 t
= "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
2917 for action
in actions
:
2919 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
2920 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
2921 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
2922 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:0",
2923 "status": "SCHEDULED",
2926 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0",
2927 "target_record_id": t
,
2929 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
2933 extra_dict
["params"] = {
2934 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
2937 task
= self
.ns
.rebuild_start_stop_task(
2947 self
.assertEqual(task
.get("action_id"), action_id
)
2948 self
.assertEqual(task
.get("nsr_id"), nsr_id
)
2949 self
.assertEqual(task
.get("target_id"), target_vim
)
2950 self
.assertDictEqual(task
, expected_result
)
2952 @patch("osm_ng_ro.ns.Ns._assign_vim")
2953 def test_verticalscale_task(self
, assign_vim
):
2957 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
2958 nsr_id
= "993166fe-723e-4680-ac4b-b1af2541ae31"
2960 target_record_id
= (
2961 "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
2962 "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
2966 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
2967 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
2968 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
2969 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
2970 "status": "SCHEDULED",
2972 "item": "verticalscale",
2973 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1",
2974 "target_record_id": target_record_id
,
2976 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
2977 "flavor_dict": "flavor_dict",
2981 "id": "bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
2983 "vim:f9f370ac-0d44-41a7-9000-457f2332bc35": {"interfaces": []}
2986 vnf
= {"_id": "665b4165-ce24-4320-bf19-b9a45bade49f"}
2987 extra_dict
["params"] = {
2988 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
2989 "flavor_dict": "flavor_dict",
2991 task
= self
.ns
.verticalscale_task(
2992 vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
2995 self
.assertDictEqual(task
, expected_result
)
2997 @patch("osm_ng_ro.ns.Ns._assign_vim")
2998 def test_migrate_task(self
, assign_vim
):
3002 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
3003 nsr_id
= "993166fe-723e-4680-ac4b-b1af2541ae31"
3005 target_record_id
= (
3006 "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
3007 "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3011 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3012 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3013 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3014 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
3015 "status": "SCHEDULED",
3018 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1",
3019 "target_record_id": target_record_id
,
3021 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3022 "migrate_host": "migrateToHost",
3026 "id": "bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
3028 "vim:f9f370ac-0d44-41a7-9000-457f2332bc35": {"interfaces": []}
3031 vnf
= {"_id": "665b4165-ce24-4320-bf19-b9a45bade49f"}
3032 extra_dict
["params"] = {
3033 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3034 "migrate_host": "migrateToHost",
3036 task
= self
.ns
.migrate_task(
3037 vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3040 self
.assertDictEqual(task
, expected_result
)
3043 class TestProcessVduParams(unittest
.TestCase
):
3046 self
.logger
= CopyingMock(autospec
=True)
3048 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3049 def test_find_persistent_root_volumes_empty_instantiation_vol_list(
3050 self
, mock_volume_keeping_required
3052 """Find persistent root volume, instantiation_vol_list is empty."""
3053 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3054 target_vdu
= target_vdu_wth_persistent_storage
3055 vdu_instantiation_volumes_list
= []
3057 mock_volume_keeping_required
.return_value
= True
3058 expected_root_disk
= {
3059 "id": "persistent-root-volume",
3060 "type-of-storage": "persistent-storage:persistent-storage",
3061 "size-of-storage": "10",
3062 "vdu-storage-requirements": [{"key": "keep-volume", "value": "true"}],
3064 expected_persist_root_disk
= {
3065 "persistent-root-volume": {
3066 "image_id": "ubuntu20.04",
3071 expected_disk_list
= [
3073 "image_id": "ubuntu20.04",
3078 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3079 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3081 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3082 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3083 self
.assertEqual(disk_list
, expected_disk_list
)
3084 self
.assertEqual(len(disk_list
), 1)
3086 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3087 def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(
3088 self
, mock_volume_keeping_required
3090 """Find persistent root volume, always selects the first vsd as root volume."""
3091 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3092 vnfd
["vdu"][0]["virtual-storage-desc"] = [
3093 "persistent-volume2",
3094 "persistent-root-volume",
3097 target_vdu
= target_vdu_wth_persistent_storage
3098 vdu_instantiation_volumes_list
= []
3100 mock_volume_keeping_required
.return_value
= True
3101 expected_root_disk
= {
3102 "id": "persistent-volume2",
3103 "type-of-storage": "persistent-storage:persistent-storage",
3104 "size-of-storage": "10",
3106 expected_persist_root_disk
= {
3107 "persistent-volume2": {
3108 "image_id": "ubuntu20.04",
3113 expected_disk_list
= [
3115 "image_id": "ubuntu20.04",
3120 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3121 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3123 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3124 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3125 self
.assertEqual(disk_list
, expected_disk_list
)
3126 self
.assertEqual(len(disk_list
), 1)
3128 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3129 def test_find_persistent_root_volumes_empty_size_of_storage(
3130 self
, mock_volume_keeping_required
3132 """Find persistent root volume, size of storage is empty."""
3133 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3134 vnfd
["virtual-storage-desc"][0]["size-of-storage"] = ""
3135 vnfd
["vdu"][0]["virtual-storage-desc"] = [
3136 "persistent-volume2",
3137 "persistent-root-volume",
3140 target_vdu
= target_vdu_wth_persistent_storage
3141 vdu_instantiation_volumes_list
= []
3143 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3144 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3146 self
.assertEqual(persist_root_disk
, None)
3147 mock_volume_keeping_required
.assert_not_called()
3148 self
.assertEqual(disk_list
, [])
3150 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3151 def test_find_persistent_root_volumes_keeping_is_not_required(
3152 self
, mock_volume_keeping_required
3154 """Find persistent root volume, volume keeping is not required."""
3155 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3156 vnfd
["virtual-storage-desc"][1]["vdu-storage-requirements"] = [
3157 {"key": "keep-volume", "value": "false"},
3159 target_vdu
= target_vdu_wth_persistent_storage
3160 vdu_instantiation_volumes_list
= []
3162 mock_volume_keeping_required
.return_value
= False
3163 expected_root_disk
= {
3164 "id": "persistent-root-volume",
3165 "type-of-storage": "persistent-storage:persistent-storage",
3166 "size-of-storage": "10",
3167 "vdu-storage-requirements": [{"key": "keep-volume", "value": "false"}],
3169 expected_persist_root_disk
= {
3170 "persistent-root-volume": {
3171 "image_id": "ubuntu20.04",
3176 expected_disk_list
= [
3178 "image_id": "ubuntu20.04",
3183 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3184 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3186 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3187 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3188 self
.assertEqual(disk_list
, expected_disk_list
)
3189 self
.assertEqual(len(disk_list
), 1)
3191 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3192 def test_find_persistent_root_volumes_target_vdu_mismatch(
3193 self
, mock_volume_keeping_required
3195 """Find persistent root volume, target vdu name is not matching."""
3196 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3197 vnfd
["vdu"][0]["name"] = "Several_Volumes-VM"
3198 target_vdu
= target_vdu_wth_persistent_storage
3199 vdu_instantiation_volumes_list
= []
3201 result
= self
.ns
.find_persistent_root_volumes(
3202 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3204 self
.assertEqual(result
, None)
3205 mock_volume_keeping_required
.assert_not_called()
3206 self
.assertEqual(disk_list
, [])
3207 self
.assertEqual(len(disk_list
), 0)
3209 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3210 def test_find_persistent_root_volumes_with_instantiation_vol_list(
3211 self
, mock_volume_keeping_required
3213 """Find persistent root volume, existing volume needs to be used."""
3214 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3215 target_vdu
= target_vdu_wth_persistent_storage
3216 vdu_instantiation_volumes_list
= [
3218 "vim-volume-id": vim_volume_id
,
3219 "name": "persistent-root-volume",
3223 expected_persist_root_disk
= {
3224 "persistent-root-volume": {
3225 "vim_volume_id": vim_volume_id
,
3226 "image_id": "ubuntu20.04",
3229 expected_disk_list
= [
3231 "vim_volume_id": vim_volume_id
,
3232 "image_id": "ubuntu20.04",
3235 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3236 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3238 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3239 mock_volume_keeping_required
.assert_not_called()
3240 self
.assertEqual(disk_list
, expected_disk_list
)
3241 self
.assertEqual(len(disk_list
), 1)
3243 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3244 def test_find_persistent_root_volumes_invalid_instantiation_params(
3245 self
, mock_volume_keeping_required
3247 """Find persistent root volume, existing volume id keyword is invalid."""
3248 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3249 target_vdu
= target_vdu_wth_persistent_storage
3250 vdu_instantiation_volumes_list
= [
3252 "volume-id": vim_volume_id
,
3253 "name": "persistent-root-volume",
3257 with self
.assertRaises(KeyError):
3258 self
.ns
.find_persistent_root_volumes(
3259 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3261 mock_volume_keeping_required
.assert_not_called()
3262 self
.assertEqual(disk_list
, [])
3263 self
.assertEqual(len(disk_list
), 0)
3265 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3266 def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list(
3267 self
, mock_volume_keeping_required
3269 """Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty."""
3270 persistent_root_disk
= {
3271 "persistent-root-volume": {
3272 "image_id": "ubuntu20.04",
3277 mock_volume_keeping_required
.return_value
= False
3278 target_vdu
= target_vdu_wth_persistent_storage
3279 vdu_instantiation_volumes_list
= []
3282 "image_id": "ubuntu20.04",
3288 "id": "persistent-volume2",
3289 "size-of-storage": "10",
3290 "type-of-storage": "persistent-storage:persistent-storage",
3292 expected_disk_list
= [
3294 "image_id": "ubuntu20.04",
3303 self
.ns
.find_persistent_volumes(
3304 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3306 self
.assertEqual(disk_list
, expected_disk_list
)
3307 mock_volume_keeping_required
.assert_called_once_with(expected_disk
)
3309 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3310 def test_find_persistent_volumes_vdu_wth_inst_vol_list(
3311 self
, mock_volume_keeping_required
3313 """Find persistent ordinary volume, vim-volume-id is given as instantiation parameter."""
3314 persistent_root_disk
= {
3315 "persistent-root-volume": {
3316 "image_id": "ubuntu20.04",
3321 vdu_instantiation_volumes_list
= [
3323 "vim-volume-id": vim_volume_id
,
3324 "name": "persistent-volume2",
3327 target_vdu
= target_vdu_wth_persistent_storage
3330 "image_id": "ubuntu20.04",
3335 expected_disk_list
= [
3337 "image_id": "ubuntu20.04",
3342 "vim_volume_id": vim_volume_id
,
3345 self
.ns
.find_persistent_volumes(
3346 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3348 self
.assertEqual(disk_list
, expected_disk_list
)
3349 mock_volume_keeping_required
.assert_not_called()
3351 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3352 def test_find_persistent_volumes_vdu_wthout_persistent_storage(
3353 self
, mock_volume_keeping_required
3355 """Find persistent ordinary volume, there is not any persistent disk."""
3356 persistent_root_disk
= {}
3357 vdu_instantiation_volumes_list
= []
3358 mock_volume_keeping_required
.return_value
= False
3359 target_vdu
= target_vdu_wthout_persistent_storage
3361 self
.ns
.find_persistent_volumes(
3362 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3364 self
.assertEqual(disk_list
, disk_list
)
3365 mock_volume_keeping_required
.assert_not_called()
3367 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3368 def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk(
3369 self
, mock_volume_keeping_required
3371 """There is persistent root disk, but there is not ordinary persistent disk."""
3372 persistent_root_disk
= {
3373 "persistent-root-volume": {
3374 "image_id": "ubuntu20.04",
3379 vdu_instantiation_volumes_list
= []
3380 mock_volume_keeping_required
.return_value
= False
3381 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3382 target_vdu
["virtual-storages"] = [
3384 "id": "persistent-root-volume",
3385 "size-of-storage": "10",
3386 "type-of-storage": "persistent-storage:persistent-storage",
3387 "vdu-storage-requirements": [
3388 {"key": "keep-volume", "value": "true"},
3392 "id": "ephemeral-volume",
3393 "size-of-storage": "1",
3394 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
3399 "image_id": "ubuntu20.04",
3404 self
.ns
.find_persistent_volumes(
3405 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3407 self
.assertEqual(disk_list
, disk_list
)
3408 mock_volume_keeping_required
.assert_not_called()
3410 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3411 def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(
3412 self
, mock_volume_keeping_required
3414 """Find persistent ordinary volume, volume id is not persistent_root_disk dict,
3415 vim-volume-id is given as instantiation parameter but disk id is not matching.
3417 mock_volume_keeping_required
.return_value
= True
3418 vdu_instantiation_volumes_list
= [
3420 "vim-volume-id": vim_volume_id
,
3421 "name": "persistent-volume3",
3424 persistent_root_disk
= {
3425 "persistent-root-volume": {
3426 "image_id": "ubuntu20.04",
3433 "image_id": "ubuntu20.04",
3438 expected_disk_list
= [
3440 "image_id": "ubuntu20.04",
3450 "id": "persistent-volume2",
3451 "size-of-storage": "10",
3452 "type-of-storage": "persistent-storage:persistent-storage",
3454 target_vdu
= target_vdu_wth_persistent_storage
3455 self
.ns
.find_persistent_volumes(
3456 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3458 self
.assertEqual(disk_list
, expected_disk_list
)
3459 mock_volume_keeping_required
.assert_called_once_with(expected_disk
)
3461 def test_is_volume_keeping_required_true(self
):
3462 """Volume keeping is required."""
3463 virtual_storage_descriptor
= {
3464 "id": "persistent-root-volume",
3465 "type-of-storage": "persistent-storage:persistent-storage",
3466 "size-of-storage": "10",
3467 "vdu-storage-requirements": [
3468 {"key": "keep-volume", "value": "true"},
3471 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3472 self
.assertEqual(result
, True)
3474 def test_is_volume_keeping_required_false(self
):
3475 """Volume keeping is not required."""
3476 virtual_storage_descriptor
= {
3477 "id": "persistent-root-volume",
3478 "type-of-storage": "persistent-storage:persistent-storage",
3479 "size-of-storage": "10",
3480 "vdu-storage-requirements": [
3481 {"key": "keep-volume", "value": "false"},
3484 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3485 self
.assertEqual(result
, False)
3487 def test_is_volume_keeping_required_wthout_vdu_storage_reqirement(self
):
3488 """Volume keeping is not required, vdu-storage-requirements key does not exist."""
3489 virtual_storage_descriptor
= {
3490 "id": "persistent-root-volume",
3491 "type-of-storage": "persistent-storage:persistent-storage",
3492 "size-of-storage": "10",
3494 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3495 self
.assertEqual(result
, False)
3497 def test_is_volume_keeping_required_wrong_keyword(self
):
3498 """vdu-storage-requirements key to indicate keeping-volume is wrong."""
3499 virtual_storage_descriptor
= {
3500 "id": "persistent-root-volume",
3501 "type-of-storage": "persistent-storage:persistent-storage",
3502 "size-of-storage": "10",
3503 "vdu-storage-requirements": [
3504 {"key": "hold-volume", "value": "true"},
3507 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3508 self
.assertEqual(result
, False)
3510 def test_sort_vdu_interfaces_position_all_wth_positions(self
):
3511 """Interfaces are sorted according to position, all have positions."""
3512 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3513 target_vdu
["interfaces"] = [
3516 "ns-vld-id": "datanet",
3521 "ns-vld-id": "mgmtnet",
3525 sorted_interfaces
= [
3528 "ns-vld-id": "mgmtnet",
3533 "ns-vld-id": "datanet",
3537 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3538 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3540 def test_sort_vdu_interfaces_position_some_wth_position(self
):
3541 """Interfaces are sorted according to position, some of them have positions."""
3542 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3543 target_vdu
["interfaces"] = [
3546 "ns-vld-id": "mgmtnet",
3550 "ns-vld-id": "datanet",
3554 sorted_interfaces
= [
3557 "ns-vld-id": "datanet",
3562 "ns-vld-id": "mgmtnet",
3565 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3566 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3568 def test_sort_vdu_interfaces_position_empty_interface_list(self
):
3569 """Interface list is empty."""
3570 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3571 target_vdu
["interfaces"] = []
3572 sorted_interfaces
= []
3573 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3574 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3576 def test_partially_locate_vdu_interfaces(self
):
3577 """Some interfaces have positions."""
3578 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3579 target_vdu
["interfaces"] = [
3582 "ns-vld-id": "net1",
3584 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3587 "ns-vld-id": "mgmtnet",
3591 "ns-vld-id": "datanet",
3595 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3596 self
.assertDictEqual(
3597 target_vdu
["interfaces"][0],
3600 "ns-vld-id": "datanet",
3604 self
.assertDictEqual(
3605 target_vdu
["interfaces"][2],
3606 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3609 def test_partially_locate_vdu_interfaces_position_start_from_0(self
):
3610 """Some interfaces have positions, position start from 0."""
3611 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3612 target_vdu
["interfaces"] = [
3615 "ns-vld-id": "net1",
3617 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3620 "ns-vld-id": "mgmtnet",
3624 "ns-vld-id": "datanet",
3628 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3629 self
.assertDictEqual(
3630 target_vdu
["interfaces"][0],
3633 "ns-vld-id": "datanet",
3637 self
.assertDictEqual(
3638 target_vdu
["interfaces"][3],
3639 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3642 def test_partially_locate_vdu_interfaces_wthout_position(self
):
3643 """Interfaces do not have positions."""
3644 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3645 target_vdu
["interfaces"] = interfaces_wthout_positions
3646 expected_result
= deepcopy(target_vdu
["interfaces"])
3647 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3648 self
.assertEqual(target_vdu
["interfaces"], expected_result
)
3650 def test_partially_locate_vdu_interfaces_all_has_position(self
):
3651 """All interfaces have position."""
3652 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3653 target_vdu
["interfaces"] = interfaces_wth_all_positions
3654 expected_interfaces
= [
3657 "ns-vld-id": "net2",
3662 "ns-vld-id": "mgmtnet",
3667 "ns-vld-id": "net1",
3671 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3672 self
.assertEqual(target_vdu
["interfaces"], expected_interfaces
)
3674 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3675 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3676 def test_prepare_vdu_cloud_init(self
, mock_parse_jinja2
, mock_get_cloud_init
):
3677 """Target_vdu has cloud-init and boot-data-drive."""
3678 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3679 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3680 target_vdu
["boot-data-drive"] = "vda"
3682 mock_get_cloud_init
.return_value
= cloud_init_content
3683 mock_parse_jinja2
.return_value
= user_data
3685 "user-data": user_data
,
3686 "boot-data-drive": "vda",
3688 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3689 self
.assertDictEqual(result
, expected_result
)
3690 mock_get_cloud_init
.assert_called_once_with(
3691 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3693 mock_parse_jinja2
.assert_called_once_with(
3694 cloud_init_content
=cloud_init_content
,
3696 context
="sample-cloud-init-path",
3699 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3700 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3701 def test_prepare_vdu_cloud_init_get_cloud_init_raise_exception(
3702 self
, mock_parse_jinja2
, mock_get_cloud_init
3704 """Target_vdu has cloud-init and boot-data-drive, get_cloud_init method raises exception."""
3705 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3706 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3707 target_vdu
["boot-data-drive"] = "vda"
3709 mock_get_cloud_init
.side_effect
= NsException(
3710 "Mismatch descriptor for cloud init."
3713 with self
.assertRaises(NsException
) as err
:
3714 self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3715 self
.assertEqual(str(err
.exception
), "Mismatch descriptor for cloud init.")
3717 mock_get_cloud_init
.assert_called_once_with(
3718 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3720 mock_parse_jinja2
.assert_not_called()
3722 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3723 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3724 def test_prepare_vdu_cloud_init_parse_jinja2_raise_exception(
3725 self
, mock_parse_jinja2
, mock_get_cloud_init
3727 """Target_vdu has cloud-init and boot-data-drive, parse_jinja2 method raises exception."""
3728 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3729 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3730 target_vdu
["boot-data-drive"] = "vda"
3732 mock_get_cloud_init
.return_value
= cloud_init_content
3733 mock_parse_jinja2
.side_effect
= NsException("Error parsing cloud-init content.")
3735 with self
.assertRaises(NsException
) as err
:
3736 self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3737 self
.assertEqual(str(err
.exception
), "Error parsing cloud-init content.")
3738 mock_get_cloud_init
.assert_called_once_with(
3739 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3741 mock_parse_jinja2
.assert_called_once_with(
3742 cloud_init_content
=cloud_init_content
,
3744 context
="sample-cloud-init-path",
3747 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3748 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3749 def test_prepare_vdu_cloud_init_vdu_wthout_boot_data_drive(
3750 self
, mock_parse_jinja2
, mock_get_cloud_init
3752 """Target_vdu has cloud-init but do not have boot-data-drive."""
3753 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3754 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3756 mock_get_cloud_init
.return_value
= cloud_init_content
3757 mock_parse_jinja2
.return_value
= user_data
3759 "user-data": user_data
,
3761 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3762 self
.assertDictEqual(result
, expected_result
)
3763 mock_get_cloud_init
.assert_called_once_with(
3764 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3766 mock_parse_jinja2
.assert_called_once_with(
3767 cloud_init_content
=cloud_init_content
,
3769 context
="sample-cloud-init-path",
3772 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3773 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3774 def test_prepare_vdu_cloud_init_exists_in_vdu2cloud_init(
3775 self
, mock_parse_jinja2
, mock_get_cloud_init
3777 """Target_vdu has cloud-init, vdu2cloud_init dict has cloud-init_content."""
3778 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3779 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3780 target_vdu
["boot-data-drive"] = "vda"
3781 vdu2cloud_init
= {"sample-cloud-init-path": cloud_init_content
}
3782 mock_parse_jinja2
.return_value
= user_data
3784 "user-data": user_data
,
3785 "boot-data-drive": "vda",
3787 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3788 self
.assertDictEqual(result
, expected_result
)
3789 mock_get_cloud_init
.assert_not_called()
3790 mock_parse_jinja2
.assert_called_once_with(
3791 cloud_init_content
=cloud_init_content
,
3793 context
="sample-cloud-init-path",
3796 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3797 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3798 def test_prepare_vdu_cloud_init_no_cloud_init(
3799 self
, mock_parse_jinja2
, mock_get_cloud_init
3801 """Target_vdu do not have cloud-init."""
3802 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3803 target_vdu
["boot-data-drive"] = "vda"
3806 "boot-data-drive": "vda",
3808 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3809 self
.assertDictEqual(result
, expected_result
)
3810 mock_get_cloud_init
.assert_not_called()
3811 mock_parse_jinja2
.assert_not_called()
3813 def test_check_vld_information_of_interfaces_ns_vld_vnf_vld_both_exist(self
):
3814 """ns_vld and vnf_vld both exist."""
3817 "ns-vld-id": "mgmtnet",
3818 "vnf-vld-id": "mgmt_cp_int",
3820 expected_result
= f
"{ns_preffix}:vld.mgmtnet"
3821 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
3822 interface
, ns_preffix
, vnf_preffix
3824 self
.assertEqual(result
, expected_result
)
3826 def test_check_vld_information_of_interfaces_empty_interfaces(self
):
3827 """Interface dict is empty."""
3829 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
3830 interface
, ns_preffix
, vnf_preffix
3832 self
.assertEqual(result
, "")
3834 def test_check_vld_information_of_interfaces_has_only_vnf_vld(self
):
3835 """Interface dict has only vnf_vld."""
3838 "vnf-vld-id": "mgmt_cp_int",
3840 expected_result
= f
"{vnf_preffix}:vld.mgmt_cp_int"
3841 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
3842 interface
, ns_preffix
, vnf_preffix
3844 self
.assertEqual(result
, expected_result
)
3846 def test_check_vld_information_of_interfaces_has_vnf_vld_wthout_vnf_prefix(
3849 """Interface dict has only vnf_vld but vnf_preffix does not exist."""
3852 "vnf-vld-id": "mgmt_cp_int",
3855 with self
.assertRaises(Exception) as err
:
3856 self
.ns
._check
_vld
_information
_of
_interfaces
(
3857 interface
, ns_preffix
, vnf_preffix
3859 self
.assertEqual(type(err
), TypeError)
3861 def test_prepare_interface_port_security_has_security_details(self
):
3862 """Interface dict has port security details."""
3865 "ns-vld-id": "mgmtnet",
3866 "vnf-vld-id": "mgmt_cp_int",
3867 "port-security-enabled": True,
3868 "port-security-disable-strategy": "allow-address-pairs",
3870 expected_interface
= {
3872 "ns-vld-id": "mgmtnet",
3873 "vnf-vld-id": "mgmt_cp_int",
3874 "port_security": True,
3875 "port_security_disable_strategy": "allow-address-pairs",
3877 self
.ns
._prepare
_interface
_port
_security
(interface
)
3878 self
.assertDictEqual(interface
, expected_interface
)
3880 def test_prepare_interface_port_security_empty_interfaces(self
):
3881 """Interface dict is empty."""
3883 expected_interface
= {}
3884 self
.ns
._prepare
_interface
_port
_security
(interface
)
3885 self
.assertDictEqual(interface
, expected_interface
)
3887 def test_prepare_interface_port_security_wthout_port_security(self
):
3888 """Interface dict does not have port security details."""
3891 "ns-vld-id": "mgmtnet",
3892 "vnf-vld-id": "mgmt_cp_int",
3894 expected_interface
= {
3896 "ns-vld-id": "mgmtnet",
3897 "vnf-vld-id": "mgmt_cp_int",
3899 self
.ns
._prepare
_interface
_port
_security
(interface
)
3900 self
.assertDictEqual(interface
, expected_interface
)
3902 def test_create_net_item_of_interface_floating_ip_port_security(self
):
3903 """Interface dict has floating ip, port-security details."""
3906 "vcpi": "sample_vcpi",
3907 "port_security": True,
3908 "port_security_disable_strategy": "allow-address-pairs",
3909 "floating_ip": "10.1.1.12",
3910 "ns-vld-id": "mgmtnet",
3911 "vnf-vld-id": "mgmt_cp_int",
3913 net_text
= f
"{ns_preffix}"
3914 expected_net_item
= {
3916 "port_security": True,
3917 "port_security_disable_strategy": "allow-address-pairs",
3918 "floating_ip": "10.1.1.12",
3919 "net_id": f
"TASK-{ns_preffix}",
3922 result
= self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
3923 self
.assertDictEqual(result
, expected_net_item
)
3925 def test_create_net_item_of_interface_invalid_net_text(self
):
3926 """net-text is invalid."""
3929 "vcpi": "sample_vcpi",
3930 "port_security": True,
3931 "port_security_disable_strategy": "allow-address-pairs",
3932 "floating_ip": "10.1.1.12",
3933 "ns-vld-id": "mgmtnet",
3934 "vnf-vld-id": "mgmt_cp_int",
3937 with self
.assertRaises(TypeError):
3938 self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
3940 def test_create_net_item_of_interface_empty_interface(self
):
3941 """Interface dict is empty."""
3943 net_text
= ns_preffix
3944 expected_net_item
= {
3945 "net_id": f
"TASK-{ns_preffix}",
3948 result
= self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
3949 self
.assertDictEqual(result
, expected_net_item
)
3951 @patch("osm_ng_ro.ns.deep_get")
3952 def test_prepare_type_of_interface_type_sriov(self
, mock_deep_get
):
3953 """Interface type is SR-IOV."""
3956 "vcpi": "sample_vcpi",
3957 "port_security": True,
3958 "port_security_disable_strategy": "allow-address-pairs",
3959 "floating_ip": "10.1.1.12",
3960 "ns-vld-id": "mgmtnet",
3961 "vnf-vld-id": "mgmt_cp_int",
3964 mock_deep_get
.return_value
= "SR-IOV"
3965 net_text
= ns_preffix
3967 expected_net_item
= {
3972 self
.ns
._prepare
_type
_of
_interface
(
3973 interface
, tasks_by_target_record_id
, net_text
, net_item
3975 self
.assertDictEqual(net_item
, expected_net_item
)
3978 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"]["net_type"],
3980 mock_deep_get
.assert_called_once_with(
3981 tasks_by_target_record_id
, net_text
, "extra_dict", "params", "net_type"
3984 @patch("osm_ng_ro.ns.deep_get")
3985 def test_prepare_type_of_interface_type_pic_passthrough_deep_get_return_empty_dict(
3988 """Interface type is PCI-PASSTHROUGH, deep_get method return empty dict."""
3991 "vcpi": "sample_vcpi",
3992 "port_security": True,
3993 "port_security_disable_strategy": "allow-address-pairs",
3994 "floating_ip": "10.1.1.12",
3995 "ns-vld-id": "mgmtnet",
3996 "vnf-vld-id": "mgmt_cp_int",
3997 "type": "PCI-PASSTHROUGH",
3999 mock_deep_get
.return_value
= {}
4000 tasks_by_target_record_id
= {}
4001 net_text
= ns_preffix
4003 expected_net_item
= {
4005 "model": "PCI-PASSTHROUGH",
4006 "type": "PCI-PASSTHROUGH",
4008 self
.ns
._prepare
_type
_of
_interface
(
4009 interface
, tasks_by_target_record_id
, net_text
, net_item
4011 self
.assertDictEqual(net_item
, expected_net_item
)
4012 mock_deep_get
.assert_called_once_with(
4013 tasks_by_target_record_id
, net_text
, "extra_dict", "params", "net_type"
4016 @patch("osm_ng_ro.ns.deep_get")
4017 def test_prepare_type_of_interface_type_mgmt(self
, mock_deep_get
):
4018 """Interface type is mgmt."""
4021 "vcpi": "sample_vcpi",
4022 "port_security": True,
4023 "port_security_disable_strategy": "allow-address-pairs",
4024 "floating_ip": "10.1.1.12",
4025 "ns-vld-id": "mgmtnet",
4026 "vnf-vld-id": "mgmt_cp_int",
4029 tasks_by_target_record_id
= {}
4030 net_text
= ns_preffix
4032 expected_net_item
= {
4035 self
.ns
._prepare
_type
_of
_interface
(
4036 interface
, tasks_by_target_record_id
, net_text
, net_item
4038 self
.assertDictEqual(net_item
, expected_net_item
)
4039 mock_deep_get
.assert_not_called()
4041 @patch("osm_ng_ro.ns.deep_get")
4042 def test_prepare_type_of_interface_type_bridge(self
, mock_deep_get
):
4043 """Interface type is bridge."""
4046 "vcpi": "sample_vcpi",
4047 "port_security": True,
4048 "port_security_disable_strategy": "allow-address-pairs",
4049 "floating_ip": "10.1.1.12",
4050 "ns-vld-id": "mgmtnet",
4051 "vnf-vld-id": "mgmt_cp_int",
4053 tasks_by_target_record_id
= {}
4054 net_text
= ns_preffix
4056 expected_net_item
= {
4060 self
.ns
._prepare
_type
_of
_interface
(
4061 interface
, tasks_by_target_record_id
, net_text
, net_item
4063 self
.assertDictEqual(net_item
, expected_net_item
)
4064 mock_deep_get
.assert_not_called()
4066 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4067 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4068 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4069 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4070 def test_prepare_vdu_interfaces(
4072 mock_type_of_interface
,
4073 mock_item_of_interface
,
4075 mock_vld_information_of_interface
,
4077 """Prepare vdu interfaces successfully."""
4078 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4081 "ns-vld-id": "net1",
4082 "ip-address": "13.2.12.31",
4083 "mgmt-interface": True,
4087 "vnf-vld-id": "net2",
4088 "mac-address": "d0:94:66:ed:fc:e2",
4092 "ns-vld-id": "mgmtnet",
4094 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4096 "params": "test_params",
4097 "find_params": "test_find_params",
4101 net_text_1
= f
"{ns_preffix}:net1"
4102 net_text_2
= f
"{vnf_preffix}:net2"
4103 net_text_3
= f
"{ns_preffix}:mgmtnet"
4106 "net_id": f
"TASK-{ns_preffix}",
4111 "net_id": f
"TASK-{ns_preffix}",
4116 "net_id": f
"TASK-{ns_preffix}",
4119 mock_item_of_interface
.side_effect
= [net_item_1
, net_item_2
, net_item_3
]
4120 mock_vld_information_of_interface
.side_effect
= [
4126 expected_extra_dict
= {
4127 "params": "test_params",
4128 "find_params": "test_find_params",
4129 "depends_on": [net_text_1
, net_text_2
, net_text_3
],
4130 "mgmt_vdu_interface": 0,
4132 updated_net_item1
= deepcopy(net_item_1
)
4133 updated_net_item1
.update({"ip_address": "13.2.12.31"})
4134 updated_net_item2
= deepcopy(net_item_2
)
4135 updated_net_item2
.update({"mac_address": "d0:94:66:ed:fc:e2"})
4136 expected_net_list
= [updated_net_item1
, updated_net_item2
, net_item_3
]
4137 self
.ns
._prepare
_vdu
_interfaces
(
4143 tasks_by_target_record_id
,
4146 _call_mock_vld_information_of_interface
= (
4147 mock_vld_information_of_interface
.call_args_list
4150 _call_mock_vld_information_of_interface
[0][0],
4151 (interface_1
, ns_preffix
, vnf_preffix
),
4154 _call_mock_vld_information_of_interface
[1][0],
4155 (interface_2
, ns_preffix
, vnf_preffix
),
4158 _call_mock_vld_information_of_interface
[2][0],
4159 (interface_3
, ns_preffix
, vnf_preffix
),
4162 _call_mock_port_security
= mock_port_security
.call_args_list
4163 self
.assertEqual(_call_mock_port_security
[0].args
[0], interface_1
)
4164 self
.assertEqual(_call_mock_port_security
[1].args
[0], interface_2
)
4165 self
.assertEqual(_call_mock_port_security
[2].args
[0], interface_3
)
4167 _call_mock_item_of_interface
= mock_item_of_interface
.call_args_list
4168 self
.assertEqual(_call_mock_item_of_interface
[0][0], (interface_1
, net_text_1
))
4169 self
.assertEqual(_call_mock_item_of_interface
[1][0], (interface_2
, net_text_2
))
4170 self
.assertEqual(_call_mock_item_of_interface
[2][0], (interface_3
, net_text_3
))
4172 _call_mock_type_of_interface
= mock_type_of_interface
.call_args_list
4174 _call_mock_type_of_interface
[0][0],
4175 (interface_1
, tasks_by_target_record_id
, net_text_1
, net_item_1
),
4178 _call_mock_type_of_interface
[1][0],
4179 (interface_2
, tasks_by_target_record_id
, net_text_2
, net_item_2
),
4182 _call_mock_type_of_interface
[2][0],
4183 (interface_3
, tasks_by_target_record_id
, net_text_3
, net_item_3
),
4185 self
.assertEqual(net_list
, expected_net_list
)
4186 self
.assertEqual(extra_dict
, expected_extra_dict
)
4187 self
.logger
.error
.assert_not_called()
4189 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4190 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4191 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4192 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4193 def test_prepare_vdu_interfaces_create_net_item_raise_exception(
4195 mock_type_of_interface
,
4196 mock_item_of_interface
,
4198 mock_vld_information_of_interface
,
4200 """Prepare vdu interfaces, create_net_item_of_interface method raise exception."""
4201 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4204 "ns-vld-id": "net1",
4205 "ip-address": "13.2.12.31",
4206 "mgmt-interface": True,
4210 "vnf-vld-id": "net2",
4211 "mac-address": "d0:94:66:ed:fc:e2",
4215 "ns-vld-id": "mgmtnet",
4217 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4219 "params": "test_params",
4220 "find_params": "test_find_params",
4223 net_text_1
= f
"{ns_preffix}:net1"
4224 mock_item_of_interface
.side_effect
= [TypeError, TypeError, TypeError]
4226 mock_vld_information_of_interface
.side_effect
= [net_text_1
]
4228 expected_extra_dict
= {
4229 "params": "test_params",
4230 "find_params": "test_find_params",
4231 "depends_on": [net_text_1
],
4233 with self
.assertRaises(TypeError):
4234 self
.ns
._prepare
_vdu
_interfaces
(
4240 tasks_by_target_record_id
,
4244 _call_mock_vld_information_of_interface
= (
4245 mock_vld_information_of_interface
.call_args_list
4248 _call_mock_vld_information_of_interface
[0][0],
4249 (interface_1
, ns_preffix
, vnf_preffix
),
4252 _call_mock_port_security
= mock_port_security
.call_args_list
4253 self
.assertEqual(_call_mock_port_security
[0].args
[0], interface_1
)
4255 _call_mock_item_of_interface
= mock_item_of_interface
.call_args_list
4256 self
.assertEqual(_call_mock_item_of_interface
[0][0], (interface_1
, net_text_1
))
4258 mock_type_of_interface
.assert_not_called()
4259 self
.logger
.error
.assert_not_called()
4260 self
.assertEqual(net_list
, [])
4261 self
.assertEqual(extra_dict
, expected_extra_dict
)
4263 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4264 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4265 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4266 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4267 def test_prepare_vdu_interfaces_vld_information_is_empty(
4269 mock_type_of_interface
,
4270 mock_item_of_interface
,
4272 mock_vld_information_of_interface
,
4274 """Prepare vdu interfaces, check_vld_information_of_interface method returns empty result."""
4275 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4278 "ns-vld-id": "net1",
4279 "ip-address": "13.2.12.31",
4280 "mgmt-interface": True,
4284 "vnf-vld-id": "net2",
4285 "mac-address": "d0:94:66:ed:fc:e2",
4289 "ns-vld-id": "mgmtnet",
4291 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4293 "params": "test_params",
4294 "find_params": "test_find_params",
4297 mock_vld_information_of_interface
.side_effect
= ["", "", ""]
4299 self
.ns
._prepare
_vdu
_interfaces
(
4305 tasks_by_target_record_id
,
4309 _call_mock_vld_information_of_interface
= (
4310 mock_vld_information_of_interface
.call_args_list
4313 _call_mock_vld_information_of_interface
[0][0],
4314 (interface_1
, ns_preffix
, vnf_preffix
),
4317 _call_mock_vld_information_of_interface
[1][0],
4318 (interface_2
, ns_preffix
, vnf_preffix
),
4321 _call_mock_vld_information_of_interface
[2][0],
4322 (interface_3
, ns_preffix
, vnf_preffix
),
4325 _call_logger
= self
.logger
.error
.call_args_list
4328 ("Interface 0 from vdu several_volumes-VM not connected to any vld",),
4332 ("Interface 1 from vdu several_volumes-VM not connected to any vld",),
4336 ("Interface 2 from vdu several_volumes-VM not connected to any vld",),
4338 self
.assertEqual(net_list
, [])
4342 "params": "test_params",
4343 "find_params": "test_find_params",
4348 mock_item_of_interface
.assert_not_called()
4349 mock_port_security
.assert_not_called()
4350 mock_type_of_interface
.assert_not_called()
4352 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4353 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4354 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4355 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4356 def test_prepare_vdu_interfaces_empty_interface_list(
4358 mock_type_of_interface
,
4359 mock_item_of_interface
,
4361 mock_vld_information_of_interface
,
4363 """Prepare vdu interfaces, interface list is empty."""
4364 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4365 target_vdu
["interfaces"] = []
4368 self
.ns
._prepare
_vdu
_interfaces
(
4374 tasks_by_target_record_id
,
4377 mock_type_of_interface
.assert_not_called()
4378 mock_vld_information_of_interface
.assert_not_called()
4379 mock_item_of_interface
.assert_not_called()
4380 mock_port_security
.assert_not_called()
4382 def test_prepare_vdu_ssh_keys(self
):
4383 """Target_vdu has ssh-keys and ro_nsr_public_key exists."""
4384 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4385 target_vdu
["ssh-keys"] = ["sample-ssh-key"]
4386 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4387 target_vdu
["ssh-access-required"] = True
4389 expected_cloud_config
= {
4390 "key-pairs": ["sample-ssh-key", {"public_key": "path_of_public_key"}]
4392 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4393 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4395 def test_prepare_vdu_ssh_keys_target_vdu_wthout_ssh_keys(self
):
4396 """Target_vdu does not have ssh-keys."""
4397 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4398 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4399 target_vdu
["ssh-access-required"] = True
4401 expected_cloud_config
= {"key-pairs": [{"public_key": "path_of_public_key"}]}
4402 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4403 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4405 def test_prepare_vdu_ssh_keys_ssh_access_is_not_required(self
):
4406 """Target_vdu has ssh-keys, ssh-access is not required."""
4407 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4408 target_vdu
["ssh-keys"] = ["sample-ssh-key"]
4409 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4410 target_vdu
["ssh-access-required"] = False
4412 expected_cloud_config
= {"key-pairs": ["sample-ssh-key"]}
4413 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4414 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4416 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4417 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4418 def test_add_persistent_root_disk_to_disk_list_keep_false(
4419 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4421 """Add persistent root disk to disk_list, keep volume set to False."""
4423 "id": "persistent-root-volume",
4424 "type-of-storage": "persistent-storage:persistent-storage",
4425 "size-of-storage": "10",
4427 mock_select_persistent_root_disk
.return_value
= root_disk
4428 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4429 vnfd
["virtual-storage-desc"][1] = root_disk
4430 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4431 persistent_root_disk
= {}
4433 mock_volume_keeping_required
.return_value
= False
4434 expected_disk_list
= [
4436 "image_id": "ubuntu20.04",
4441 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4442 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4444 self
.assertEqual(disk_list
, expected_disk_list
)
4445 mock_select_persistent_root_disk
.assert_called_once()
4446 mock_volume_keeping_required
.assert_called_once()
4448 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4449 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4450 def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises(
4451 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4453 """Add persistent root disk to disk_list"""
4455 "id": "persistent-root-volume",
4456 "type-of-storage": "persistent-storage:persistent-storage",
4457 "size-of-storage": "10",
4459 mock_select_persistent_root_disk
.side_effect
= AttributeError
4460 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4461 vnfd
["virtual-storage-desc"][1] = root_disk
4462 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4463 persistent_root_disk
= {}
4465 with self
.assertRaises(AttributeError):
4466 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4467 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4469 self
.assertEqual(disk_list
, [])
4470 mock_select_persistent_root_disk
.assert_called_once()
4471 mock_volume_keeping_required
.assert_not_called()
4473 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4474 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4475 def test_add_persistent_root_disk_to_disk_list_keep_true(
4476 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4478 """Add persistent root disk, keeo volume set to True."""
4479 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4480 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4481 mock_volume_keeping_required
.return_value
= True
4483 "id": "persistent-root-volume",
4484 "type-of-storage": "persistent-storage:persistent-storage",
4485 "size-of-storage": "10",
4486 "vdu-storage-requirements": [
4487 {"key": "keep-volume", "value": "true"},
4490 mock_select_persistent_root_disk
.return_value
= root_disk
4491 persistent_root_disk
= {}
4493 expected_disk_list
= [
4495 "image_id": "ubuntu20.04",
4500 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4501 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4503 self
.assertEqual(disk_list
, expected_disk_list
)
4504 mock_volume_keeping_required
.assert_called_once_with(root_disk
)
4506 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4507 def test_add_persistent_ordinary_disk_to_disk_list(
4508 self
, mock_volume_keeping_required
4510 """Add persistent ordinary disk, keeo volume set to True."""
4511 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4512 mock_volume_keeping_required
.return_value
= False
4513 persistent_root_disk
= {
4514 "persistent-root-volume": {
4515 "image_id": "ubuntu20.04",
4521 "id": "persistent-volume2",
4522 "type-of-storage": "persistent-storage:persistent-storage",
4523 "size-of-storage": "10",
4525 persistent_ordinary_disk
= {}
4527 expected_disk_list
= [
4533 self
.ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
4534 target_vdu
, persistent_root_disk
, persistent_ordinary_disk
, disk_list
4536 self
.assertEqual(disk_list
, expected_disk_list
)
4537 mock_volume_keeping_required
.assert_called_once_with(ordinary_disk
)
4539 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4540 def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(
4541 self
, mock_volume_keeping_required
4543 """Add persistent ordinary disk, vsd id is in root_disk dict."""
4544 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4545 mock_volume_keeping_required
.return_value
= False
4546 persistent_root_disk
= {
4547 "persistent-root-volume": {
4548 "image_id": "ubuntu20.04",
4552 "persistent-volume2": {
4556 persistent_ordinary_disk
= {}
4559 self
.ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
4560 target_vdu
, persistent_root_disk
, persistent_ordinary_disk
, disk_list
4562 self
.assertEqual(disk_list
, [])
4563 mock_volume_keeping_required
.assert_not_called()
4565 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4566 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4567 def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage(
4568 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4570 """VNFD does not have persistent storage."""
4571 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4572 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4573 mock_select_persistent_root_disk
.return_value
= None
4574 persistent_root_disk
= {}
4576 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4577 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4579 self
.assertEqual(disk_list
, [])
4580 self
.assertEqual(mock_select_persistent_root_disk
.call_count
, 2)
4581 mock_volume_keeping_required
.assert_not_called()
4583 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4584 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4585 def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk(
4586 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4588 """Persistent_root_disk dict is empty."""
4589 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4590 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4591 mock_select_persistent_root_disk
.return_value
= None
4592 persistent_root_disk
= {}
4594 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4595 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4597 self
.assertEqual(disk_list
, [])
4598 self
.assertEqual(mock_select_persistent_root_disk
.call_count
, 2)
4599 mock_volume_keeping_required
.assert_not_called()
4601 def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self
):
4602 """Invalid extra dict."""
4603 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4604 target_vdu
["affinity-or-anti-affinity-group-id"] = "sample_affinity-group-id"
4606 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4607 with self
.assertRaises(NsException
) as err
:
4608 self
.ns
._prepare
_vdu
_affinity
_group
_list
(target_vdu
, extra_dict
, ns_preffix
)
4609 self
.assertEqual(str(err
.exception
), "Invalid extra_dict format.")
4611 def test_prepare_vdu_affinity_group_list_one_affinity_group(self
):
4612 """There is one affinity-group."""
4613 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4614 target_vdu
["affinity-or-anti-affinity-group-id"] = ["sample_affinity-group-id"]
4615 extra_dict
= {"depends_on": []}
4616 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4617 affinity_group_txt
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.sample_affinity-group-id"
4618 expected_result
= [{"affinity_group_id": "TASK-" + affinity_group_txt
}]
4619 expected_extra_dict
= {"depends_on": [affinity_group_txt
]}
4620 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4621 target_vdu
, extra_dict
, ns_preffix
4623 self
.assertDictEqual(extra_dict
, expected_extra_dict
)
4624 self
.assertEqual(result
, expected_result
)
4626 def test_prepare_vdu_affinity_group_list_several_affinity_groups(self
):
4627 """There are two affinity-groups."""
4628 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4629 target_vdu
["affinity-or-anti-affinity-group-id"] = [
4630 "affinity-group-id1",
4631 "affinity-group-id2",
4633 extra_dict
= {"depends_on": []}
4634 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4635 affinity_group_txt1
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id1"
4636 affinity_group_txt2
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id2"
4638 {"affinity_group_id": "TASK-" + affinity_group_txt1
},
4639 {"affinity_group_id": "TASK-" + affinity_group_txt2
},
4641 expected_extra_dict
= {"depends_on": [affinity_group_txt1
, affinity_group_txt2
]}
4642 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4643 target_vdu
, extra_dict
, ns_preffix
4645 self
.assertDictEqual(extra_dict
, expected_extra_dict
)
4646 self
.assertEqual(result
, expected_result
)
4648 def test_prepare_vdu_affinity_group_list_no_affinity_group(self
):
4649 """There is not any affinity-group."""
4650 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4651 extra_dict
= {"depends_on": []}
4652 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4653 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4654 target_vdu
, extra_dict
, ns_preffix
4656 self
.assertDictEqual(extra_dict
, {"depends_on": []})
4657 self
.assertEqual(result
, [])
4659 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4660 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4661 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4662 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4663 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4664 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4665 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4666 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4667 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4668 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4669 def test_process_vdu_params_with_inst_vol_list(
4671 mock_prepare_vdu_affinity_group_list
,
4672 mock_add_persistent_ordinary_disks_to_disk_list
,
4673 mock_add_persistent_root_disk_to_disk_list
,
4674 mock_find_persistent_volumes
,
4675 mock_find_persistent_root_volumes
,
4676 mock_prepare_vdu_ssh_keys
,
4677 mock_prepare_vdu_cloud_init
,
4678 mock_prepare_vdu_interfaces
,
4679 mock_locate_vdu_interfaces
,
4680 mock_sort_vdu_interfaces
,
4682 """Instantiation volume list is empty."""
4683 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4685 target_vdu
["interfaces"] = interfaces_wth_all_positions
4687 vdu_instantiation_vol_list
= [
4689 "vim-volume-id": vim_volume_id
,
4690 "name": "persistent-volume2",
4693 target_vdu
["additionalParams"] = {
4694 "OSM": {"vdu_volumes": vdu_instantiation_vol_list
}
4696 mock_prepare_vdu_cloud_init
.return_value
= {}
4697 mock_prepare_vdu_affinity_group_list
.return_value
= []
4698 persistent_root_disk
= {
4699 "persistent-root-volume": {
4700 "image_id": "ubuntu20.04",
4704 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
4706 new_kwargs
= deepcopy(kwargs
)
4711 "tasks_by_target_record_id": {},
4715 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
4716 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4717 db
.get_one
.return_value
= vnfd
4718 result
= Ns
._process
_vdu
_params
(
4719 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
4721 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
4722 mock_locate_vdu_interfaces
.assert_not_called()
4723 mock_prepare_vdu_cloud_init
.assert_called_once()
4724 mock_add_persistent_root_disk_to_disk_list
.assert_not_called()
4725 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
4726 mock_prepare_vdu_interfaces
.assert_called_once_with(
4728 expected_extra_dict_copy
,
4735 self
.assertDictEqual(result
, expected_extra_dict_copy
)
4736 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
4737 mock_prepare_vdu_affinity_group_list
.assert_called_once()
4738 mock_find_persistent_volumes
.assert_called_once_with(
4739 persistent_root_disk
, target_vdu
, vdu_instantiation_vol_list
, []
4742 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4743 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4744 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4745 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4746 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4747 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4748 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4749 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4750 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4751 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4752 def test_process_vdu_params_wth_affinity_groups(
4754 mock_prepare_vdu_affinity_group_list
,
4755 mock_add_persistent_ordinary_disks_to_disk_list
,
4756 mock_add_persistent_root_disk_to_disk_list
,
4757 mock_find_persistent_volumes
,
4758 mock_find_persistent_root_volumes
,
4759 mock_prepare_vdu_ssh_keys
,
4760 mock_prepare_vdu_cloud_init
,
4761 mock_prepare_vdu_interfaces
,
4762 mock_locate_vdu_interfaces
,
4763 mock_sort_vdu_interfaces
,
4765 """There is cloud-config."""
4766 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4769 target_vdu
["interfaces"] = interfaces_wth_all_positions
4770 mock_prepare_vdu_cloud_init
.return_value
= {}
4771 mock_prepare_vdu_affinity_group_list
.return_value
= [
4776 new_kwargs
= deepcopy(kwargs
)
4781 "tasks_by_target_record_id": {},
4785 expected_extra_dict3
= deepcopy(expected_extra_dict2
)
4786 expected_extra_dict3
["params"]["affinity_group_list"] = [
4790 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4791 db
.get_one
.return_value
= vnfd
4792 result
= Ns
._process
_vdu
_params
(
4793 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
4795 self
.assertDictEqual(result
, expected_extra_dict3
)
4796 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
4797 mock_locate_vdu_interfaces
.assert_not_called()
4798 mock_prepare_vdu_cloud_init
.assert_called_once()
4799 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
4800 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
4801 mock_prepare_vdu_interfaces
.assert_called_once_with(
4803 expected_extra_dict3
,
4811 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
4812 mock_prepare_vdu_affinity_group_list
.assert_called_once()
4813 mock_find_persistent_volumes
.assert_not_called()
4815 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4816 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4817 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4818 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4819 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4820 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4821 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4822 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4823 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4824 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4825 def test_process_vdu_params_wth_cloud_config(
4827 mock_prepare_vdu_affinity_group_list
,
4828 mock_add_persistent_ordinary_disks_to_disk_list
,
4829 mock_add_persistent_root_disk_to_disk_list
,
4830 mock_find_persistent_volumes
,
4831 mock_find_persistent_root_volumes
,
4832 mock_prepare_vdu_ssh_keys
,
4833 mock_prepare_vdu_cloud_init
,
4834 mock_prepare_vdu_interfaces
,
4835 mock_locate_vdu_interfaces
,
4836 mock_sort_vdu_interfaces
,
4838 """There is cloud-config."""
4839 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4842 target_vdu
["interfaces"] = interfaces_wth_all_positions
4843 mock_prepare_vdu_cloud_init
.return_value
= {
4844 "user-data": user_data
,
4845 "boot-data-drive": "vda",
4847 mock_prepare_vdu_affinity_group_list
.return_value
= []
4849 new_kwargs
= deepcopy(kwargs
)
4854 "tasks_by_target_record_id": {},
4858 expected_extra_dict3
= deepcopy(expected_extra_dict2
)
4859 expected_extra_dict3
["params"]["cloud_config"] = {
4860 "user-data": user_data
,
4861 "boot-data-drive": "vda",
4863 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4864 db
.get_one
.return_value
= vnfd
4865 result
= Ns
._process
_vdu
_params
(
4866 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
4868 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
4869 mock_locate_vdu_interfaces
.assert_not_called()
4870 mock_prepare_vdu_cloud_init
.assert_called_once()
4871 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
4872 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
4873 mock_prepare_vdu_interfaces
.assert_called_once_with(
4875 expected_extra_dict3
,
4882 self
.assertDictEqual(result
, expected_extra_dict3
)
4883 mock_prepare_vdu_ssh_keys
.assert_called_once_with(
4884 target_vdu
, None, {"user-data": user_data
, "boot-data-drive": "vda"}
4886 mock_prepare_vdu_affinity_group_list
.assert_called_once()
4887 mock_find_persistent_volumes
.assert_not_called()
4889 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4890 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4891 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4892 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4893 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4894 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4895 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4896 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4897 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4898 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4899 def test_process_vdu_params_wthout_persistent_storage(
4901 mock_prepare_vdu_affinity_group_list
,
4902 mock_add_persistent_ordinary_disks_to_disk_list
,
4903 mock_add_persistent_root_disk_to_disk_list
,
4904 mock_find_persistent_volumes
,
4905 mock_find_persistent_root_volumes
,
4906 mock_prepare_vdu_ssh_keys
,
4907 mock_prepare_vdu_cloud_init
,
4908 mock_prepare_vdu_interfaces
,
4909 mock_locate_vdu_interfaces
,
4910 mock_sort_vdu_interfaces
,
4912 """There is not any persistent storage."""
4913 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4916 target_vdu
["interfaces"] = interfaces_wth_all_positions
4917 mock_prepare_vdu_cloud_init
.return_value
= {}
4918 mock_prepare_vdu_affinity_group_list
.return_value
= []
4920 new_kwargs
= deepcopy(kwargs
)
4925 "tasks_by_target_record_id": {},
4929 expected_extra_dict_copy
= deepcopy(expected_extra_dict2
)
4930 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4931 db
.get_one
.return_value
= vnfd
4932 result
= Ns
._process
_vdu
_params
(
4933 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
4935 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
4936 mock_locate_vdu_interfaces
.assert_not_called()
4937 mock_prepare_vdu_cloud_init
.assert_called_once()
4938 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
4939 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
4940 mock_prepare_vdu_interfaces
.assert_called_once_with(
4942 expected_extra_dict_copy
,
4949 self
.assertDictEqual(result
, expected_extra_dict_copy
)
4950 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
4951 mock_prepare_vdu_affinity_group_list
.assert_called_once()
4952 mock_find_persistent_volumes
.assert_not_called()
4954 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4955 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4956 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4957 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4958 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4959 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4960 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4961 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4962 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4963 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4964 def test_process_vdu_params_interfaces_partially_located(
4966 mock_prepare_vdu_affinity_group_list
,
4967 mock_add_persistent_ordinary_disks_to_disk_list
,
4968 mock_add_persistent_root_disk_to_disk_list
,
4969 mock_find_persistent_volumes
,
4970 mock_find_persistent_root_volumes
,
4971 mock_prepare_vdu_ssh_keys
,
4972 mock_prepare_vdu_cloud_init
,
4973 mock_prepare_vdu_interfaces
,
4974 mock_locate_vdu_interfaces
,
4975 mock_sort_vdu_interfaces
,
4977 """Some interfaces have position."""
4978 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4981 target_vdu
["interfaces"] = [
4984 "ns-vld-id": "net1",
4986 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 2},
4989 "ns-vld-id": "mgmtnet",
4992 mock_prepare_vdu_cloud_init
.return_value
= {}
4993 mock_prepare_vdu_affinity_group_list
.return_value
= []
4994 persistent_root_disk
= {
4995 "persistent-root-volume": {
4996 "image_id": "ubuntu20.04",
5001 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5003 new_kwargs
= deepcopy(kwargs
)
5008 "tasks_by_target_record_id": {},
5013 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5014 db
.get_one
.return_value
= vnfd
5015 result
= Ns
._process
_vdu
_params
(
5016 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5018 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
5019 mock_sort_vdu_interfaces
.assert_not_called()
5020 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5021 mock_prepare_vdu_cloud_init
.assert_called_once()
5022 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5023 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5024 mock_prepare_vdu_interfaces
.assert_called_once_with(
5026 expected_extra_dict_copy
,
5033 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5034 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5035 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5036 mock_find_persistent_volumes
.assert_not_called()
5037 mock_find_persistent_root_volumes
.assert_not_called()
5039 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5040 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5041 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5042 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5043 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5044 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5045 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5046 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5047 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5048 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5049 def test_process_vdu_params_no_interface_position(
5051 mock_prepare_vdu_affinity_group_list
,
5052 mock_add_persistent_ordinary_disks_to_disk_list
,
5053 mock_add_persistent_root_disk_to_disk_list
,
5054 mock_find_persistent_volumes
,
5055 mock_find_persistent_root_volumes
,
5056 mock_prepare_vdu_ssh_keys
,
5057 mock_prepare_vdu_cloud_init
,
5058 mock_prepare_vdu_interfaces
,
5059 mock_locate_vdu_interfaces
,
5060 mock_sort_vdu_interfaces
,
5062 """Interfaces do not have position."""
5063 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5066 target_vdu
["interfaces"] = interfaces_wthout_positions
5067 mock_prepare_vdu_cloud_init
.return_value
= {}
5068 mock_prepare_vdu_affinity_group_list
.return_value
= []
5069 persistent_root_disk
= {
5070 "persistent-root-volume": {
5071 "image_id": "ubuntu20.04",
5076 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5077 new_kwargs
= deepcopy(kwargs
)
5082 "tasks_by_target_record_id": {},
5087 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5088 db
.get_one
.return_value
= vnfd
5089 result
= Ns
._process
_vdu
_params
(
5090 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5092 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
5093 mock_sort_vdu_interfaces
.assert_not_called()
5094 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5095 mock_prepare_vdu_cloud_init
.assert_called_once()
5096 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5097 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5098 mock_prepare_vdu_interfaces
.assert_called_once_with(
5100 expected_extra_dict_copy
,
5107 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5108 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5109 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5110 mock_find_persistent_volumes
.assert_not_called()
5111 mock_find_persistent_root_volumes
.assert_not_called()
5113 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5114 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5115 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5116 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5117 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5118 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5119 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5120 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5121 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5122 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5123 def test_process_vdu_params_prepare_vdu_interfaces_raises_exception(
5125 mock_prepare_vdu_affinity_group_list
,
5126 mock_add_persistent_ordinary_disks_to_disk_list
,
5127 mock_add_persistent_root_disk_to_disk_list
,
5128 mock_find_persistent_volumes
,
5129 mock_find_persistent_root_volumes
,
5130 mock_prepare_vdu_ssh_keys
,
5131 mock_prepare_vdu_cloud_init
,
5132 mock_prepare_vdu_interfaces
,
5133 mock_locate_vdu_interfaces
,
5134 mock_sort_vdu_interfaces
,
5136 """Prepare vdu interfaces method raises exception."""
5137 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5140 target_vdu
["interfaces"] = interfaces_wthout_positions
5141 mock_prepare_vdu_cloud_init
.return_value
= {}
5142 mock_prepare_vdu_affinity_group_list
.return_value
= []
5143 persistent_root_disk
= {
5144 "persistent-root-volume": {
5145 "image_id": "ubuntu20.04",
5150 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5151 new_kwargs
= deepcopy(kwargs
)
5156 "tasks_by_target_record_id": {},
5160 mock_prepare_vdu_interfaces
.side_effect
= TypeError
5162 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5163 db
.get_one
.return_value
= vnfd
5164 with self
.assertRaises(Exception) as err
:
5165 Ns
._process
_vdu
_params
(
5166 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5168 self
.assertEqual(type(err
), TypeError)
5169 mock_sort_vdu_interfaces
.assert_not_called()
5170 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5171 mock_prepare_vdu_cloud_init
.assert_not_called()
5172 mock_add_persistent_root_disk_to_disk_list
.assert_not_called()
5173 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5174 mock_prepare_vdu_interfaces
.assert_called_once()
5175 mock_prepare_vdu_ssh_keys
.assert_not_called()
5176 mock_prepare_vdu_affinity_group_list
.assert_not_called()
5177 mock_find_persistent_volumes
.assert_not_called()
5178 mock_find_persistent_root_volumes
.assert_not_called()
5180 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5181 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5182 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5183 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5184 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5185 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5186 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5187 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5188 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5189 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5190 def test_process_vdu_params_add_persistent_root_disk_raises_exception(
5192 mock_prepare_vdu_affinity_group_list
,
5193 mock_add_persistent_ordinary_disks_to_disk_list
,
5194 mock_add_persistent_root_disk_to_disk_list
,
5195 mock_find_persistent_volumes
,
5196 mock_find_persistent_root_volumes
,
5197 mock_prepare_vdu_ssh_keys
,
5198 mock_prepare_vdu_cloud_init
,
5199 mock_prepare_vdu_interfaces
,
5200 mock_locate_vdu_interfaces
,
5201 mock_sort_vdu_interfaces
,
5203 """Add persistent root disk method raises exception."""
5204 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5207 target_vdu
["interfaces"] = interfaces_wthout_positions
5208 mock_prepare_vdu_cloud_init
.return_value
= {}
5209 mock_prepare_vdu_affinity_group_list
.return_value
= []
5210 mock_add_persistent_root_disk_to_disk_list
.side_effect
= KeyError
5211 new_kwargs
= deepcopy(kwargs
)
5216 "tasks_by_target_record_id": {},
5221 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5222 db
.get_one
.return_value
= vnfd
5223 with self
.assertRaises(Exception) as err
:
5224 Ns
._process
_vdu
_params
(
5225 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5227 self
.assertEqual(type(err
), KeyError)
5228 mock_sort_vdu_interfaces
.assert_not_called()
5229 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5230 mock_prepare_vdu_cloud_init
.assert_called_once()
5231 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5232 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5233 mock_prepare_vdu_interfaces
.assert_called_once_with(
5237 f
"{ns_preffix}:image.0",
5238 f
"{ns_preffix}:flavor.0",
5248 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5249 mock_prepare_vdu_affinity_group_list
.assert_not_called()
5250 mock_find_persistent_volumes
.assert_not_called()
5251 mock_find_persistent_root_volumes
.assert_not_called()
5253 def test_select_persistent_root_disk(self
):
5254 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5255 vdu
["virtual-storage-desc"] = [
5256 "persistent-root-volume",
5257 "persistent-volume2",
5260 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5261 expected_result
= vsd
5262 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5263 self
.assertEqual(result
, expected_result
)
5265 def test_select_persistent_root_disk_first_vsd_is_different(self
):
5266 """VDU first virtual-storage-desc is different than vsd id."""
5267 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5268 vdu
["virtual-storage-desc"] = [
5269 "persistent-volume2",
5270 "persistent-root-volume",
5273 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5274 expected_result
= None
5275 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5276 self
.assertEqual(result
, expected_result
)
5278 def test_select_persistent_root_disk_vsd_is_not_persistent(self
):
5279 """vsd type is not persistent."""
5280 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5281 vdu
["virtual-storage-desc"] = [
5282 "persistent-volume2",
5283 "persistent-root-volume",
5286 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5287 vsd
["type-of-storage"] = "etsi-nfv-descriptors:ephemeral-storage"
5288 expected_result
= None
5289 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5290 self
.assertEqual(result
, expected_result
)
5292 def test_select_persistent_root_disk_vsd_does_not_have_size(self
):
5293 """vsd size is None."""
5294 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5295 vdu
["virtual-storage-desc"] = [
5296 "persistent-volume2",
5297 "persistent-root-volume",
5300 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5301 vsd
["size-of-storage"] = None
5302 expected_result
= None
5303 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5304 self
.assertEqual(result
, expected_result
)
5306 def test_select_persistent_root_disk_vdu_wthout_vsd(self
):
5307 """VDU does not have virtual-storage-desc."""
5308 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5309 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5310 expected_result
= None
5311 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5312 self
.assertEqual(result
, expected_result
)
5314 def test_select_persistent_root_disk_invalid_vsd_type(self
):
5315 """vsd is list, expected to be a dict."""
5316 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5317 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"]
5318 with self
.assertRaises(AttributeError):
5319 Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)