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_threads(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(self
):
1517 expected_numa_result
= {"cores": 3}
1518 expected_epa_vcpu_set_result
= True
1520 "cpu-pinning-policy": "DEDICATED",
1523 epa_vcpu_set
= False
1525 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1526 guest_epa_quota
=guest_epa_quota
,
1527 vcpu_count
=vcpu_count
,
1528 epa_vcpu_set
=epa_vcpu_set
,
1531 self
.assertDictEqual(expected_numa_result
, numa_result
)
1532 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1534 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1535 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1536 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1537 def test__process_guest_epa_params_with_empty_params(
1539 guest_epa_numa_params
,
1540 guest_epa_cpu_pinning_params
,
1541 guest_epa_quota_params
,
1543 expected_result
= {}
1546 result
= Ns
._process
_epa
_params
(
1547 target_flavor
=target_flavor
,
1550 self
.assertDictEqual(expected_result
, result
)
1551 self
.assertFalse(guest_epa_numa_params
.called
)
1552 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1553 self
.assertFalse(guest_epa_quota_params
.called
)
1555 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1556 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1557 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1558 def test__process_guest_epa_params_with_wrong_params(
1560 guest_epa_numa_params
,
1561 guest_epa_cpu_pinning_params
,
1562 guest_epa_quota_params
,
1564 expected_result
= {}
1566 "no-guest-epa": "here",
1569 result
= Ns
._process
_epa
_params
(
1570 target_flavor
=target_flavor
,
1573 self
.assertDictEqual(expected_result
, result
)
1574 self
.assertFalse(guest_epa_numa_params
.called
)
1575 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1576 self
.assertFalse(guest_epa_quota_params
.called
)
1578 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1579 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1580 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1581 def test__process_guest_epa_params(
1583 guest_epa_numa_params
,
1584 guest_epa_cpu_pinning_params
,
1585 guest_epa_quota_params
,
1588 "mem-policy": "STRICT",
1593 "numa-node-policy": {
1594 "mem-policy": "STRICT",
1599 guest_epa_numa_params
.return_value
= ({}, False)
1600 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1601 guest_epa_quota_params
.return_value
= {}
1603 result
= Ns
._process
_epa
_params
(
1604 target_flavor
=target_flavor
,
1607 self
.assertDictEqual(expected_result
, result
)
1608 self
.assertTrue(guest_epa_numa_params
.called
)
1609 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1610 self
.assertTrue(guest_epa_quota_params
.called
)
1612 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1613 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1614 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1615 def test__process_guest_epa_params_with_mempage_size(
1617 guest_epa_numa_params
,
1618 guest_epa_cpu_pinning_params
,
1619 guest_epa_quota_params
,
1622 "mempage-size": "1G",
1623 "mem-policy": "STRICT",
1628 "mempage-size": "1G",
1629 "numa-node-policy": {
1630 "mem-policy": "STRICT",
1635 guest_epa_numa_params
.return_value
= ({}, False)
1636 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1637 guest_epa_quota_params
.return_value
= {}
1639 result
= Ns
._process
_epa
_params
(
1640 target_flavor
=target_flavor
,
1643 self
.assertDictEqual(expected_result
, result
)
1644 self
.assertTrue(guest_epa_numa_params
.called
)
1645 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1646 self
.assertTrue(guest_epa_quota_params
.called
)
1648 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1649 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1650 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1651 def test__process_guest_epa_params_with_numa(
1653 guest_epa_numa_params
,
1654 guest_epa_cpu_pinning_params
,
1655 guest_epa_quota_params
,
1658 "mempage-size": "1G",
1659 "cpu-pinning-policy": "DEDICATED",
1660 "cpu-thread-pinning-policy": "PREFER",
1665 "paired-threads": 3,
1666 "paired-threads-id": [("0", "1"), ("4", "5")],
1670 "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1671 "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1672 "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1673 "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1678 "mempage-size": "1G",
1679 "cpu-pinning-policy": "DEDICATED",
1680 "cpu-thread-pinning-policy": "PREFER",
1681 "numa-node-policy": {
1686 "num-paired-threads": "3",
1687 "paired-thread-ids": [
1726 guest_epa_numa_params
.return_value
= (
1730 "paired-threads": 3,
1731 "paired-threads-id": [("0", "1"), ("4", "5")],
1738 guest_epa_cpu_pinning_params
.return_value
= (
1744 guest_epa_quota_params
.return_value
= {
1767 result
= Ns
._process
_epa
_params
(
1768 target_flavor
=target_flavor
,
1770 self
.assertEqual(expected_result
, result
)
1771 self
.assertTrue(guest_epa_numa_params
.called
)
1772 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1773 self
.assertTrue(guest_epa_quota_params
.called
)
1775 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1776 def test__process_flavor_params_with_empty_target_flavor(
1785 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1790 target_record_id
= ""
1792 with self
.assertRaises(KeyError):
1793 Ns
._process
_flavor
_params
(
1794 target_flavor
=target_flavor
,
1797 target_record_id
=target_record_id
,
1800 self
.assertFalse(epa_params
.called
)
1802 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1803 def test__process_flavor_params_with_wrong_target_flavor(
1809 "no-target-flavor": "here",
1813 target_record_id
= ""
1815 with self
.assertRaises(KeyError):
1816 Ns
._process
_flavor
_params
(
1817 target_flavor
=target_flavor
,
1820 target_record_id
=target_record_id
,
1823 self
.assertFalse(epa_params
.called
)
1825 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1826 def test__process_flavor_params_with_empty_indata(
1851 "memory-mb": "1024",
1856 target_record_id
= ""
1858 epa_params
.return_value
= {}
1860 result
= Ns
._process
_flavor
_params
(
1861 target_flavor
=target_flavor
,
1864 target_record_id
=target_record_id
,
1867 self
.assertTrue(epa_params
.called
)
1868 self
.assertDictEqual(result
, expected_result
)
1870 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1871 def test__process_flavor_params_with_wrong_indata(
1896 "memory-mb": "1024",
1903 target_record_id
= ""
1905 epa_params
.return_value
= {}
1907 result
= Ns
._process
_flavor
_params
(
1908 target_flavor
=target_flavor
,
1911 target_record_id
=target_record_id
,
1914 self
.assertTrue(epa_params
.called
)
1915 self
.assertDictEqual(result
, expected_result
)
1917 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1918 def test__process_flavor_params_with_ephemeral_disk(
1926 db
.get_one
.return_value
= {
1927 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1932 {"id": "without_volumes-VM", "min-number-of-instances": 1}
1936 "id": "without_volumes-vnf",
1937 "product-name": "without_volumes-vnf",
1940 "id": "without_volumes-VM",
1941 "name": "without_volumes-VM",
1942 "sw-image-desc": "ubuntu20.04",
1943 "alternative-sw-image-desc": [
1945 "ubuntu20.04-azure",
1947 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
1951 "virtual-storage-desc": [
1952 {"id": "root-volume", "size-of-storage": "10"},
1954 "id": "ephemeral-volume",
1955 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1956 "size-of-storage": "1",
1962 "path": "/app/storage/",
1990 "memory-mb": "1024",
1998 "ns-flavor-id": "test_id",
1999 "virtual-storages": [
2001 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2002 "size-of-storage": "10",
2007 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2012 target_record_id
= ""
2014 epa_params
.return_value
= {}
2016 result
= Ns
._process
_flavor
_params
(
2017 target_flavor
=target_flavor
,
2020 target_record_id
=target_record_id
,
2024 self
.assertTrue(epa_params
.called
)
2025 self
.assertDictEqual(result
, expected_result
)
2027 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2028 def test__process_flavor_params_with_swap_disk(
2056 "memory-mb": "1024",
2064 "ns-flavor-id": "test_id",
2065 "virtual-storages": [
2067 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2068 "size-of-storage": "20",
2073 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2078 target_record_id
= ""
2080 epa_params
.return_value
= {}
2082 result
= Ns
._process
_flavor
_params
(
2083 target_flavor
=target_flavor
,
2086 target_record_id
=target_record_id
,
2089 self
.assertTrue(epa_params
.called
)
2090 self
.assertDictEqual(result
, expected_result
)
2092 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2093 def test__process_flavor_params_with_persistent_root_disk(
2102 db
.get_one
.return_value
= {
2103 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2108 {"id": "several_volumes-VM", "min-number-of-instances": 1}
2112 "id": "several_volumes-vnf",
2113 "product-name": "several_volumes-vnf",
2116 "id": "several_volumes-VM",
2117 "name": "several_volumes-VM",
2118 "sw-image-desc": "ubuntu20.04",
2119 "alternative-sw-image-desc": [
2121 "ubuntu20.04-azure",
2123 "virtual-storage-desc": [
2124 "persistent-root-volume",
2129 "virtual-storage-desc": [
2131 "id": "persistent-root-volume",
2132 "type-of-storage": "persistent-storage:persistent-storage",
2133 "size-of-storage": "10",
2139 "path": "/app/storage/",
2165 "memory-mb": "1024",
2173 "vdu-name": "several_volumes-VM",
2174 "ns-flavor-id": "test_id",
2175 "virtual-storages": [
2177 "type-of-storage": "persistent-storage:persistent-storage",
2178 "size-of-storage": "10",
2183 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2188 target_record_id
= ""
2190 epa_params
.return_value
= {}
2192 result
= Ns
._process
_flavor
_params
(
2193 target_flavor
=target_flavor
,
2196 target_record_id
=target_record_id
,
2200 self
.assertTrue(epa_params
.called
)
2201 self
.assertDictEqual(result
, expected_result
)
2203 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2204 def test__process_flavor_params_with_epa_params(
2216 "numa": "there-is-numa-here",
2227 "numa": "there-is-numa-here",
2236 "memory-mb": "1024",
2244 "ns-flavor-id": "test_id",
2247 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2252 target_record_id
= ""
2254 epa_params
.return_value
= {
2255 "numa": "there-is-numa-here",
2258 result
= Ns
._process
_flavor
_params
(
2259 target_flavor
=target_flavor
,
2262 target_record_id
=target_record_id
,
2265 self
.assertTrue(epa_params
.called
)
2266 self
.assertDictEqual(result
, expected_result
)
2268 @patch("osm_ng_ro.ns.Ns._process_epa_params")
2269 def test__process_flavor_params(
2278 db
.get_one
.return_value
= {
2279 "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2284 {"id": "without_volumes-VM", "min-number-of-instances": 1}
2288 "id": "without_volumes-vnf",
2289 "product-name": "without_volumes-vnf",
2292 "id": "without_volumes-VM",
2293 "name": "without_volumes-VM",
2294 "sw-image-desc": "ubuntu20.04",
2295 "alternative-sw-image-desc": [
2297 "ubuntu20.04-azure",
2299 "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2303 "virtual-storage-desc": [
2304 {"id": "root-volume", "size-of-storage": "10"},
2306 "id": "ephemeral-volume",
2307 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2308 "size-of-storage": "1",
2314 "path": "/app/storage/",
2329 "numa": "there-is-numa-here",
2342 "numa": "there-is-numa-here",
2351 "memory-mb": "1024",
2359 "ns-flavor-id": "test_id",
2360 "virtual-storages": [
2362 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2363 "size-of-storage": "10",
2366 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2367 "size-of-storage": "20",
2372 "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2377 target_record_id
= ""
2379 epa_params
.return_value
= {
2380 "numa": "there-is-numa-here",
2383 result
= Ns
._process
_flavor
_params
(
2384 target_flavor
=target_flavor
,
2387 target_record_id
=target_record_id
,
2391 self
.assertTrue(epa_params
.called
)
2392 self
.assertDictEqual(result
, expected_result
)
2394 def test__ip_profile_to_ro_with_none(self
):
2397 result
= Ns
._ip
_profile
_to
_ro
(
2398 ip_profile
=ip_profile
,
2401 self
.assertIsNone(result
)
2403 def test__ip_profile_to_ro_with_empty_profile(self
):
2406 result
= Ns
._ip
_profile
_to
_ro
(
2407 ip_profile
=ip_profile
,
2410 self
.assertIsNone(result
)
2412 def test__ip_profile_to_ro_with_wrong_profile(self
):
2414 "no-profile": "here",
2417 "ip_version": "IPv4",
2418 "subnet_address": None,
2419 "gateway_address": None,
2420 "dhcp_enabled": False,
2421 "dhcp_start_address": None,
2425 result
= Ns
._ip
_profile
_to
_ro
(
2426 ip_profile
=ip_profile
,
2429 self
.assertDictEqual(expected_result
, result
)
2431 def test__ip_profile_to_ro_with_ipv4_profile(self
):
2433 "ip-version": "ipv4",
2434 "subnet-address": "192.168.0.0/24",
2435 "gateway-address": "192.168.0.254",
2438 "start-address": "192.168.0.10",
2443 "ip_version": "IPv4",
2444 "subnet_address": "192.168.0.0/24",
2445 "gateway_address": "192.168.0.254",
2446 "dhcp_enabled": True,
2447 "dhcp_start_address": "192.168.0.10",
2451 result
= Ns
._ip
_profile
_to
_ro
(
2452 ip_profile
=ip_profile
,
2455 self
.assertDictEqual(expected_result
, result
)
2457 def test__ip_profile_to_ro_with_ipv6_profile(self
):
2459 "ip-version": "ipv6",
2460 "subnet-address": "2001:0200:0001::/48",
2461 "gateway-address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
2464 "start-address": "2001:0200:0001::0010",
2469 "ip_version": "IPv6",
2470 "subnet_address": "2001:0200:0001::/48",
2471 "gateway_address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
2472 "dhcp_enabled": True,
2473 "dhcp_start_address": "2001:0200:0001::0010",
2477 result
= Ns
._ip
_profile
_to
_ro
(
2478 ip_profile
=ip_profile
,
2481 self
.assertDictEqual(expected_result
, result
)
2483 def test__ip_profile_to_ro_with_dns_server(self
):
2485 "ip-version": "ipv4",
2486 "subnet-address": "192.168.0.0/24",
2487 "gateway-address": "192.168.0.254",
2490 "start-address": "192.168.0.10",
2495 "address": "8.8.8.8",
2498 "address": "1.1.1.1",
2501 "address": "1.0.0.1",
2506 "ip_version": "IPv4",
2507 "subnet_address": "192.168.0.0/24",
2508 "gateway_address": "192.168.0.254",
2509 "dhcp_enabled": True,
2510 "dhcp_start_address": "192.168.0.10",
2512 "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
2515 result
= Ns
._ip
_profile
_to
_ro
(
2516 ip_profile
=ip_profile
,
2519 self
.assertDictEqual(expected_result
, result
)
2521 def test__ip_profile_to_ro_with_security_group(self
):
2523 "ip-version": "ipv4",
2524 "subnet-address": "192.168.0.0/24",
2525 "gateway-address": "192.168.0.254",
2528 "start-address": "192.168.0.10",
2532 "some-security-group": "here",
2536 "ip_version": "IPv4",
2537 "subnet_address": "192.168.0.0/24",
2538 "gateway_address": "192.168.0.254",
2539 "dhcp_enabled": True,
2540 "dhcp_start_address": "192.168.0.10",
2543 "some-security-group": "here",
2547 result
= Ns
._ip
_profile
_to
_ro
(
2548 ip_profile
=ip_profile
,
2551 self
.assertDictEqual(expected_result
, result
)
2553 def test__ip_profile_to_ro(self
):
2555 "ip-version": "ipv4",
2556 "subnet-address": "192.168.0.0/24",
2557 "gateway-address": "192.168.0.254",
2560 "start-address": "192.168.0.10",
2565 "address": "8.8.8.8",
2568 "address": "1.1.1.1",
2571 "address": "1.0.0.1",
2575 "some-security-group": "here",
2579 "ip_version": "IPv4",
2580 "subnet_address": "192.168.0.0/24",
2581 "gateway_address": "192.168.0.254",
2582 "dhcp_enabled": True,
2583 "dhcp_start_address": "192.168.0.10",
2585 "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
2587 "some-security-group": "here",
2591 result
= Ns
._ip
_profile
_to
_ro
(
2592 ip_profile
=ip_profile
,
2595 self
.assertDictEqual(expected_result
, result
)
2597 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2598 def test__process_net_params_with_empty_params(
2609 "provider_network": "some-profile-here",
2611 target_record_id
= ""
2614 "net_name": "ns-name-vld-name",
2615 "net_type": "bridge",
2617 "some_ip_profile": "here",
2619 "provider_network_profile": "some-profile-here",
2623 ip_profile_to_ro
.return_value
= {
2624 "some_ip_profile": "here",
2627 result
= Ns
._process
_net
_params
(
2628 target_vld
=target_vld
,
2631 target_record_id
=target_record_id
,
2634 self
.assertDictEqual(expected_result
, result
)
2635 self
.assertTrue(ip_profile_to_ro
.called
)
2637 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2638 def test__process_net_params_with_vim_info_sdn(
2650 "sdn-ports": ["some", "ports", "here"],
2651 "vlds": ["some", "vlds", "here"],
2654 target_record_id
= "vld.sdn.something"
2657 "sdn-ports": ["some", "ports", "here"],
2658 "vlds": ["some", "vlds", "here"],
2663 result
= Ns
._process
_net
_params
(
2664 target_vld
=target_vld
,
2667 target_record_id
=target_record_id
,
2670 self
.assertDictEqual(expected_result
, result
)
2671 self
.assertFalse(ip_profile_to_ro
.called
)
2673 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2674 def test__process_net_params_with_vim_info_sdn_target_vim(
2686 "sdn-ports": ["some", "ports", "here"],
2687 "vlds": ["some", "vlds", "here"],
2688 "target_vim": "some-vim",
2691 target_record_id
= "vld.sdn.something"
2693 "depends_on": ["some-vim vld.sdn"],
2695 "sdn-ports": ["some", "ports", "here"],
2696 "vlds": ["some", "vlds", "here"],
2697 "target_vim": "some-vim",
2702 result
= Ns
._process
_net
_params
(
2703 target_vld
=target_vld
,
2706 target_record_id
=target_record_id
,
2709 self
.assertDictEqual(expected_result
, result
)
2710 self
.assertFalse(ip_profile_to_ro
.called
)
2712 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2713 def test__process_net_params_with_vim_network_name(
2724 "vim_network_name": "some-network-name",
2726 target_record_id
= "vld.sdn.something"
2730 "name": "some-network-name",
2735 result
= Ns
._process
_net
_params
(
2736 target_vld
=target_vld
,
2739 target_record_id
=target_record_id
,
2742 self
.assertDictEqual(expected_result
, result
)
2743 self
.assertFalse(ip_profile_to_ro
.called
)
2745 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2746 def test__process_net_params_with_vim_network_id(
2757 "vim_network_id": "some-network-id",
2759 target_record_id
= "vld.sdn.something"
2763 "id": "some-network-id",
2768 result
= Ns
._process
_net
_params
(
2769 target_vld
=target_vld
,
2772 target_record_id
=target_record_id
,
2775 self
.assertDictEqual(expected_result
, result
)
2776 self
.assertFalse(ip_profile_to_ro
.called
)
2778 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2779 def test__process_net_params_with_mgmt_network(
2786 "mgmt-network": "some-mgmt-network",
2792 target_record_id
= "vld.sdn.something"
2800 result
= Ns
._process
_net
_params
(
2801 target_vld
=target_vld
,
2804 target_record_id
=target_record_id
,
2807 self
.assertDictEqual(expected_result
, result
)
2808 self
.assertFalse(ip_profile_to_ro
.called
)
2810 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2811 def test__process_net_params_with_underlay_eline(
2817 "underlay": "some-underlay-here",
2824 "provider_network": "some-profile-here",
2826 target_record_id
= ""
2830 "some_ip_profile": "here",
2832 "net_name": "ns-name-vld-name",
2834 "provider_network_profile": "some-profile-here",
2838 ip_profile_to_ro
.return_value
= {
2839 "some_ip_profile": "here",
2842 result
= Ns
._process
_net
_params
(
2843 target_vld
=target_vld
,
2846 target_record_id
=target_record_id
,
2849 self
.assertDictEqual(expected_result
, result
)
2850 self
.assertTrue(ip_profile_to_ro
.called
)
2852 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2853 def test__process_net_params_with_underlay_elan(
2859 "underlay": "some-underlay-here",
2866 "provider_network": "some-profile-here",
2868 target_record_id
= ""
2872 "some_ip_profile": "here",
2874 "net_name": "ns-name-vld-name",
2876 "provider_network_profile": "some-profile-here",
2880 ip_profile_to_ro
.return_value
= {
2881 "some_ip_profile": "here",
2884 result
= Ns
._process
_net
_params
(
2885 target_vld
=target_vld
,
2888 target_record_id
=target_record_id
,
2891 self
.assertDictEqual(expected_result
, result
)
2892 self
.assertTrue(ip_profile_to_ro
.called
)
2894 def test__get_cloud_init_exception(self
):
2895 db_mock
= MagicMock(name
="database mock")
2900 with self
.assertRaises(NsException
):
2901 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2903 def test__get_cloud_init_file_fs_exception(self
):
2904 db_mock
= MagicMock(name
="database mock")
2907 location
= "vnfr_id_123456:file:test_file"
2908 db_mock
.get_one
.return_value
= {
2911 "folder": "/home/osm",
2912 "pkg-dir": "vnfr_test_dir",
2917 with self
.assertRaises(NsException
):
2918 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2920 def test__get_cloud_init_file(self
):
2921 db_mock
= MagicMock(name
="database mock")
2922 fs_mock
= MagicMock(name
="filesystem mock")
2923 file_mock
= MagicMock(name
="file mock")
2925 location
= "vnfr_id_123456:file:test_file"
2926 cloud_init_content
= "this is a cloud init file content"
2928 db_mock
.get_one
.return_value
= {
2931 "folder": "/home/osm",
2932 "pkg-dir": "vnfr_test_dir",
2936 fs_mock
.file_open
.return_value
= file_mock
2937 file_mock
.__enter
__.return_value
.read
.return_value
= cloud_init_content
2939 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2941 self
.assertEqual(cloud_init_content
, result
)
2943 def test__get_cloud_init_vdu(self
):
2944 db_mock
= MagicMock(name
="database mock")
2947 location
= "vnfr_id_123456:vdu:0"
2948 cloud_init_content
= "this is a cloud init file content"
2950 db_mock
.get_one
.return_value
= {
2953 "cloud-init": cloud_init_content
,
2958 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2960 self
.assertEqual(cloud_init_content
, result
)
2962 @patch("jinja2.Environment.__init__")
2963 def test__parse_jinja2_undefined_error(self
, env_mock
: Mock
):
2964 cloud_init_content
= None
2968 env_mock
.side_effect
= UndefinedError("UndefinedError occurred.")
2970 with self
.assertRaises(NsException
):
2972 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2975 @patch("jinja2.Environment.__init__")
2976 def test__parse_jinja2_template_error(self
, env_mock
: Mock
):
2977 cloud_init_content
= None
2981 env_mock
.side_effect
= TemplateError("TemplateError occurred.")
2983 with self
.assertRaises(NsException
):
2985 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2988 @patch("jinja2.Environment.__init__")
2989 def test__parse_jinja2_template_not_found(self
, env_mock
: Mock
):
2990 cloud_init_content
= None
2994 env_mock
.side_effect
= TemplateNotFound("TemplateNotFound occurred.")
2996 with self
.assertRaises(NsException
):
2998 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
3001 def test_rendering_jinja2_temp_without_special_characters(self
):
3002 cloud_init_content
= """
3005 table_type: {{type}}
3007 overwrite: {{is_override}}
3010 - [ sh, -xc, "echo $(date) '{{command}}'" ]
3014 "is_override": "False",
3015 "command": "; mkdir abc",
3017 context
= "cloud-init for VM"
3018 expected_result
= """
3026 - [ sh, -xc, "echo $(date) '; mkdir abc'" ]
3028 result
= Ns
._parse
_jinja
2(
3029 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
3031 self
.assertEqual(result
, expected_result
)
3033 def test_rendering_jinja2_temp_with_special_characters(self
):
3034 cloud_init_content
= """
3037 table_type: {{type}}
3039 overwrite: {{is_override}}
3042 - [ sh, -xc, "echo $(date) '{{command}}'" ]
3046 "is_override": "False",
3047 "command": "& rm -rf",
3049 context
= "cloud-init for VM"
3050 expected_result
= """
3058 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
3060 result
= Ns
._parse
_jinja
2(
3061 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
3063 self
.assertNotEqual(result
, expected_result
)
3065 def test_rendering_jinja2_temp_with_special_characters_autoescape_is_false(self
):
3066 with
patch("osm_ng_ro.ns.Environment") as mock_environment
:
3067 mock_environment
.return_value
= Environment(
3068 undefined
=StrictUndefined
,
3069 autoescape
=select_autoescape(default_for_string
=False, default
=False),
3071 cloud_init_content
= """
3074 table_type: {{type}}
3076 overwrite: {{is_override}}
3079 - [ sh, -xc, "echo $(date) '{{command}}'" ]
3083 "is_override": "False",
3084 "command": "& rm -rf /",
3086 context
= "cloud-init for VM"
3087 expected_result
= """
3095 - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
3097 result
= Ns
._parse
_jinja
2(
3098 cloud_init_content
=cloud_init_content
,
3102 self
.assertEqual(result
, expected_result
)
3104 @patch("osm_ng_ro.ns.Ns._assign_vim")
3105 def test__rebuild_start_stop_task(self
, assign_vim
):
3108 actions
= ["start", "stop", "rebuild"]
3109 vdu_id
= "bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3110 vnf_id
= "665b4165-ce24-4320-bf19-b9a45bade49f"
3112 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
3113 nsr_id
= "993166fe-723e-4680-ac4b-b1af2541ae31"
3115 target_vim
= "vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3116 t
= "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3117 for action
in actions
:
3119 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3120 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3121 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3122 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:0",
3123 "status": "SCHEDULED",
3126 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0",
3127 "target_record_id": t
,
3129 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3133 extra_dict
["params"] = {
3134 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3137 task
= self
.ns
.rebuild_start_stop_task(
3147 self
.assertEqual(task
.get("action_id"), action_id
)
3148 self
.assertEqual(task
.get("nsr_id"), nsr_id
)
3149 self
.assertEqual(task
.get("target_id"), target_vim
)
3150 self
.assertDictEqual(task
, expected_result
)
3152 @patch("osm_ng_ro.ns.Ns._assign_vim")
3153 def test_verticalscale_task(self
, assign_vim
):
3157 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
3158 nsr_id
= "993166fe-723e-4680-ac4b-b1af2541ae31"
3160 target_record_id
= (
3161 "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
3162 "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3166 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3167 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3168 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3169 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
3170 "status": "SCHEDULED",
3172 "item": "verticalscale",
3173 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1",
3174 "target_record_id": target_record_id
,
3176 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3177 "flavor_dict": "flavor_dict",
3181 "id": "bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
3183 "vim:f9f370ac-0d44-41a7-9000-457f2332bc35": {"interfaces": []}
3186 vnf
= {"_id": "665b4165-ce24-4320-bf19-b9a45bade49f"}
3187 extra_dict
["params"] = {
3188 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3189 "flavor_dict": "flavor_dict",
3191 task
= self
.ns
.verticalscale_task(
3192 vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3195 self
.assertDictEqual(task
, expected_result
)
3197 @patch("osm_ng_ro.ns.Ns._assign_vim")
3198 def test_migrate_task(self
, assign_vim
):
3202 action_id
= "bb937f49-3870-4169-b758-9732e1ff40f3"
3203 nsr_id
= "993166fe-723e-4680-ac4b-b1af2541ae31"
3205 target_record_id
= (
3206 "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
3207 "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3211 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3212 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3213 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3214 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
3215 "status": "SCHEDULED",
3218 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1",
3219 "target_record_id": target_record_id
,
3221 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3222 "migrate_host": "migrateToHost",
3226 "id": "bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
3228 "vim:f9f370ac-0d44-41a7-9000-457f2332bc35": {"interfaces": []}
3231 vnf
= {"_id": "665b4165-ce24-4320-bf19-b9a45bade49f"}
3232 extra_dict
["params"] = {
3233 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3234 "migrate_host": "migrateToHost",
3236 task
= self
.ns
.migrate_task(
3237 vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3240 self
.assertDictEqual(task
, expected_result
)
3243 class TestProcessVduParams(unittest
.TestCase
):
3246 self
.logger
= CopyingMock(autospec
=True)
3248 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3249 def test_find_persistent_root_volumes_empty_instantiation_vol_list(
3250 self
, mock_volume_keeping_required
3252 """Find persistent root volume, instantiation_vol_list is empty."""
3253 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3254 target_vdu
= target_vdu_wth_persistent_storage
3255 vdu_instantiation_volumes_list
= []
3257 mock_volume_keeping_required
.return_value
= True
3258 expected_root_disk
= {
3259 "id": "persistent-root-volume",
3260 "type-of-storage": "persistent-storage:persistent-storage",
3261 "size-of-storage": "10",
3262 "vdu-storage-requirements": [{"key": "keep-volume", "value": "true"}],
3264 expected_persist_root_disk
= {
3265 "persistent-root-volume": {
3266 "image_id": "ubuntu20.04",
3271 expected_disk_list
= [
3273 "image_id": "ubuntu20.04",
3278 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3279 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3281 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3282 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3283 self
.assertEqual(disk_list
, expected_disk_list
)
3284 self
.assertEqual(len(disk_list
), 1)
3286 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3287 def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(
3288 self
, mock_volume_keeping_required
3290 """Find persistent root volume, always selects the first vsd as root volume."""
3291 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3292 vnfd
["vdu"][0]["virtual-storage-desc"] = [
3293 "persistent-volume2",
3294 "persistent-root-volume",
3297 target_vdu
= target_vdu_wth_persistent_storage
3298 vdu_instantiation_volumes_list
= []
3300 mock_volume_keeping_required
.return_value
= True
3301 expected_root_disk
= {
3302 "id": "persistent-volume2",
3303 "type-of-storage": "persistent-storage:persistent-storage",
3304 "size-of-storage": "10",
3306 expected_persist_root_disk
= {
3307 "persistent-volume2": {
3308 "image_id": "ubuntu20.04",
3313 expected_disk_list
= [
3315 "image_id": "ubuntu20.04",
3320 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3321 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3323 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3324 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3325 self
.assertEqual(disk_list
, expected_disk_list
)
3326 self
.assertEqual(len(disk_list
), 1)
3328 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3329 def test_find_persistent_root_volumes_empty_size_of_storage(
3330 self
, mock_volume_keeping_required
3332 """Find persistent root volume, size of storage is empty."""
3333 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3334 vnfd
["virtual-storage-desc"][0]["size-of-storage"] = ""
3335 vnfd
["vdu"][0]["virtual-storage-desc"] = [
3336 "persistent-volume2",
3337 "persistent-root-volume",
3340 target_vdu
= target_vdu_wth_persistent_storage
3341 vdu_instantiation_volumes_list
= []
3343 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3344 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3346 self
.assertEqual(persist_root_disk
, None)
3347 mock_volume_keeping_required
.assert_not_called()
3348 self
.assertEqual(disk_list
, [])
3350 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3351 def test_find_persistent_root_volumes_keeping_is_not_required(
3352 self
, mock_volume_keeping_required
3354 """Find persistent root volume, volume keeping is not required."""
3355 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3356 vnfd
["virtual-storage-desc"][1]["vdu-storage-requirements"] = [
3357 {"key": "keep-volume", "value": "false"},
3359 target_vdu
= target_vdu_wth_persistent_storage
3360 vdu_instantiation_volumes_list
= []
3362 mock_volume_keeping_required
.return_value
= False
3363 expected_root_disk
= {
3364 "id": "persistent-root-volume",
3365 "type-of-storage": "persistent-storage:persistent-storage",
3366 "size-of-storage": "10",
3367 "vdu-storage-requirements": [{"key": "keep-volume", "value": "false"}],
3369 expected_persist_root_disk
= {
3370 "persistent-root-volume": {
3371 "image_id": "ubuntu20.04",
3376 expected_disk_list
= [
3378 "image_id": "ubuntu20.04",
3383 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3384 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3386 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3387 mock_volume_keeping_required
.assert_called_once_with(expected_root_disk
)
3388 self
.assertEqual(disk_list
, expected_disk_list
)
3389 self
.assertEqual(len(disk_list
), 1)
3391 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3392 def test_find_persistent_root_volumes_target_vdu_mismatch(
3393 self
, mock_volume_keeping_required
3395 """Find persistent root volume, target vdu name is not matching."""
3396 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3397 vnfd
["vdu"][0]["name"] = "Several_Volumes-VM"
3398 target_vdu
= target_vdu_wth_persistent_storage
3399 vdu_instantiation_volumes_list
= []
3401 result
= self
.ns
.find_persistent_root_volumes(
3402 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3404 self
.assertEqual(result
, None)
3405 mock_volume_keeping_required
.assert_not_called()
3406 self
.assertEqual(disk_list
, [])
3407 self
.assertEqual(len(disk_list
), 0)
3409 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3410 def test_find_persistent_root_volumes_with_instantiation_vol_list(
3411 self
, mock_volume_keeping_required
3413 """Find persistent root volume, existing volume needs to be used."""
3414 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3415 target_vdu
= target_vdu_wth_persistent_storage
3416 vdu_instantiation_volumes_list
= [
3418 "vim-volume-id": vim_volume_id
,
3419 "name": "persistent-root-volume",
3423 expected_persist_root_disk
= {
3424 "persistent-root-volume": {
3425 "vim_volume_id": vim_volume_id
,
3426 "image_id": "ubuntu20.04",
3429 expected_disk_list
= [
3431 "vim_volume_id": vim_volume_id
,
3432 "image_id": "ubuntu20.04",
3435 persist_root_disk
= self
.ns
.find_persistent_root_volumes(
3436 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3438 self
.assertEqual(persist_root_disk
, expected_persist_root_disk
)
3439 mock_volume_keeping_required
.assert_not_called()
3440 self
.assertEqual(disk_list
, expected_disk_list
)
3441 self
.assertEqual(len(disk_list
), 1)
3443 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3444 def test_find_persistent_root_volumes_invalid_instantiation_params(
3445 self
, mock_volume_keeping_required
3447 """Find persistent root volume, existing volume id keyword is invalid."""
3448 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
3449 target_vdu
= target_vdu_wth_persistent_storage
3450 vdu_instantiation_volumes_list
= [
3452 "volume-id": vim_volume_id
,
3453 "name": "persistent-root-volume",
3457 with self
.assertRaises(KeyError):
3458 self
.ns
.find_persistent_root_volumes(
3459 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3461 mock_volume_keeping_required
.assert_not_called()
3462 self
.assertEqual(disk_list
, [])
3463 self
.assertEqual(len(disk_list
), 0)
3465 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3466 def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list(
3467 self
, mock_volume_keeping_required
3469 """Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty."""
3470 persistent_root_disk
= {
3471 "persistent-root-volume": {
3472 "image_id": "ubuntu20.04",
3477 mock_volume_keeping_required
.return_value
= False
3478 target_vdu
= target_vdu_wth_persistent_storage
3479 vdu_instantiation_volumes_list
= []
3482 "image_id": "ubuntu20.04",
3488 "id": "persistent-volume2",
3489 "size-of-storage": "10",
3490 "type-of-storage": "persistent-storage:persistent-storage",
3492 expected_disk_list
= [
3494 "image_id": "ubuntu20.04",
3503 self
.ns
.find_persistent_volumes(
3504 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3506 self
.assertEqual(disk_list
, expected_disk_list
)
3507 mock_volume_keeping_required
.assert_called_once_with(expected_disk
)
3509 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3510 def test_find_persistent_volumes_vdu_wth_inst_vol_list(
3511 self
, mock_volume_keeping_required
3513 """Find persistent ordinary volume, vim-volume-id is given as instantiation parameter."""
3514 persistent_root_disk
= {
3515 "persistent-root-volume": {
3516 "image_id": "ubuntu20.04",
3521 vdu_instantiation_volumes_list
= [
3523 "vim-volume-id": vim_volume_id
,
3524 "name": "persistent-volume2",
3527 target_vdu
= target_vdu_wth_persistent_storage
3530 "image_id": "ubuntu20.04",
3535 expected_disk_list
= [
3537 "image_id": "ubuntu20.04",
3542 "vim_volume_id": vim_volume_id
,
3545 self
.ns
.find_persistent_volumes(
3546 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3548 self
.assertEqual(disk_list
, expected_disk_list
)
3549 mock_volume_keeping_required
.assert_not_called()
3551 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3552 def test_find_persistent_volumes_vdu_wthout_persistent_storage(
3553 self
, mock_volume_keeping_required
3555 """Find persistent ordinary volume, there is not any persistent disk."""
3556 persistent_root_disk
= {}
3557 vdu_instantiation_volumes_list
= []
3558 mock_volume_keeping_required
.return_value
= False
3559 target_vdu
= target_vdu_wthout_persistent_storage
3561 self
.ns
.find_persistent_volumes(
3562 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3564 self
.assertEqual(disk_list
, disk_list
)
3565 mock_volume_keeping_required
.assert_not_called()
3567 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3568 def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk(
3569 self
, mock_volume_keeping_required
3571 """There is persistent root disk, but there is not ordinary persistent disk."""
3572 persistent_root_disk
= {
3573 "persistent-root-volume": {
3574 "image_id": "ubuntu20.04",
3579 vdu_instantiation_volumes_list
= []
3580 mock_volume_keeping_required
.return_value
= False
3581 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3582 target_vdu
["virtual-storages"] = [
3584 "id": "persistent-root-volume",
3585 "size-of-storage": "10",
3586 "type-of-storage": "persistent-storage:persistent-storage",
3587 "vdu-storage-requirements": [
3588 {"key": "keep-volume", "value": "true"},
3592 "id": "ephemeral-volume",
3593 "size-of-storage": "1",
3594 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
3599 "image_id": "ubuntu20.04",
3604 self
.ns
.find_persistent_volumes(
3605 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3607 self
.assertEqual(disk_list
, disk_list
)
3608 mock_volume_keeping_required
.assert_not_called()
3610 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3611 def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(
3612 self
, mock_volume_keeping_required
3614 """Find persistent ordinary volume, volume id is not persistent_root_disk dict,
3615 vim-volume-id is given as instantiation parameter but disk id is not matching."""
3616 mock_volume_keeping_required
.return_value
= True
3617 vdu_instantiation_volumes_list
= [
3619 "vim-volume-id": vim_volume_id
,
3620 "name": "persistent-volume3",
3623 persistent_root_disk
= {
3624 "persistent-root-volume": {
3625 "image_id": "ubuntu20.04",
3632 "image_id": "ubuntu20.04",
3637 expected_disk_list
= [
3639 "image_id": "ubuntu20.04",
3649 "id": "persistent-volume2",
3650 "size-of-storage": "10",
3651 "type-of-storage": "persistent-storage:persistent-storage",
3653 target_vdu
= target_vdu_wth_persistent_storage
3654 self
.ns
.find_persistent_volumes(
3655 persistent_root_disk
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
3657 self
.assertEqual(disk_list
, expected_disk_list
)
3658 mock_volume_keeping_required
.assert_called_once_with(expected_disk
)
3660 def test_is_volume_keeping_required_true(self
):
3661 """Volume keeping is required."""
3662 virtual_storage_descriptor
= {
3663 "id": "persistent-root-volume",
3664 "type-of-storage": "persistent-storage:persistent-storage",
3665 "size-of-storage": "10",
3666 "vdu-storage-requirements": [
3667 {"key": "keep-volume", "value": "true"},
3670 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3671 self
.assertEqual(result
, True)
3673 def test_is_volume_keeping_required_false(self
):
3674 """Volume keeping is not required."""
3675 virtual_storage_descriptor
= {
3676 "id": "persistent-root-volume",
3677 "type-of-storage": "persistent-storage:persistent-storage",
3678 "size-of-storage": "10",
3679 "vdu-storage-requirements": [
3680 {"key": "keep-volume", "value": "false"},
3683 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3684 self
.assertEqual(result
, False)
3686 def test_is_volume_keeping_required_wthout_vdu_storage_reqirement(self
):
3687 """Volume keeping is not required, vdu-storage-requirements key does not exist."""
3688 virtual_storage_descriptor
= {
3689 "id": "persistent-root-volume",
3690 "type-of-storage": "persistent-storage:persistent-storage",
3691 "size-of-storage": "10",
3693 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3694 self
.assertEqual(result
, False)
3696 def test_is_volume_keeping_required_wrong_keyword(self
):
3697 """vdu-storage-requirements key to indicate keeping-volume is wrong."""
3698 virtual_storage_descriptor
= {
3699 "id": "persistent-root-volume",
3700 "type-of-storage": "persistent-storage:persistent-storage",
3701 "size-of-storage": "10",
3702 "vdu-storage-requirements": [
3703 {"key": "hold-volume", "value": "true"},
3706 result
= self
.ns
.is_volume_keeping_required(virtual_storage_descriptor
)
3707 self
.assertEqual(result
, False)
3709 def test_sort_vdu_interfaces_position_all_wth_positions(self
):
3710 """Interfaces are sorted according to position, all have positions."""
3711 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3712 target_vdu
["interfaces"] = [
3715 "ns-vld-id": "datanet",
3720 "ns-vld-id": "mgmtnet",
3724 sorted_interfaces
= [
3727 "ns-vld-id": "mgmtnet",
3732 "ns-vld-id": "datanet",
3736 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3737 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3739 def test_sort_vdu_interfaces_position_some_wth_position(self
):
3740 """Interfaces are sorted according to position, some of them have positions."""
3741 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3742 target_vdu
["interfaces"] = [
3745 "ns-vld-id": "mgmtnet",
3749 "ns-vld-id": "datanet",
3753 sorted_interfaces
= [
3756 "ns-vld-id": "datanet",
3761 "ns-vld-id": "mgmtnet",
3764 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3765 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3767 def test_sort_vdu_interfaces_position_empty_interface_list(self
):
3768 """Interface list is empty."""
3769 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3770 target_vdu
["interfaces"] = []
3771 sorted_interfaces
= []
3772 self
.ns
._sort
_vdu
_interfaces
(target_vdu
)
3773 self
.assertEqual(target_vdu
["interfaces"], sorted_interfaces
)
3775 def test_partially_locate_vdu_interfaces(self
):
3776 """Some interfaces have positions."""
3777 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3778 target_vdu
["interfaces"] = [
3781 "ns-vld-id": "net1",
3783 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3786 "ns-vld-id": "mgmtnet",
3790 "ns-vld-id": "datanet",
3794 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3795 self
.assertDictEqual(
3796 target_vdu
["interfaces"][0],
3799 "ns-vld-id": "datanet",
3803 self
.assertDictEqual(
3804 target_vdu
["interfaces"][2],
3805 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3808 def test_partially_locate_vdu_interfaces_position_start_from_0(self
):
3809 """Some interfaces have positions, position start from 0."""
3810 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3811 target_vdu
["interfaces"] = [
3814 "ns-vld-id": "net1",
3816 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3819 "ns-vld-id": "mgmtnet",
3823 "ns-vld-id": "datanet",
3827 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3828 self
.assertDictEqual(
3829 target_vdu
["interfaces"][0],
3832 "ns-vld-id": "datanet",
3836 self
.assertDictEqual(
3837 target_vdu
["interfaces"][3],
3838 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3841 def test_partially_locate_vdu_interfaces_wthout_position(self
):
3842 """Interfaces do not have positions."""
3843 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3844 target_vdu
["interfaces"] = interfaces_wthout_positions
3845 expected_result
= deepcopy(target_vdu
["interfaces"])
3846 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3847 self
.assertEqual(target_vdu
["interfaces"], expected_result
)
3849 def test_partially_locate_vdu_interfaces_all_has_position(self
):
3850 """All interfaces have position."""
3851 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3852 target_vdu
["interfaces"] = interfaces_wth_all_positions
3853 expected_interfaces
= [
3856 "ns-vld-id": "net2",
3861 "ns-vld-id": "mgmtnet",
3866 "ns-vld-id": "net1",
3870 self
.ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
3871 self
.assertEqual(target_vdu
["interfaces"], expected_interfaces
)
3873 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3874 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3875 def test_prepare_vdu_cloud_init(self
, mock_parse_jinja2
, mock_get_cloud_init
):
3876 """Target_vdu has cloud-init and boot-data-drive."""
3877 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3878 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3879 target_vdu
["boot-data-drive"] = "vda"
3881 mock_get_cloud_init
.return_value
= cloud_init_content
3882 mock_parse_jinja2
.return_value
= user_data
3884 "user-data": user_data
,
3885 "boot-data-drive": "vda",
3887 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3888 self
.assertDictEqual(result
, expected_result
)
3889 mock_get_cloud_init
.assert_called_once_with(
3890 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3892 mock_parse_jinja2
.assert_called_once_with(
3893 cloud_init_content
=cloud_init_content
,
3895 context
="sample-cloud-init-path",
3898 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3899 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3900 def test_prepare_vdu_cloud_init_get_cloud_init_raise_exception(
3901 self
, mock_parse_jinja2
, mock_get_cloud_init
3903 """Target_vdu has cloud-init and boot-data-drive, get_cloud_init method raises exception."""
3904 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3905 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3906 target_vdu
["boot-data-drive"] = "vda"
3908 mock_get_cloud_init
.side_effect
= NsException(
3909 "Mismatch descriptor for cloud init."
3912 with self
.assertRaises(NsException
) as err
:
3913 self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3914 self
.assertEqual(str(err
.exception
), "Mismatch descriptor for cloud init.")
3916 mock_get_cloud_init
.assert_called_once_with(
3917 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3919 mock_parse_jinja2
.assert_not_called()
3921 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3922 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3923 def test_prepare_vdu_cloud_init_parse_jinja2_raise_exception(
3924 self
, mock_parse_jinja2
, mock_get_cloud_init
3926 """Target_vdu has cloud-init and boot-data-drive, parse_jinja2 method raises exception."""
3927 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3928 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3929 target_vdu
["boot-data-drive"] = "vda"
3931 mock_get_cloud_init
.return_value
= cloud_init_content
3932 mock_parse_jinja2
.side_effect
= NsException("Error parsing cloud-init content.")
3934 with self
.assertRaises(NsException
) as err
:
3935 self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3936 self
.assertEqual(str(err
.exception
), "Error parsing cloud-init content.")
3937 mock_get_cloud_init
.assert_called_once_with(
3938 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3940 mock_parse_jinja2
.assert_called_once_with(
3941 cloud_init_content
=cloud_init_content
,
3943 context
="sample-cloud-init-path",
3946 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3947 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3948 def test_prepare_vdu_cloud_init_vdu_wthout_boot_data_drive(
3949 self
, mock_parse_jinja2
, mock_get_cloud_init
3951 """Target_vdu has cloud-init but do not have boot-data-drive."""
3952 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3953 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3955 mock_get_cloud_init
.return_value
= cloud_init_content
3956 mock_parse_jinja2
.return_value
= user_data
3958 "user-data": user_data
,
3960 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3961 self
.assertDictEqual(result
, expected_result
)
3962 mock_get_cloud_init
.assert_called_once_with(
3963 db
=db
, fs
=fs
, location
="sample-cloud-init-path"
3965 mock_parse_jinja2
.assert_called_once_with(
3966 cloud_init_content
=cloud_init_content
,
3968 context
="sample-cloud-init-path",
3971 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3972 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3973 def test_prepare_vdu_cloud_init_exists_in_vdu2cloud_init(
3974 self
, mock_parse_jinja2
, mock_get_cloud_init
3976 """Target_vdu has cloud-init, vdu2cloud_init dict has cloud-init_content."""
3977 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
3978 target_vdu
["cloud-init"] = "sample-cloud-init-path"
3979 target_vdu
["boot-data-drive"] = "vda"
3980 vdu2cloud_init
= {"sample-cloud-init-path": cloud_init_content
}
3981 mock_parse_jinja2
.return_value
= user_data
3983 "user-data": user_data
,
3984 "boot-data-drive": "vda",
3986 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
3987 self
.assertDictEqual(result
, expected_result
)
3988 mock_get_cloud_init
.assert_not_called()
3989 mock_parse_jinja2
.assert_called_once_with(
3990 cloud_init_content
=cloud_init_content
,
3992 context
="sample-cloud-init-path",
3995 @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3996 @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3997 def test_prepare_vdu_cloud_init_no_cloud_init(
3998 self
, mock_parse_jinja2
, mock_get_cloud_init
4000 """Target_vdu do not have cloud-init."""
4001 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4002 target_vdu
["boot-data-drive"] = "vda"
4005 "boot-data-drive": "vda",
4007 result
= self
.ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
4008 self
.assertDictEqual(result
, expected_result
)
4009 mock_get_cloud_init
.assert_not_called()
4010 mock_parse_jinja2
.assert_not_called()
4012 def test_check_vld_information_of_interfaces_ns_vld_vnf_vld_both_exist(self
):
4013 """ns_vld and vnf_vld both exist."""
4016 "ns-vld-id": "mgmtnet",
4017 "vnf-vld-id": "mgmt_cp_int",
4019 expected_result
= f
"{ns_preffix}:vld.mgmtnet"
4020 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
4021 interface
, ns_preffix
, vnf_preffix
4023 self
.assertEqual(result
, expected_result
)
4025 def test_check_vld_information_of_interfaces_empty_interfaces(self
):
4026 """Interface dict is empty."""
4028 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
4029 interface
, ns_preffix
, vnf_preffix
4031 self
.assertEqual(result
, "")
4033 def test_check_vld_information_of_interfaces_has_only_vnf_vld(self
):
4034 """Interface dict has only vnf_vld."""
4037 "vnf-vld-id": "mgmt_cp_int",
4039 expected_result
= f
"{vnf_preffix}:vld.mgmt_cp_int"
4040 result
= self
.ns
._check
_vld
_information
_of
_interfaces
(
4041 interface
, ns_preffix
, vnf_preffix
4043 self
.assertEqual(result
, expected_result
)
4045 def test_check_vld_information_of_interfaces_has_vnf_vld_wthout_vnf_prefix(
4048 """Interface dict has only vnf_vld but vnf_preffix does not exist."""
4051 "vnf-vld-id": "mgmt_cp_int",
4054 with self
.assertRaises(Exception) as err
:
4055 self
.ns
._check
_vld
_information
_of
_interfaces
(
4056 interface
, ns_preffix
, vnf_preffix
4058 self
.assertEqual(type(err
), TypeError)
4060 def test_prepare_interface_port_security_has_security_details(self
):
4061 """Interface dict has port security details."""
4064 "ns-vld-id": "mgmtnet",
4065 "vnf-vld-id": "mgmt_cp_int",
4066 "port-security-enabled": True,
4067 "port-security-disable-strategy": "allow-address-pairs",
4069 expected_interface
= {
4071 "ns-vld-id": "mgmtnet",
4072 "vnf-vld-id": "mgmt_cp_int",
4073 "port_security": True,
4074 "port_security_disable_strategy": "allow-address-pairs",
4076 self
.ns
._prepare
_interface
_port
_security
(interface
)
4077 self
.assertDictEqual(interface
, expected_interface
)
4079 def test_prepare_interface_port_security_empty_interfaces(self
):
4080 """Interface dict is empty."""
4082 expected_interface
= {}
4083 self
.ns
._prepare
_interface
_port
_security
(interface
)
4084 self
.assertDictEqual(interface
, expected_interface
)
4086 def test_prepare_interface_port_security_wthout_port_security(self
):
4087 """Interface dict does not have port security details."""
4090 "ns-vld-id": "mgmtnet",
4091 "vnf-vld-id": "mgmt_cp_int",
4093 expected_interface
= {
4095 "ns-vld-id": "mgmtnet",
4096 "vnf-vld-id": "mgmt_cp_int",
4098 self
.ns
._prepare
_interface
_port
_security
(interface
)
4099 self
.assertDictEqual(interface
, expected_interface
)
4101 def test_create_net_item_of_interface_floating_ip_port_security(self
):
4102 """Interface dict has floating ip, port-security details."""
4105 "vcpi": "sample_vcpi",
4106 "port_security": True,
4107 "port_security_disable_strategy": "allow-address-pairs",
4108 "floating_ip": "10.1.1.12",
4109 "ns-vld-id": "mgmtnet",
4110 "vnf-vld-id": "mgmt_cp_int",
4112 net_text
= f
"{ns_preffix}"
4113 expected_net_item
= {
4115 "port_security": True,
4116 "port_security_disable_strategy": "allow-address-pairs",
4117 "floating_ip": "10.1.1.12",
4118 "net_id": f
"TASK-{ns_preffix}",
4121 result
= self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
4122 self
.assertDictEqual(result
, expected_net_item
)
4124 def test_create_net_item_of_interface_invalid_net_text(self
):
4125 """net-text is invalid."""
4128 "vcpi": "sample_vcpi",
4129 "port_security": True,
4130 "port_security_disable_strategy": "allow-address-pairs",
4131 "floating_ip": "10.1.1.12",
4132 "ns-vld-id": "mgmtnet",
4133 "vnf-vld-id": "mgmt_cp_int",
4136 with self
.assertRaises(TypeError):
4137 self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
4139 def test_create_net_item_of_interface_empty_interface(self
):
4140 """Interface dict is empty."""
4142 net_text
= ns_preffix
4143 expected_net_item
= {
4144 "net_id": f
"TASK-{ns_preffix}",
4147 result
= self
.ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
4148 self
.assertDictEqual(result
, expected_net_item
)
4150 @patch("osm_ng_ro.ns.deep_get")
4151 def test_prepare_type_of_interface_type_sriov(self
, mock_deep_get
):
4152 """Interface type is SR-IOV."""
4155 "vcpi": "sample_vcpi",
4156 "port_security": True,
4157 "port_security_disable_strategy": "allow-address-pairs",
4158 "floating_ip": "10.1.1.12",
4159 "ns-vld-id": "mgmtnet",
4160 "vnf-vld-id": "mgmt_cp_int",
4163 mock_deep_get
.return_value
= "SR-IOV"
4164 net_text
= ns_preffix
4166 expected_net_item
= {
4171 self
.ns
._prepare
_type
_of
_interface
(
4172 interface
, tasks_by_target_record_id
, net_text
, net_item
4174 self
.assertDictEqual(net_item
, expected_net_item
)
4177 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"]["net_type"],
4179 mock_deep_get
.assert_called_once_with(
4180 tasks_by_target_record_id
, net_text
, "extra_dict", "params", "net_type"
4183 @patch("osm_ng_ro.ns.deep_get")
4184 def test_prepare_type_of_interface_type_pic_passthrough_deep_get_return_empty_dict(
4187 """Interface type is PCI-PASSTHROUGH, deep_get method return empty dict."""
4190 "vcpi": "sample_vcpi",
4191 "port_security": True,
4192 "port_security_disable_strategy": "allow-address-pairs",
4193 "floating_ip": "10.1.1.12",
4194 "ns-vld-id": "mgmtnet",
4195 "vnf-vld-id": "mgmt_cp_int",
4196 "type": "PCI-PASSTHROUGH",
4198 mock_deep_get
.return_value
= {}
4199 tasks_by_target_record_id
= {}
4200 net_text
= ns_preffix
4202 expected_net_item
= {
4204 "model": "PCI-PASSTHROUGH",
4205 "type": "PCI-PASSTHROUGH",
4207 self
.ns
._prepare
_type
_of
_interface
(
4208 interface
, tasks_by_target_record_id
, net_text
, net_item
4210 self
.assertDictEqual(net_item
, expected_net_item
)
4211 mock_deep_get
.assert_called_once_with(
4212 tasks_by_target_record_id
, net_text
, "extra_dict", "params", "net_type"
4215 @patch("osm_ng_ro.ns.deep_get")
4216 def test_prepare_type_of_interface_type_mgmt(self
, mock_deep_get
):
4217 """Interface type is mgmt."""
4220 "vcpi": "sample_vcpi",
4221 "port_security": True,
4222 "port_security_disable_strategy": "allow-address-pairs",
4223 "floating_ip": "10.1.1.12",
4224 "ns-vld-id": "mgmtnet",
4225 "vnf-vld-id": "mgmt_cp_int",
4228 tasks_by_target_record_id
= {}
4229 net_text
= ns_preffix
4231 expected_net_item
= {
4234 self
.ns
._prepare
_type
_of
_interface
(
4235 interface
, tasks_by_target_record_id
, net_text
, net_item
4237 self
.assertDictEqual(net_item
, expected_net_item
)
4238 mock_deep_get
.assert_not_called()
4240 @patch("osm_ng_ro.ns.deep_get")
4241 def test_prepare_type_of_interface_type_bridge(self
, mock_deep_get
):
4242 """Interface type is bridge."""
4245 "vcpi": "sample_vcpi",
4246 "port_security": True,
4247 "port_security_disable_strategy": "allow-address-pairs",
4248 "floating_ip": "10.1.1.12",
4249 "ns-vld-id": "mgmtnet",
4250 "vnf-vld-id": "mgmt_cp_int",
4252 tasks_by_target_record_id
= {}
4253 net_text
= ns_preffix
4255 expected_net_item
= {
4259 self
.ns
._prepare
_type
_of
_interface
(
4260 interface
, tasks_by_target_record_id
, net_text
, net_item
4262 self
.assertDictEqual(net_item
, expected_net_item
)
4263 mock_deep_get
.assert_not_called()
4265 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4266 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4267 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4268 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4269 def test_prepare_vdu_interfaces(
4271 mock_type_of_interface
,
4272 mock_item_of_interface
,
4274 mock_vld_information_of_interface
,
4276 """Prepare vdu interfaces successfully."""
4277 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4280 "ns-vld-id": "net1",
4281 "ip-address": "13.2.12.31",
4282 "mgmt-interface": True,
4286 "vnf-vld-id": "net2",
4287 "mac-address": "d0:94:66:ed:fc:e2",
4291 "ns-vld-id": "mgmtnet",
4293 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4295 "params": "test_params",
4296 "find_params": "test_find_params",
4300 net_text_1
= f
"{ns_preffix}:net1"
4301 net_text_2
= f
"{vnf_preffix}:net2"
4302 net_text_3
= f
"{ns_preffix}:mgmtnet"
4305 "net_id": f
"TASK-{ns_preffix}",
4310 "net_id": f
"TASK-{ns_preffix}",
4315 "net_id": f
"TASK-{ns_preffix}",
4318 mock_item_of_interface
.side_effect
= [net_item_1
, net_item_2
, net_item_3
]
4319 mock_vld_information_of_interface
.side_effect
= [
4325 expected_extra_dict
= {
4326 "params": "test_params",
4327 "find_params": "test_find_params",
4328 "depends_on": [net_text_1
, net_text_2
, net_text_3
],
4329 "mgmt_vdu_interface": 0,
4331 updated_net_item1
= deepcopy(net_item_1
)
4332 updated_net_item1
.update({"ip_address": "13.2.12.31"})
4333 updated_net_item2
= deepcopy(net_item_2
)
4334 updated_net_item2
.update({"mac_address": "d0:94:66:ed:fc:e2"})
4335 expected_net_list
= [updated_net_item1
, updated_net_item2
, net_item_3
]
4336 self
.ns
._prepare
_vdu
_interfaces
(
4342 tasks_by_target_record_id
,
4345 _call_mock_vld_information_of_interface
= (
4346 mock_vld_information_of_interface
.call_args_list
4349 _call_mock_vld_information_of_interface
[0][0],
4350 (interface_1
, ns_preffix
, vnf_preffix
),
4353 _call_mock_vld_information_of_interface
[1][0],
4354 (interface_2
, ns_preffix
, vnf_preffix
),
4357 _call_mock_vld_information_of_interface
[2][0],
4358 (interface_3
, ns_preffix
, vnf_preffix
),
4361 _call_mock_port_security
= mock_port_security
.call_args_list
4362 self
.assertEqual(_call_mock_port_security
[0].args
[0], interface_1
)
4363 self
.assertEqual(_call_mock_port_security
[1].args
[0], interface_2
)
4364 self
.assertEqual(_call_mock_port_security
[2].args
[0], interface_3
)
4366 _call_mock_item_of_interface
= mock_item_of_interface
.call_args_list
4367 self
.assertEqual(_call_mock_item_of_interface
[0][0], (interface_1
, net_text_1
))
4368 self
.assertEqual(_call_mock_item_of_interface
[1][0], (interface_2
, net_text_2
))
4369 self
.assertEqual(_call_mock_item_of_interface
[2][0], (interface_3
, net_text_3
))
4371 _call_mock_type_of_interface
= mock_type_of_interface
.call_args_list
4373 _call_mock_type_of_interface
[0][0],
4374 (interface_1
, tasks_by_target_record_id
, net_text_1
, net_item_1
),
4377 _call_mock_type_of_interface
[1][0],
4378 (interface_2
, tasks_by_target_record_id
, net_text_2
, net_item_2
),
4381 _call_mock_type_of_interface
[2][0],
4382 (interface_3
, tasks_by_target_record_id
, net_text_3
, net_item_3
),
4384 self
.assertEqual(net_list
, expected_net_list
)
4385 self
.assertEqual(extra_dict
, expected_extra_dict
)
4386 self
.logger
.error
.assert_not_called()
4388 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4389 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4390 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4391 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4392 def test_prepare_vdu_interfaces_create_net_item_raise_exception(
4394 mock_type_of_interface
,
4395 mock_item_of_interface
,
4397 mock_vld_information_of_interface
,
4399 """Prepare vdu interfaces, create_net_item_of_interface method raise exception."""
4400 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4403 "ns-vld-id": "net1",
4404 "ip-address": "13.2.12.31",
4405 "mgmt-interface": True,
4409 "vnf-vld-id": "net2",
4410 "mac-address": "d0:94:66:ed:fc:e2",
4414 "ns-vld-id": "mgmtnet",
4416 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4418 "params": "test_params",
4419 "find_params": "test_find_params",
4422 net_text_1
= f
"{ns_preffix}:net1"
4423 mock_item_of_interface
.side_effect
= [TypeError, TypeError, TypeError]
4425 mock_vld_information_of_interface
.side_effect
= [net_text_1
]
4427 expected_extra_dict
= {
4428 "params": "test_params",
4429 "find_params": "test_find_params",
4430 "depends_on": [net_text_1
],
4432 with self
.assertRaises(TypeError):
4433 self
.ns
._prepare
_vdu
_interfaces
(
4439 tasks_by_target_record_id
,
4443 _call_mock_vld_information_of_interface
= (
4444 mock_vld_information_of_interface
.call_args_list
4447 _call_mock_vld_information_of_interface
[0][0],
4448 (interface_1
, ns_preffix
, vnf_preffix
),
4451 _call_mock_port_security
= mock_port_security
.call_args_list
4452 self
.assertEqual(_call_mock_port_security
[0].args
[0], interface_1
)
4454 _call_mock_item_of_interface
= mock_item_of_interface
.call_args_list
4455 self
.assertEqual(_call_mock_item_of_interface
[0][0], (interface_1
, net_text_1
))
4457 mock_type_of_interface
.assert_not_called()
4458 self
.logger
.error
.assert_not_called()
4459 self
.assertEqual(net_list
, [])
4460 self
.assertEqual(extra_dict
, expected_extra_dict
)
4462 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4463 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4464 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4465 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4466 def test_prepare_vdu_interfaces_vld_information_is_empty(
4468 mock_type_of_interface
,
4469 mock_item_of_interface
,
4471 mock_vld_information_of_interface
,
4473 """Prepare vdu interfaces, check_vld_information_of_interface method returns empty result."""
4474 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4477 "ns-vld-id": "net1",
4478 "ip-address": "13.2.12.31",
4479 "mgmt-interface": True,
4483 "vnf-vld-id": "net2",
4484 "mac-address": "d0:94:66:ed:fc:e2",
4488 "ns-vld-id": "mgmtnet",
4490 target_vdu
["interfaces"] = [interface_1
, interface_2
, interface_3
]
4492 "params": "test_params",
4493 "find_params": "test_find_params",
4496 mock_vld_information_of_interface
.side_effect
= ["", "", ""]
4498 self
.ns
._prepare
_vdu
_interfaces
(
4504 tasks_by_target_record_id
,
4508 _call_mock_vld_information_of_interface
= (
4509 mock_vld_information_of_interface
.call_args_list
4512 _call_mock_vld_information_of_interface
[0][0],
4513 (interface_1
, ns_preffix
, vnf_preffix
),
4516 _call_mock_vld_information_of_interface
[1][0],
4517 (interface_2
, ns_preffix
, vnf_preffix
),
4520 _call_mock_vld_information_of_interface
[2][0],
4521 (interface_3
, ns_preffix
, vnf_preffix
),
4524 _call_logger
= self
.logger
.error
.call_args_list
4527 ("Interface 0 from vdu several_volumes-VM not connected to any vld",),
4531 ("Interface 1 from vdu several_volumes-VM not connected to any vld",),
4535 ("Interface 2 from vdu several_volumes-VM not connected to any vld",),
4537 self
.assertEqual(net_list
, [])
4541 "params": "test_params",
4542 "find_params": "test_find_params",
4547 mock_item_of_interface
.assert_not_called()
4548 mock_port_security
.assert_not_called()
4549 mock_type_of_interface
.assert_not_called()
4551 @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4552 @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4553 @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4554 @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4555 def test_prepare_vdu_interfaces_empty_interface_list(
4557 mock_type_of_interface
,
4558 mock_item_of_interface
,
4560 mock_vld_information_of_interface
,
4562 """Prepare vdu interfaces, interface list is empty."""
4563 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4564 target_vdu
["interfaces"] = []
4567 self
.ns
._prepare
_vdu
_interfaces
(
4573 tasks_by_target_record_id
,
4576 mock_type_of_interface
.assert_not_called()
4577 mock_vld_information_of_interface
.assert_not_called()
4578 mock_item_of_interface
.assert_not_called()
4579 mock_port_security
.assert_not_called()
4581 def test_prepare_vdu_ssh_keys(self
):
4582 """Target_vdu has ssh-keys and ro_nsr_public_key exists."""
4583 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4584 target_vdu
["ssh-keys"] = ["sample-ssh-key"]
4585 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4586 target_vdu
["ssh-access-required"] = True
4588 expected_cloud_config
= {
4589 "key-pairs": ["sample-ssh-key", {"public_key": "path_of_public_key"}]
4591 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4592 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4594 def test_prepare_vdu_ssh_keys_target_vdu_wthout_ssh_keys(self
):
4595 """Target_vdu does not have ssh-keys."""
4596 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4597 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4598 target_vdu
["ssh-access-required"] = True
4600 expected_cloud_config
= {"key-pairs": [{"public_key": "path_of_public_key"}]}
4601 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4602 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4604 def test_prepare_vdu_ssh_keys_ssh_access_is_not_required(self
):
4605 """Target_vdu has ssh-keys, ssh-access is not required."""
4606 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4607 target_vdu
["ssh-keys"] = ["sample-ssh-key"]
4608 ro_nsr_public_key
= {"public_key": "path_of_public_key"}
4609 target_vdu
["ssh-access-required"] = False
4611 expected_cloud_config
= {"key-pairs": ["sample-ssh-key"]}
4612 self
.ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
4613 self
.assertDictEqual(cloud_config
, expected_cloud_config
)
4615 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4616 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4617 def test_add_persistent_root_disk_to_disk_list_keep_false(
4618 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4620 """Add persistent root disk to disk_list, keep volume set to False."""
4622 "id": "persistent-root-volume",
4623 "type-of-storage": "persistent-storage:persistent-storage",
4624 "size-of-storage": "10",
4626 mock_select_persistent_root_disk
.return_value
= root_disk
4627 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4628 vnfd
["virtual-storage-desc"][1] = root_disk
4629 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4630 persistent_root_disk
= {}
4632 mock_volume_keeping_required
.return_value
= False
4633 expected_disk_list
= [
4635 "image_id": "ubuntu20.04",
4640 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4641 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4643 self
.assertEqual(disk_list
, expected_disk_list
)
4644 mock_select_persistent_root_disk
.assert_called_once()
4645 mock_volume_keeping_required
.assert_called_once()
4647 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4648 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4649 def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises(
4650 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4652 """Add persistent root disk to disk_list"""
4654 "id": "persistent-root-volume",
4655 "type-of-storage": "persistent-storage:persistent-storage",
4656 "size-of-storage": "10",
4658 mock_select_persistent_root_disk
.side_effect
= AttributeError
4659 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4660 vnfd
["virtual-storage-desc"][1] = root_disk
4661 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4662 persistent_root_disk
= {}
4664 with self
.assertRaises(AttributeError):
4665 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4666 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4668 self
.assertEqual(disk_list
, [])
4669 mock_select_persistent_root_disk
.assert_called_once()
4670 mock_volume_keeping_required
.assert_not_called()
4672 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4673 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4674 def test_add_persistent_root_disk_to_disk_list_keep_true(
4675 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4677 """Add persistent root disk, keeo volume set to True."""
4678 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4679 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4680 mock_volume_keeping_required
.return_value
= True
4682 "id": "persistent-root-volume",
4683 "type-of-storage": "persistent-storage:persistent-storage",
4684 "size-of-storage": "10",
4685 "vdu-storage-requirements": [
4686 {"key": "keep-volume", "value": "true"},
4689 mock_select_persistent_root_disk
.return_value
= root_disk
4690 persistent_root_disk
= {}
4692 expected_disk_list
= [
4694 "image_id": "ubuntu20.04",
4699 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4700 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4702 self
.assertEqual(disk_list
, expected_disk_list
)
4703 mock_volume_keeping_required
.assert_called_once_with(root_disk
)
4705 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4706 def test_add_persistent_ordinary_disk_to_disk_list(
4707 self
, mock_volume_keeping_required
4709 """Add persistent ordinary disk, keeo volume set to True."""
4710 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4711 mock_volume_keeping_required
.return_value
= False
4712 persistent_root_disk
= {
4713 "persistent-root-volume": {
4714 "image_id": "ubuntu20.04",
4720 "id": "persistent-volume2",
4721 "type-of-storage": "persistent-storage:persistent-storage",
4722 "size-of-storage": "10",
4724 persistent_ordinary_disk
= {}
4726 expected_disk_list
= [
4732 self
.ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
4733 target_vdu
, persistent_root_disk
, persistent_ordinary_disk
, disk_list
4735 self
.assertEqual(disk_list
, expected_disk_list
)
4736 mock_volume_keeping_required
.assert_called_once_with(ordinary_disk
)
4738 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4739 def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(
4740 self
, mock_volume_keeping_required
4742 """Add persistent ordinary disk, vsd id is in root_disk dict."""
4743 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4744 mock_volume_keeping_required
.return_value
= False
4745 persistent_root_disk
= {
4746 "persistent-root-volume": {
4747 "image_id": "ubuntu20.04",
4751 "persistent-volume2": {
4755 persistent_ordinary_disk
= {}
4758 self
.ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
4759 target_vdu
, persistent_root_disk
, persistent_ordinary_disk
, disk_list
4761 self
.assertEqual(disk_list
, [])
4762 mock_volume_keeping_required
.assert_not_called()
4764 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4765 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4766 def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage(
4767 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4769 """VNFD does not have persistent storage."""
4770 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4771 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4772 mock_select_persistent_root_disk
.return_value
= None
4773 persistent_root_disk
= {}
4775 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4776 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4778 self
.assertEqual(disk_list
, [])
4779 self
.assertEqual(mock_select_persistent_root_disk
.call_count
, 2)
4780 mock_volume_keeping_required
.assert_not_called()
4782 @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4783 @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4784 def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk(
4785 self
, mock_volume_keeping_required
, mock_select_persistent_root_disk
4787 """Persistent_root_disk dict is empty."""
4788 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
4789 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4790 mock_select_persistent_root_disk
.return_value
= None
4791 persistent_root_disk
= {}
4793 self
.ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
4794 vnfd
, target_vdu
, persistent_root_disk
, disk_list
4796 self
.assertEqual(disk_list
, [])
4797 self
.assertEqual(mock_select_persistent_root_disk
.call_count
, 2)
4798 mock_volume_keeping_required
.assert_not_called()
4800 def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self
):
4801 """Invalid extra dict."""
4802 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4803 target_vdu
["affinity-or-anti-affinity-group-id"] = "sample_affinity-group-id"
4805 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4806 with self
.assertRaises(NsException
) as err
:
4807 self
.ns
._prepare
_vdu
_affinity
_group
_list
(target_vdu
, extra_dict
, ns_preffix
)
4808 self
.assertEqual(str(err
.exception
), "Invalid extra_dict format.")
4810 def test_prepare_vdu_affinity_group_list_one_affinity_group(self
):
4811 """There is one affinity-group."""
4812 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4813 target_vdu
["affinity-or-anti-affinity-group-id"] = ["sample_affinity-group-id"]
4814 extra_dict
= {"depends_on": []}
4815 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4816 affinity_group_txt
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.sample_affinity-group-id"
4817 expected_result
= [{"affinity_group_id": "TASK-" + affinity_group_txt
}]
4818 expected_extra_dict
= {"depends_on": [affinity_group_txt
]}
4819 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4820 target_vdu
, extra_dict
, ns_preffix
4822 self
.assertDictEqual(extra_dict
, expected_extra_dict
)
4823 self
.assertEqual(result
, expected_result
)
4825 def test_prepare_vdu_affinity_group_list_several_affinity_groups(self
):
4826 """There are two affinity-groups."""
4827 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4828 target_vdu
["affinity-or-anti-affinity-group-id"] = [
4829 "affinity-group-id1",
4830 "affinity-group-id2",
4832 extra_dict
= {"depends_on": []}
4833 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4834 affinity_group_txt1
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id1"
4835 affinity_group_txt2
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id2"
4837 {"affinity_group_id": "TASK-" + affinity_group_txt1
},
4838 {"affinity_group_id": "TASK-" + affinity_group_txt2
},
4840 expected_extra_dict
= {"depends_on": [affinity_group_txt1
, affinity_group_txt2
]}
4841 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4842 target_vdu
, extra_dict
, ns_preffix
4844 self
.assertDictEqual(extra_dict
, expected_extra_dict
)
4845 self
.assertEqual(result
, expected_result
)
4847 def test_prepare_vdu_affinity_group_list_no_affinity_group(self
):
4848 """There is not any affinity-group."""
4849 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4850 extra_dict
= {"depends_on": []}
4851 ns_preffix
= "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4852 result
= self
.ns
._prepare
_vdu
_affinity
_group
_list
(
4853 target_vdu
, extra_dict
, ns_preffix
4855 self
.assertDictEqual(extra_dict
, {"depends_on": []})
4856 self
.assertEqual(result
, [])
4858 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4859 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4860 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4861 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4862 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4863 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4864 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4865 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4866 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4867 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4868 def test_process_vdu_params_with_inst_vol_list(
4870 mock_prepare_vdu_affinity_group_list
,
4871 mock_add_persistent_ordinary_disks_to_disk_list
,
4872 mock_add_persistent_root_disk_to_disk_list
,
4873 mock_find_persistent_volumes
,
4874 mock_find_persistent_root_volumes
,
4875 mock_prepare_vdu_ssh_keys
,
4876 mock_prepare_vdu_cloud_init
,
4877 mock_prepare_vdu_interfaces
,
4878 mock_locate_vdu_interfaces
,
4879 mock_sort_vdu_interfaces
,
4881 """Instantiation volume list is empty."""
4882 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
4884 target_vdu
["interfaces"] = interfaces_wth_all_positions
4886 vdu_instantiation_vol_list
= [
4888 "vim-volume-id": vim_volume_id
,
4889 "name": "persistent-volume2",
4892 target_vdu
["additionalParams"] = {
4893 "OSM": {"vdu_volumes": vdu_instantiation_vol_list
}
4895 mock_prepare_vdu_cloud_init
.return_value
= {}
4896 mock_prepare_vdu_affinity_group_list
.return_value
= []
4897 persistent_root_disk
= {
4898 "persistent-root-volume": {
4899 "image_id": "ubuntu20.04",
4903 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
4905 new_kwargs
= deepcopy(kwargs
)
4910 "tasks_by_target_record_id": {},
4914 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
4915 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4916 db
.get_one
.return_value
= vnfd
4917 result
= Ns
._process
_vdu
_params
(
4918 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
4920 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
4921 mock_locate_vdu_interfaces
.assert_not_called()
4922 mock_prepare_vdu_cloud_init
.assert_called_once()
4923 mock_add_persistent_root_disk_to_disk_list
.assert_not_called()
4924 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
4925 mock_prepare_vdu_interfaces
.assert_called_once_with(
4927 expected_extra_dict_copy
,
4934 self
.assertDictEqual(result
, expected_extra_dict_copy
)
4935 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
4936 mock_prepare_vdu_affinity_group_list
.assert_called_once()
4937 mock_find_persistent_volumes
.assert_called_once_with(
4938 persistent_root_disk
, target_vdu
, vdu_instantiation_vol_list
, []
4941 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4942 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4943 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4944 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4945 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4946 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4947 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4948 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4949 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4950 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4951 def test_process_vdu_params_wth_affinity_groups(
4953 mock_prepare_vdu_affinity_group_list
,
4954 mock_add_persistent_ordinary_disks_to_disk_list
,
4955 mock_add_persistent_root_disk_to_disk_list
,
4956 mock_find_persistent_volumes
,
4957 mock_find_persistent_root_volumes
,
4958 mock_prepare_vdu_ssh_keys
,
4959 mock_prepare_vdu_cloud_init
,
4960 mock_prepare_vdu_interfaces
,
4961 mock_locate_vdu_interfaces
,
4962 mock_sort_vdu_interfaces
,
4964 """There is cloud-config."""
4965 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
4968 target_vdu
["interfaces"] = interfaces_wth_all_positions
4969 mock_prepare_vdu_cloud_init
.return_value
= {}
4970 mock_prepare_vdu_affinity_group_list
.return_value
= [
4975 new_kwargs
= deepcopy(kwargs
)
4980 "tasks_by_target_record_id": {},
4984 expected_extra_dict3
= deepcopy(expected_extra_dict2
)
4985 expected_extra_dict3
["params"]["affinity_group_list"] = [
4989 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
4990 db
.get_one
.return_value
= vnfd
4991 result
= Ns
._process
_vdu
_params
(
4992 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
4994 self
.assertDictEqual(result
, expected_extra_dict3
)
4995 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
4996 mock_locate_vdu_interfaces
.assert_not_called()
4997 mock_prepare_vdu_cloud_init
.assert_called_once()
4998 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
4999 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5000 mock_prepare_vdu_interfaces
.assert_called_once_with(
5002 expected_extra_dict3
,
5010 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5011 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5012 mock_find_persistent_volumes
.assert_not_called()
5014 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5015 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5016 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5017 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5018 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5019 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5020 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5021 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5022 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5023 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5024 def test_process_vdu_params_wth_cloud_config(
5026 mock_prepare_vdu_affinity_group_list
,
5027 mock_add_persistent_ordinary_disks_to_disk_list
,
5028 mock_add_persistent_root_disk_to_disk_list
,
5029 mock_find_persistent_volumes
,
5030 mock_find_persistent_root_volumes
,
5031 mock_prepare_vdu_ssh_keys
,
5032 mock_prepare_vdu_cloud_init
,
5033 mock_prepare_vdu_interfaces
,
5034 mock_locate_vdu_interfaces
,
5035 mock_sort_vdu_interfaces
,
5037 """There is cloud-config."""
5038 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
5041 target_vdu
["interfaces"] = interfaces_wth_all_positions
5042 mock_prepare_vdu_cloud_init
.return_value
= {
5043 "user-data": user_data
,
5044 "boot-data-drive": "vda",
5046 mock_prepare_vdu_affinity_group_list
.return_value
= []
5048 new_kwargs
= deepcopy(kwargs
)
5053 "tasks_by_target_record_id": {},
5057 expected_extra_dict3
= deepcopy(expected_extra_dict2
)
5058 expected_extra_dict3
["params"]["cloud_config"] = {
5059 "user-data": user_data
,
5060 "boot-data-drive": "vda",
5062 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5063 db
.get_one
.return_value
= vnfd
5064 result
= Ns
._process
_vdu
_params
(
5065 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5067 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
5068 mock_locate_vdu_interfaces
.assert_not_called()
5069 mock_prepare_vdu_cloud_init
.assert_called_once()
5070 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5071 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5072 mock_prepare_vdu_interfaces
.assert_called_once_with(
5074 expected_extra_dict3
,
5081 self
.assertDictEqual(result
, expected_extra_dict3
)
5082 mock_prepare_vdu_ssh_keys
.assert_called_once_with(
5083 target_vdu
, None, {"user-data": user_data
, "boot-data-drive": "vda"}
5085 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5086 mock_find_persistent_volumes
.assert_not_called()
5088 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5089 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5090 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5091 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5092 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5093 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5094 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5095 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5096 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5097 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5098 def test_process_vdu_params_wthout_persistent_storage(
5100 mock_prepare_vdu_affinity_group_list
,
5101 mock_add_persistent_ordinary_disks_to_disk_list
,
5102 mock_add_persistent_root_disk_to_disk_list
,
5103 mock_find_persistent_volumes
,
5104 mock_find_persistent_root_volumes
,
5105 mock_prepare_vdu_ssh_keys
,
5106 mock_prepare_vdu_cloud_init
,
5107 mock_prepare_vdu_interfaces
,
5108 mock_locate_vdu_interfaces
,
5109 mock_sort_vdu_interfaces
,
5111 """There is not any persistent storage."""
5112 target_vdu
= deepcopy(target_vdu_wthout_persistent_storage
)
5115 target_vdu
["interfaces"] = interfaces_wth_all_positions
5116 mock_prepare_vdu_cloud_init
.return_value
= {}
5117 mock_prepare_vdu_affinity_group_list
.return_value
= []
5119 new_kwargs
= deepcopy(kwargs
)
5124 "tasks_by_target_record_id": {},
5128 expected_extra_dict_copy
= deepcopy(expected_extra_dict2
)
5129 vnfd
= deepcopy(vnfd_wthout_persistent_storage
)
5130 db
.get_one
.return_value
= vnfd
5131 result
= Ns
._process
_vdu
_params
(
5132 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5134 mock_sort_vdu_interfaces
.assert_called_once_with(target_vdu
)
5135 mock_locate_vdu_interfaces
.assert_not_called()
5136 mock_prepare_vdu_cloud_init
.assert_called_once()
5137 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5138 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5139 mock_prepare_vdu_interfaces
.assert_called_once_with(
5141 expected_extra_dict_copy
,
5148 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5149 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5150 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5151 mock_find_persistent_volumes
.assert_not_called()
5153 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5154 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5155 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5156 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5157 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5158 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5159 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5160 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5161 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5162 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5163 def test_process_vdu_params_interfaces_partially_located(
5165 mock_prepare_vdu_affinity_group_list
,
5166 mock_add_persistent_ordinary_disks_to_disk_list
,
5167 mock_add_persistent_root_disk_to_disk_list
,
5168 mock_find_persistent_volumes
,
5169 mock_find_persistent_root_volumes
,
5170 mock_prepare_vdu_ssh_keys
,
5171 mock_prepare_vdu_cloud_init
,
5172 mock_prepare_vdu_interfaces
,
5173 mock_locate_vdu_interfaces
,
5174 mock_sort_vdu_interfaces
,
5176 """Some interfaces have position."""
5177 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5180 target_vdu
["interfaces"] = [
5183 "ns-vld-id": "net1",
5185 {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 2},
5188 "ns-vld-id": "mgmtnet",
5191 mock_prepare_vdu_cloud_init
.return_value
= {}
5192 mock_prepare_vdu_affinity_group_list
.return_value
= []
5193 persistent_root_disk
= {
5194 "persistent-root-volume": {
5195 "image_id": "ubuntu20.04",
5200 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5202 new_kwargs
= deepcopy(kwargs
)
5207 "tasks_by_target_record_id": {},
5212 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5213 db
.get_one
.return_value
= vnfd
5214 result
= Ns
._process
_vdu
_params
(
5215 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5217 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
5218 mock_sort_vdu_interfaces
.assert_not_called()
5219 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5220 mock_prepare_vdu_cloud_init
.assert_called_once()
5221 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5222 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5223 mock_prepare_vdu_interfaces
.assert_called_once_with(
5225 expected_extra_dict_copy
,
5232 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5233 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5234 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5235 mock_find_persistent_volumes
.assert_not_called()
5236 mock_find_persistent_root_volumes
.assert_not_called()
5238 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5239 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5240 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5241 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5242 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5243 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5244 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5245 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5246 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5247 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5248 def test_process_vdu_params_no_interface_position(
5250 mock_prepare_vdu_affinity_group_list
,
5251 mock_add_persistent_ordinary_disks_to_disk_list
,
5252 mock_add_persistent_root_disk_to_disk_list
,
5253 mock_find_persistent_volumes
,
5254 mock_find_persistent_root_volumes
,
5255 mock_prepare_vdu_ssh_keys
,
5256 mock_prepare_vdu_cloud_init
,
5257 mock_prepare_vdu_interfaces
,
5258 mock_locate_vdu_interfaces
,
5259 mock_sort_vdu_interfaces
,
5261 """Interfaces do not have position."""
5262 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5265 target_vdu
["interfaces"] = interfaces_wthout_positions
5266 mock_prepare_vdu_cloud_init
.return_value
= {}
5267 mock_prepare_vdu_affinity_group_list
.return_value
= []
5268 persistent_root_disk
= {
5269 "persistent-root-volume": {
5270 "image_id": "ubuntu20.04",
5275 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5276 new_kwargs
= deepcopy(kwargs
)
5281 "tasks_by_target_record_id": {},
5286 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5287 db
.get_one
.return_value
= vnfd
5288 result
= Ns
._process
_vdu
_params
(
5289 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5291 expected_extra_dict_copy
= deepcopy(expected_extra_dict
)
5292 mock_sort_vdu_interfaces
.assert_not_called()
5293 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5294 mock_prepare_vdu_cloud_init
.assert_called_once()
5295 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5296 mock_add_persistent_ordinary_disks_to_disk_list
.assert_called_once()
5297 mock_prepare_vdu_interfaces
.assert_called_once_with(
5299 expected_extra_dict_copy
,
5306 self
.assertDictEqual(result
, expected_extra_dict_copy
)
5307 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5308 mock_prepare_vdu_affinity_group_list
.assert_called_once()
5309 mock_find_persistent_volumes
.assert_not_called()
5310 mock_find_persistent_root_volumes
.assert_not_called()
5312 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5313 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5314 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5315 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5316 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5317 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5318 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5319 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5320 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5321 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5322 def test_process_vdu_params_prepare_vdu_interfaces_raises_exception(
5324 mock_prepare_vdu_affinity_group_list
,
5325 mock_add_persistent_ordinary_disks_to_disk_list
,
5326 mock_add_persistent_root_disk_to_disk_list
,
5327 mock_find_persistent_volumes
,
5328 mock_find_persistent_root_volumes
,
5329 mock_prepare_vdu_ssh_keys
,
5330 mock_prepare_vdu_cloud_init
,
5331 mock_prepare_vdu_interfaces
,
5332 mock_locate_vdu_interfaces
,
5333 mock_sort_vdu_interfaces
,
5335 """Prepare vdu interfaces method raises exception."""
5336 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5339 target_vdu
["interfaces"] = interfaces_wthout_positions
5340 mock_prepare_vdu_cloud_init
.return_value
= {}
5341 mock_prepare_vdu_affinity_group_list
.return_value
= []
5342 persistent_root_disk
= {
5343 "persistent-root-volume": {
5344 "image_id": "ubuntu20.04",
5349 mock_find_persistent_root_volumes
.return_value
= persistent_root_disk
5350 new_kwargs
= deepcopy(kwargs
)
5355 "tasks_by_target_record_id": {},
5359 mock_prepare_vdu_interfaces
.side_effect
= TypeError
5361 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5362 db
.get_one
.return_value
= vnfd
5363 with self
.assertRaises(Exception) as err
:
5364 Ns
._process
_vdu
_params
(
5365 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5367 self
.assertEqual(type(err
), TypeError)
5368 mock_sort_vdu_interfaces
.assert_not_called()
5369 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5370 mock_prepare_vdu_cloud_init
.assert_not_called()
5371 mock_add_persistent_root_disk_to_disk_list
.assert_not_called()
5372 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5373 mock_prepare_vdu_interfaces
.assert_called_once()
5374 mock_prepare_vdu_ssh_keys
.assert_not_called()
5375 mock_prepare_vdu_affinity_group_list
.assert_not_called()
5376 mock_find_persistent_volumes
.assert_not_called()
5377 mock_find_persistent_root_volumes
.assert_not_called()
5379 @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5380 @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5381 @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5382 @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5383 @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5384 @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5385 @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5386 @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5387 @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5388 @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5389 def test_process_vdu_params_add_persistent_root_disk_raises_exception(
5391 mock_prepare_vdu_affinity_group_list
,
5392 mock_add_persistent_ordinary_disks_to_disk_list
,
5393 mock_add_persistent_root_disk_to_disk_list
,
5394 mock_find_persistent_volumes
,
5395 mock_find_persistent_root_volumes
,
5396 mock_prepare_vdu_ssh_keys
,
5397 mock_prepare_vdu_cloud_init
,
5398 mock_prepare_vdu_interfaces
,
5399 mock_locate_vdu_interfaces
,
5400 mock_sort_vdu_interfaces
,
5402 """Add persistent root disk method raises exception."""
5403 target_vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5406 target_vdu
["interfaces"] = interfaces_wthout_positions
5407 mock_prepare_vdu_cloud_init
.return_value
= {}
5408 mock_prepare_vdu_affinity_group_list
.return_value
= []
5409 mock_add_persistent_root_disk_to_disk_list
.side_effect
= KeyError
5410 new_kwargs
= deepcopy(kwargs
)
5415 "tasks_by_target_record_id": {},
5420 vnfd
= deepcopy(vnfd_wth_persistent_storage
)
5421 db
.get_one
.return_value
= vnfd
5422 with self
.assertRaises(Exception) as err
:
5423 Ns
._process
_vdu
_params
(
5424 target_vdu
, indata
, vim_info
=None, target_record_id
=None, **new_kwargs
5426 self
.assertEqual(type(err
), KeyError)
5427 mock_sort_vdu_interfaces
.assert_not_called()
5428 mock_locate_vdu_interfaces
.assert_called_once_with(target_vdu
)
5429 mock_prepare_vdu_cloud_init
.assert_called_once()
5430 mock_add_persistent_root_disk_to_disk_list
.assert_called_once()
5431 mock_add_persistent_ordinary_disks_to_disk_list
.assert_not_called()
5432 mock_prepare_vdu_interfaces
.assert_called_once_with(
5436 f
"{ns_preffix}:image.0",
5437 f
"{ns_preffix}:flavor.0",
5447 mock_prepare_vdu_ssh_keys
.assert_called_once_with(target_vdu
, None, {})
5448 mock_prepare_vdu_affinity_group_list
.assert_not_called()
5449 mock_find_persistent_volumes
.assert_not_called()
5450 mock_find_persistent_root_volumes
.assert_not_called()
5452 def test_select_persistent_root_disk(self
):
5453 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5454 vdu
["virtual-storage-desc"] = [
5455 "persistent-root-volume",
5456 "persistent-volume2",
5459 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5460 expected_result
= vsd
5461 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5462 self
.assertEqual(result
, expected_result
)
5464 def test_select_persistent_root_disk_first_vsd_is_different(self
):
5465 """VDU first virtual-storage-desc is different than vsd id."""
5466 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5467 vdu
["virtual-storage-desc"] = [
5468 "persistent-volume2",
5469 "persistent-root-volume",
5472 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5473 expected_result
= None
5474 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5475 self
.assertEqual(result
, expected_result
)
5477 def test_select_persistent_root_disk_vsd_is_not_persistent(self
):
5478 """vsd type is not persistent."""
5479 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5480 vdu
["virtual-storage-desc"] = [
5481 "persistent-volume2",
5482 "persistent-root-volume",
5485 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5486 vsd
["type-of-storage"] = "etsi-nfv-descriptors:ephemeral-storage"
5487 expected_result
= None
5488 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5489 self
.assertEqual(result
, expected_result
)
5491 def test_select_persistent_root_disk_vsd_does_not_have_size(self
):
5492 """vsd size is None."""
5493 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5494 vdu
["virtual-storage-desc"] = [
5495 "persistent-volume2",
5496 "persistent-root-volume",
5499 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5500 vsd
["size-of-storage"] = None
5501 expected_result
= None
5502 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5503 self
.assertEqual(result
, expected_result
)
5505 def test_select_persistent_root_disk_vdu_wthout_vsd(self
):
5506 """VDU does not have virtual-storage-desc."""
5507 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5508 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"][1]
5509 expected_result
= None
5510 result
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
5511 self
.assertEqual(result
, expected_result
)
5513 def test_select_persistent_root_disk_invalid_vsd_type(self
):
5514 """vsd is list, expected to be a dict."""
5515 vdu
= deepcopy(target_vdu_wth_persistent_storage
)
5516 vsd
= deepcopy(vnfd_wth_persistent_storage
)["virtual-storage-desc"]
5517 with self
.assertRaises(AttributeError):
5518 Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)