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 #######################################################################################
19 from unittest
.mock
import MagicMock
, Mock
, patch
21 from jinja2
import TemplateError
, TemplateNotFound
, UndefinedError
22 from osm_ng_ro
.ns
import Ns
, NsException
25 __author__
= "Eduardo Sousa"
26 __date__
= "$19-NOV-2021 00:00:00$"
29 class TestNs(unittest
.TestCase
):
33 def test__create_task_without_extra_dict(self
):
35 "target_id": "vim_openstack_1",
36 "action_id": "123456",
38 "task_id": "123456:1",
39 "status": "SCHEDULED",
42 "target_record": "test_target_record",
43 "target_record_id": "test_target_record_id",
46 "action_id": "123456",
51 task
= Ns
._create
_task
(
52 deployment_info
=deployment_info
,
53 target_id
="vim_openstack_1",
56 target_record
="test_target_record",
57 target_record_id
="test_target_record_id",
60 self
.assertEqual(deployment_info
.get("task_index"), 2)
61 self
.assertDictEqual(task
, expected_result
)
63 def test__create_task(self
):
65 "target_id": "vim_openstack_1",
66 "action_id": "123456",
68 "task_id": "123456:1",
69 "status": "SCHEDULED",
72 "target_record": "test_target_record",
73 "target_record_id": "test_target_record_id",
74 # values coming from extra_dict
75 "params": "test_params",
76 "find_params": "test_find_params",
77 "depends_on": "test_depends_on",
80 "action_id": "123456",
85 task
= Ns
._create
_task
(
86 deployment_info
=deployment_info
,
87 target_id
="vim_openstack_1",
90 target_record
="test_target_record",
91 target_record_id
="test_target_record_id",
93 "params": "test_params",
94 "find_params": "test_find_params",
95 "depends_on": "test_depends_on",
99 self
.assertEqual(deployment_info
.get("task_index"), 2)
100 self
.assertDictEqual(task
, expected_result
)
102 @patch("osm_ng_ro.ns.time")
103 def test__create_ro_task(self
, mock_time
: Mock
):
104 now
= 1637324838.994551
105 mock_time
.return_value
= now
107 "target_id": "vim_openstack_1",
108 "action_id": "123456",
110 "task_id": "123456:1",
111 "status": "SCHEDULED",
114 "target_record": "test_target_record",
115 "target_record_id": "test_target_record_id",
116 # values coming from extra_dict
117 "params": "test_params",
118 "find_params": "test_find_params",
119 "depends_on": "test_depends_on",
125 "target_id": "vim_openstack_1",
128 "created_items": None,
141 ro_task
= Ns
._create
_ro
_task
(
142 target_id
="vim_openstack_1",
146 self
.assertDictEqual(ro_task
, expected_result
)
148 def test__process_image_params_with_empty_target_image(self
):
154 result
= Ns
._process
_image
_params
(
155 target_image
=target_image
,
158 target_record_id
=None,
161 self
.assertDictEqual(expected_result
, result
)
163 def test__process_image_params_with_wrong_target_image(self
):
168 "no_image": "to_see_here",
171 result
= Ns
._process
_image
_params
(
172 target_image
=target_image
,
175 target_record_id
=None,
178 self
.assertDictEqual(expected_result
, result
)
180 def test__process_image_params_with_image(self
):
192 result
= Ns
._process
_image
_params
(
193 target_image
=target_image
,
196 target_record_id
=None,
199 self
.assertDictEqual(expected_result
, result
)
201 def test__process_image_params_with_vim_image_id(self
):
210 "vim_image_id": "123456",
213 result
= Ns
._process
_image
_params
(
214 target_image
=target_image
,
217 target_record_id
=None,
220 self
.assertDictEqual(expected_result
, result
)
222 def test__process_image_params_with_image_checksum(self
):
226 "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
231 "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
234 result
= Ns
._process
_image
_params
(
235 target_image
=target_image
,
238 target_record_id
=None,
241 self
.assertDictEqual(expected_result
, result
)
243 def test__get_resource_allocation_params_with_empty_target_image(self
):
245 quota_descriptor
= {}
247 result
= Ns
._get
_resource
_allocation
_params
(
248 quota_descriptor
=quota_descriptor
,
251 self
.assertDictEqual(expected_result
, result
)
253 def test__get_resource_allocation_params_with_wrong_target_image(self
):
256 "no_quota": "present_here",
259 result
= Ns
._get
_resource
_allocation
_params
(
260 quota_descriptor
=quota_descriptor
,
263 self
.assertDictEqual(expected_result
, result
)
265 def test__get_resource_allocation_params_with_limit(self
):
273 result
= Ns
._get
_resource
_allocation
_params
(
274 quota_descriptor
=quota_descriptor
,
277 self
.assertDictEqual(expected_result
, result
)
279 def test__get_resource_allocation_params_with_reserve(self
):
287 result
= Ns
._get
_resource
_allocation
_params
(
288 quota_descriptor
=quota_descriptor
,
291 self
.assertDictEqual(expected_result
, result
)
293 def test__get_resource_allocation_params_with_shares(self
):
301 result
= Ns
._get
_resource
_allocation
_params
(
302 quota_descriptor
=quota_descriptor
,
305 self
.assertDictEqual(expected_result
, result
)
307 def test__get_resource_allocation_params(self
):
319 result
= Ns
._get
_resource
_allocation
_params
(
320 quota_descriptor
=quota_descriptor
,
323 self
.assertDictEqual(expected_result
, result
)
325 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
326 def test__process_guest_epa_quota_params_with_empty_quota_epa_cpu(
334 result
= Ns
._process
_guest
_epa
_quota
_params
(
335 guest_epa_quota
=guest_epa_quota
,
336 epa_vcpu_set
=epa_vcpu_set
,
339 self
.assertDictEqual(expected_result
, result
)
340 self
.assertFalse(resource_allocation
.called
)
342 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
343 def test__process_guest_epa_quota_params_with_empty_quota_false_epa_cpu(
351 result
= Ns
._process
_guest
_epa
_quota
_params
(
352 guest_epa_quota
=guest_epa_quota
,
353 epa_vcpu_set
=epa_vcpu_set
,
356 self
.assertDictEqual(expected_result
, result
)
357 self
.assertFalse(resource_allocation
.called
)
359 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
360 def test__process_guest_epa_quota_params_with_wrong_quota_epa_cpu(
366 "no-quota": "nothing",
370 result
= Ns
._process
_guest
_epa
_quota
_params
(
371 guest_epa_quota
=guest_epa_quota
,
372 epa_vcpu_set
=epa_vcpu_set
,
375 self
.assertDictEqual(expected_result
, result
)
376 self
.assertFalse(resource_allocation
.called
)
378 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
379 def test__process_guest_epa_quota_params_with_wrong_quota_false_epa_cpu(
385 "no-quota": "nothing",
389 result
= Ns
._process
_guest
_epa
_quota
_params
(
390 guest_epa_quota
=guest_epa_quota
,
391 epa_vcpu_set
=epa_vcpu_set
,
394 self
.assertDictEqual(expected_result
, result
)
395 self
.assertFalse(resource_allocation
.called
)
397 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
398 def test__process_guest_epa_quota_params_with_cpu_quota_epa_cpu(
412 result
= Ns
._process
_guest
_epa
_quota
_params
(
413 guest_epa_quota
=guest_epa_quota
,
414 epa_vcpu_set
=epa_vcpu_set
,
417 self
.assertDictEqual(expected_result
, result
)
418 self
.assertFalse(resource_allocation
.called
)
420 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
421 def test__process_guest_epa_quota_params_with_cpu_quota_false_epa_cpu(
441 resource_allocation_param
= {
446 resource_allocation
.return_value
= {
452 result
= Ns
._process
_guest
_epa
_quota
_params
(
453 guest_epa_quota
=guest_epa_quota
,
454 epa_vcpu_set
=epa_vcpu_set
,
457 resource_allocation
.assert_called_once_with(resource_allocation_param
)
458 self
.assertDictEqual(expected_result
, result
)
460 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
461 def test__process_guest_epa_quota_params_with_mem_quota_epa_cpu(
481 resource_allocation_param
= {
486 resource_allocation
.return_value
= {
492 result
= Ns
._process
_guest
_epa
_quota
_params
(
493 guest_epa_quota
=guest_epa_quota
,
494 epa_vcpu_set
=epa_vcpu_set
,
497 resource_allocation
.assert_called_once_with(resource_allocation_param
)
498 self
.assertDictEqual(expected_result
, result
)
500 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
501 def test__process_guest_epa_quota_params_with_mem_quota_false_epa_cpu(
521 resource_allocation_param
= {
526 resource_allocation
.return_value
= {
532 result
= Ns
._process
_guest
_epa
_quota
_params
(
533 guest_epa_quota
=guest_epa_quota
,
534 epa_vcpu_set
=epa_vcpu_set
,
537 resource_allocation
.assert_called_once_with(resource_allocation_param
)
538 self
.assertDictEqual(expected_result
, result
)
540 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
541 def test__process_guest_epa_quota_params_with_disk_io_quota_epa_cpu(
561 resource_allocation_param
= {
566 resource_allocation
.return_value
= {
572 result
= Ns
._process
_guest
_epa
_quota
_params
(
573 guest_epa_quota
=guest_epa_quota
,
574 epa_vcpu_set
=epa_vcpu_set
,
577 resource_allocation
.assert_called_once_with(resource_allocation_param
)
578 self
.assertDictEqual(expected_result
, result
)
580 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
581 def test__process_guest_epa_quota_params_with_disk_io_quota_false_epa_cpu(
601 resource_allocation_param
= {
606 resource_allocation
.return_value
= {
612 result
= Ns
._process
_guest
_epa
_quota
_params
(
613 guest_epa_quota
=guest_epa_quota
,
614 epa_vcpu_set
=epa_vcpu_set
,
617 resource_allocation
.assert_called_once_with(resource_allocation_param
)
618 self
.assertDictEqual(expected_result
, result
)
620 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
621 def test__process_guest_epa_quota_params_with_vif_quota_epa_cpu(
641 resource_allocation_param
= {
646 resource_allocation
.return_value
= {
652 result
= Ns
._process
_guest
_epa
_quota
_params
(
653 guest_epa_quota
=guest_epa_quota
,
654 epa_vcpu_set
=epa_vcpu_set
,
657 resource_allocation
.assert_called_once_with(resource_allocation_param
)
658 self
.assertDictEqual(expected_result
, result
)
660 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
661 def test__process_guest_epa_quota_params_with_vif_quota_false_epa_cpu(
681 resource_allocation_param
= {
686 resource_allocation
.return_value
= {
692 result
= Ns
._process
_guest
_epa
_quota
_params
(
693 guest_epa_quota
=guest_epa_quota
,
694 epa_vcpu_set
=epa_vcpu_set
,
697 resource_allocation
.assert_called_once_with(resource_allocation_param
)
698 self
.assertDictEqual(expected_result
, result
)
700 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
701 def test__process_guest_epa_quota_params_with_quota_epa_cpu(
746 resource_allocation
.return_value
= {
752 result
= Ns
._process
_guest
_epa
_quota
_params
(
753 guest_epa_quota
=guest_epa_quota
,
754 epa_vcpu_set
=epa_vcpu_set
,
757 self
.assertTrue(resource_allocation
.called
)
758 self
.assertDictEqual(expected_result
, result
)
760 @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
761 def test__process_guest_epa_quota_params_with_quota_epa_cpu_no_set(
811 resource_allocation
.return_value
= {
817 result
= Ns
._process
_guest
_epa
_quota
_params
(
818 guest_epa_quota
=guest_epa_quota
,
819 epa_vcpu_set
=epa_vcpu_set
,
822 self
.assertTrue(resource_allocation
.called
)
823 self
.assertDictEqual(expected_result
, result
)
825 def test__process_guest_epa_numa_params_with_empty_numa_params(self
):
826 expected_numa_result
= {}
827 expected_epa_vcpu_set_result
= False
830 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
831 guest_epa_quota
=guest_epa_quota
,
834 self
.assertDictEqual(expected_numa_result
, numa_result
)
835 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
837 def test__process_guest_epa_numa_params_with_wrong_numa_params(self
):
838 expected_numa_result
= {}
839 expected_epa_vcpu_set_result
= False
840 guest_epa_quota
= {"no_nume": "here"}
842 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
843 guest_epa_quota
=guest_epa_quota
,
846 self
.assertDictEqual(expected_numa_result
, numa_result
)
847 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
849 def test__process_guest_epa_numa_params_with_numa_node_policy(self
):
850 expected_numa_result
= {}
851 expected_epa_vcpu_set_result
= False
852 guest_epa_quota
= {"numa-node-policy": {}}
854 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
855 guest_epa_quota
=guest_epa_quota
,
858 self
.assertDictEqual(expected_numa_result
, numa_result
)
859 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
861 def test__process_guest_epa_numa_params_with_no_node(self
):
862 expected_numa_result
= {}
863 expected_epa_vcpu_set_result
= False
865 "numa-node-policy": {
870 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
871 guest_epa_quota
=guest_epa_quota
,
874 self
.assertDictEqual(expected_numa_result
, numa_result
)
875 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
877 def test__process_guest_epa_numa_params_with_1_node_num_cores(self
):
878 expected_numa_result
= {"cores": 3}
879 expected_epa_vcpu_set_result
= True
881 "numa-node-policy": {
890 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
891 guest_epa_quota
=guest_epa_quota
,
894 self
.assertDictEqual(expected_numa_result
, numa_result
)
895 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
897 def test__process_guest_epa_numa_params_with_1_node_paired_threads(self
):
898 expected_numa_result
= {"paired-threads": 3}
899 expected_epa_vcpu_set_result
= True
901 "numa-node-policy": {
904 "paired-threads": {"num-paired-threads": "3"},
910 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
911 guest_epa_quota
=guest_epa_quota
,
914 self
.assertDictEqual(expected_numa_result
, numa_result
)
915 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
917 def test__process_guest_epa_numa_params_with_1_node_paired_threads_ids(self
):
918 expected_numa_result
= {
919 "paired-threads-id": [("0", "1"), ("4", "5")],
921 expected_epa_vcpu_set_result
= False
923 "numa-node-policy": {
927 "paired-thread-ids": [
943 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
944 guest_epa_quota
=guest_epa_quota
,
947 self
.assertDictEqual(expected_numa_result
, numa_result
)
948 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
950 def test__process_guest_epa_numa_params_with_1_node_num_threads(self
):
951 expected_numa_result
= {"threads": 3}
952 expected_epa_vcpu_set_result
= True
954 "numa-node-policy": {
963 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
964 guest_epa_quota
=guest_epa_quota
,
967 self
.assertDictEqual(expected_numa_result
, numa_result
)
968 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
970 def test__process_guest_epa_numa_params_with_1_node_memory_mb(self
):
971 expected_numa_result
= {"memory": 2}
972 expected_epa_vcpu_set_result
= False
974 "numa-node-policy": {
983 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
984 guest_epa_quota
=guest_epa_quota
,
987 self
.assertDictEqual(expected_numa_result
, numa_result
)
988 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
990 def test__process_guest_epa_numa_params_with_1_node(self
):
991 expected_numa_result
= {
994 "paired-threads-id": [("0", "1"), ("4", "5")],
998 expected_epa_vcpu_set_result
= True
1000 "numa-node-policy": {
1005 "num-paired-threads": "3",
1006 "paired-thread-ids": [
1024 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1025 guest_epa_quota
=guest_epa_quota
,
1028 self
.assertDictEqual(expected_numa_result
, numa_result
)
1029 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1031 def test__process_guest_epa_numa_params_with_2_nodes(self
):
1032 expected_numa_result
= {
1034 "paired-threads": 3,
1035 "paired-threads-id": [("0", "1"), ("4", "5")],
1039 expected_epa_vcpu_set_result
= True
1041 "numa-node-policy": {
1046 "num-paired-threads": "3",
1047 "paired-thread-ids": [
1064 "num-paired-threads": "7",
1065 "paired-thread-ids": [
1083 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_numa
_params
(
1084 guest_epa_quota
=guest_epa_quota
,
1087 self
.assertDictEqual(expected_numa_result
, numa_result
)
1088 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1090 def test__process_guest_epa_cpu_pinning_params_with_empty_params(self
):
1091 expected_numa_result
= {}
1092 expected_epa_vcpu_set_result
= False
1093 guest_epa_quota
= {}
1095 epa_vcpu_set
= False
1097 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1098 guest_epa_quota
=guest_epa_quota
,
1099 vcpu_count
=vcpu_count
,
1100 epa_vcpu_set
=epa_vcpu_set
,
1103 self
.assertDictEqual(expected_numa_result
, numa_result
)
1104 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1106 def test__process_guest_epa_cpu_pinning_params_with_wrong_params(self
):
1107 expected_numa_result
= {}
1108 expected_epa_vcpu_set_result
= False
1110 "no-cpu-pinning-policy": "here",
1113 epa_vcpu_set
= False
1115 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1116 guest_epa_quota
=guest_epa_quota
,
1117 vcpu_count
=vcpu_count
,
1118 epa_vcpu_set
=epa_vcpu_set
,
1121 self
.assertDictEqual(expected_numa_result
, numa_result
)
1122 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1124 def test__process_guest_epa_cpu_pinning_params_with_epa_vcpu_set(self
):
1125 expected_numa_result
= {}
1126 expected_epa_vcpu_set_result
= True
1128 "cpu-pinning-policy": "DEDICATED",
1133 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1134 guest_epa_quota
=guest_epa_quota
,
1135 vcpu_count
=vcpu_count
,
1136 epa_vcpu_set
=epa_vcpu_set
,
1139 self
.assertDictEqual(expected_numa_result
, numa_result
)
1140 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1142 def test__process_guest_epa_cpu_pinning_params_with_threads(self
):
1143 expected_numa_result
= {"threads": 3}
1144 expected_epa_vcpu_set_result
= True
1146 "cpu-pinning-policy": "DEDICATED",
1147 "cpu-thread-pinning-policy": "PREFER",
1150 epa_vcpu_set
= False
1152 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1153 guest_epa_quota
=guest_epa_quota
,
1154 vcpu_count
=vcpu_count
,
1155 epa_vcpu_set
=epa_vcpu_set
,
1158 self
.assertDictEqual(expected_numa_result
, numa_result
)
1159 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1161 def test__process_guest_epa_cpu_pinning_params(self
):
1162 expected_numa_result
= {"cores": 3}
1163 expected_epa_vcpu_set_result
= True
1165 "cpu-pinning-policy": "DEDICATED",
1168 epa_vcpu_set
= False
1170 numa_result
, epa_vcpu_set_result
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
1171 guest_epa_quota
=guest_epa_quota
,
1172 vcpu_count
=vcpu_count
,
1173 epa_vcpu_set
=epa_vcpu_set
,
1176 self
.assertDictEqual(expected_numa_result
, numa_result
)
1177 self
.assertEqual(expected_epa_vcpu_set_result
, epa_vcpu_set_result
)
1179 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1180 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1181 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1182 def test__process_guest_epa_params_with_empty_params(
1184 guest_epa_numa_params
,
1185 guest_epa_cpu_pinning_params
,
1186 guest_epa_quota_params
,
1188 expected_result
= {}
1191 result
= Ns
._process
_epa
_params
(
1192 target_flavor
=target_flavor
,
1195 self
.assertDictEqual(expected_result
, result
)
1196 self
.assertFalse(guest_epa_numa_params
.called
)
1197 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1198 self
.assertFalse(guest_epa_quota_params
.called
)
1200 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1201 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1202 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1203 def test__process_guest_epa_params_with_wrong_params(
1205 guest_epa_numa_params
,
1206 guest_epa_cpu_pinning_params
,
1207 guest_epa_quota_params
,
1209 expected_result
= {}
1211 "no-guest-epa": "here",
1214 result
= Ns
._process
_epa
_params
(
1215 target_flavor
=target_flavor
,
1218 self
.assertDictEqual(expected_result
, result
)
1219 self
.assertFalse(guest_epa_numa_params
.called
)
1220 self
.assertFalse(guest_epa_cpu_pinning_params
.called
)
1221 self
.assertFalse(guest_epa_quota_params
.called
)
1223 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1224 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1225 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1226 def test__process_guest_epa_params(
1228 guest_epa_numa_params
,
1229 guest_epa_cpu_pinning_params
,
1230 guest_epa_quota_params
,
1232 expected_result
= {}
1239 guest_epa_numa_params
.return_value
= ({}, False)
1240 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1241 guest_epa_quota_params
.return_value
= {}
1243 result
= Ns
._process
_epa
_params
(
1244 target_flavor
=target_flavor
,
1247 self
.assertDictEqual(expected_result
, result
)
1248 self
.assertTrue(guest_epa_numa_params
.called
)
1249 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1250 self
.assertTrue(guest_epa_quota_params
.called
)
1252 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1253 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1254 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1255 def test__process_guest_epa_params_with_mempage_size(
1257 guest_epa_numa_params
,
1258 guest_epa_cpu_pinning_params
,
1259 guest_epa_quota_params
,
1262 "mempage-size": "1G",
1265 "guest-epa": {"vcpu-count": 1, "mempage-size": "1G"},
1268 guest_epa_numa_params
.return_value
= ({}, False)
1269 guest_epa_cpu_pinning_params
.return_value
= ({}, False)
1270 guest_epa_quota_params
.return_value
= {}
1272 result
= Ns
._process
_epa
_params
(
1273 target_flavor
=target_flavor
,
1276 self
.assertDictEqual(expected_result
, result
)
1277 self
.assertTrue(guest_epa_numa_params
.called
)
1278 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1279 self
.assertTrue(guest_epa_quota_params
.called
)
1281 @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1282 @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1283 @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1284 def test__process_guest_epa_params_with_numa(
1286 guest_epa_numa_params
,
1287 guest_epa_cpu_pinning_params
,
1288 guest_epa_quota_params
,
1291 "mempage-size": "1G",
1296 "paired-threads": 3,
1297 "paired-threads-id": [("0", "1"), ("4", "5")],
1301 "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1302 "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1303 "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1304 "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1309 "mempage-size": "1G",
1310 "cpu-pinning-policy": "DEDICATED",
1311 "cpu-thread-pinning-policy": "PREFER",
1312 "numa-node-policy": {
1317 "num-paired-threads": "3",
1318 "paired-thread-ids": [
1357 guest_epa_numa_params
.return_value
= (
1360 "paired-threads": 3,
1361 "paired-threads-id": [("0", "1"), ("4", "5")],
1367 guest_epa_cpu_pinning_params
.return_value
= (
1373 guest_epa_quota_params
.return_value
= {
1396 result
= Ns
._process
_epa
_params
(
1397 target_flavor
=target_flavor
,
1400 self
.assertDictEqual(expected_result
, result
)
1401 self
.assertTrue(guest_epa_numa_params
.called
)
1402 self
.assertTrue(guest_epa_cpu_pinning_params
.called
)
1403 self
.assertTrue(guest_epa_quota_params
.called
)
1405 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1406 def test__process_flavor_params_with_empty_target_flavor(
1413 target_record_id
= ""
1415 with self
.assertRaises(KeyError):
1416 Ns
._process
_flavor
_params
(
1417 target_flavor
=target_flavor
,
1420 target_record_id
=target_record_id
,
1423 self
.assertFalse(epa_params
.called
)
1425 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1426 def test__process_flavor_params_with_wrong_target_flavor(
1431 "no-target-flavor": "here",
1435 target_record_id
= ""
1437 with self
.assertRaises(KeyError):
1438 Ns
._process
_flavor
_params
(
1439 target_flavor
=target_flavor
,
1442 target_record_id
=target_record_id
,
1445 self
.assertFalse(epa_params
.called
)
1447 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1448 def test__process_flavor_params_with_empty_indata(
1472 "memory-mb": "1024",
1477 target_record_id
= ""
1479 epa_params
.return_value
= {}
1481 result
= Ns
._process
_flavor
_params
(
1482 target_flavor
=target_flavor
,
1485 target_record_id
=target_record_id
,
1488 self
.assertTrue(epa_params
.called
)
1489 self
.assertDictEqual(result
, expected_result
)
1491 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1492 def test__process_flavor_params_with_wrong_indata(
1516 "memory-mb": "1024",
1523 target_record_id
= ""
1525 epa_params
.return_value
= {}
1527 result
= Ns
._process
_flavor
_params
(
1528 target_flavor
=target_flavor
,
1531 target_record_id
=target_record_id
,
1534 self
.assertTrue(epa_params
.called
)
1535 self
.assertDictEqual(result
, expected_result
)
1537 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1538 def test__process_flavor_params_with_ephemeral_disk(
1565 "memory-mb": "1024",
1573 "ns-flavor-id": "test_id",
1574 "virtual-storages": [
1576 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1577 "size-of-storage": "10",
1586 target_record_id
= ""
1588 epa_params
.return_value
= {}
1590 result
= Ns
._process
_flavor
_params
(
1591 target_flavor
=target_flavor
,
1594 target_record_id
=target_record_id
,
1597 self
.assertTrue(epa_params
.called
)
1598 self
.assertDictEqual(result
, expected_result
)
1600 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1601 def test__process_flavor_params_with_swap_disk(
1628 "memory-mb": "1024",
1636 "ns-flavor-id": "test_id",
1637 "virtual-storages": [
1639 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
1640 "size-of-storage": "20",
1649 target_record_id
= ""
1651 epa_params
.return_value
= {}
1653 result
= Ns
._process
_flavor
_params
(
1654 target_flavor
=target_flavor
,
1657 target_record_id
=target_record_id
,
1660 self
.assertTrue(epa_params
.called
)
1661 self
.assertDictEqual(result
, expected_result
)
1663 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1664 def test__process_flavor_params_with_epa_params(
1675 "numa": "there-is-numa-here",
1686 "numa": "there-is-numa-here",
1695 "memory-mb": "1024",
1700 target_record_id
= ""
1702 epa_params
.return_value
= {
1703 "numa": "there-is-numa-here",
1706 result
= Ns
._process
_flavor
_params
(
1707 target_flavor
=target_flavor
,
1710 target_record_id
=target_record_id
,
1713 self
.assertTrue(epa_params
.called
)
1714 self
.assertDictEqual(result
, expected_result
)
1716 @patch("osm_ng_ro.ns.Ns._process_epa_params")
1717 def test__process_flavor_params(
1730 "numa": "there-is-numa-here",
1743 "numa": "there-is-numa-here",
1752 "memory-mb": "1024",
1760 "ns-flavor-id": "test_id",
1761 "virtual-storages": [
1763 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1764 "size-of-storage": "10",
1767 "type-of-storage": "etsi-nfv-descriptors:swap-storage",
1768 "size-of-storage": "20",
1777 target_record_id
= ""
1779 epa_params
.return_value
= {
1780 "numa": "there-is-numa-here",
1783 result
= Ns
._process
_flavor
_params
(
1784 target_flavor
=target_flavor
,
1787 target_record_id
=target_record_id
,
1790 self
.assertTrue(epa_params
.called
)
1791 self
.assertDictEqual(result
, expected_result
)
1793 def test__ip_profile_to_ro_with_none(self
):
1796 result
= Ns
._ip
_profile
_to
_ro
(
1797 ip_profile
=ip_profile
,
1800 self
.assertIsNone(result
)
1802 def test__ip_profile_to_ro_with_empty_profile(self
):
1805 result
= Ns
._ip
_profile
_to
_ro
(
1806 ip_profile
=ip_profile
,
1809 self
.assertIsNone(result
)
1811 def test__ip_profile_to_ro_with_wrong_profile(self
):
1813 "no-profile": "here",
1816 "ip_version": "IPv4",
1817 "subnet_address": None,
1818 "gateway_address": None,
1819 "dhcp_enabled": False,
1820 "dhcp_start_address": None,
1824 result
= Ns
._ip
_profile
_to
_ro
(
1825 ip_profile
=ip_profile
,
1828 self
.assertDictEqual(expected_result
, result
)
1830 def test__ip_profile_to_ro_with_ipv4_profile(self
):
1832 "ip-version": "ipv4",
1833 "subnet-address": "192.168.0.0/24",
1834 "gateway-address": "192.168.0.254",
1837 "start-address": "192.168.0.10",
1842 "ip_version": "IPv4",
1843 "subnet_address": "192.168.0.0/24",
1844 "gateway_address": "192.168.0.254",
1845 "dhcp_enabled": True,
1846 "dhcp_start_address": "192.168.0.10",
1850 result
= Ns
._ip
_profile
_to
_ro
(
1851 ip_profile
=ip_profile
,
1854 self
.assertDictEqual(expected_result
, result
)
1856 def test__ip_profile_to_ro_with_ipv6_profile(self
):
1858 "ip-version": "ipv6",
1859 "subnet-address": "2001:0200:0001::/48",
1860 "gateway-address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
1863 "start-address": "2001:0200:0001::0010",
1868 "ip_version": "IPv6",
1869 "subnet_address": "2001:0200:0001::/48",
1870 "gateway_address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
1871 "dhcp_enabled": True,
1872 "dhcp_start_address": "2001:0200:0001::0010",
1876 result
= Ns
._ip
_profile
_to
_ro
(
1877 ip_profile
=ip_profile
,
1880 self
.assertDictEqual(expected_result
, result
)
1882 def test__ip_profile_to_ro_with_dns_server(self
):
1884 "ip-version": "ipv4",
1885 "subnet-address": "192.168.0.0/24",
1886 "gateway-address": "192.168.0.254",
1889 "start-address": "192.168.0.10",
1894 "address": "8.8.8.8",
1897 "address": "1.1.1.1",
1900 "address": "1.0.0.1",
1905 "ip_version": "IPv4",
1906 "subnet_address": "192.168.0.0/24",
1907 "gateway_address": "192.168.0.254",
1908 "dhcp_enabled": True,
1909 "dhcp_start_address": "192.168.0.10",
1911 "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
1914 result
= Ns
._ip
_profile
_to
_ro
(
1915 ip_profile
=ip_profile
,
1918 self
.assertDictEqual(expected_result
, result
)
1920 def test__ip_profile_to_ro_with_security_group(self
):
1922 "ip-version": "ipv4",
1923 "subnet-address": "192.168.0.0/24",
1924 "gateway-address": "192.168.0.254",
1927 "start-address": "192.168.0.10",
1931 "some-security-group": "here",
1935 "ip_version": "IPv4",
1936 "subnet_address": "192.168.0.0/24",
1937 "gateway_address": "192.168.0.254",
1938 "dhcp_enabled": True,
1939 "dhcp_start_address": "192.168.0.10",
1942 "some-security-group": "here",
1946 result
= Ns
._ip
_profile
_to
_ro
(
1947 ip_profile
=ip_profile
,
1950 self
.assertDictEqual(expected_result
, result
)
1952 def test__ip_profile_to_ro(self
):
1954 "ip-version": "ipv4",
1955 "subnet-address": "192.168.0.0/24",
1956 "gateway-address": "192.168.0.254",
1959 "start-address": "192.168.0.10",
1964 "address": "8.8.8.8",
1967 "address": "1.1.1.1",
1970 "address": "1.0.0.1",
1974 "some-security-group": "here",
1978 "ip_version": "IPv4",
1979 "subnet_address": "192.168.0.0/24",
1980 "gateway_address": "192.168.0.254",
1981 "dhcp_enabled": True,
1982 "dhcp_start_address": "192.168.0.10",
1984 "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
1986 "some-security-group": "here",
1990 result
= Ns
._ip
_profile
_to
_ro
(
1991 ip_profile
=ip_profile
,
1994 self
.assertDictEqual(expected_result
, result
)
1996 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
1997 def test__process_net_params_with_empty_params(
2008 "provider_network": "some-profile-here",
2010 target_record_id
= ""
2013 "net_name": "ns-name-vld-name",
2014 "net_type": "bridge",
2016 "some_ip_profile": "here",
2018 "provider_network_profile": "some-profile-here",
2022 ip_profile_to_ro
.return_value
= {
2023 "some_ip_profile": "here",
2026 result
= Ns
._process
_net
_params
(
2027 target_vld
=target_vld
,
2030 target_record_id
=target_record_id
,
2033 self
.assertDictEqual(expected_result
, result
)
2034 self
.assertTrue(ip_profile_to_ro
.called
)
2036 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2037 def test__process_net_params_with_vim_info_sdn(
2049 "sdn-ports": ["some", "ports", "here"],
2050 "vlds": ["some", "vlds", "here"],
2053 target_record_id
= "vld.sdn.something"
2056 "sdn-ports": ["some", "ports", "here"],
2057 "vlds": ["some", "vlds", "here"],
2062 result
= Ns
._process
_net
_params
(
2063 target_vld
=target_vld
,
2066 target_record_id
=target_record_id
,
2069 self
.assertDictEqual(expected_result
, result
)
2070 self
.assertFalse(ip_profile_to_ro
.called
)
2072 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2073 def test__process_net_params_with_vim_info_sdn_target_vim(
2085 "sdn-ports": ["some", "ports", "here"],
2086 "vlds": ["some", "vlds", "here"],
2087 "target_vim": "some-vim",
2090 target_record_id
= "vld.sdn.something"
2092 "depends_on": ["some-vim vld.sdn"],
2094 "sdn-ports": ["some", "ports", "here"],
2095 "vlds": ["some", "vlds", "here"],
2096 "target_vim": "some-vim",
2101 result
= Ns
._process
_net
_params
(
2102 target_vld
=target_vld
,
2105 target_record_id
=target_record_id
,
2108 self
.assertDictEqual(expected_result
, result
)
2109 self
.assertFalse(ip_profile_to_ro
.called
)
2111 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2112 def test__process_net_params_with_vim_network_name(
2123 "vim_network_name": "some-network-name",
2125 target_record_id
= "vld.sdn.something"
2129 "name": "some-network-name",
2134 result
= Ns
._process
_net
_params
(
2135 target_vld
=target_vld
,
2138 target_record_id
=target_record_id
,
2141 self
.assertDictEqual(expected_result
, result
)
2142 self
.assertFalse(ip_profile_to_ro
.called
)
2144 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2145 def test__process_net_params_with_vim_network_id(
2156 "vim_network_id": "some-network-id",
2158 target_record_id
= "vld.sdn.something"
2162 "id": "some-network-id",
2167 result
= Ns
._process
_net
_params
(
2168 target_vld
=target_vld
,
2171 target_record_id
=target_record_id
,
2174 self
.assertDictEqual(expected_result
, result
)
2175 self
.assertFalse(ip_profile_to_ro
.called
)
2177 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2178 def test__process_net_params_with_mgmt_network(
2185 "mgmt-network": "some-mgmt-network",
2191 target_record_id
= "vld.sdn.something"
2199 result
= Ns
._process
_net
_params
(
2200 target_vld
=target_vld
,
2203 target_record_id
=target_record_id
,
2206 self
.assertDictEqual(expected_result
, result
)
2207 self
.assertFalse(ip_profile_to_ro
.called
)
2209 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2210 def test__process_net_params_with_underlay_eline(
2216 "underlay": "some-underlay-here",
2223 "provider_network": "some-profile-here",
2225 target_record_id
= ""
2229 "some_ip_profile": "here",
2231 "net_name": "ns-name-vld-name",
2233 "provider_network_profile": "some-profile-here",
2237 ip_profile_to_ro
.return_value
= {
2238 "some_ip_profile": "here",
2241 result
= Ns
._process
_net
_params
(
2242 target_vld
=target_vld
,
2245 target_record_id
=target_record_id
,
2248 self
.assertDictEqual(expected_result
, result
)
2249 self
.assertTrue(ip_profile_to_ro
.called
)
2251 @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2252 def test__process_net_params_with_underlay_elan(
2258 "underlay": "some-underlay-here",
2265 "provider_network": "some-profile-here",
2267 target_record_id
= ""
2271 "some_ip_profile": "here",
2273 "net_name": "ns-name-vld-name",
2275 "provider_network_profile": "some-profile-here",
2279 ip_profile_to_ro
.return_value
= {
2280 "some_ip_profile": "here",
2283 result
= Ns
._process
_net
_params
(
2284 target_vld
=target_vld
,
2287 target_record_id
=target_record_id
,
2290 self
.assertDictEqual(expected_result
, result
)
2291 self
.assertTrue(ip_profile_to_ro
.called
)
2293 def test__get_cloud_init_exception(self
):
2294 db_mock
= MagicMock(name
="database mock")
2299 with self
.assertRaises(NsException
):
2300 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2302 def test__get_cloud_init_file_fs_exception(self
):
2303 db_mock
= MagicMock(name
="database mock")
2306 location
= "vnfr_id_123456:file:test_file"
2307 db_mock
.get_one
.return_value
= {
2310 "folder": "/home/osm",
2311 "pkg-dir": "vnfr_test_dir",
2316 with self
.assertRaises(NsException
):
2317 Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2319 def test__get_cloud_init_file(self
):
2320 db_mock
= MagicMock(name
="database mock")
2321 fs_mock
= MagicMock(name
="filesystem mock")
2322 file_mock
= MagicMock(name
="file mock")
2324 location
= "vnfr_id_123456:file:test_file"
2325 cloud_init_content
= "this is a cloud init file content"
2327 db_mock
.get_one
.return_value
= {
2330 "folder": "/home/osm",
2331 "pkg-dir": "vnfr_test_dir",
2335 fs_mock
.file_open
.return_value
= file_mock
2336 file_mock
.__enter
__.return_value
.read
.return_value
= cloud_init_content
2338 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2340 self
.assertEqual(cloud_init_content
, result
)
2342 def test__get_cloud_init_vdu(self
):
2343 db_mock
= MagicMock(name
="database mock")
2346 location
= "vnfr_id_123456:vdu:0"
2347 cloud_init_content
= "this is a cloud init file content"
2349 db_mock
.get_one
.return_value
= {
2352 "cloud-init": cloud_init_content
,
2357 result
= Ns
._get
_cloud
_init
(db
=db_mock
, fs
=fs_mock
, location
=location
)
2359 self
.assertEqual(cloud_init_content
, result
)
2361 @patch("jinja2.Environment.__init__")
2362 def test__parse_jinja2_undefined_error(self
, env_mock
: Mock
):
2363 cloud_init_content
= None
2367 env_mock
.side_effect
= UndefinedError("UndefinedError occurred.")
2369 with self
.assertRaises(NsException
):
2371 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2374 @patch("jinja2.Environment.__init__")
2375 def test__parse_jinja2_template_error(self
, env_mock
: Mock
):
2376 cloud_init_content
= None
2380 env_mock
.side_effect
= TemplateError("TemplateError occurred.")
2382 with self
.assertRaises(NsException
):
2384 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2387 @patch("jinja2.Environment.__init__")
2388 def test__parse_jinja2_template_not_found(self
, env_mock
: Mock
):
2389 cloud_init_content
= None
2393 env_mock
.side_effect
= TemplateNotFound("TemplateNotFound occurred.")
2395 with self
.assertRaises(NsException
):
2397 cloud_init_content
=cloud_init_content
, params
=params
, context
=context
2400 def test__parse_jinja2(self
):
2403 def test__process_vdu_params_empty_kargs(self
):
2406 def test__process_vdu_params_interface_ns_vld_id(self
):
2409 def test__process_vdu_params_interface_vnf_vld_id(self
):
2412 def test__process_vdu_params_interface_unknown(self
):
2415 def test__process_vdu_params_interface_port_security_enabled(self
):
2418 def test__process_vdu_params_interface_port_security_disable_strategy(self
):
2421 def test__process_vdu_params_interface_sriov(self
):
2424 def test__process_vdu_params_interface_pci_passthrough(self
):
2427 def test__process_vdu_params_interface_om_mgmt(self
):
2430 def test__process_vdu_params_interface_mgmt_interface(self
):
2433 def test__process_vdu_params_interface_mgmt_vnf(self
):
2436 def test__process_vdu_params_interface_bridge(self
):
2439 def test__process_vdu_params_interface_ip_address(self
):
2442 def test__process_vdu_params_interface_mac_address(self
):
2445 def test__process_vdu_params_vdu_cloud_init_missing(self
):
2448 def test__process_vdu_params_vdu_cloud_init_present(self
):
2451 def test__process_vdu_params_vdu_boot_data_drive(self
):
2454 def test__process_vdu_params_vdu_ssh_keys(self
):
2457 def test__process_vdu_params_vdu_ssh_access_required(self
):
2460 def test__process_vdu_params_vdu_virtual_storages(self
):
2463 def test__process_vdu_params(self
):