Code Coverage

Cobertura Coverage Report > NG-RO.osm_ng_ro.tests >

test_ns.py

Trend

File Coverage summary

NameClassesLinesConditionals
test_ns.py
100%
1/1
99%
1831/1843
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
test_ns.py
99%
1831/1843
N/A

Source

NG-RO/osm_ng_ro/tests/test_ns.py
1 #######################################################################################
2 # Copyright ETSI Contributors and Others.
3 #
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
7 #
8 #    http://www.apache.org/licenses/LICENSE-2.0
9 #
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
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #######################################################################################
17 1 from copy import deepcopy
18 1 import unittest
19 1 from unittest.mock import MagicMock, Mock, patch
20
21 1 from jinja2 import (
22     Environment,
23     select_autoescape,
24     StrictUndefined,
25     TemplateError,
26     TemplateNotFound,
27     UndefinedError,
28 )
29 1 from osm_ng_ro.ns import Ns, NsException
30
31
32 1 __author__ = "Eduardo Sousa"
33 1 __date__ = "$19-NOV-2021 00:00:00$"
34
35
36 # Variables used in Tests
37 1 vnfd_wth_persistent_storage = {
38     "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
39     "df": [
40         {
41             "id": "default-df",
42             "vdu-profile": [{"id": "several_volumes-VM", "min-number-of-instances": 1}],
43         }
44     ],
45     "id": "several_volumes-vnf",
46     "product-name": "several_volumes-vnf",
47     "vdu": [
48         {
49             "id": "several_volumes-VM",
50             "name": "several_volumes-VM",
51             "sw-image-desc": "ubuntu20.04",
52             "alternative-sw-image-desc": [
53                 "ubuntu20.04-aws",
54                 "ubuntu20.04-azure",
55             ],
56             "virtual-storage-desc": [
57                 "persistent-root-volume",
58                 "persistent-volume2",
59                 "ephemeral-volume",
60             ],
61         }
62     ],
63     "version": "1.0",
64     "virtual-storage-desc": [
65         {
66             "id": "persistent-volume2",
67             "type-of-storage": "persistent-storage:persistent-storage",
68             "size-of-storage": "10",
69         },
70         {
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"},
76             ],
77         },
78         {
79             "id": "ephemeral-volume",
80             "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
81             "size-of-storage": "1",
82         },
83     ],
84     "_admin": {
85         "storage": {
86             "fs": "mongo",
87             "path": "/app/storage/",
88         },
89         "type": "vnfd",
90     },
91 }
92 1 vim_volume_id = "ru937f49-3870-4169-b758-9732e1ff40f3"
93 1 task_by_target_record_id = {
94     "nsrs:th47f48-9870-4169-b758-9732e1ff40f3": {
95         "extra_dict": {"params": {"net_type": "SR-IOV"}}
96     }
97 }
98 1 interfaces_wthout_positions = [
99     {
100         "name": "vdu-eth1",
101         "ns-vld-id": "net1",
102     },
103     {
104         "name": "vdu-eth2",
105         "ns-vld-id": "net2",
106     },
107     {
108         "name": "vdu-eth3",
109         "ns-vld-id": "mgmtnet",
110     },
111 ]
112 1 interfaces_wth_all_positions = [
113     {
114         "name": "vdu-eth1",
115         "ns-vld-id": "net1",
116         "position": 2,
117     },
118     {
119         "name": "vdu-eth2",
120         "ns-vld-id": "net2",
121         "position": 0,
122     },
123     {
124         "name": "vdu-eth3",
125         "ns-vld-id": "mgmtnet",
126         "position": 1,
127     },
128 ]
129 1 target_vdu_wth_persistent_storage = {
130     "_id": "09a0baa7-b7cb-4924-bd63-9f04a1c23960",
131     "ns-flavor-id": "0",
132     "ns-image-id": "0",
133     "vdu-name": "several_volumes-VM",
134     "interfaces": [
135         {
136             "name": "vdu-eth0",
137             "ns-vld-id": "mgmtnet",
138         }
139     ],
140     "virtual-storages": [
141         {
142             "id": "persistent-volume2",
143             "size-of-storage": "10",
144             "type-of-storage": "persistent-storage:persistent-storage",
145         },
146         {
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"},
152             ],
153         },
154         {
155             "id": "ephemeral-volume",
156             "size-of-storage": "1",
157             "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
158         },
159     ],
160 }
161 1 db = MagicMock(name="database mock")
162 1 fs = MagicMock(name="database mock")
163 1 ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
164 1 vnf_preffix = "vnfrs:wh47f48-y870-4169-b758-5732e1ff40f5"
165 1 vnfr_id = "wh47f48-y870-4169-b758-5732e1ff40f5"
166 1 nsr_id = "th47f48-9870-4169-b758-9732e1ff40f3"
167 1 indata = {
168     "name": "sample_name",
169 }
170 1 expected_extra_dict = {
171     "depends_on": [
172         f"{ns_preffix}:image.0",
173         f"{ns_preffix}:flavor.0",
174     ],
175     "params": {
176         "affinity_group_list": [],
177         "availability_zone_index": None,
178         "availability_zone_list": None,
179         "cloud_config": None,
180         "description": "several_volumes-VM",
181         "disk_list": [],
182         "flavor_id": f"TASK-{ns_preffix}:flavor.0",
183         "image_id": f"TASK-{ns_preffix}:image.0",
184         "name": "sample_name-vnf-several-volumes-several_volumes-VM-0",
185         "net_list": [],
186         "start": True,
187     },
188 }
189
190 1 expected_extra_dict2 = {
191     "depends_on": [
192         f"{ns_preffix}:image.0",
193         f"{ns_preffix}:flavor.0",
194     ],
195     "params": {
196         "affinity_group_list": [],
197         "availability_zone_index": None,
198         "availability_zone_list": None,
199         "cloud_config": None,
200         "description": "without_volumes-VM",
201         "disk_list": [],
202         "flavor_id": f"TASK-{ns_preffix}:flavor.0",
203         "image_id": f"TASK-{ns_preffix}:image.0",
204         "name": "sample_name-vnf-several-volumes-without_volumes-VM-0",
205         "net_list": [],
206         "start": True,
207     },
208 }
209
210 1 tasks_by_target_record_id = {
211     "nsrs:th47f48-9870-4169-b758-9732e1ff40f3": {
212         "extra_dict": {
213             "params": {
214                 "net_type": "SR-IOV",
215             }
216         }
217     }
218 }
219 1 kwargs = {
220     "db": MagicMock(),
221     "vdu2cloud_init": {},
222     "vnfr": {
223         "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
224         "member-vnf-index-ref": "vnf-several-volumes",
225     },
226 }
227 1 vnfd_wthout_persistent_storage = {
228     "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
229     "df": [
230         {
231             "id": "default-df",
232             "vdu-profile": [{"id": "without_volumes-VM", "min-number-of-instances": 1}],
233         }
234     ],
235     "id": "without_volumes-vnf",
236     "product-name": "without_volumes-vnf",
237     "vdu": [
238         {
239             "id": "without_volumes-VM",
240             "name": "without_volumes-VM",
241             "sw-image-desc": "ubuntu20.04",
242             "alternative-sw-image-desc": [
243                 "ubuntu20.04-aws",
244                 "ubuntu20.04-azure",
245             ],
246             "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
247         }
248     ],
249     "version": "1.0",
250     "virtual-storage-desc": [
251         {"id": "root-volume", "size-of-storage": "10"},
252         {
253             "id": "ephemeral-volume",
254             "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
255             "size-of-storage": "1",
256         },
257     ],
258     "_admin": {
259         "storage": {
260             "fs": "mongo",
261             "path": "/app/storage/",
262         },
263         "type": "vnfd",
264     },
265 }
266
267 1 target_vdu_wthout_persistent_storage = {
268     "_id": "09a0baa7-b7cb-4924-bd63-9f04a1c23960",
269     "ns-flavor-id": "0",
270     "ns-image-id": "0",
271     "vdu-name": "without_volumes-VM",
272     "interfaces": [
273         {
274             "name": "vdu-eth0",
275             "ns-vld-id": "mgmtnet",
276         }
277     ],
278     "virtual-storages": [
279         {
280             "id": "root-volume",
281             "size-of-storage": "10",
282         },
283         {
284             "id": "ephemeral-volume",
285             "size-of-storage": "1",
286             "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
287         },
288     ],
289 }
290 1 cloud_init_content = """
291 disk_setup:
292     ephemeral0:
293         table_type: {{type}}
294         layout: True
295         overwrite: {{is_override}}
296 runcmd:
297      - [ ls, -l, / ]
298      - [ sh, -xc, "echo $(date) '{{command}}'" ]
299 """
300
301 1 user_data = """
302 disk_setup:
303     ephemeral0:
304         table_type: mbr
305         layout: True
306         overwrite: False
307 runcmd:
308      - [ ls, -l, / ]
309      - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
310 """
311 1 vdu_id = "bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
312 1 vnf_id = "665b4165-ce24-4320-bf19-b9a45bade49f"
313 1 target_vim = "vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
314 1 action_id = "bb937f49-3870-4169-b758-9732e1ff40f3"
315 1 nsr_id_2 = "993166fe-723e-4680-ac4b-b1af2541ae31"
316 1 target_record_1 = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
317 1 target_record_id = (
318     "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
319     "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
320 )
321 1 expected_result_vertical_scale = {
322     "target_id": target_vim,
323     "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
324     "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
325     "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
326     "status": "SCHEDULED",
327     "action": "EXEC",
328     "item": "verticalscale",
329     "target_record": target_record_1,
330     "target_record_id": target_record_id,
331     "params": {
332         "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
333         "flavor_dict": "flavor_dict",
334     },
335 }
336 1 vdu = {
337     "id": vdu_id,
338     "vim_info": {target_vim: {"interfaces": []}},
339 }
340 1 vnf = {"_id": vnf_id}
341 1 extra_dict_vertical_scale = {
342     "params": {
343         "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
344         "flavor_dict": "flavor_dict",
345     },
346 }
347 1 extra_dict_migrate = {
348     "params": {
349         "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
350         "migrate_host": "migrateToHost",
351     },
352 }
353 1 expected_result_migrate = {
354     "target_id": target_vim,
355     "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
356     "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
357     "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
358     "status": "SCHEDULED",
359     "action": "EXEC",
360     "item": "migrate",
361     "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
362     "target_record_id": target_record_id,
363     "params": {
364         "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
365         "migrate_host": "migrateToHost",
366     },
367 }
368 1 expected_result_rebuild_start_stop = {
369     "target_id": target_vim,
370     "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
371     "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
372     "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:0",
373     "status": "SCHEDULED",
374     "action": "EXEC",
375     "item": "update",
376     "target_record_id": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
377 }
378
379
380 1 class TestException(Exception):
381 1     pass
382
383
384 1 class CopyingMock(MagicMock):
385 1     def __call__(self, *args, **kwargs):
386 1         args = deepcopy(args)
387 1         kwargs = deepcopy(kwargs)
388 1         return super(CopyingMock, self).__call__(*args, **kwargs)
389
390
391 1 class TestNs(unittest.TestCase):
392 1     def setUp(self):
393 1         pass
394
395 1     def test__create_task_without_extra_dict(self):
396 1         expected_result = {
397             "target_id": "vim_openstack_1",
398             "action_id": "123456",
399             "nsr_id": "654321",
400             "task_id": "123456:1",
401             "status": "SCHEDULED",
402             "action": "CREATE",
403             "item": "test_item",
404             "target_record": "test_target_record",
405             "target_record_id": "test_target_record_id",
406         }
407 1         deployment_info = {
408             "action_id": "123456",
409             "nsr_id": "654321",
410             "task_index": 1,
411         }
412
413 1         task = Ns._create_task(
414             deployment_info=deployment_info,
415             target_id="vim_openstack_1",
416             item="test_item",
417             action="CREATE",
418             target_record="test_target_record",
419             target_record_id="test_target_record_id",
420         )
421
422 1         self.assertEqual(deployment_info.get("task_index"), 2)
423 1         self.assertDictEqual(task, expected_result)
424
425 1     def test__create_task(self):
426 1         expected_result = {
427             "target_id": "vim_openstack_1",
428             "action_id": "123456",
429             "nsr_id": "654321",
430             "task_id": "123456:1",
431             "status": "SCHEDULED",
432             "action": "CREATE",
433             "item": "test_item",
434             "target_record": "test_target_record",
435             "target_record_id": "test_target_record_id",
436             # values coming from extra_dict
437             "params": "test_params",
438             "find_params": "test_find_params",
439             "depends_on": "test_depends_on",
440         }
441 1         deployment_info = {
442             "action_id": "123456",
443             "nsr_id": "654321",
444             "task_index": 1,
445         }
446
447 1         task = Ns._create_task(
448             deployment_info=deployment_info,
449             target_id="vim_openstack_1",
450             item="test_item",
451             action="CREATE",
452             target_record="test_target_record",
453             target_record_id="test_target_record_id",
454             extra_dict={
455                 "params": "test_params",
456                 "find_params": "test_find_params",
457                 "depends_on": "test_depends_on",
458             },
459         )
460
461 1         self.assertEqual(deployment_info.get("task_index"), 2)
462 1         self.assertDictEqual(task, expected_result)
463
464 1     @patch("osm_ng_ro.ns.time")
465 1     def test__create_ro_task(self, mock_time: Mock):
466 1         now = 1637324838.994551
467 1         mock_time.return_value = now
468 1         task = {
469             "target_id": "vim_openstack_1",
470             "action_id": "123456",
471             "nsr_id": "654321",
472             "task_id": "123456:1",
473             "status": "SCHEDULED",
474             "action": "CREATE",
475             "item": "test_item",
476             "target_record": "test_target_record",
477             "target_record_id": "test_target_record_id",
478             # values coming from extra_dict
479             "params": "test_params",
480             "find_params": "test_find_params",
481             "depends_on": "test_depends_on",
482         }
483 1         expected_result = {
484             "_id": "123456:1",
485             "locked_by": None,
486             "locked_at": 0.0,
487             "target_id": "vim_openstack_1",
488             "vim_info": {
489                 "created": False,
490                 "created_items": None,
491                 "vim_id": None,
492                 "vim_name": None,
493                 "vim_status": None,
494                 "vim_details": None,
495                 "vim_message": None,
496                 "refresh_at": None,
497             },
498             "modified_at": now,
499             "created_at": now,
500             "to_check_at": now,
501             "tasks": [task],
502         }
503
504 1         ro_task = Ns._create_ro_task(
505             target_id="vim_openstack_1",
506             task=task,
507         )
508
509 1         self.assertDictEqual(ro_task, expected_result)
510
511 1     def test__process_image_params_with_empty_target_image(self):
512 1         expected_result = {
513             "find_params": {},
514         }
515 1         target_image = {}
516
517 1         result = Ns._process_image_params(
518             target_image=target_image,
519             indata=None,
520             vim_info=None,
521             target_record_id=None,
522         )
523
524 1         self.assertDictEqual(expected_result, result)
525
526 1     def test__process_image_params_with_wrong_target_image(self):
527 1         expected_result = {
528             "find_params": {},
529         }
530 1         target_image = {
531             "no_image": "to_see_here",
532         }
533
534 1         result = Ns._process_image_params(
535             target_image=target_image,
536             indata=None,
537             vim_info=None,
538             target_record_id=None,
539         )
540
541 1         self.assertDictEqual(expected_result, result)
542
543 1     def test__process_image_params_with_image(self):
544 1         expected_result = {
545             "find_params": {
546                 "filter_dict": {
547                     "name": "cirros",
548                 },
549             },
550         }
551 1         target_image = {
552             "image": "cirros",
553         }
554
555 1         result = Ns._process_image_params(
556             target_image=target_image,
557             indata=None,
558             vim_info=None,
559             target_record_id=None,
560         )
561
562 1         self.assertDictEqual(expected_result, result)
563
564 1     def test__process_image_params_with_vim_image_id(self):
565 1         expected_result = {
566             "find_params": {
567                 "filter_dict": {
568                     "id": "123456",
569                 },
570             },
571         }
572 1         target_image = {
573             "vim_image_id": "123456",
574         }
575
576 1         result = Ns._process_image_params(
577             target_image=target_image,
578             indata=None,
579             vim_info=None,
580             target_record_id=None,
581         )
582
583 1         self.assertDictEqual(expected_result, result)
584
585 1     def test__process_image_params_with_image_checksum(self):
586 1         expected_result = {
587             "find_params": {
588                 "filter_dict": {
589                     "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
590                 },
591             },
592         }
593 1         target_image = {
594             "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
595         }
596
597 1         result = Ns._process_image_params(
598             target_image=target_image,
599             indata=None,
600             vim_info=None,
601             target_record_id=None,
602         )
603
604 1         self.assertDictEqual(expected_result, result)
605
606 1     def test__get_resource_allocation_params_with_empty_target_image(self):
607 1         expected_result = {}
608 1         quota_descriptor = {}
609
610 1         result = Ns._get_resource_allocation_params(
611             quota_descriptor=quota_descriptor,
612         )
613
614 1         self.assertDictEqual(expected_result, result)
615
616 1     def test__get_resource_allocation_params_with_wrong_target_image(self):
617 1         expected_result = {}
618 1         quota_descriptor = {
619             "no_quota": "present_here",
620         }
621
622 1         result = Ns._get_resource_allocation_params(
623             quota_descriptor=quota_descriptor,
624         )
625
626 1         self.assertDictEqual(expected_result, result)
627
628 1     def test__get_resource_allocation_params_with_limit(self):
629 1         expected_result = {
630             "limit": 10,
631         }
632 1         quota_descriptor = {
633             "limit": "10",
634         }
635
636 1         result = Ns._get_resource_allocation_params(
637             quota_descriptor=quota_descriptor,
638         )
639
640 1         self.assertDictEqual(expected_result, result)
641
642 1     def test__get_resource_allocation_params_with_reserve(self):
643 1         expected_result = {
644             "reserve": 20,
645         }
646 1         quota_descriptor = {
647             "reserve": "20",
648         }
649
650 1         result = Ns._get_resource_allocation_params(
651             quota_descriptor=quota_descriptor,
652         )
653
654 1         self.assertDictEqual(expected_result, result)
655
656 1     def test__get_resource_allocation_params_with_shares(self):
657 1         expected_result = {
658             "shares": 30,
659         }
660 1         quota_descriptor = {
661             "shares": "30",
662         }
663
664 1         result = Ns._get_resource_allocation_params(
665             quota_descriptor=quota_descriptor,
666         )
667
668 1         self.assertDictEqual(expected_result, result)
669
670 1     def test__get_resource_allocation_params(self):
671 1         expected_result = {
672             "limit": 10,
673             "reserve": 20,
674             "shares": 30,
675         }
676 1         quota_descriptor = {
677             "limit": "10",
678             "reserve": "20",
679             "shares": "30",
680         }
681
682 1         result = Ns._get_resource_allocation_params(
683             quota_descriptor=quota_descriptor,
684         )
685
686 1         self.assertDictEqual(expected_result, result)
687
688 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
689 1     def test__process_guest_epa_quota_params_with_empty_quota_epa_cpu(
690         self,
691         resource_allocation,
692     ):
693 1         expected_result = {}
694 1         guest_epa_quota = {}
695 1         epa_vcpu_set = True
696
697 1         result = Ns._process_guest_epa_quota_params(
698             guest_epa_quota=guest_epa_quota,
699             epa_vcpu_set=epa_vcpu_set,
700         )
701
702 1         self.assertDictEqual(expected_result, result)
703 1         self.assertFalse(resource_allocation.called)
704
705 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
706 1     def test__process_guest_epa_quota_params_with_empty_quota_false_epa_cpu(
707         self,
708         resource_allocation,
709     ):
710 1         expected_result = {}
711 1         guest_epa_quota = {}
712 1         epa_vcpu_set = False
713
714 1         result = Ns._process_guest_epa_quota_params(
715             guest_epa_quota=guest_epa_quota,
716             epa_vcpu_set=epa_vcpu_set,
717         )
718
719 1         self.assertDictEqual(expected_result, result)
720 1         self.assertFalse(resource_allocation.called)
721
722 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
723 1     def test__process_guest_epa_quota_params_with_wrong_quota_epa_cpu(
724         self,
725         resource_allocation,
726     ):
727 1         expected_result = {}
728 1         guest_epa_quota = {
729             "no-quota": "nothing",
730         }
731 1         epa_vcpu_set = True
732
733 1         result = Ns._process_guest_epa_quota_params(
734             guest_epa_quota=guest_epa_quota,
735             epa_vcpu_set=epa_vcpu_set,
736         )
737
738 1         self.assertDictEqual(expected_result, result)
739 1         self.assertFalse(resource_allocation.called)
740
741 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
742 1     def test__process_guest_epa_quota_params_with_wrong_quota_false_epa_cpu(
743         self,
744         resource_allocation,
745     ):
746 1         expected_result = {}
747 1         guest_epa_quota = {
748             "no-quota": "nothing",
749         }
750 1         epa_vcpu_set = False
751
752 1         result = Ns._process_guest_epa_quota_params(
753             guest_epa_quota=guest_epa_quota,
754             epa_vcpu_set=epa_vcpu_set,
755         )
756
757 1         self.assertDictEqual(expected_result, result)
758 1         self.assertFalse(resource_allocation.called)
759
760 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
761 1     def test__process_guest_epa_quota_params_with_cpu_quota_epa_cpu(
762         self,
763         resource_allocation,
764     ):
765 1         expected_result = {}
766 1         guest_epa_quota = {
767             "cpu-quota": {
768                 "limit": "10",
769                 "reserve": "20",
770                 "shares": "30",
771             },
772         }
773 1         epa_vcpu_set = True
774
775 1         result = Ns._process_guest_epa_quota_params(
776             guest_epa_quota=guest_epa_quota,
777             epa_vcpu_set=epa_vcpu_set,
778         )
779
780 1         self.assertDictEqual(expected_result, result)
781 1         self.assertFalse(resource_allocation.called)
782
783 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
784 1     def test__process_guest_epa_quota_params_with_cpu_quota_false_epa_cpu(
785         self,
786         resource_allocation,
787     ):
788 1         expected_result = {
789             "cpu-quota": {
790                 "limit": 10,
791                 "reserve": 20,
792                 "shares": 30,
793             },
794         }
795 1         guest_epa_quota = {
796             "cpu-quota": {
797                 "limit": "10",
798                 "reserve": "20",
799                 "shares": "30",
800             },
801         }
802 1         epa_vcpu_set = False
803
804 1         resource_allocation_param = {
805             "limit": "10",
806             "reserve": "20",
807             "shares": "30",
808         }
809 1         resource_allocation.return_value = {
810             "limit": 10,
811             "reserve": 20,
812             "shares": 30,
813         }
814
815 1         result = Ns._process_guest_epa_quota_params(
816             guest_epa_quota=guest_epa_quota,
817             epa_vcpu_set=epa_vcpu_set,
818         )
819
820 1         resource_allocation.assert_called_once_with(resource_allocation_param)
821 1         self.assertDictEqual(expected_result, result)
822
823 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
824 1     def test__process_guest_epa_quota_params_with_mem_quota_epa_cpu(
825         self,
826         resource_allocation,
827     ):
828 1         expected_result = {
829             "mem-quota": {
830                 "limit": 10,
831                 "reserve": 20,
832                 "shares": 30,
833             },
834         }
835 1         guest_epa_quota = {
836             "mem-quota": {
837                 "limit": "10",
838                 "reserve": "20",
839                 "shares": "30",
840             },
841         }
842 1         epa_vcpu_set = True
843
844 1         resource_allocation_param = {
845             "limit": "10",
846             "reserve": "20",
847             "shares": "30",
848         }
849 1         resource_allocation.return_value = {
850             "limit": 10,
851             "reserve": 20,
852             "shares": 30,
853         }
854
855 1         result = Ns._process_guest_epa_quota_params(
856             guest_epa_quota=guest_epa_quota,
857             epa_vcpu_set=epa_vcpu_set,
858         )
859
860 1         resource_allocation.assert_called_once_with(resource_allocation_param)
861 1         self.assertDictEqual(expected_result, result)
862
863 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
864 1     def test__process_guest_epa_quota_params_with_mem_quota_false_epa_cpu(
865         self,
866         resource_allocation,
867     ):
868 1         expected_result = {
869             "mem-quota": {
870                 "limit": 10,
871                 "reserve": 20,
872                 "shares": 30,
873             },
874         }
875 1         guest_epa_quota = {
876             "mem-quota": {
877                 "limit": "10",
878                 "reserve": "20",
879                 "shares": "30",
880             },
881         }
882 1         epa_vcpu_set = False
883
884 1         resource_allocation_param = {
885             "limit": "10",
886             "reserve": "20",
887             "shares": "30",
888         }
889 1         resource_allocation.return_value = {
890             "limit": 10,
891             "reserve": 20,
892             "shares": 30,
893         }
894
895 1         result = Ns._process_guest_epa_quota_params(
896             guest_epa_quota=guest_epa_quota,
897             epa_vcpu_set=epa_vcpu_set,
898         )
899
900 1         resource_allocation.assert_called_once_with(resource_allocation_param)
901 1         self.assertDictEqual(expected_result, result)
902
903 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
904 1     def test__process_guest_epa_quota_params_with_disk_io_quota_epa_cpu(
905         self,
906         resource_allocation,
907     ):
908 1         expected_result = {
909             "disk-io-quota": {
910                 "limit": 10,
911                 "reserve": 20,
912                 "shares": 30,
913             },
914         }
915 1         guest_epa_quota = {
916             "disk-io-quota": {
917                 "limit": "10",
918                 "reserve": "20",
919                 "shares": "30",
920             },
921         }
922 1         epa_vcpu_set = True
923
924 1         resource_allocation_param = {
925             "limit": "10",
926             "reserve": "20",
927             "shares": "30",
928         }
929 1         resource_allocation.return_value = {
930             "limit": 10,
931             "reserve": 20,
932             "shares": 30,
933         }
934
935 1         result = Ns._process_guest_epa_quota_params(
936             guest_epa_quota=guest_epa_quota,
937             epa_vcpu_set=epa_vcpu_set,
938         )
939
940 1         resource_allocation.assert_called_once_with(resource_allocation_param)
941 1         self.assertDictEqual(expected_result, result)
942
943 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
944 1     def test__process_guest_epa_quota_params_with_disk_io_quota_false_epa_cpu(
945         self,
946         resource_allocation,
947     ):
948 1         expected_result = {
949             "disk-io-quota": {
950                 "limit": 10,
951                 "reserve": 20,
952                 "shares": 30,
953             },
954         }
955 1         guest_epa_quota = {
956             "disk-io-quota": {
957                 "limit": "10",
958                 "reserve": "20",
959                 "shares": "30",
960             },
961         }
962 1         epa_vcpu_set = False
963
964 1         resource_allocation_param = {
965             "limit": "10",
966             "reserve": "20",
967             "shares": "30",
968         }
969 1         resource_allocation.return_value = {
970             "limit": 10,
971             "reserve": 20,
972             "shares": 30,
973         }
974
975 1         result = Ns._process_guest_epa_quota_params(
976             guest_epa_quota=guest_epa_quota,
977             epa_vcpu_set=epa_vcpu_set,
978         )
979
980 1         resource_allocation.assert_called_once_with(resource_allocation_param)
981 1         self.assertDictEqual(expected_result, result)
982
983 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
984 1     def test__process_guest_epa_quota_params_with_vif_quota_epa_cpu(
985         self,
986         resource_allocation,
987     ):
988 1         expected_result = {
989             "vif-quota": {
990                 "limit": 10,
991                 "reserve": 20,
992                 "shares": 30,
993             },
994         }
995 1         guest_epa_quota = {
996             "vif-quota": {
997                 "limit": "10",
998                 "reserve": "20",
999                 "shares": "30",
1000             },
1001         }
1002 1         epa_vcpu_set = True
1003
1004 1         resource_allocation_param = {
1005             "limit": "10",
1006             "reserve": "20",
1007             "shares": "30",
1008         }
1009 1         resource_allocation.return_value = {
1010             "limit": 10,
1011             "reserve": 20,
1012             "shares": 30,
1013         }
1014
1015 1         result = Ns._process_guest_epa_quota_params(
1016             guest_epa_quota=guest_epa_quota,
1017             epa_vcpu_set=epa_vcpu_set,
1018         )
1019
1020 1         resource_allocation.assert_called_once_with(resource_allocation_param)
1021 1         self.assertDictEqual(expected_result, result)
1022
1023 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1024 1     def test__process_guest_epa_quota_params_with_vif_quota_false_epa_cpu(
1025         self,
1026         resource_allocation,
1027     ):
1028 1         expected_result = {
1029             "vif-quota": {
1030                 "limit": 10,
1031                 "reserve": 20,
1032                 "shares": 30,
1033             },
1034         }
1035 1         guest_epa_quota = {
1036             "vif-quota": {
1037                 "limit": "10",
1038                 "reserve": "20",
1039                 "shares": "30",
1040             },
1041         }
1042 1         epa_vcpu_set = False
1043
1044 1         resource_allocation_param = {
1045             "limit": "10",
1046             "reserve": "20",
1047             "shares": "30",
1048         }
1049 1         resource_allocation.return_value = {
1050             "limit": 10,
1051             "reserve": 20,
1052             "shares": 30,
1053         }
1054
1055 1         result = Ns._process_guest_epa_quota_params(
1056             guest_epa_quota=guest_epa_quota,
1057             epa_vcpu_set=epa_vcpu_set,
1058         )
1059
1060 1         resource_allocation.assert_called_once_with(resource_allocation_param)
1061 1         self.assertDictEqual(expected_result, result)
1062
1063 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1064 1     def test__process_guest_epa_quota_params_with_quota_epa_cpu(
1065         self,
1066         resource_allocation,
1067     ):
1068 1         expected_result = {
1069             "mem-quota": {
1070                 "limit": 10,
1071                 "reserve": 20,
1072                 "shares": 30,
1073             },
1074             "disk-io-quota": {
1075                 "limit": 10,
1076                 "reserve": 20,
1077                 "shares": 30,
1078             },
1079             "vif-quota": {
1080                 "limit": 10,
1081                 "reserve": 20,
1082                 "shares": 30,
1083             },
1084         }
1085 1         guest_epa_quota = {
1086             "cpu-quota": {
1087                 "limit": "10",
1088                 "reserve": "20",
1089                 "shares": "30",
1090             },
1091             "mem-quota": {
1092                 "limit": "10",
1093                 "reserve": "20",
1094                 "shares": "30",
1095             },
1096             "disk-io-quota": {
1097                 "limit": "10",
1098                 "reserve": "20",
1099                 "shares": "30",
1100             },
1101             "vif-quota": {
1102                 "limit": "10",
1103                 "reserve": "20",
1104                 "shares": "30",
1105             },
1106         }
1107 1         epa_vcpu_set = True
1108
1109 1         resource_allocation.return_value = {
1110             "limit": 10,
1111             "reserve": 20,
1112             "shares": 30,
1113         }
1114
1115 1         result = Ns._process_guest_epa_quota_params(
1116             guest_epa_quota=guest_epa_quota,
1117             epa_vcpu_set=epa_vcpu_set,
1118         )
1119
1120 1         self.assertTrue(resource_allocation.called)
1121 1         self.assertDictEqual(expected_result, result)
1122
1123 1     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
1124 1     def test__process_guest_epa_quota_params_with_quota_epa_cpu_no_set(
1125         self,
1126         resource_allocation,
1127     ):
1128 1         expected_result = {
1129             "cpu-quota": {
1130                 "limit": 10,
1131                 "reserve": 20,
1132                 "shares": 30,
1133             },
1134             "mem-quota": {
1135                 "limit": 10,
1136                 "reserve": 20,
1137                 "shares": 30,
1138             },
1139             "disk-io-quota": {
1140                 "limit": 10,
1141                 "reserve": 20,
1142                 "shares": 30,
1143             },
1144             "vif-quota": {
1145                 "limit": 10,
1146                 "reserve": 20,
1147                 "shares": 30,
1148             },
1149         }
1150 1         guest_epa_quota = {
1151             "cpu-quota": {
1152                 "limit": "10",
1153                 "reserve": "20",
1154                 "shares": "30",
1155             },
1156             "mem-quota": {
1157                 "limit": "10",
1158                 "reserve": "20",
1159                 "shares": "30",
1160             },
1161             "disk-io-quota": {
1162                 "limit": "10",
1163                 "reserve": "20",
1164                 "shares": "30",
1165             },
1166             "vif-quota": {
1167                 "limit": "10",
1168                 "reserve": "20",
1169                 "shares": "30",
1170             },
1171         }
1172 1         epa_vcpu_set = False
1173
1174 1         resource_allocation.return_value = {
1175             "limit": 10,
1176             "reserve": 20,
1177             "shares": 30,
1178         }
1179
1180 1         result = Ns._process_guest_epa_quota_params(
1181             guest_epa_quota=guest_epa_quota,
1182             epa_vcpu_set=epa_vcpu_set,
1183         )
1184
1185 1         self.assertTrue(resource_allocation.called)
1186 1         self.assertDictEqual(expected_result, result)
1187
1188 1     def test__process_guest_epa_numa_params_with_empty_numa_params(self):
1189 1         expected_numa_result = []
1190 1         expected_epa_vcpu_set_result = False
1191 1         guest_epa_quota = {}
1192
1193 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1194             guest_epa_quota=guest_epa_quota,
1195         )
1196 1         self.assertEqual(expected_numa_result, numa_result)
1197 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1198
1199 1     def test__process_guest_epa_numa_params_with_wrong_numa_params(self):
1200 1         expected_numa_result = []
1201 1         expected_epa_vcpu_set_result = False
1202 1         guest_epa_quota = {"no_nume": "here"}
1203
1204 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1205             guest_epa_quota=guest_epa_quota,
1206         )
1207
1208 1         self.assertEqual(expected_numa_result, numa_result)
1209 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1210
1211 1     def test__process_guest_epa_numa_params_with_numa_node_policy(self):
1212 1         expected_numa_result = []
1213 1         expected_epa_vcpu_set_result = False
1214 1         guest_epa_quota = {"numa-node-policy": {}}
1215
1216 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1217             guest_epa_quota=guest_epa_quota,
1218         )
1219
1220 1         self.assertEqual(expected_numa_result, numa_result)
1221 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1222
1223 1     def test__process_guest_epa_numa_params_with_no_node(self):
1224 1         expected_numa_result = []
1225 1         expected_epa_vcpu_set_result = False
1226 1         guest_epa_quota = {
1227             "numa-node-policy": {
1228                 "node": [],
1229             },
1230         }
1231
1232 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1233             guest_epa_quota=guest_epa_quota,
1234         )
1235
1236 1         self.assertEqual(expected_numa_result, numa_result)
1237 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1238
1239 1     def test__process_guest_epa_numa_params_with_1_node_num_cores(self):
1240 1         expected_numa_result = [{"cores": 3}]
1241 1         expected_epa_vcpu_set_result = True
1242 1         guest_epa_quota = {
1243             "numa-node-policy": {
1244                 "node": [
1245                     {
1246                         "num-cores": 3,
1247                     },
1248                 ],
1249             },
1250         }
1251
1252 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1253             guest_epa_quota=guest_epa_quota,
1254         )
1255
1256 1         self.assertEqual(expected_numa_result, numa_result)
1257 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1258
1259 1     def test__process_guest_epa_numa_params_with_1_node_paired_threads(self):
1260 1         expected_numa_result = [{"paired_threads": 3}]
1261 1         expected_epa_vcpu_set_result = True
1262 1         guest_epa_quota = {
1263             "numa-node-policy": {
1264                 "node": [
1265                     {
1266                         "paired-threads": {"num-paired-threads": "3"},
1267                     },
1268                 ],
1269             },
1270         }
1271
1272 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1273             guest_epa_quota=guest_epa_quota,
1274         )
1275
1276 1         self.assertEqual(expected_numa_result, numa_result)
1277 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1278
1279 1     def test__process_guest_epa_numa_params_with_1_node_paired_threads_ids(self):
1280 1         expected_numa_result = [
1281             {
1282                 "paired-threads-id": [("0", "1"), ("4", "5")],
1283             }
1284         ]
1285 1         expected_epa_vcpu_set_result = False
1286 1         guest_epa_quota = {
1287             "numa-node-policy": {
1288                 "node": [
1289                     {
1290                         "paired-threads": {
1291                             "paired-thread-ids": [
1292                                 {
1293                                     "thread-a": 0,
1294                                     "thread-b": 1,
1295                                 },
1296                                 {
1297                                     "thread-a": 4,
1298                                     "thread-b": 5,
1299                                 },
1300                             ],
1301                         },
1302                     },
1303                 ],
1304             },
1305         }
1306
1307 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1308             guest_epa_quota=guest_epa_quota,
1309         )
1310
1311 1         self.assertEqual(expected_numa_result, numa_result)
1312 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1313
1314 1     def test__process_guest_epa_numa_params_with_1_node_num_threads(self):
1315 1         expected_numa_result = [{"threads": 3}]
1316 1         expected_epa_vcpu_set_result = True
1317 1         guest_epa_quota = {
1318             "numa-node-policy": {
1319                 "node": [
1320                     {
1321                         "num-threads": "3",
1322                     },
1323                 ],
1324             },
1325         }
1326
1327 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1328             guest_epa_quota=guest_epa_quota,
1329         )
1330
1331 1         self.assertEqual(expected_numa_result, numa_result)
1332 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1333
1334 1     def test__process_guest_epa_numa_params_with_1_node_memory_mb(self):
1335 1         expected_numa_result = [{"memory": 2}]
1336 1         expected_epa_vcpu_set_result = False
1337 1         guest_epa_quota = {
1338             "numa-node-policy": {
1339                 "node": [
1340                     {
1341                         "memory-mb": 2048,
1342                     },
1343                 ],
1344             },
1345         }
1346
1347 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1348             guest_epa_quota=guest_epa_quota,
1349         )
1350
1351 1         self.assertEqual(expected_numa_result, numa_result)
1352 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1353
1354 1     def test__process_guest_epa_numa_params_with_1_node_vcpu(self):
1355 1         expected_numa_result = [
1356             {
1357                 "id": 0,
1358                 "vcpu": [0, 1],
1359             }
1360         ]
1361 1         expected_epa_vcpu_set_result = False
1362 1         guest_epa_quota = {
1363             "numa-node-policy": {
1364                 "node": [{"id": "0", "vcpu": [{"id": "0"}, {"id": "1"}]}],
1365             },
1366         }
1367
1368 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1369             guest_epa_quota=guest_epa_quota,
1370         )
1371
1372 1         self.assertEqual(expected_numa_result, numa_result)
1373 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1374
1375 1     def test__process_guest_epa_numa_params_with_2_node_vcpu(self):
1376 1         expected_numa_result = [
1377             {
1378                 "id": 0,
1379                 "vcpu": [0, 1],
1380             },
1381             {
1382                 "id": 1,
1383                 "vcpu": [2, 3],
1384             },
1385         ]
1386
1387 1         expected_epa_vcpu_set_result = False
1388 1         guest_epa_quota = {
1389             "numa-node-policy": {
1390                 "node": [
1391                     {"id": "0", "vcpu": [{"id": "0"}, {"id": "1"}]},
1392                     {"id": "1", "vcpu": [{"id": "2"}, {"id": "3"}]},
1393                 ],
1394             },
1395         }
1396
1397 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1398             guest_epa_quota=guest_epa_quota,
1399         )
1400
1401 1         self.assertEqual(expected_numa_result, numa_result)
1402 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1403
1404 1     def test__process_guest_epa_numa_params_with_1_node(self):
1405 1         expected_numa_result = [
1406             {
1407                 # "id": 0,
1408                 # "vcpu": [0, 1],
1409                 "cores": 3,
1410                 "paired_threads": 3,
1411                 "paired-threads-id": [("0", "1"), ("4", "5")],
1412                 "threads": 3,
1413                 "memory": 2,
1414             }
1415         ]
1416 1         expected_epa_vcpu_set_result = True
1417 1         guest_epa_quota = {
1418             "numa-node-policy": {
1419                 "node": [
1420                     {
1421                         "num-cores": 3,
1422                         "paired-threads": {
1423                             "num-paired-threads": "3",
1424                             "paired-thread-ids": [
1425                                 {
1426                                     "thread-a": 0,
1427                                     "thread-b": 1,
1428                                 },
1429                                 {
1430                                     "thread-a": 4,
1431                                     "thread-b": 5,
1432                                 },
1433                             ],
1434                         },
1435                         "num-threads": "3",
1436                         "memory-mb": 2048,
1437                     },
1438                 ],
1439             },
1440         }
1441
1442 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1443             guest_epa_quota=guest_epa_quota,
1444         )
1445
1446 1         self.assertEqual(expected_numa_result, numa_result)
1447 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1448
1449 1     def test__process_guest_epa_numa_params_with_2_nodes(self):
1450 1         expected_numa_result = [
1451             {
1452                 "cores": 3,
1453                 "paired_threads": 3,
1454                 "paired-threads-id": [("0", "1"), ("4", "5")],
1455                 "threads": 3,
1456                 "memory": 2,
1457             },
1458             {
1459                 "cores": 7,
1460                 "paired_threads": 7,
1461                 "paired-threads-id": [("2", "3"), ("5", "6")],
1462                 "threads": 4,
1463                 "memory": 4,
1464             },
1465         ]
1466 1         expected_epa_vcpu_set_result = True
1467 1         guest_epa_quota = {
1468             "numa-node-policy": {
1469                 "node": [
1470                     {
1471                         "num-cores": 3,
1472                         "paired-threads": {
1473                             "num-paired-threads": "3",
1474                             "paired-thread-ids": [
1475                                 {
1476                                     "thread-a": 0,
1477                                     "thread-b": 1,
1478                                 },
1479                                 {
1480                                     "thread-a": 4,
1481                                     "thread-b": 5,
1482                                 },
1483                             ],
1484                         },
1485                         "num-threads": "3",
1486                         "memory-mb": 2048,
1487                     },
1488                     {
1489                         "num-cores": 7,
1490                         "paired-threads": {
1491                             "num-paired-threads": "7",
1492                             "paired-thread-ids": [
1493                                 {
1494                                     "thread-a": 2,
1495                                     "thread-b": 3,
1496                                 },
1497                                 {
1498                                     "thread-a": 5,
1499                                     "thread-b": 6,
1500                                 },
1501                             ],
1502                         },
1503                         "num-threads": "4",
1504                         "memory-mb": 4096,
1505                     },
1506                 ],
1507             },
1508         }
1509
1510 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1511             guest_epa_quota=guest_epa_quota,
1512         )
1513
1514 1         self.assertEqual(expected_numa_result, numa_result)
1515 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1516
1517 1     def test__process_guest_epa_cpu_pinning_params_with_empty_params(self):
1518 1         expected_numa_result = {}
1519 1         expected_epa_vcpu_set_result = False
1520 1         guest_epa_quota = {}
1521 1         vcpu_count = 0
1522 1         epa_vcpu_set = False
1523
1524 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1525             guest_epa_quota=guest_epa_quota,
1526             vcpu_count=vcpu_count,
1527             epa_vcpu_set=epa_vcpu_set,
1528         )
1529
1530 1         self.assertDictEqual(expected_numa_result, numa_result)
1531 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1532
1533 1     def test__process_guest_epa_cpu_pinning_params_with_wrong_params(self):
1534 1         expected_numa_result = {}
1535 1         expected_epa_vcpu_set_result = False
1536 1         guest_epa_quota = {
1537             "no-cpu-pinning-policy": "here",
1538         }
1539 1         vcpu_count = 0
1540 1         epa_vcpu_set = False
1541
1542 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1543             guest_epa_quota=guest_epa_quota,
1544             vcpu_count=vcpu_count,
1545             epa_vcpu_set=epa_vcpu_set,
1546         )
1547
1548 1         self.assertDictEqual(expected_numa_result, numa_result)
1549 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1550
1551 1     def test__process_guest_epa_cpu_pinning_params_with_epa_vcpu_set(self):
1552 1         expected_numa_result = {}
1553 1         expected_epa_vcpu_set_result = True
1554 1         guest_epa_quota = {
1555             "cpu-pinning-policy": "DEDICATED",
1556         }
1557 1         vcpu_count = 0
1558 1         epa_vcpu_set = True
1559
1560 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1561             guest_epa_quota=guest_epa_quota,
1562             vcpu_count=vcpu_count,
1563             epa_vcpu_set=epa_vcpu_set,
1564         )
1565
1566 1         self.assertDictEqual(expected_numa_result, numa_result)
1567 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1568
1569 1     def test__process_guest_epa_cpu_pinning_params_with_policy_prefer(self):
1570 1         expected_numa_result = {"threads": 3}
1571 1         expected_epa_vcpu_set_result = True
1572 1         guest_epa_quota = {
1573             "cpu-pinning-policy": "DEDICATED",
1574             "cpu-thread-pinning-policy": "PREFER",
1575         }
1576 1         vcpu_count = 3
1577 1         epa_vcpu_set = False
1578
1579 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1580             guest_epa_quota=guest_epa_quota,
1581             vcpu_count=vcpu_count,
1582             epa_vcpu_set=epa_vcpu_set,
1583         )
1584
1585 1         self.assertDictEqual(expected_numa_result, numa_result)
1586 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1587
1588 1     def test__process_guest_epa_cpu_pinning_params_with_policy_isolate(self):
1589 1         expected_numa_result = {"cores": 3}
1590 1         expected_epa_vcpu_set_result = True
1591 1         guest_epa_quota = {
1592             "cpu-pinning-policy": "DEDICATED",
1593             "cpu-thread-pinning-policy": "ISOLATE",
1594         }
1595 1         vcpu_count = 3
1596 1         epa_vcpu_set = False
1597
1598 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1599             guest_epa_quota=guest_epa_quota,
1600             vcpu_count=vcpu_count,
1601             epa_vcpu_set=epa_vcpu_set,
1602         )
1603
1604 1         self.assertDictEqual(expected_numa_result, numa_result)
1605 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1606
1607 1     def test__process_guest_epa_cpu_pinning_params_with_policy_require(self):
1608 1         expected_numa_result = {"threads": 3}
1609 1         expected_epa_vcpu_set_result = True
1610 1         guest_epa_quota = {
1611             "cpu-pinning-policy": "DEDICATED",
1612             "cpu-thread-pinning-policy": "REQUIRE",
1613         }
1614 1         vcpu_count = 3
1615 1         epa_vcpu_set = False
1616
1617 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1618             guest_epa_quota=guest_epa_quota,
1619             vcpu_count=vcpu_count,
1620             epa_vcpu_set=epa_vcpu_set,
1621         )
1622
1623 1         self.assertDictEqual(expected_numa_result, numa_result)
1624 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1625
1626 1     def test__process_guest_epa_cpu_pinning_params(self):
1627 1         expected_numa_result = {"threads": 3}
1628 1         expected_epa_vcpu_set_result = True
1629 1         guest_epa_quota = {
1630             "cpu-pinning-policy": "DEDICATED",
1631         }
1632 1         vcpu_count = 3
1633 1         epa_vcpu_set = False
1634
1635 1         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1636             guest_epa_quota=guest_epa_quota,
1637             vcpu_count=vcpu_count,
1638             epa_vcpu_set=epa_vcpu_set,
1639         )
1640
1641 1         self.assertDictEqual(expected_numa_result, numa_result)
1642 1         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1643
1644 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1645 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1646 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1647 1     def test__process_guest_epa_params_with_empty_params(
1648         self,
1649         guest_epa_numa_params,
1650         guest_epa_cpu_pinning_params,
1651         guest_epa_quota_params,
1652     ):
1653 1         expected_result = {}
1654 1         target_flavor = {}
1655
1656 1         result = Ns._process_epa_params(
1657             target_flavor=target_flavor,
1658         )
1659
1660 1         self.assertDictEqual(expected_result, result)
1661 1         self.assertFalse(guest_epa_numa_params.called)
1662 1         self.assertFalse(guest_epa_cpu_pinning_params.called)
1663 1         self.assertFalse(guest_epa_quota_params.called)
1664
1665 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1666 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1667 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1668 1     def test__process_guest_epa_params_with_wrong_params(
1669         self,
1670         guest_epa_numa_params,
1671         guest_epa_cpu_pinning_params,
1672         guest_epa_quota_params,
1673     ):
1674 1         expected_result = {}
1675 1         target_flavor = {
1676             "no-guest-epa": "here",
1677         }
1678
1679 1         result = Ns._process_epa_params(
1680             target_flavor=target_flavor,
1681         )
1682
1683 1         self.assertDictEqual(expected_result, result)
1684 1         self.assertFalse(guest_epa_numa_params.called)
1685 1         self.assertFalse(guest_epa_cpu_pinning_params.called)
1686 1         self.assertFalse(guest_epa_quota_params.called)
1687
1688 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1689 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1690 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1691 1     def test__process_guest_epa_params(
1692         self,
1693         guest_epa_numa_params,
1694         guest_epa_cpu_pinning_params,
1695         guest_epa_quota_params,
1696     ):
1697 1         expected_result = {
1698             "mem-policy": "STRICT",
1699         }
1700 1         target_flavor = {
1701             "guest-epa": {
1702                 "vcpu-count": 1,
1703                 "numa-node-policy": {
1704                     "mem-policy": "STRICT",
1705                 },
1706             },
1707         }
1708
1709 1         guest_epa_numa_params.return_value = ({}, False)
1710 1         guest_epa_cpu_pinning_params.return_value = ({}, False)
1711 1         guest_epa_quota_params.return_value = {}
1712
1713 1         result = Ns._process_epa_params(
1714             target_flavor=target_flavor,
1715         )
1716
1717 1         self.assertDictEqual(expected_result, result)
1718 1         self.assertTrue(guest_epa_numa_params.called)
1719 1         self.assertTrue(guest_epa_cpu_pinning_params.called)
1720 1         self.assertTrue(guest_epa_quota_params.called)
1721
1722 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1723 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1724 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1725 1     def test__process_guest_epa_params_with_mempage_size(
1726         self,
1727         guest_epa_numa_params,
1728         guest_epa_cpu_pinning_params,
1729         guest_epa_quota_params,
1730     ):
1731 1         expected_result = {
1732             "mempage-size": "1G",
1733             "mem-policy": "STRICT",
1734         }
1735 1         target_flavor = {
1736             "guest-epa": {
1737                 "vcpu-count": 1,
1738                 "mempage-size": "1G",
1739                 "numa-node-policy": {
1740                     "mem-policy": "STRICT",
1741                 },
1742             },
1743         }
1744
1745 1         guest_epa_numa_params.return_value = ({}, False)
1746 1         guest_epa_cpu_pinning_params.return_value = ({}, False)
1747 1         guest_epa_quota_params.return_value = {}
1748
1749 1         result = Ns._process_epa_params(
1750             target_flavor=target_flavor,
1751         )
1752
1753 1         self.assertDictEqual(expected_result, result)
1754 1         self.assertTrue(guest_epa_numa_params.called)
1755 1         self.assertTrue(guest_epa_cpu_pinning_params.called)
1756 1         self.assertTrue(guest_epa_quota_params.called)
1757
1758 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1759 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1760 1     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1761 1     def test__process_guest_epa_params_with_numa(
1762         self,
1763         guest_epa_numa_params,
1764         guest_epa_cpu_pinning_params,
1765         guest_epa_quota_params,
1766     ):
1767 1         expected_result = {
1768             "mempage-size": "1G",
1769             "cpu-pinning-policy": "DEDICATED",
1770             "cpu-thread-pinning-policy": "PREFER",
1771             "numas": [
1772                 {
1773                     "cores": 3,
1774                     "memory": 2,
1775                     "paired-threads": 3,
1776                     "paired-threads-id": [("0", "1"), ("4", "5")],
1777                     "threads": 3,
1778                 }
1779             ],
1780             "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1781             "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1782             "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1783             "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1784         }
1785 1         target_flavor = {
1786             "guest-epa": {
1787                 "vcpu-count": 1,
1788                 "mempage-size": "1G",
1789                 "cpu-pinning-policy": "DEDICATED",
1790                 "cpu-thread-pinning-policy": "PREFER",
1791                 "numa-node-policy": {
1792                     "node": [
1793                         {
1794                             "num-cores": 3,
1795                             "paired-threads": {
1796                                 "num-paired-threads": "3",
1797                                 "paired-thread-ids": [
1798                                     {
1799                                         "thread-a": 0,
1800                                         "thread-b": 1,
1801                                     },
1802                                     {
1803                                         "thread-a": 4,
1804                                         "thread-b": 5,
1805                                     },
1806                                 ],
1807                             },
1808                             "num-threads": "3",
1809                             "memory-mb": 2048,
1810                         },
1811                     ],
1812                 },
1813                 "cpu-quota": {
1814                     "limit": "10",
1815                     "reserve": "20",
1816                     "shares": "30",
1817                 },
1818                 "mem-quota": {
1819                     "limit": "10",
1820                     "reserve": "20",
1821                     "shares": "30",
1822                 },
1823                 "disk-io-quota": {
1824                     "limit": "10",
1825                     "reserve": "20",
1826                     "shares": "30",
1827                 },
1828                 "vif-quota": {
1829                     "limit": "10",
1830                     "reserve": "20",
1831                     "shares": "30",
1832                 },
1833             },
1834         }
1835
1836 1         guest_epa_numa_params.return_value = (
1837             [
1838                 {
1839                     "cores": 3,
1840                     "paired-threads": 3,
1841                     "paired-threads-id": [("0", "1"), ("4", "5")],
1842                     "threads": 3,
1843                     "memory": 2,
1844                 },
1845             ],
1846             True,
1847         )
1848 1         guest_epa_cpu_pinning_params.return_value = (
1849             {
1850                 "threads": 3,
1851             },
1852             True,
1853         )
1854 1         guest_epa_quota_params.return_value = {
1855             "cpu-quota": {
1856                 "limit": 10,
1857                 "reserve": 20,
1858                 "shares": 30,
1859             },
1860             "mem-quota": {
1861                 "limit": 10,
1862                 "reserve": 20,
1863                 "shares": 30,
1864             },
1865             "disk-io-quota": {
1866                 "limit": 10,
1867                 "reserve": 20,
1868                 "shares": 30,
1869             },
1870             "vif-quota": {
1871                 "limit": 10,
1872                 "reserve": 20,
1873                 "shares": 30,
1874             },
1875         }
1876
1877 1         result = Ns._process_epa_params(
1878             target_flavor=target_flavor,
1879         )
1880 1         self.assertEqual(expected_result, result)
1881 1         self.assertTrue(guest_epa_numa_params.called)
1882 1         self.assertTrue(guest_epa_cpu_pinning_params.called)
1883 1         self.assertTrue(guest_epa_quota_params.called)
1884
1885 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1886 1     def test__process_flavor_params_with_empty_target_flavor(
1887         self,
1888         epa_params,
1889     ):
1890 1         target_flavor = {}
1891 1         indata = {
1892             "vnf": [
1893                 {
1894                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1895                 },
1896             ],
1897         }
1898 1         vim_info = {}
1899 1         target_record_id = ""
1900
1901 1         with self.assertRaises(KeyError):
1902 1             Ns._process_flavor_params(
1903                 target_flavor=target_flavor,
1904                 indata=indata,
1905                 vim_info=vim_info,
1906                 target_record_id=target_record_id,
1907             )
1908
1909 1         self.assertFalse(epa_params.called)
1910
1911 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1912 1     def test__process_flavor_params_with_wrong_target_flavor(
1913         self,
1914         epa_params,
1915     ):
1916 1         target_flavor = {
1917             "no-target-flavor": "here",
1918         }
1919 1         indata = {}
1920 1         vim_info = {}
1921 1         target_record_id = ""
1922
1923 1         with self.assertRaises(KeyError):
1924 1             Ns._process_flavor_params(
1925                 target_flavor=target_flavor,
1926                 indata=indata,
1927                 vim_info=vim_info,
1928                 target_record_id=target_record_id,
1929             )
1930
1931 1         self.assertFalse(epa_params.called)
1932
1933 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1934 1     def test__process_flavor_params_with_empty_indata(
1935         self,
1936         epa_params,
1937     ):
1938 1         expected_result = {
1939             "find_params": {
1940                 "flavor_data": {
1941                     "disk": 10,
1942                     "ram": 1024,
1943                     "vcpus": 2,
1944                 },
1945             },
1946             "params": {
1947                 "flavor_data": {
1948                     "disk": 10,
1949                     "name": "test",
1950                     "ram": 1024,
1951                     "vcpus": 2,
1952                 },
1953             },
1954         }
1955 1         target_flavor = {
1956             "name": "test",
1957             "storage-gb": "10",
1958             "memory-mb": "1024",
1959             "vcpu-count": "2",
1960         }
1961 1         indata = {}
1962 1         vim_info = {}
1963 1         target_record_id = ""
1964
1965 1         epa_params.return_value = {}
1966
1967 1         result = Ns._process_flavor_params(
1968             target_flavor=target_flavor,
1969             indata=indata,
1970             vim_info=vim_info,
1971             target_record_id=target_record_id,
1972         )
1973
1974 1         self.assertTrue(epa_params.called)
1975 1         self.assertDictEqual(result, expected_result)
1976
1977 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1978 1     def test__process_flavor_params_with_wrong_indata(
1979         self,
1980         epa_params,
1981     ):
1982 1         expected_result = {
1983             "find_params": {
1984                 "flavor_data": {
1985                     "disk": 10,
1986                     "ram": 1024,
1987                     "vcpus": 2,
1988                 },
1989             },
1990             "params": {
1991                 "flavor_data": {
1992                     "disk": 10,
1993                     "name": "test",
1994                     "ram": 1024,
1995                     "vcpus": 2,
1996                 },
1997             },
1998         }
1999 1         target_flavor = {
2000             "name": "test",
2001             "storage-gb": "10",
2002             "memory-mb": "1024",
2003             "vcpu-count": "2",
2004         }
2005 1         indata = {
2006             "no-vnf": "here",
2007         }
2008 1         vim_info = {}
2009 1         target_record_id = ""
2010
2011 1         epa_params.return_value = {}
2012
2013 1         result = Ns._process_flavor_params(
2014             target_flavor=target_flavor,
2015             indata=indata,
2016             vim_info=vim_info,
2017             target_record_id=target_record_id,
2018         )
2019
2020 1         self.assertTrue(epa_params.called)
2021 1         self.assertDictEqual(result, expected_result)
2022
2023 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2024 1     def test__process_flavor_params_with_ephemeral_disk(
2025         self,
2026         epa_params,
2027     ):
2028 1         kwargs = {
2029             "db": db,
2030         }
2031
2032 1         db.get_one.return_value = {
2033             "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2034             "df": [
2035                 {
2036                     "id": "default-df",
2037                     "vdu-profile": [
2038                         {"id": "without_volumes-VM", "min-number-of-instances": 1}
2039                     ],
2040                 }
2041             ],
2042             "id": "without_volumes-vnf",
2043             "product-name": "without_volumes-vnf",
2044             "vdu": [
2045                 {
2046                     "id": "without_volumes-VM",
2047                     "name": "without_volumes-VM",
2048                     "sw-image-desc": "ubuntu20.04",
2049                     "alternative-sw-image-desc": [
2050                         "ubuntu20.04-aws",
2051                         "ubuntu20.04-azure",
2052                     ],
2053                     "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2054                 }
2055             ],
2056             "version": "1.0",
2057             "virtual-storage-desc": [
2058                 {"id": "root-volume", "size-of-storage": "10"},
2059                 {
2060                     "id": "ephemeral-volume",
2061                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2062                     "size-of-storage": "1",
2063                 },
2064             ],
2065             "_admin": {
2066                 "storage": {
2067                     "fs": "mongo",
2068                     "path": "/app/storage/",
2069                 },
2070                 "type": "vnfd",
2071             },
2072         }
2073 1         expected_result = {
2074             "find_params": {
2075                 "flavor_data": {
2076                     "disk": 10,
2077                     "ram": 1024,
2078                     "vcpus": 2,
2079                     "ephemeral": 10,
2080                 },
2081             },
2082             "params": {
2083                 "flavor_data": {
2084                     "disk": 10,
2085                     "name": "test",
2086                     "ram": 1024,
2087                     "vcpus": 2,
2088                     "ephemeral": 10,
2089                 },
2090             },
2091         }
2092 1         target_flavor = {
2093             "id": "test_id",
2094             "name": "test",
2095             "storage-gb": "10",
2096             "memory-mb": "1024",
2097             "vcpu-count": "2",
2098         }
2099 1         indata = {
2100             "vnf": [
2101                 {
2102                     "vdur": [
2103                         {
2104                             "ns-flavor-id": "test_id",
2105                             "virtual-storages": [
2106                                 {
2107                                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2108                                     "size-of-storage": "10",
2109                                 },
2110                             ],
2111                         },
2112                     ],
2113                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2114                 },
2115             ],
2116         }
2117 1         vim_info = {}
2118 1         target_record_id = ""
2119
2120 1         epa_params.return_value = {}
2121
2122 1         result = Ns._process_flavor_params(
2123             target_flavor=target_flavor,
2124             indata=indata,
2125             vim_info=vim_info,
2126             target_record_id=target_record_id,
2127             **kwargs,
2128         )
2129
2130 1         self.assertTrue(epa_params.called)
2131 1         self.assertDictEqual(result, expected_result)
2132
2133 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2134 1     def test__process_flavor_params_with_swap_disk(
2135         self,
2136         epa_params,
2137     ):
2138 1         expected_result = {
2139             "find_params": {
2140                 "flavor_data": {
2141                     "disk": 10,
2142                     "ram": 1024,
2143                     "vcpus": 2,
2144                     "swap": 20,
2145                 },
2146             },
2147             "params": {
2148                 "flavor_data": {
2149                     "disk": 10,
2150                     "name": "test",
2151                     "ram": 1024,
2152                     "vcpus": 2,
2153                     "swap": 20,
2154                 },
2155             },
2156         }
2157 1         target_flavor = {
2158             "id": "test_id",
2159             "name": "test",
2160             "storage-gb": "10",
2161             "memory-mb": "1024",
2162             "vcpu-count": "2",
2163         }
2164 1         indata = {
2165             "vnf": [
2166                 {
2167                     "vdur": [
2168                         {
2169                             "ns-flavor-id": "test_id",
2170                             "virtual-storages": [
2171                                 {
2172                                     "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2173                                     "size-of-storage": "20",
2174                                 },
2175                             ],
2176                         },
2177                     ],
2178                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2179                 },
2180             ],
2181         }
2182 1         vim_info = {}
2183 1         target_record_id = ""
2184
2185 1         epa_params.return_value = {}
2186
2187 1         result = Ns._process_flavor_params(
2188             target_flavor=target_flavor,
2189             indata=indata,
2190             vim_info=vim_info,
2191             target_record_id=target_record_id,
2192         )
2193
2194 1         self.assertTrue(epa_params.called)
2195 1         self.assertDictEqual(result, expected_result)
2196
2197 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2198 1     def test__process_flavor_params_with_persistent_root_disk(
2199         self,
2200         epa_params,
2201     ):
2202 1         kwargs = {
2203             "db": db,
2204         }
2205
2206 1         db.get_one.return_value = {
2207             "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2208             "df": [
2209                 {
2210                     "id": "default-df",
2211                     "vdu-profile": [
2212                         {"id": "several_volumes-VM", "min-number-of-instances": 1}
2213                     ],
2214                 }
2215             ],
2216             "id": "several_volumes-vnf",
2217             "product-name": "several_volumes-vnf",
2218             "vdu": [
2219                 {
2220                     "id": "several_volumes-VM",
2221                     "name": "several_volumes-VM",
2222                     "sw-image-desc": "ubuntu20.04",
2223                     "alternative-sw-image-desc": [
2224                         "ubuntu20.04-aws",
2225                         "ubuntu20.04-azure",
2226                     ],
2227                     "virtual-storage-desc": [
2228                         "persistent-root-volume",
2229                     ],
2230                 }
2231             ],
2232             "version": "1.0",
2233             "virtual-storage-desc": [
2234                 {
2235                     "id": "persistent-root-volume",
2236                     "type-of-storage": "persistent-storage:persistent-storage",
2237                     "size-of-storage": "10",
2238                 },
2239             ],
2240             "_admin": {
2241                 "storage": {
2242                     "fs": "mongo",
2243                     "path": "/app/storage/",
2244                 },
2245                 "type": "vnfd",
2246             },
2247         }
2248 1         expected_result = {
2249             "find_params": {
2250                 "flavor_data": {
2251                     "disk": 0,
2252                     "ram": 1024,
2253                     "vcpus": 2,
2254                 },
2255             },
2256             "params": {
2257                 "flavor_data": {
2258                     "disk": 0,
2259                     "name": "test",
2260                     "ram": 1024,
2261                     "vcpus": 2,
2262                 },
2263             },
2264         }
2265 1         target_flavor = {
2266             "id": "test_id",
2267             "name": "test",
2268             "storage-gb": "10",
2269             "memory-mb": "1024",
2270             "vcpu-count": "2",
2271         }
2272 1         indata = {
2273             "vnf": [
2274                 {
2275                     "vdur": [
2276                         {
2277                             "vdu-name": "several_volumes-VM",
2278                             "ns-flavor-id": "test_id",
2279                             "virtual-storages": [
2280                                 {
2281                                     "type-of-storage": "persistent-storage:persistent-storage",
2282                                     "size-of-storage": "10",
2283                                 },
2284                             ],
2285                         },
2286                     ],
2287                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2288                 },
2289             ],
2290         }
2291 1         vim_info = {}
2292 1         target_record_id = ""
2293
2294 1         epa_params.return_value = {}
2295
2296 1         result = Ns._process_flavor_params(
2297             target_flavor=target_flavor,
2298             indata=indata,
2299             vim_info=vim_info,
2300             target_record_id=target_record_id,
2301             **kwargs,
2302         )
2303
2304 1         self.assertTrue(epa_params.called)
2305 1         self.assertDictEqual(result, expected_result)
2306
2307 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2308 1     def test__process_flavor_params_with_epa_params(
2309         self,
2310         epa_params,
2311     ):
2312 1         expected_result = {
2313             "find_params": {
2314                 "flavor_data": {
2315                     "disk": 10,
2316                     "ram": 1024,
2317                     "vcpus": 2,
2318                     "extended": {
2319                         "numa": "there-is-numa-here",
2320                     },
2321                 },
2322             },
2323             "params": {
2324                 "flavor_data": {
2325                     "disk": 10,
2326                     "name": "test",
2327                     "ram": 1024,
2328                     "vcpus": 2,
2329                     "extended": {
2330                         "numa": "there-is-numa-here",
2331                     },
2332                 },
2333             },
2334         }
2335 1         target_flavor = {
2336             "id": "test_id",
2337             "name": "test",
2338             "storage-gb": "10",
2339             "memory-mb": "1024",
2340             "vcpu-count": "2",
2341         }
2342 1         indata = {
2343             "vnf": [
2344                 {
2345                     "vdur": [
2346                         {
2347                             "ns-flavor-id": "test_id",
2348                         },
2349                     ],
2350                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2351                 },
2352             ],
2353         }
2354 1         vim_info = {}
2355 1         target_record_id = ""
2356
2357 1         epa_params.return_value = {
2358             "numa": "there-is-numa-here",
2359         }
2360
2361 1         result = Ns._process_flavor_params(
2362             target_flavor=target_flavor,
2363             indata=indata,
2364             vim_info=vim_info,
2365             target_record_id=target_record_id,
2366         )
2367
2368 1         self.assertTrue(epa_params.called)
2369 1         self.assertDictEqual(result, expected_result)
2370
2371 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2372 1     def test__process_flavor_params_with_vim_flavor_id(
2373         self,
2374         epa_params,
2375     ):
2376 1         expected_result = {
2377             "find_params": {
2378                 "vim_flavor_id": "test.flavor",
2379             },
2380         }
2381 1         target_flavor = {
2382             "id": "test_id",
2383             "name": "test",
2384             "storage-gb": "10",
2385             "memory-mb": "1024",
2386             "vcpu-count": "2",
2387         }
2388 1         indata = {
2389             "vnf": [
2390                 {
2391                     "vdur": [
2392                         {
2393                             "ns-flavor-id": "test_id",
2394                             "additionalParams": {
2395                                 "OSM": {"vim_flavor_id": "test.flavor"}
2396                             },
2397                         },
2398                     ],
2399                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2400                 },
2401             ],
2402         }
2403 1         vim_info = {}
2404 1         target_record_id = ""
2405
2406 1         epa_params.return_value = {}
2407
2408 1         result = Ns._process_flavor_params(
2409             target_flavor=target_flavor,
2410             indata=indata,
2411             vim_info=vim_info,
2412             target_record_id=target_record_id,
2413         )
2414
2415 1         self.assertFalse(epa_params.called)
2416 1         self.assertDictEqual(result, expected_result)
2417
2418 1     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2419 1     def test__process_flavor_params(
2420         self,
2421         epa_params,
2422     ):
2423 1         kwargs = {
2424             "db": db,
2425         }
2426
2427 1         db.get_one.return_value = {
2428             "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2429             "df": [
2430                 {
2431                     "id": "default-df",
2432                     "vdu-profile": [
2433                         {"id": "without_volumes-VM", "min-number-of-instances": 1}
2434                     ],
2435                 }
2436             ],
2437             "id": "without_volumes-vnf",
2438             "product-name": "without_volumes-vnf",
2439             "vdu": [
2440                 {
2441                     "id": "without_volumes-VM",
2442                     "name": "without_volumes-VM",
2443                     "sw-image-desc": "ubuntu20.04",
2444                     "alternative-sw-image-desc": [
2445                         "ubuntu20.04-aws",
2446                         "ubuntu20.04-azure",
2447                     ],
2448                     "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2449                 }
2450             ],
2451             "version": "1.0",
2452             "virtual-storage-desc": [
2453                 {"id": "root-volume", "size-of-storage": "10"},
2454                 {
2455                     "id": "ephemeral-volume",
2456                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2457                     "size-of-storage": "1",
2458                 },
2459             ],
2460             "_admin": {
2461                 "storage": {
2462                     "fs": "mongo",
2463                     "path": "/app/storage/",
2464                 },
2465                 "type": "vnfd",
2466             },
2467         }
2468
2469 1         expected_result = {
2470             "find_params": {
2471                 "flavor_data": {
2472                     "disk": 10,
2473                     "ram": 1024,
2474                     "vcpus": 2,
2475                     "ephemeral": 10,
2476                     "swap": 20,
2477                     "extended": {
2478                         "numa": "there-is-numa-here",
2479                     },
2480                 },
2481             },
2482             "params": {
2483                 "flavor_data": {
2484                     "disk": 10,
2485                     "name": "test",
2486                     "ram": 1024,
2487                     "vcpus": 2,
2488                     "ephemeral": 10,
2489                     "swap": 20,
2490                     "extended": {
2491                         "numa": "there-is-numa-here",
2492                     },
2493                 },
2494             },
2495         }
2496 1         target_flavor = {
2497             "id": "test_id",
2498             "name": "test",
2499             "storage-gb": "10",
2500             "memory-mb": "1024",
2501             "vcpu-count": "2",
2502         }
2503 1         indata = {
2504             "vnf": [
2505                 {
2506                     "vdur": [
2507                         {
2508                             "ns-flavor-id": "test_id",
2509                             "virtual-storages": [
2510                                 {
2511                                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2512                                     "size-of-storage": "10",
2513                                 },
2514                                 {
2515                                     "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2516                                     "size-of-storage": "20",
2517                                 },
2518                             ],
2519                         },
2520                     ],
2521                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2522                 },
2523             ],
2524         }
2525 1         vim_info = {}
2526 1         target_record_id = ""
2527
2528 1         epa_params.return_value = {
2529             "numa": "there-is-numa-here",
2530         }
2531
2532 1         result = Ns._process_flavor_params(
2533             target_flavor=target_flavor,
2534             indata=indata,
2535             vim_info=vim_info,
2536             target_record_id=target_record_id,
2537             **kwargs,
2538         )
2539
2540 1         self.assertTrue(epa_params.called)
2541 1         self.assertDictEqual(result, expected_result)
2542
2543 1     def test__process_net_params_with_empty_params(
2544         self,
2545     ):
2546 1         target_vld = {
2547             "name": "vld-name",
2548         }
2549 1         indata = {
2550             "name": "ns-name",
2551         }
2552 1         vim_info = {
2553             "provider_network": "some-profile-here",
2554             "ip_profile": {
2555                 "some_ip_profile": "here",
2556             },
2557         }
2558 1         target_record_id = ""
2559 1         expected_result = {
2560             "params": {
2561                 "net_name": "ns-name-vld-name",
2562                 "net_type": "bridge",
2563                 "ip_profile": {
2564                     "some_ip_profile": "here",
2565                 },
2566                 "provider_network_profile": "some-profile-here",
2567             }
2568         }
2569
2570 1         result = Ns._process_net_params(
2571             target_vld=target_vld,
2572             indata=indata,
2573             vim_info=vim_info,
2574             target_record_id=target_record_id,
2575         )
2576
2577 1         self.assertDictEqual(expected_result, result)
2578
2579 1     def test__process_net_params_with_vim_info_sdn(
2580         self,
2581     ):
2582 1         target_vld = {
2583             "name": "vld-name",
2584         }
2585 1         indata = {
2586             "name": "ns-name",
2587         }
2588 1         vim_info = {
2589             "sdn": "some-sdn",
2590             "sdn-ports": ["some", "ports", "here"],
2591             "vlds": ["some", "vlds", "here"],
2592             "type": "sdn-type",
2593         }
2594 1         target_record_id = "vld.sdn.something"
2595 1         expected_result = {
2596             "params": {
2597                 "sdn-ports": ["some", "ports", "here"],
2598                 "vlds": ["some", "vlds", "here"],
2599                 "type": "sdn-type",
2600             }
2601         }
2602
2603 1         result = Ns._process_net_params(
2604             target_vld=target_vld,
2605             indata=indata,
2606             vim_info=vim_info,
2607             target_record_id=target_record_id,
2608         )
2609
2610 1         self.assertDictEqual(expected_result, result)
2611
2612 1     def test__process_net_params_with_vim_info_sdn_target_vim(
2613         self,
2614     ):
2615 1         target_vld = {
2616             "name": "vld-name",
2617         }
2618 1         indata = {
2619             "name": "ns-name",
2620         }
2621 1         vim_info = {
2622             "sdn": "some-sdn",
2623             "sdn-ports": ["some", "ports", "here"],
2624             "vlds": ["some", "vlds", "here"],
2625             "target_vim": "some-vim",
2626             "type": "sdn-type",
2627         }
2628 1         target_record_id = "vld.sdn.something"
2629 1         expected_result = {
2630             "depends_on": ["some-vim vld.sdn"],
2631             "params": {
2632                 "sdn-ports": ["some", "ports", "here"],
2633                 "vlds": ["some", "vlds", "here"],
2634                 "target_vim": "some-vim",
2635                 "type": "sdn-type",
2636             },
2637         }
2638
2639 1         result = Ns._process_net_params(
2640             target_vld=target_vld,
2641             indata=indata,
2642             vim_info=vim_info,
2643             target_record_id=target_record_id,
2644         )
2645
2646 1         self.assertDictEqual(expected_result, result)
2647
2648 1     def test__process_net_params_with_vim_network_name(
2649         self,
2650     ):
2651 1         target_vld = {
2652             "name": "vld-name",
2653         }
2654 1         indata = {
2655             "name": "ns-name",
2656         }
2657 1         vim_info = {
2658             "vim_network_name": "some-network-name",
2659         }
2660 1         target_record_id = "vld.sdn.something"
2661 1         expected_result = {
2662             "find_params": {
2663                 "filter_dict": {
2664                     "name": "some-network-name",
2665                 },
2666             },
2667         }
2668
2669 1         result = Ns._process_net_params(
2670             target_vld=target_vld,
2671             indata=indata,
2672             vim_info=vim_info,
2673             target_record_id=target_record_id,
2674         )
2675
2676 1         self.assertDictEqual(expected_result, result)
2677
2678 1     def test__process_net_params_with_vim_network_id(
2679         self,
2680     ):
2681 1         target_vld = {
2682             "name": "vld-name",
2683         }
2684 1         indata = {
2685             "name": "ns-name",
2686         }
2687 1         vim_info = {
2688             "vim_network_id": "some-network-id",
2689         }
2690 1         target_record_id = "vld.sdn.something"
2691 1         expected_result = {
2692             "find_params": {
2693                 "filter_dict": {
2694                     "id": "some-network-id",
2695                 },
2696             },
2697         }
2698
2699 1         result = Ns._process_net_params(
2700             target_vld=target_vld,
2701             indata=indata,
2702             vim_info=vim_info,
2703             target_record_id=target_record_id,
2704         )
2705
2706 1         self.assertDictEqual(expected_result, result)
2707
2708 1     def test__process_net_params_with_mgmt_network(
2709         self,
2710     ):
2711 1         target_vld = {
2712             "id": "vld-id",
2713             "name": "vld-name",
2714             "mgmt-network": "some-mgmt-network",
2715         }
2716 1         indata = {
2717             "name": "ns-name",
2718         }
2719 1         vim_info = {}
2720 1         target_record_id = "vld.sdn.something"
2721 1         expected_result = {
2722             "find_params": {
2723                 "mgmt": True,
2724                 "name": "vld-id",
2725             },
2726         }
2727
2728 1         result = Ns._process_net_params(
2729             target_vld=target_vld,
2730             indata=indata,
2731             vim_info=vim_info,
2732             target_record_id=target_record_id,
2733         )
2734
2735 1         self.assertDictEqual(expected_result, result)
2736
2737 1     def test__process_net_params_with_underlay_eline(
2738         self,
2739     ):
2740 1         target_vld = {
2741             "name": "vld-name",
2742             "underlay": "some-underlay-here",
2743             "type": "ELINE",
2744         }
2745 1         indata = {
2746             "name": "ns-name",
2747         }
2748 1         vim_info = {
2749             "provider_network": "some-profile-here",
2750             "ip_profile": {
2751                 "some_ip_profile": "here",
2752             },
2753         }
2754 1         target_record_id = ""
2755 1         expected_result = {
2756             "params": {
2757                 "ip_profile": {
2758                     "some_ip_profile": "here",
2759                 },
2760                 "net_name": "ns-name-vld-name",
2761                 "net_type": "ptp",
2762                 "provider_network_profile": "some-profile-here",
2763             }
2764         }
2765
2766 1         result = Ns._process_net_params(
2767             target_vld=target_vld,
2768             indata=indata,
2769             vim_info=vim_info,
2770             target_record_id=target_record_id,
2771         )
2772
2773 1         self.assertDictEqual(expected_result, result)
2774
2775 1     def test__process_net_params_with_underlay_elan(
2776         self,
2777     ):
2778 1         target_vld = {
2779             "name": "vld-name",
2780             "underlay": "some-underlay-here",
2781             "type": "ELAN",
2782         }
2783 1         indata = {
2784             "name": "ns-name",
2785         }
2786 1         vim_info = {
2787             "provider_network": "some-profile-here",
2788             "ip_profile": {
2789                 "some_ip_profile": "here",
2790             },
2791         }
2792 1         target_record_id = ""
2793 1         expected_result = {
2794             "params": {
2795                 "ip_profile": {
2796                     "some_ip_profile": "here",
2797                 },
2798                 "net_name": "ns-name-vld-name",
2799                 "net_type": "data",
2800                 "provider_network_profile": "some-profile-here",
2801             }
2802         }
2803
2804 1         result = Ns._process_net_params(
2805             target_vld=target_vld,
2806             indata=indata,
2807             vim_info=vim_info,
2808             target_record_id=target_record_id,
2809         )
2810
2811 1         self.assertDictEqual(expected_result, result)
2812
2813 1     def test__get_cloud_init_exception(self):
2814 1         db_mock = MagicMock(name="database mock")
2815 1         fs_mock = None
2816
2817 1         location = ""
2818
2819 1         with self.assertRaises(NsException):
2820 1             Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2821
2822 1     def test__get_cloud_init_file_fs_exception(self):
2823 1         db_mock = MagicMock(name="database mock")
2824 1         fs_mock = None
2825
2826 1         location = "vnfr_id_123456:file:test_file"
2827 1         db_mock.get_one.return_value = {
2828             "_admin": {
2829                 "storage": {
2830                     "folder": "/home/osm",
2831                     "pkg-dir": "vnfr_test_dir",
2832                 },
2833             },
2834         }
2835
2836 1         with self.assertRaises(NsException):
2837 1             Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2838
2839 1     def test__get_cloud_init_file(self):
2840 1         db_mock = MagicMock(name="database mock")
2841 1         fs_mock = MagicMock(name="filesystem mock")
2842 1         file_mock = MagicMock(name="file mock")
2843
2844 1         location = "vnfr_id_123456:file:test_file"
2845 1         cloud_init_content = "this is a cloud init file content"
2846
2847 1         db_mock.get_one.return_value = {
2848             "_admin": {
2849                 "storage": {
2850                     "folder": "/home/osm",
2851                     "pkg-dir": "vnfr_test_dir",
2852                 },
2853             },
2854         }
2855 1         fs_mock.file_open.return_value = file_mock
2856 1         file_mock.__enter__.return_value.read.return_value = cloud_init_content
2857
2858 1         result = Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2859
2860 1         self.assertEqual(cloud_init_content, result)
2861
2862 1     def test__get_cloud_init_vdu(self):
2863 1         db_mock = MagicMock(name="database mock")
2864 1         fs_mock = None
2865
2866 1         location = "vnfr_id_123456:vdu:0"
2867 1         cloud_init_content = "this is a cloud init file content"
2868
2869 1         db_mock.get_one.return_value = {
2870             "vdu": {
2871                 0: {
2872                     "cloud-init": cloud_init_content,
2873                 },
2874             },
2875         }
2876
2877 1         result = Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2878
2879 1         self.assertEqual(cloud_init_content, result)
2880
2881 1     @patch("jinja2.Environment.__init__")
2882 1     def test__parse_jinja2_undefined_error(self, env_mock: Mock):
2883 1         cloud_init_content = None
2884 1         params = None
2885 1         context = None
2886
2887 1         env_mock.side_effect = UndefinedError("UndefinedError occurred.")
2888
2889 1         with self.assertRaises(NsException):
2890 1             Ns._parse_jinja2(
2891                 cloud_init_content=cloud_init_content, params=params, context=context
2892             )
2893
2894 1     @patch("jinja2.Environment.__init__")
2895 1     def test__parse_jinja2_template_error(self, env_mock: Mock):
2896 1         cloud_init_content = None
2897 1         params = None
2898 1         context = None
2899
2900 1         env_mock.side_effect = TemplateError("TemplateError occurred.")
2901
2902 1         with self.assertRaises(NsException):
2903 1             Ns._parse_jinja2(
2904                 cloud_init_content=cloud_init_content, params=params, context=context
2905             )
2906
2907 1     @patch("jinja2.Environment.__init__")
2908 1     def test__parse_jinja2_template_not_found(self, env_mock: Mock):
2909 1         cloud_init_content = None
2910 1         params = None
2911 1         context = None
2912
2913 1         env_mock.side_effect = TemplateNotFound("TemplateNotFound occurred.")
2914
2915 1         with self.assertRaises(NsException):
2916 1             Ns._parse_jinja2(
2917                 cloud_init_content=cloud_init_content, params=params, context=context
2918             )
2919
2920 1     def test_rendering_jinja2_temp_without_special_characters(self):
2921 1         cloud_init_content = """
2922         disk_setup:
2923             ephemeral0:
2924                 table_type: {{type}}
2925                 layout: True
2926                 overwrite: {{is_override}}
2927         runcmd:
2928              - [ ls, -l, / ]
2929              - [ sh, -xc, "echo $(date) '{{command}}'" ]
2930         """
2931 1         params = {
2932             "type": "mbr",
2933             "is_override": "False",
2934             "command": "; mkdir abc",
2935         }
2936 1         context = "cloud-init for VM"
2937 1         expected_result = """
2938         disk_setup:
2939             ephemeral0:
2940                 table_type: mbr
2941                 layout: True
2942                 overwrite: False
2943         runcmd:
2944              - [ ls, -l, / ]
2945              - [ sh, -xc, "echo $(date) '; mkdir abc'" ]
2946         """
2947 1         result = Ns._parse_jinja2(
2948             cloud_init_content=cloud_init_content, params=params, context=context
2949         )
2950 1         self.assertEqual(result, expected_result)
2951
2952 1     def test_rendering_jinja2_temp_with_special_characters(self):
2953 1         cloud_init_content = """
2954         disk_setup:
2955             ephemeral0:
2956                 table_type: {{type}}
2957                 layout: True
2958                 overwrite: {{is_override}}
2959         runcmd:
2960              - [ ls, -l, / ]
2961              - [ sh, -xc, "echo $(date) '{{command}}'" ]
2962         """
2963 1         params = {
2964             "type": "mbr",
2965             "is_override": "False",
2966             "command": "& rm -rf",
2967         }
2968 1         context = "cloud-init for VM"
2969 1         expected_result = """
2970         disk_setup:
2971             ephemeral0:
2972                 table_type: mbr
2973                 layout: True
2974                 overwrite: False
2975         runcmd:
2976              - [ ls, -l, / ]
2977              - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
2978         """
2979 1         result = Ns._parse_jinja2(
2980             cloud_init_content=cloud_init_content, params=params, context=context
2981         )
2982 1         self.assertNotEqual(result, expected_result)
2983
2984 1     def test_rendering_jinja2_temp_with_special_characters_autoescape_is_false(self):
2985 1         with patch("osm_ng_ro.ns.Environment") as mock_environment:
2986 1             mock_environment.return_value = Environment(
2987                 undefined=StrictUndefined,
2988                 autoescape=select_autoescape(default_for_string=False, default=False),
2989             )
2990 1             cloud_init_content = """
2991                 disk_setup:
2992                     ephemeral0:
2993                         table_type: {{type}}
2994                         layout: True
2995                         overwrite: {{is_override}}
2996                 runcmd:
2997                      - [ ls, -l, / ]
2998                      - [ sh, -xc, "echo $(date) '{{command}}'" ]
2999                 """
3000 1             params = {
3001                 "type": "mbr",
3002                 "is_override": "False",
3003                 "command": "& rm -rf /",
3004             }
3005 1             context = "cloud-init for VM"
3006 1             expected_result = """
3007                 disk_setup:
3008                     ephemeral0:
3009                         table_type: mbr
3010                         layout: True
3011                         overwrite: False
3012                 runcmd:
3013                      - [ ls, -l, / ]
3014                      - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
3015                 """
3016 1             result = Ns._parse_jinja2(
3017                 cloud_init_content=cloud_init_content,
3018                 params=params,
3019                 context=context,
3020             )
3021 1             self.assertEqual(result, expected_result)
3022
3023 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3024 1     def test__rebuild_start_stop_task__successful(self, assign_vim):
3025 1         self.ns = Ns()
3026 1         extra_dict = {}
3027 1         actions = ["start", "stop", "rebuild"]
3028 1         vdu_index = "0"
3029 1         task_index = 0
3030 1         for action in actions:
3031 1             params = {
3032                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3033                 "action": action,
3034             }
3035 1             extra_dict["params"] = params
3036 1             expected_result = deepcopy(expected_result_rebuild_start_stop)
3037 1             expected_result[
3038                 "target_record"
3039             ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3040 1             expected_result["params"] = params
3041 1             task = self.ns.rebuild_start_stop_task(
3042                 vdu_id,
3043                 vnf_id,
3044                 vdu_index,
3045                 action_id,
3046                 nsr_id_2,
3047                 task_index,
3048                 target_vim,
3049                 extra_dict,
3050             )
3051 1             self.assertDictEqual(task, expected_result)
3052
3053 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3054 1     def test__rebuild_start_stop_task__empty_extra_dict__task_without_params(
3055         self, assign_vim
3056     ):
3057 1         self.ns = Ns()
3058 1         extra_dict = {}
3059 1         actions = ["start", "stop", "rebuild"]
3060 1         vdu_index = "0"
3061 1         task_index = 0
3062 1         expected_result = deepcopy(expected_result_rebuild_start_stop)
3063 1         expected_result[
3064             "target_record"
3065         ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3066 1         for _ in actions:
3067 1             task = self.ns.rebuild_start_stop_task(
3068                 vdu_id,
3069                 vnf_id,
3070                 vdu_index,
3071                 action_id,
3072                 nsr_id_2,
3073                 task_index,
3074                 target_vim,
3075                 extra_dict,
3076             )
3077 1             self.assertDictEqual(task, expected_result)
3078
3079 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3080 1     def test__rebuild_start_stop_task__different_vdu_index__target_record_changes(
3081         self, assign_vim
3082     ):
3083 1         self.ns = Ns()
3084 1         extra_dict = {}
3085 1         actions = ["start", "stop", "rebuild"]
3086 1         vdu_index = "4"
3087 1         task_index = 0
3088 1         for action in actions:
3089 1             params = {
3090                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3091                 "action": action,
3092             }
3093 1             extra_dict["params"] = params
3094 1             expected_result = deepcopy(expected_result_rebuild_start_stop)
3095 1             expected_result[
3096                 "target_record"
3097             ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3098 1             expected_result["params"] = params
3099 1             task = self.ns.rebuild_start_stop_task(
3100                 vdu_id,
3101                 vnf_id,
3102                 vdu_index,
3103                 action_id,
3104                 nsr_id_2,
3105                 task_index,
3106                 target_vim,
3107                 extra_dict,
3108             )
3109 1             self.assertDictEqual(task, expected_result)
3110
3111 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3112 1     def test__rebuild_start_stop_task__different_task_index__task_id_changes(
3113         self, assign_vim
3114     ):
3115 1         self.ns = Ns()
3116 1         extra_dict = {}
3117 1         actions = ["start", "stop", "rebuild"]
3118 1         vdu_index = "0"
3119 1         task_index = 3
3120 1         for action in actions:
3121 1             params = {
3122                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3123                 "action": action,
3124             }
3125 1             extra_dict["params"] = params
3126 1             expected_result = deepcopy(expected_result_rebuild_start_stop)
3127 1             expected_result[
3128                 "target_record"
3129             ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3130 1             expected_result["params"] = params
3131 1             expected_result["task_id"] = "bb937f49-3870-4169-b758-9732e1ff40f3:3"
3132 1             task = self.ns.rebuild_start_stop_task(
3133                 vdu_id,
3134                 vnf_id,
3135                 vdu_index,
3136                 action_id,
3137                 nsr_id_2,
3138                 task_index,
3139                 target_vim,
3140                 extra_dict,
3141             )
3142 1             self.assertDictEqual(task, expected_result)
3143
3144 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3145 1     def test__rebuild_start_stop_task__assign_vim_raises__task_is_not_created(
3146         self, assign_vim
3147     ):
3148 1         self.ns = Ns()
3149 1         extra_dict = {}
3150 1         actions = ["start", "stop", "rebuild"]
3151 1         vdu_index = "0"
3152 1         task_index = 0
3153 1         for action in actions:
3154 1             params = {
3155                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3156                 "action": action,
3157             }
3158 1             extra_dict["params"] = params
3159 1             assign_vim.side_effect = TestException("Can not connect to VIM.")
3160 1             with self.assertRaises(TestException) as err:
3161 1                 task = self.ns.rebuild_start_stop_task(
3162                     vdu_id,
3163                     vnf_id,
3164                     vdu_index,
3165                     action_id,
3166                     nsr_id_2,
3167                     task_index,
3168                     target_vim,
3169                     extra_dict,
3170                 )
3171 0                 self.assertEqual(task, None)
3172 0                 self.assertEqual(str(err.exception), "Can not connect to VIM.")
3173
3174 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3175 1     def test_verticalscale_task__successful(self, assign_vim):
3176 1         self.ns = Ns()
3177 1         vdu_index = "1"
3178 1         task_index = 1
3179 1         task = self.ns.verticalscale_task(
3180             vdu,
3181             vnf,
3182             vdu_index,
3183             action_id,
3184             nsr_id_2,
3185             task_index,
3186             extra_dict_vertical_scale,
3187         )
3188 1         self.assertDictEqual(task, expected_result_vertical_scale)
3189
3190 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3191 1     def test_verticalscale_task__task_index_changes__task_id_changes(self, assign_vim):
3192 1         self.ns = Ns()
3193 1         vdu_index = "1"
3194 1         task_index = 2
3195 1         expected_result = deepcopy(expected_result_vertical_scale)
3196 1         expected_result["task_id"] = "bb937f49-3870-4169-b758-9732e1ff40f3:2"
3197 1         task = self.ns.verticalscale_task(
3198             vdu,
3199             vnf,
3200             vdu_index,
3201             action_id,
3202             nsr_id_2,
3203             task_index,
3204             extra_dict_vertical_scale,
3205         )
3206 1         self.assertDictEqual(task, expected_result)
3207
3208 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3209 1     def test_verticalscale_task__empty_extra_dict__expected_result_without_params(
3210         self, assign_vim
3211     ):
3212 1         self.ns = Ns()
3213 1         extra_dict = {}
3214 1         vdu_index = "1"
3215 1         task_index = 1
3216 1         expected_result = deepcopy(expected_result_vertical_scale)
3217 1         expected_result.pop("params")
3218 1         task = self.ns.verticalscale_task(
3219             vdu, vnf, vdu_index, action_id, nsr_id_2, task_index, extra_dict
3220         )
3221 1         self.assertDictEqual(task, expected_result)
3222
3223 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3224 1     def test_verticalscale_task__assign_vim_raises__task_is_not_created(
3225         self, assign_vim
3226     ):
3227 1         self.ns = Ns()
3228 1         vdu_index = "1"
3229 1         task_index = 1
3230 1         assign_vim.side_effect = TestException("Can not connect to VIM.")
3231 1         with self.assertRaises(TestException) as err:
3232 1             task = self.ns.verticalscale_task(
3233                 vdu,
3234                 vnf,
3235                 vdu_index,
3236                 action_id,
3237                 nsr_id_2,
3238                 task_index,
3239                 extra_dict_vertical_scale,
3240             )
3241 0             self.assertEqual(task, {})
3242 0             self.assertEqual(str(err.exception), "Can not connect to VIM.")
3243
3244 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3245 1     def test_migrate_task__successful(self, assign_vim):
3246 1         self.ns = Ns()
3247 1         vdu_index = "1"
3248 1         task_index = 1
3249 1         task = self.ns.migrate_task(
3250             vdu, vnf, vdu_index, action_id, nsr_id_2, task_index, extra_dict_migrate
3251         )
3252 1         self.assertDictEqual(task, expected_result_migrate)
3253
3254 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3255 1     def test_migrate_task__empty_extra_dict__task_without_params(self, assign_vim):
3256 1         self.ns = Ns()
3257 1         extra_dict = {}
3258 1         vdu_index = "1"
3259 1         task_index = 1
3260 1         expected_result = deepcopy(expected_result_migrate)
3261 1         expected_result.pop("params")
3262 1         task = self.ns.migrate_task(
3263             vdu, vnf, vdu_index, action_id, nsr_id_2, task_index, extra_dict
3264         )
3265 1         self.assertDictEqual(task, expected_result)
3266
3267 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3268 1     def test_migrate_task__different_vdu_index__target_record_with_different_vdu_index(
3269         self, assign_vim
3270     ):
3271 1         self.ns = Ns()
3272 1         vdu_index = "4"
3273 1         task_index = 1
3274 1         expected_result = deepcopy(expected_result_migrate)
3275 1         expected_result[
3276             "target_record"
3277         ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3278 1         task = self.ns.migrate_task(
3279             vdu, vnf, vdu_index, action_id, nsr_id_2, task_index, extra_dict_migrate
3280         )
3281 1         self.assertDictEqual(task, expected_result)
3282
3283 1     @patch("osm_ng_ro.ns.Ns._assign_vim")
3284 1     def test_migrate_task__assign_vim_raises__task_is_not_created(self, assign_vim):
3285 1         self.ns = Ns()
3286 1         vdu_index = "1"
3287 1         task_index = 1
3288 1         assign_vim.side_effect = TestException("Can not connect to VIM.")
3289 1         with self.assertRaises(TestException) as err:
3290 1             task = self.ns.migrate_task(
3291                 vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict_migrate
3292             )
3293 0             self.assertDictEqual(task, {})
3294 0             self.assertEqual(str(err.exception), "Can not connect to VIM.")
3295
3296
3297 1 class TestProcessVduParams(unittest.TestCase):
3298 1     def setUp(self):
3299 1         self.ns = Ns()
3300 1         self.logger = CopyingMock(autospec=True)
3301
3302 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3303 1     def test_find_persistent_root_volumes_empty_instantiation_vol_list(
3304         self, mock_volume_keeping_required
3305     ):
3306         """Find persistent root volume, instantiation_vol_list is empty."""
3307 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3308 1         target_vdu = target_vdu_wth_persistent_storage
3309 1         vdu_instantiation_volumes_list = []
3310 1         disk_list = []
3311 1         mock_volume_keeping_required.return_value = True
3312 1         expected_root_disk = {
3313             "id": "persistent-root-volume",
3314             "type-of-storage": "persistent-storage:persistent-storage",
3315             "size-of-storage": "10",
3316             "vdu-storage-requirements": [{"key": "keep-volume", "value": "true"}],
3317         }
3318 1         expected_persist_root_disk = {
3319             "persistent-root-volume": {
3320                 "image_id": "ubuntu20.04",
3321                 "size": "10",
3322                 "keep": True,
3323             }
3324         }
3325 1         expected_disk_list = [
3326             {
3327                 "image_id": "ubuntu20.04",
3328                 "size": "10",
3329                 "keep": True,
3330             },
3331         ]
3332 1         persist_root_disk = self.ns.find_persistent_root_volumes(
3333             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3334         )
3335 1         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3336 1         mock_volume_keeping_required.assert_called_once_with(expected_root_disk)
3337 1         self.assertEqual(disk_list, expected_disk_list)
3338 1         self.assertEqual(len(disk_list), 1)
3339
3340 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3341 1     def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(
3342         self, mock_volume_keeping_required
3343     ):
3344         """Find persistent root volume, always selects the first vsd as root volume."""
3345 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3346 1         vnfd["vdu"][0]["virtual-storage-desc"] = [
3347             "persistent-volume2",
3348             "persistent-root-volume",
3349             "ephemeral-volume",
3350         ]
3351 1         target_vdu = target_vdu_wth_persistent_storage
3352 1         vdu_instantiation_volumes_list = []
3353 1         disk_list = []
3354 1         mock_volume_keeping_required.return_value = True
3355 1         expected_root_disk = {
3356             "id": "persistent-volume2",
3357             "type-of-storage": "persistent-storage:persistent-storage",
3358             "size-of-storage": "10",
3359         }
3360 1         expected_persist_root_disk = {
3361             "persistent-volume2": {
3362                 "image_id": "ubuntu20.04",
3363                 "size": "10",
3364                 "keep": True,
3365             }
3366         }
3367 1         expected_disk_list = [
3368             {
3369                 "image_id": "ubuntu20.04",
3370                 "size": "10",
3371                 "keep": True,
3372             },
3373         ]
3374 1         persist_root_disk = self.ns.find_persistent_root_volumes(
3375             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3376         )
3377 1         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3378 1         mock_volume_keeping_required.assert_called_once_with(expected_root_disk)
3379 1         self.assertEqual(disk_list, expected_disk_list)
3380 1         self.assertEqual(len(disk_list), 1)
3381
3382 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3383 1     def test_find_persistent_root_volumes_empty_size_of_storage(
3384         self, mock_volume_keeping_required
3385     ):
3386         """Find persistent root volume, size of storage is empty."""
3387 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3388 1         vnfd["virtual-storage-desc"][0]["size-of-storage"] = ""
3389 1         vnfd["vdu"][0]["virtual-storage-desc"] = [
3390             "persistent-volume2",
3391             "persistent-root-volume",
3392             "ephemeral-volume",
3393         ]
3394 1         target_vdu = target_vdu_wth_persistent_storage
3395 1         vdu_instantiation_volumes_list = []
3396 1         disk_list = []
3397 1         persist_root_disk = self.ns.find_persistent_root_volumes(
3398             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3399         )
3400 1         self.assertEqual(persist_root_disk, {})
3401 1         mock_volume_keeping_required.assert_not_called()
3402 1         self.assertEqual(disk_list, [])
3403
3404 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3405 1     def test_find_persistent_root_volumes_keeping_is_not_required(
3406         self, mock_volume_keeping_required
3407     ):
3408         """Find persistent root volume, volume keeping is not required."""
3409 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3410 1         vnfd["virtual-storage-desc"][1]["vdu-storage-requirements"] = [
3411             {"key": "keep-volume", "value": "false"},
3412         ]
3413 1         target_vdu = target_vdu_wth_persistent_storage
3414 1         vdu_instantiation_volumes_list = []
3415 1         disk_list = []
3416 1         mock_volume_keeping_required.return_value = False
3417 1         expected_root_disk = {
3418             "id": "persistent-root-volume",
3419             "type-of-storage": "persistent-storage:persistent-storage",
3420             "size-of-storage": "10",
3421             "vdu-storage-requirements": [{"key": "keep-volume", "value": "false"}],
3422         }
3423 1         expected_persist_root_disk = {
3424             "persistent-root-volume": {
3425                 "image_id": "ubuntu20.04",
3426                 "size": "10",
3427                 "keep": False,
3428             }
3429         }
3430 1         expected_disk_list = [
3431             {
3432                 "image_id": "ubuntu20.04",
3433                 "size": "10",
3434                 "keep": False,
3435             },
3436         ]
3437 1         persist_root_disk = self.ns.find_persistent_root_volumes(
3438             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3439         )
3440 1         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3441 1         mock_volume_keeping_required.assert_called_once_with(expected_root_disk)
3442 1         self.assertEqual(disk_list, expected_disk_list)
3443 1         self.assertEqual(len(disk_list), 1)
3444
3445 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3446 1     def test_find_persistent_root_volumes_target_vdu_mismatch(
3447         self, mock_volume_keeping_required
3448     ):
3449         """Find persistent root volume, target vdu name is not matching."""
3450 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3451 1         vnfd["vdu"][0]["name"] = "Several_Volumes-VM"
3452 1         target_vdu = target_vdu_wth_persistent_storage
3453 1         vdu_instantiation_volumes_list = []
3454 1         disk_list = []
3455 1         result = self.ns.find_persistent_root_volumes(
3456             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3457         )
3458 1         self.assertEqual(result, None)
3459 1         mock_volume_keeping_required.assert_not_called()
3460 1         self.assertEqual(disk_list, [])
3461 1         self.assertEqual(len(disk_list), 0)
3462
3463 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3464 1     def test_find_persistent_root_volumes_with_instantiation_vol_list(
3465         self, mock_volume_keeping_required
3466     ):
3467         """Find persistent root volume, existing volume needs to be used."""
3468 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3469 1         target_vdu = target_vdu_wth_persistent_storage
3470 1         vdu_instantiation_volumes_list = [
3471             {
3472                 "vim-volume-id": vim_volume_id,
3473                 "name": "persistent-root-volume",
3474             }
3475         ]
3476 1         disk_list = []
3477 1         expected_persist_root_disk = {
3478             "persistent-root-volume": {
3479                 "vim_volume_id": vim_volume_id,
3480                 "image_id": "ubuntu20.04",
3481             },
3482         }
3483 1         expected_disk_list = [
3484             {
3485                 "vim_volume_id": vim_volume_id,
3486                 "image_id": "ubuntu20.04",
3487             },
3488         ]
3489 1         persist_root_disk = self.ns.find_persistent_root_volumes(
3490             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3491         )
3492 1         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3493 1         mock_volume_keeping_required.assert_not_called()
3494 1         self.assertEqual(disk_list, expected_disk_list)
3495 1         self.assertEqual(len(disk_list), 1)
3496
3497 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3498 1     def test_find_persistent_root_volumes_invalid_instantiation_params(
3499         self, mock_volume_keeping_required
3500     ):
3501         """Find persistent root volume, existing volume id keyword is invalid."""
3502 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
3503 1         target_vdu = target_vdu_wth_persistent_storage
3504 1         vdu_instantiation_volumes_list = [
3505             {
3506                 "volume-id": vim_volume_id,
3507                 "name": "persistent-root-volume",
3508             }
3509         ]
3510 1         disk_list = []
3511 1         with self.assertRaises(KeyError):
3512 1             self.ns.find_persistent_root_volumes(
3513                 vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3514             )
3515 1         mock_volume_keeping_required.assert_not_called()
3516 1         self.assertEqual(disk_list, [])
3517 1         self.assertEqual(len(disk_list), 0)
3518
3519 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3520 1     def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list(
3521         self, mock_volume_keeping_required
3522     ):
3523         """Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty."""
3524 1         persistent_root_disk = {
3525             "persistent-root-volume": {
3526                 "image_id": "ubuntu20.04",
3527                 "size": "10",
3528                 "keep": False,
3529             }
3530         }
3531 1         mock_volume_keeping_required.return_value = False
3532 1         target_vdu = target_vdu_wth_persistent_storage
3533 1         vdu_instantiation_volumes_list = []
3534 1         disk_list = [
3535             {
3536                 "image_id": "ubuntu20.04",
3537                 "size": "10",
3538                 "keep": False,
3539             },
3540         ]
3541 1         expected_disk = {
3542             "id": "persistent-volume2",
3543             "size-of-storage": "10",
3544             "type-of-storage": "persistent-storage:persistent-storage",
3545         }
3546 1         expected_disk_list = [
3547             {
3548                 "image_id": "ubuntu20.04",
3549                 "size": "10",
3550                 "keep": False,
3551             },
3552             {
3553                 "size": "10",
3554                 "keep": False,
3555             },
3556         ]
3557 1         self.ns.find_persistent_volumes(
3558             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3559         )
3560 1         self.assertEqual(disk_list, expected_disk_list)
3561 1         mock_volume_keeping_required.assert_called_once_with(expected_disk)
3562
3563 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3564 1     def test_find_persistent_volumes_vdu_wth_inst_vol_list(
3565         self, mock_volume_keeping_required
3566     ):
3567         """Find persistent ordinary volume, vim-volume-id is given as instantiation parameter."""
3568 1         persistent_root_disk = {
3569             "persistent-root-volume": {
3570                 "image_id": "ubuntu20.04",
3571                 "size": "10",
3572                 "keep": False,
3573             }
3574         }
3575 1         vdu_instantiation_volumes_list = [
3576             {
3577                 "vim-volume-id": vim_volume_id,
3578                 "name": "persistent-volume2",
3579             }
3580         ]
3581 1         target_vdu = target_vdu_wth_persistent_storage
3582 1         disk_list = [
3583             {
3584                 "image_id": "ubuntu20.04",
3585                 "size": "10",
3586                 "keep": False,
3587             },
3588         ]
3589 1         expected_disk_list = [
3590             {
3591                 "image_id": "ubuntu20.04",
3592                 "size": "10",
3593                 "keep": False,
3594             },
3595             {
3596                 "vim_volume_id": vim_volume_id,
3597             },
3598         ]
3599 1         self.ns.find_persistent_volumes(
3600             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3601         )
3602 1         self.assertEqual(disk_list, expected_disk_list)
3603 1         mock_volume_keeping_required.assert_not_called()
3604
3605 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3606 1     def test_find_persistent_volumes_vdu_wthout_persistent_storage(
3607         self, mock_volume_keeping_required
3608     ):
3609         """Find persistent ordinary volume, there is not any persistent disk."""
3610 1         persistent_root_disk = {}
3611 1         vdu_instantiation_volumes_list = []
3612 1         mock_volume_keeping_required.return_value = False
3613 1         target_vdu = target_vdu_wthout_persistent_storage
3614 1         disk_list = []
3615 1         self.ns.find_persistent_volumes(
3616             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3617         )
3618 1         self.assertEqual(disk_list, disk_list)
3619 1         mock_volume_keeping_required.assert_not_called()
3620
3621 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3622 1     def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk(
3623         self, mock_volume_keeping_required
3624     ):
3625         """There is persistent root disk, but there is not ordinary persistent disk."""
3626 1         persistent_root_disk = {
3627             "persistent-root-volume": {
3628                 "image_id": "ubuntu20.04",
3629                 "size": "10",
3630                 "keep": False,
3631             }
3632         }
3633 1         vdu_instantiation_volumes_list = []
3634 1         mock_volume_keeping_required.return_value = False
3635 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3636 1         target_vdu["virtual-storages"] = [
3637             {
3638                 "id": "persistent-root-volume",
3639                 "size-of-storage": "10",
3640                 "type-of-storage": "persistent-storage:persistent-storage",
3641                 "vdu-storage-requirements": [
3642                     {"key": "keep-volume", "value": "true"},
3643                 ],
3644             },
3645             {
3646                 "id": "ephemeral-volume",
3647                 "size-of-storage": "1",
3648                 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
3649             },
3650         ]
3651 1         disk_list = [
3652             {
3653                 "image_id": "ubuntu20.04",
3654                 "size": "10",
3655                 "keep": False,
3656             },
3657         ]
3658 1         self.ns.find_persistent_volumes(
3659             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3660         )
3661 1         self.assertEqual(disk_list, disk_list)
3662 1         mock_volume_keeping_required.assert_not_called()
3663
3664 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
3665 1     def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(
3666         self, mock_volume_keeping_required
3667     ):
3668         """Find persistent ordinary volume, volume id is not persistent_root_disk dict,
3669         vim-volume-id is given as instantiation parameter but disk id is not matching.
3670         """
3671 1         mock_volume_keeping_required.return_value = True
3672 1         vdu_instantiation_volumes_list = [
3673             {
3674                 "vim-volume-id": vim_volume_id,
3675                 "name": "persistent-volume3",
3676             }
3677         ]
3678 1         persistent_root_disk = {
3679             "persistent-root-volume": {
3680                 "image_id": "ubuntu20.04",
3681                 "size": "10",
3682                 "keep": False,
3683             }
3684         }
3685 1         disk_list = [
3686             {
3687                 "image_id": "ubuntu20.04",
3688                 "size": "10",
3689                 "keep": False,
3690             },
3691         ]
3692 1         expected_disk_list = [
3693             {
3694                 "image_id": "ubuntu20.04",
3695                 "size": "10",
3696                 "keep": False,
3697             },
3698             {
3699                 "size": "10",
3700                 "keep": True,
3701             },
3702         ]
3703 1         expected_disk = {
3704             "id": "persistent-volume2",
3705             "size-of-storage": "10",
3706             "type-of-storage": "persistent-storage:persistent-storage",
3707         }
3708 1         target_vdu = target_vdu_wth_persistent_storage
3709 1         self.ns.find_persistent_volumes(
3710             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3711         )
3712 1         self.assertEqual(disk_list, expected_disk_list)
3713 1         mock_volume_keeping_required.assert_called_once_with(expected_disk)
3714
3715 1     def test_is_volume_keeping_required_true(self):
3716         """Volume keeping is required."""
3717 1         virtual_storage_descriptor = {
3718             "id": "persistent-root-volume",
3719             "type-of-storage": "persistent-storage:persistent-storage",
3720             "size-of-storage": "10",
3721             "vdu-storage-requirements": [
3722                 {"key": "keep-volume", "value": "true"},
3723             ],
3724         }
3725 1         result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
3726 1         self.assertEqual(result, True)
3727
3728 1     def test_is_volume_keeping_required_false(self):
3729         """Volume keeping is not required."""
3730 1         virtual_storage_descriptor = {
3731             "id": "persistent-root-volume",
3732             "type-of-storage": "persistent-storage:persistent-storage",
3733             "size-of-storage": "10",
3734             "vdu-storage-requirements": [
3735                 {"key": "keep-volume", "value": "false"},
3736             ],
3737         }
3738 1         result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
3739 1         self.assertEqual(result, False)
3740
3741 1     def test_is_volume_keeping_required_wthout_vdu_storage_reqirement(self):
3742         """Volume keeping is not required, vdu-storage-requirements key does not exist."""
3743 1         virtual_storage_descriptor = {
3744             "id": "persistent-root-volume",
3745             "type-of-storage": "persistent-storage:persistent-storage",
3746             "size-of-storage": "10",
3747         }
3748 1         result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
3749 1         self.assertEqual(result, False)
3750
3751 1     def test_is_volume_keeping_required_wrong_keyword(self):
3752         """vdu-storage-requirements key to indicate keeping-volume is wrong."""
3753 1         virtual_storage_descriptor = {
3754             "id": "persistent-root-volume",
3755             "type-of-storage": "persistent-storage:persistent-storage",
3756             "size-of-storage": "10",
3757             "vdu-storage-requirements": [
3758                 {"key": "hold-volume", "value": "true"},
3759             ],
3760         }
3761 1         result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
3762 1         self.assertEqual(result, False)
3763
3764 1     def test_sort_vdu_interfaces_position_all_wth_positions(self):
3765         """Interfaces are sorted according to position, all have positions."""
3766 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3767 1         target_vdu["interfaces"] = [
3768             {
3769                 "name": "vdu-eth1",
3770                 "ns-vld-id": "datanet",
3771                 "position": 2,
3772             },
3773             {
3774                 "name": "vdu-eth0",
3775                 "ns-vld-id": "mgmtnet",
3776                 "position": 1,
3777             },
3778         ]
3779 1         sorted_interfaces = [
3780             {
3781                 "name": "vdu-eth0",
3782                 "ns-vld-id": "mgmtnet",
3783                 "position": 1,
3784             },
3785             {
3786                 "name": "vdu-eth1",
3787                 "ns-vld-id": "datanet",
3788                 "position": 2,
3789             },
3790         ]
3791 1         self.ns._sort_vdu_interfaces(target_vdu)
3792 1         self.assertEqual(target_vdu["interfaces"], sorted_interfaces)
3793
3794 1     def test_sort_vdu_interfaces_position_some_wth_position(self):
3795         """Interfaces are sorted according to position, some of them have positions."""
3796 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3797 1         target_vdu["interfaces"] = [
3798             {
3799                 "name": "vdu-eth0",
3800                 "ns-vld-id": "mgmtnet",
3801             },
3802             {
3803                 "name": "vdu-eth1",
3804                 "ns-vld-id": "datanet",
3805                 "position": 1,
3806             },
3807         ]
3808 1         sorted_interfaces = [
3809             {
3810                 "name": "vdu-eth1",
3811                 "ns-vld-id": "datanet",
3812                 "position": 1,
3813             },
3814             {
3815                 "name": "vdu-eth0",
3816                 "ns-vld-id": "mgmtnet",
3817             },
3818         ]
3819 1         self.ns._sort_vdu_interfaces(target_vdu)
3820 1         self.assertEqual(target_vdu["interfaces"], sorted_interfaces)
3821
3822 1     def test_sort_vdu_interfaces_position_empty_interface_list(self):
3823         """Interface list is empty."""
3824 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3825 1         target_vdu["interfaces"] = []
3826 1         sorted_interfaces = []
3827 1         self.ns._sort_vdu_interfaces(target_vdu)
3828 1         self.assertEqual(target_vdu["interfaces"], sorted_interfaces)
3829
3830 1     def test_partially_locate_vdu_interfaces(self):
3831         """Some interfaces have positions."""
3832 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3833 1         target_vdu["interfaces"] = [
3834             {
3835                 "name": "vdu-eth1",
3836                 "ns-vld-id": "net1",
3837             },
3838             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3839             {
3840                 "name": "vdu-eth3",
3841                 "ns-vld-id": "mgmtnet",
3842             },
3843             {
3844                 "name": "vdu-eth1",
3845                 "ns-vld-id": "datanet",
3846                 "position": 1,
3847             },
3848         ]
3849 1         self.ns._partially_locate_vdu_interfaces(target_vdu)
3850 1         self.assertDictEqual(
3851             target_vdu["interfaces"][0],
3852             {
3853                 "name": "vdu-eth1",
3854                 "ns-vld-id": "datanet",
3855                 "position": 1,
3856             },
3857         )
3858 1         self.assertDictEqual(
3859             target_vdu["interfaces"][2],
3860             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3861         )
3862
3863 1     def test_partially_locate_vdu_interfaces_position_start_from_0(self):
3864         """Some interfaces have positions, position start from 0."""
3865 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3866 1         target_vdu["interfaces"] = [
3867             {
3868                 "name": "vdu-eth1",
3869                 "ns-vld-id": "net1",
3870             },
3871             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3872             {
3873                 "name": "vdu-eth3",
3874                 "ns-vld-id": "mgmtnet",
3875             },
3876             {
3877                 "name": "vdu-eth1",
3878                 "ns-vld-id": "datanet",
3879                 "position": 0,
3880             },
3881         ]
3882 1         self.ns._partially_locate_vdu_interfaces(target_vdu)
3883 1         self.assertDictEqual(
3884             target_vdu["interfaces"][0],
3885             {
3886                 "name": "vdu-eth1",
3887                 "ns-vld-id": "datanet",
3888                 "position": 0,
3889             },
3890         )
3891 1         self.assertDictEqual(
3892             target_vdu["interfaces"][3],
3893             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3894         )
3895
3896 1     def test_partially_locate_vdu_interfaces_wthout_position(self):
3897         """Interfaces do not have positions."""
3898 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3899 1         target_vdu["interfaces"] = interfaces_wthout_positions
3900 1         expected_result = deepcopy(target_vdu["interfaces"])
3901 1         self.ns._partially_locate_vdu_interfaces(target_vdu)
3902 1         self.assertEqual(target_vdu["interfaces"], expected_result)
3903
3904 1     def test_partially_locate_vdu_interfaces_all_has_position(self):
3905         """All interfaces have position."""
3906 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3907 1         target_vdu["interfaces"] = interfaces_wth_all_positions
3908 1         expected_interfaces = [
3909             {
3910                 "name": "vdu-eth2",
3911                 "ns-vld-id": "net2",
3912                 "position": 0,
3913             },
3914             {
3915                 "name": "vdu-eth3",
3916                 "ns-vld-id": "mgmtnet",
3917                 "position": 1,
3918             },
3919             {
3920                 "name": "vdu-eth1",
3921                 "ns-vld-id": "net1",
3922                 "position": 2,
3923             },
3924         ]
3925 1         self.ns._partially_locate_vdu_interfaces(target_vdu)
3926 1         self.assertEqual(target_vdu["interfaces"], expected_interfaces)
3927
3928 1     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3929 1     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3930 1     def test_prepare_vdu_cloud_init(self, mock_parse_jinja2, mock_get_cloud_init):
3931         """Target_vdu has cloud-init and boot-data-drive."""
3932 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3933 1         target_vdu["cloud-init"] = "sample-cloud-init-path"
3934 1         target_vdu["boot-data-drive"] = "vda"
3935 1         vdu2cloud_init = {}
3936 1         mock_get_cloud_init.return_value = cloud_init_content
3937 1         mock_parse_jinja2.return_value = user_data
3938 1         expected_result = {
3939             "user-data": user_data,
3940             "boot-data-drive": "vda",
3941         }
3942 1         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3943 1         self.assertDictEqual(result, expected_result)
3944 1         mock_get_cloud_init.assert_called_once_with(
3945             db=db, fs=fs, location="sample-cloud-init-path"
3946         )
3947 1         mock_parse_jinja2.assert_called_once_with(
3948             cloud_init_content=cloud_init_content,
3949             params=None,
3950             context="sample-cloud-init-path",
3951         )
3952
3953 1     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3954 1     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3955 1     def test_prepare_vdu_cloud_init_get_cloud_init_raise_exception(
3956         self, mock_parse_jinja2, mock_get_cloud_init
3957     ):
3958         """Target_vdu has cloud-init and boot-data-drive, get_cloud_init method raises exception."""
3959 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3960 1         target_vdu["cloud-init"] = "sample-cloud-init-path"
3961 1         target_vdu["boot-data-drive"] = "vda"
3962 1         vdu2cloud_init = {}
3963 1         mock_get_cloud_init.side_effect = NsException(
3964             "Mismatch descriptor for cloud init."
3965         )
3966
3967 1         with self.assertRaises(NsException) as err:
3968 1             self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3969 0             self.assertEqual(str(err.exception), "Mismatch descriptor for cloud init.")
3970
3971 1         mock_get_cloud_init.assert_called_once_with(
3972             db=db, fs=fs, location="sample-cloud-init-path"
3973         )
3974 1         mock_parse_jinja2.assert_not_called()
3975
3976 1     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3977 1     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3978 1     def test_prepare_vdu_cloud_init_parse_jinja2_raise_exception(
3979         self, mock_parse_jinja2, mock_get_cloud_init
3980     ):
3981         """Target_vdu has cloud-init and boot-data-drive, parse_jinja2 method raises exception."""
3982 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3983 1         target_vdu["cloud-init"] = "sample-cloud-init-path"
3984 1         target_vdu["boot-data-drive"] = "vda"
3985 1         vdu2cloud_init = {}
3986 1         mock_get_cloud_init.return_value = cloud_init_content
3987 1         mock_parse_jinja2.side_effect = NsException("Error parsing cloud-init content.")
3988
3989 1         with self.assertRaises(NsException) as err:
3990 1             self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3991 0             self.assertEqual(str(err.exception), "Error parsing cloud-init content.")
3992 1         mock_get_cloud_init.assert_called_once_with(
3993             db=db, fs=fs, location="sample-cloud-init-path"
3994         )
3995 1         mock_parse_jinja2.assert_called_once_with(
3996             cloud_init_content=cloud_init_content,
3997             params=None,
3998             context="sample-cloud-init-path",
3999         )
4000
4001 1     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
4002 1     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
4003 1     def test_prepare_vdu_cloud_init_vdu_wthout_boot_data_drive(
4004         self, mock_parse_jinja2, mock_get_cloud_init
4005     ):
4006         """Target_vdu has cloud-init but do not have boot-data-drive."""
4007 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4008 1         target_vdu["cloud-init"] = "sample-cloud-init-path"
4009 1         vdu2cloud_init = {}
4010 1         mock_get_cloud_init.return_value = cloud_init_content
4011 1         mock_parse_jinja2.return_value = user_data
4012 1         expected_result = {
4013             "user-data": user_data,
4014         }
4015 1         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
4016 1         self.assertDictEqual(result, expected_result)
4017 1         mock_get_cloud_init.assert_called_once_with(
4018             db=db, fs=fs, location="sample-cloud-init-path"
4019         )
4020 1         mock_parse_jinja2.assert_called_once_with(
4021             cloud_init_content=cloud_init_content,
4022             params=None,
4023             context="sample-cloud-init-path",
4024         )
4025
4026 1     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
4027 1     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
4028 1     def test_prepare_vdu_cloud_init_exists_in_vdu2cloud_init(
4029         self, mock_parse_jinja2, mock_get_cloud_init
4030     ):
4031         """Target_vdu has cloud-init, vdu2cloud_init dict has cloud-init_content."""
4032 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4033 1         target_vdu["cloud-init"] = "sample-cloud-init-path"
4034 1         target_vdu["boot-data-drive"] = "vda"
4035 1         vdu2cloud_init = {"sample-cloud-init-path": cloud_init_content}
4036 1         mock_parse_jinja2.return_value = user_data
4037 1         expected_result = {
4038             "user-data": user_data,
4039             "boot-data-drive": "vda",
4040         }
4041 1         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
4042 1         self.assertDictEqual(result, expected_result)
4043 1         mock_get_cloud_init.assert_not_called()
4044 1         mock_parse_jinja2.assert_called_once_with(
4045             cloud_init_content=cloud_init_content,
4046             params=None,
4047             context="sample-cloud-init-path",
4048         )
4049
4050 1     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
4051 1     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
4052 1     def test_prepare_vdu_cloud_init_no_cloud_init(
4053         self, mock_parse_jinja2, mock_get_cloud_init
4054     ):
4055         """Target_vdu do not have cloud-init."""
4056 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4057 1         target_vdu["boot-data-drive"] = "vda"
4058 1         vdu2cloud_init = {}
4059 1         expected_result = {
4060             "boot-data-drive": "vda",
4061         }
4062 1         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
4063 1         self.assertDictEqual(result, expected_result)
4064 1         mock_get_cloud_init.assert_not_called()
4065 1         mock_parse_jinja2.assert_not_called()
4066
4067 1     def test_check_vld_information_of_interfaces_ns_vld_vnf_vld_both_exist(self):
4068         """ns_vld and vnf_vld both exist."""
4069 1         interface = {
4070             "name": "vdu-eth0",
4071             "ns-vld-id": "mgmtnet",
4072             "vnf-vld-id": "mgmt_cp_int",
4073         }
4074 1         expected_result = f"{ns_preffix}:vld.mgmtnet"
4075 1         result = self.ns._check_vld_information_of_interfaces(
4076             interface, ns_preffix, vnf_preffix
4077         )
4078 1         self.assertEqual(result, expected_result)
4079
4080 1     def test_check_vld_information_of_interfaces_empty_interfaces(self):
4081         """Interface dict is empty."""
4082 1         interface = {}
4083 1         result = self.ns._check_vld_information_of_interfaces(
4084             interface, ns_preffix, vnf_preffix
4085         )
4086 1         self.assertEqual(result, "")
4087
4088 1     def test_check_vld_information_of_interfaces_has_only_vnf_vld(self):
4089         """Interface dict has only vnf_vld."""
4090 1         interface = {
4091             "name": "vdu-eth0",
4092             "vnf-vld-id": "mgmt_cp_int",
4093         }
4094 1         expected_result = f"{vnf_preffix}:vld.mgmt_cp_int"
4095 1         result = self.ns._check_vld_information_of_interfaces(
4096             interface, ns_preffix, vnf_preffix
4097         )
4098 1         self.assertEqual(result, expected_result)
4099
4100 1     def test_check_vld_information_of_interfaces_has_vnf_vld_wthout_vnf_prefix(
4101         self,
4102     ):
4103         """Interface dict has only vnf_vld but vnf_preffix does not exist."""
4104 1         interface = {
4105             "name": "vdu-eth0",
4106             "vnf-vld-id": "mgmt_cp_int",
4107         }
4108 1         vnf_preffix = None
4109 1         with self.assertRaises(Exception) as err:
4110 1             self.ns._check_vld_information_of_interfaces(
4111                 interface, ns_preffix, vnf_preffix
4112             )
4113 0             self.assertEqual(type(err), TypeError)
4114
4115 1     def test_prepare_interface_port_security_has_security_details(self):
4116         """Interface dict has port security details."""
4117 1         interface = {
4118             "name": "vdu-eth0",
4119             "ns-vld-id": "mgmtnet",
4120             "vnf-vld-id": "mgmt_cp_int",
4121             "port-security-enabled": True,
4122             "port-security-disable-strategy": "allow-address-pairs",
4123         }
4124 1         expected_interface = {
4125             "name": "vdu-eth0",
4126             "ns-vld-id": "mgmtnet",
4127             "vnf-vld-id": "mgmt_cp_int",
4128             "port_security": True,
4129             "port_security_disable_strategy": "allow-address-pairs",
4130         }
4131 1         self.ns._prepare_interface_port_security(interface)
4132 1         self.assertDictEqual(interface, expected_interface)
4133
4134 1     def test_prepare_interface_port_security_empty_interfaces(self):
4135         """Interface dict is empty."""
4136 1         interface = {}
4137 1         expected_interface = {}
4138 1         self.ns._prepare_interface_port_security(interface)
4139 1         self.assertDictEqual(interface, expected_interface)
4140
4141 1     def test_prepare_interface_port_security_wthout_port_security(self):
4142         """Interface dict does not have port security details."""
4143 1         interface = {
4144             "name": "vdu-eth0",
4145             "ns-vld-id": "mgmtnet",
4146             "vnf-vld-id": "mgmt_cp_int",
4147         }
4148 1         expected_interface = {
4149             "name": "vdu-eth0",
4150             "ns-vld-id": "mgmtnet",
4151             "vnf-vld-id": "mgmt_cp_int",
4152         }
4153 1         self.ns._prepare_interface_port_security(interface)
4154 1         self.assertDictEqual(interface, expected_interface)
4155
4156 1     def test_create_net_item_of_interface_floating_ip_port_security(self):
4157         """Interface dict has floating ip, port-security details."""
4158 1         interface = {
4159             "name": "vdu-eth0",
4160             "vcpi": "sample_vcpi",
4161             "port_security": True,
4162             "port_security_disable_strategy": "allow-address-pairs",
4163             "floating_ip": "10.1.1.12",
4164             "ns-vld-id": "mgmtnet",
4165             "vnf-vld-id": "mgmt_cp_int",
4166         }
4167 1         net_text = f"{ns_preffix}"
4168 1         expected_net_item = {
4169             "name": "vdu-eth0",
4170             "port_security": True,
4171             "port_security_disable_strategy": "allow-address-pairs",
4172             "floating_ip": "10.1.1.12",
4173             "net_id": f"TASK-{ns_preffix}",
4174             "type": "virtual",
4175         }
4176 1         result = self.ns._create_net_item_of_interface(interface, net_text)
4177 1         self.assertDictEqual(result, expected_net_item)
4178
4179 1     def test_create_net_item_of_interface_invalid_net_text(self):
4180         """net-text is invalid."""
4181 1         interface = {
4182             "name": "vdu-eth0",
4183             "vcpi": "sample_vcpi",
4184             "port_security": True,
4185             "port_security_disable_strategy": "allow-address-pairs",
4186             "floating_ip": "10.1.1.12",
4187             "ns-vld-id": "mgmtnet",
4188             "vnf-vld-id": "mgmt_cp_int",
4189         }
4190 1         net_text = None
4191 1         with self.assertRaises(TypeError):
4192 1             self.ns._create_net_item_of_interface(interface, net_text)
4193
4194 1     def test_create_net_item_of_interface_empty_interface(self):
4195         """Interface dict is empty."""
4196 1         interface = {}
4197 1         net_text = ns_preffix
4198 1         expected_net_item = {
4199             "net_id": f"TASK-{ns_preffix}",
4200             "type": "virtual",
4201         }
4202 1         result = self.ns._create_net_item_of_interface(interface, net_text)
4203 1         self.assertDictEqual(result, expected_net_item)
4204
4205 1     @patch("osm_ng_ro.ns.deep_get")
4206 1     def test_prepare_type_of_interface_type_sriov(self, mock_deep_get):
4207         """Interface type is SR-IOV."""
4208 1         interface = {
4209             "name": "vdu-eth0",
4210             "vcpi": "sample_vcpi",
4211             "port_security": True,
4212             "port_security_disable_strategy": "allow-address-pairs",
4213             "floating_ip": "10.1.1.12",
4214             "ns-vld-id": "mgmtnet",
4215             "vnf-vld-id": "mgmt_cp_int",
4216             "type": "SR-IOV",
4217         }
4218 1         mock_deep_get.return_value = "SR-IOV"
4219 1         net_text = ns_preffix
4220 1         net_item = {}
4221 1         expected_net_item = {
4222             "use": "data",
4223             "model": "SR-IOV",
4224             "type": "SR-IOV",
4225         }
4226 1         self.ns._prepare_type_of_interface(
4227             interface, tasks_by_target_record_id, net_text, net_item
4228         )
4229 1         self.assertDictEqual(net_item, expected_net_item)
4230 1         self.assertEqual(
4231             "data",
4232             tasks_by_target_record_id[net_text]["extra_dict"]["params"]["net_type"],
4233         )
4234 1         mock_deep_get.assert_called_once_with(
4235             tasks_by_target_record_id, net_text, "extra_dict", "params", "net_type"
4236         )
4237
4238 1     @patch("osm_ng_ro.ns.deep_get")
4239 1     def test_prepare_type_of_interface_type_pic_passthrough_deep_get_return_empty_dict(
4240         self, mock_deep_get
4241     ):
4242         """Interface type is PCI-PASSTHROUGH, deep_get method return empty dict."""
4243 1         interface = {
4244             "name": "vdu-eth0",
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",
4251             "type": "PCI-PASSTHROUGH",
4252         }
4253 1         mock_deep_get.return_value = {}
4254 1         tasks_by_target_record_id = {}
4255 1         net_text = ns_preffix
4256 1         net_item = {}
4257 1         expected_net_item = {
4258             "use": "data",
4259             "model": "PCI-PASSTHROUGH",
4260             "type": "PCI-PASSTHROUGH",
4261         }
4262 1         self.ns._prepare_type_of_interface(
4263             interface, tasks_by_target_record_id, net_text, net_item
4264         )
4265 1         self.assertDictEqual(net_item, expected_net_item)
4266 1         mock_deep_get.assert_called_once_with(
4267             tasks_by_target_record_id, net_text, "extra_dict", "params", "net_type"
4268         )
4269
4270 1     @patch("osm_ng_ro.ns.deep_get")
4271 1     def test_prepare_type_of_interface_type_mgmt(self, mock_deep_get):
4272         """Interface type is mgmt."""
4273 1         interface = {
4274             "name": "vdu-eth0",
4275             "vcpi": "sample_vcpi",
4276             "port_security": True,
4277             "port_security_disable_strategy": "allow-address-pairs",
4278             "floating_ip": "10.1.1.12",
4279             "ns-vld-id": "mgmtnet",
4280             "vnf-vld-id": "mgmt_cp_int",
4281             "type": "OM-MGMT",
4282         }
4283 1         tasks_by_target_record_id = {}
4284 1         net_text = ns_preffix
4285 1         net_item = {}
4286 1         expected_net_item = {
4287             "use": "mgmt",
4288         }
4289 1         self.ns._prepare_type_of_interface(
4290             interface, tasks_by_target_record_id, net_text, net_item
4291         )
4292 1         self.assertDictEqual(net_item, expected_net_item)
4293 1         mock_deep_get.assert_not_called()
4294
4295 1     @patch("osm_ng_ro.ns.deep_get")
4296 1     def test_prepare_type_of_interface_type_bridge(self, mock_deep_get):
4297         """Interface type is bridge."""
4298 1         interface = {
4299             "name": "vdu-eth0",
4300             "vcpi": "sample_vcpi",
4301             "port_security": True,
4302             "port_security_disable_strategy": "allow-address-pairs",
4303             "floating_ip": "10.1.1.12",
4304             "ns-vld-id": "mgmtnet",
4305             "vnf-vld-id": "mgmt_cp_int",
4306         }
4307 1         tasks_by_target_record_id = {}
4308 1         net_text = ns_preffix
4309 1         net_item = {}
4310 1         expected_net_item = {
4311             "use": "bridge",
4312             "model": None,
4313         }
4314 1         self.ns._prepare_type_of_interface(
4315             interface, tasks_by_target_record_id, net_text, net_item
4316         )
4317 1         self.assertDictEqual(net_item, expected_net_item)
4318 1         mock_deep_get.assert_not_called()
4319
4320 1     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4321 1     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4322 1     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4323 1     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4324 1     def test_prepare_vdu_interfaces(
4325         self,
4326         mock_type_of_interface,
4327         mock_item_of_interface,
4328         mock_port_security,
4329         mock_vld_information_of_interface,
4330     ):
4331         """Prepare vdu interfaces successfully."""
4332 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4333 1         interface_1 = {
4334             "name": "vdu-eth1",
4335             "ns-vld-id": "net1",
4336             "ip-address": "13.2.12.31",
4337             "mgmt-interface": True,
4338         }
4339 1         interface_2 = {
4340             "name": "vdu-eth2",
4341             "vnf-vld-id": "net2",
4342             "mac-address": "d0:94:66:ed:fc:e2",
4343         }
4344 1         interface_3 = {
4345             "name": "vdu-eth3",
4346             "ns-vld-id": "mgmtnet",
4347         }
4348 1         target_vdu["interfaces"] = [interface_1, interface_2, interface_3]
4349 1         extra_dict = {
4350             "params": "test_params",
4351             "find_params": "test_find_params",
4352             "depends_on": [],
4353         }
4354
4355 1         net_text_1 = f"{ns_preffix}:net1"
4356 1         net_text_2 = f"{vnf_preffix}:net2"
4357 1         net_text_3 = f"{ns_preffix}:mgmtnet"
4358 1         net_item_1 = {
4359             "name": "vdu-eth1",
4360             "net_id": f"TASK-{ns_preffix}",
4361             "type": "virtual",
4362         }
4363 1         net_item_2 = {
4364             "name": "vdu-eth2",
4365             "net_id": f"TASK-{ns_preffix}",
4366             "type": "virtual",
4367         }
4368 1         net_item_3 = {
4369             "name": "vdu-eth3",
4370             "net_id": f"TASK-{ns_preffix}",
4371             "type": "virtual",
4372         }
4373 1         mock_item_of_interface.side_effect = [net_item_1, net_item_2, net_item_3]
4374 1         mock_vld_information_of_interface.side_effect = [
4375             net_text_1,
4376             net_text_2,
4377             net_text_3,
4378         ]
4379 1         net_list = []
4380 1         expected_extra_dict = {
4381             "params": "test_params",
4382             "find_params": "test_find_params",
4383             "depends_on": [net_text_1, net_text_2, net_text_3],
4384             "mgmt_vdu_interface": 0,
4385         }
4386 1         updated_net_item1 = deepcopy(net_item_1)
4387 1         updated_net_item1.update({"ip_address": "13.2.12.31"})
4388 1         updated_net_item2 = deepcopy(net_item_2)
4389 1         updated_net_item2.update({"mac_address": "d0:94:66:ed:fc:e2"})
4390 1         expected_net_list = [updated_net_item1, updated_net_item2, net_item_3]
4391 1         self.ns._prepare_vdu_interfaces(
4392             target_vdu,
4393             extra_dict,
4394             ns_preffix,
4395             vnf_preffix,
4396             self.logger,
4397             tasks_by_target_record_id,
4398             net_list,
4399         )
4400 1         _call_mock_vld_information_of_interface = (
4401             mock_vld_information_of_interface.call_args_list
4402         )
4403 1         self.assertEqual(
4404             _call_mock_vld_information_of_interface[0][0],
4405             (interface_1, ns_preffix, vnf_preffix),
4406         )
4407 1         self.assertEqual(
4408             _call_mock_vld_information_of_interface[1][0],
4409             (interface_2, ns_preffix, vnf_preffix),
4410         )
4411 1         self.assertEqual(
4412             _call_mock_vld_information_of_interface[2][0],
4413             (interface_3, ns_preffix, vnf_preffix),
4414         )
4415
4416 1         _call_mock_port_security = mock_port_security.call_args_list
4417 1         self.assertEqual(_call_mock_port_security[0].args[0], interface_1)
4418 1         self.assertEqual(_call_mock_port_security[1].args[0], interface_2)
4419 1         self.assertEqual(_call_mock_port_security[2].args[0], interface_3)
4420
4421 1         _call_mock_item_of_interface = mock_item_of_interface.call_args_list
4422 1         self.assertEqual(_call_mock_item_of_interface[0][0], (interface_1, net_text_1))
4423 1         self.assertEqual(_call_mock_item_of_interface[1][0], (interface_2, net_text_2))
4424 1         self.assertEqual(_call_mock_item_of_interface[2][0], (interface_3, net_text_3))
4425
4426 1         _call_mock_type_of_interface = mock_type_of_interface.call_args_list
4427 1         self.assertEqual(
4428             _call_mock_type_of_interface[0][0],
4429             (interface_1, tasks_by_target_record_id, net_text_1, net_item_1),
4430         )
4431 1         self.assertEqual(
4432             _call_mock_type_of_interface[1][0],
4433             (interface_2, tasks_by_target_record_id, net_text_2, net_item_2),
4434         )
4435 1         self.assertEqual(
4436             _call_mock_type_of_interface[2][0],
4437             (interface_3, tasks_by_target_record_id, net_text_3, net_item_3),
4438         )
4439 1         self.assertEqual(net_list, expected_net_list)
4440 1         self.assertEqual(extra_dict, expected_extra_dict)
4441 1         self.logger.error.assert_not_called()
4442
4443 1     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4444 1     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4445 1     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4446 1     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4447 1     def test_prepare_vdu_interfaces_create_net_item_raise_exception(
4448         self,
4449         mock_type_of_interface,
4450         mock_item_of_interface,
4451         mock_port_security,
4452         mock_vld_information_of_interface,
4453     ):
4454         """Prepare vdu interfaces, create_net_item_of_interface method raise exception."""
4455 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4456 1         interface_1 = {
4457             "name": "vdu-eth1",
4458             "ns-vld-id": "net1",
4459             "ip-address": "13.2.12.31",
4460             "mgmt-interface": True,
4461         }
4462 1         interface_2 = {
4463             "name": "vdu-eth2",
4464             "vnf-vld-id": "net2",
4465             "mac-address": "d0:94:66:ed:fc:e2",
4466         }
4467 1         interface_3 = {
4468             "name": "vdu-eth3",
4469             "ns-vld-id": "mgmtnet",
4470         }
4471 1         target_vdu["interfaces"] = [interface_1, interface_2, interface_3]
4472 1         extra_dict = {
4473             "params": "test_params",
4474             "find_params": "test_find_params",
4475             "depends_on": [],
4476         }
4477 1         net_text_1 = f"{ns_preffix}:net1"
4478 1         mock_item_of_interface.side_effect = [TypeError, TypeError, TypeError]
4479
4480 1         mock_vld_information_of_interface.side_effect = [net_text_1]
4481 1         net_list = []
4482 1         expected_extra_dict = {
4483             "params": "test_params",
4484             "find_params": "test_find_params",
4485             "depends_on": [net_text_1],
4486         }
4487 1         with self.assertRaises(TypeError):
4488 1             self.ns._prepare_vdu_interfaces(
4489                 target_vdu,
4490                 extra_dict,
4491                 ns_preffix,
4492                 vnf_preffix,
4493                 self.logger,
4494                 tasks_by_target_record_id,
4495                 net_list,
4496             )
4497
4498 1         _call_mock_vld_information_of_interface = (
4499             mock_vld_information_of_interface.call_args_list
4500         )
4501 1         self.assertEqual(
4502             _call_mock_vld_information_of_interface[0][0],
4503             (interface_1, ns_preffix, vnf_preffix),
4504         )
4505
4506 1         _call_mock_port_security = mock_port_security.call_args_list
4507 1         self.assertEqual(_call_mock_port_security[0].args[0], interface_1)
4508
4509 1         _call_mock_item_of_interface = mock_item_of_interface.call_args_list
4510 1         self.assertEqual(_call_mock_item_of_interface[0][0], (interface_1, net_text_1))
4511
4512 1         mock_type_of_interface.assert_not_called()
4513 1         self.logger.error.assert_not_called()
4514 1         self.assertEqual(net_list, [])
4515 1         self.assertEqual(extra_dict, expected_extra_dict)
4516
4517 1     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4518 1     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4519 1     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4520 1     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4521 1     def test_prepare_vdu_interfaces_vld_information_is_empty(
4522         self,
4523         mock_type_of_interface,
4524         mock_item_of_interface,
4525         mock_port_security,
4526         mock_vld_information_of_interface,
4527     ):
4528         """Prepare vdu interfaces, check_vld_information_of_interface method returns empty result."""
4529 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4530 1         interface_1 = {
4531             "name": "vdu-eth1",
4532             "ns-vld-id": "net1",
4533             "ip-address": "13.2.12.31",
4534             "mgmt-interface": True,
4535         }
4536 1         interface_2 = {
4537             "name": "vdu-eth2",
4538             "vnf-vld-id": "net2",
4539             "mac-address": "d0:94:66:ed:fc:e2",
4540         }
4541 1         interface_3 = {
4542             "name": "vdu-eth3",
4543             "ns-vld-id": "mgmtnet",
4544         }
4545 1         target_vdu["interfaces"] = [interface_1, interface_2, interface_3]
4546 1         extra_dict = {
4547             "params": "test_params",
4548             "find_params": "test_find_params",
4549             "depends_on": [],
4550         }
4551 1         mock_vld_information_of_interface.side_effect = ["", "", ""]
4552 1         net_list = []
4553 1         self.ns._prepare_vdu_interfaces(
4554             target_vdu,
4555             extra_dict,
4556             ns_preffix,
4557             vnf_preffix,
4558             self.logger,
4559             tasks_by_target_record_id,
4560             net_list,
4561         )
4562
4563 1         _call_mock_vld_information_of_interface = (
4564             mock_vld_information_of_interface.call_args_list
4565         )
4566 1         self.assertEqual(
4567             _call_mock_vld_information_of_interface[0][0],
4568             (interface_1, ns_preffix, vnf_preffix),
4569         )
4570 1         self.assertEqual(
4571             _call_mock_vld_information_of_interface[1][0],
4572             (interface_2, ns_preffix, vnf_preffix),
4573         )
4574 1         self.assertEqual(
4575             _call_mock_vld_information_of_interface[2][0],
4576             (interface_3, ns_preffix, vnf_preffix),
4577         )
4578
4579 1         _call_logger = self.logger.error.call_args_list
4580 1         self.assertEqual(
4581             _call_logger[0][0],
4582             ("Interface 0 from vdu several_volumes-VM not connected to any vld",),
4583         )
4584 1         self.assertEqual(
4585             _call_logger[1][0],
4586             ("Interface 1 from vdu several_volumes-VM not connected to any vld",),
4587         )
4588 1         self.assertEqual(
4589             _call_logger[2][0],
4590             ("Interface 2 from vdu several_volumes-VM not connected to any vld",),
4591         )
4592 1         self.assertEqual(net_list, [])
4593 1         self.assertEqual(
4594             extra_dict,
4595             {
4596                 "params": "test_params",
4597                 "find_params": "test_find_params",
4598                 "depends_on": [],
4599             },
4600         )
4601
4602 1         mock_item_of_interface.assert_not_called()
4603 1         mock_port_security.assert_not_called()
4604 1         mock_type_of_interface.assert_not_called()
4605
4606 1     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4607 1     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4608 1     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4609 1     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4610 1     def test_prepare_vdu_interfaces_empty_interface_list(
4611         self,
4612         mock_type_of_interface,
4613         mock_item_of_interface,
4614         mock_port_security,
4615         mock_vld_information_of_interface,
4616     ):
4617         """Prepare vdu interfaces, interface list is empty."""
4618 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4619 1         target_vdu["interfaces"] = []
4620 1         extra_dict = {}
4621 1         net_list = []
4622 1         self.ns._prepare_vdu_interfaces(
4623             target_vdu,
4624             extra_dict,
4625             ns_preffix,
4626             vnf_preffix,
4627             self.logger,
4628             tasks_by_target_record_id,
4629             net_list,
4630         )
4631 1         mock_type_of_interface.assert_not_called()
4632 1         mock_vld_information_of_interface.assert_not_called()
4633 1         mock_item_of_interface.assert_not_called()
4634 1         mock_port_security.assert_not_called()
4635
4636 1     def test_prepare_vdu_ssh_keys(self):
4637         """Target_vdu has ssh-keys and ro_nsr_public_key exists."""
4638 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4639 1         target_vdu["ssh-keys"] = ["sample-ssh-key"]
4640 1         ro_nsr_public_key = {"public_key": "path_of_public_key"}
4641 1         target_vdu["ssh-access-required"] = True
4642 1         cloud_config = {}
4643 1         expected_cloud_config = {
4644             "key-pairs": ["sample-ssh-key", {"public_key": "path_of_public_key"}]
4645         }
4646 1         self.ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
4647 1         self.assertDictEqual(cloud_config, expected_cloud_config)
4648
4649 1     def test_prepare_vdu_ssh_keys_target_vdu_wthout_ssh_keys(self):
4650         """Target_vdu does not have ssh-keys."""
4651 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4652 1         ro_nsr_public_key = {"public_key": "path_of_public_key"}
4653 1         target_vdu["ssh-access-required"] = True
4654 1         cloud_config = {}
4655 1         expected_cloud_config = {"key-pairs": [{"public_key": "path_of_public_key"}]}
4656 1         self.ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
4657 1         self.assertDictEqual(cloud_config, expected_cloud_config)
4658
4659 1     def test_prepare_vdu_ssh_keys_ssh_access_is_not_required(self):
4660         """Target_vdu has ssh-keys, ssh-access is not required."""
4661 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4662 1         target_vdu["ssh-keys"] = ["sample-ssh-key"]
4663 1         ro_nsr_public_key = {"public_key": "path_of_public_key"}
4664 1         target_vdu["ssh-access-required"] = False
4665 1         cloud_config = {}
4666 1         expected_cloud_config = {"key-pairs": ["sample-ssh-key"]}
4667 1         self.ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
4668 1         self.assertDictEqual(cloud_config, expected_cloud_config)
4669
4670 1     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4671 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4672 1     def test_add_persistent_root_disk_to_disk_list_keep_false(
4673         self, mock_volume_keeping_required, mock_select_persistent_root_disk
4674     ):
4675         """Add persistent root disk to disk_list, keep volume set to False."""
4676 1         root_disk = {
4677             "id": "persistent-root-volume",
4678             "type-of-storage": "persistent-storage:persistent-storage",
4679             "size-of-storage": "10",
4680         }
4681 1         mock_select_persistent_root_disk.return_value = root_disk
4682 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
4683 1         vnfd["virtual-storage-desc"][1] = root_disk
4684 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4685 1         persistent_root_disk = {}
4686 1         disk_list = []
4687 1         mock_volume_keeping_required.return_value = False
4688 1         expected_disk_list = [
4689             {
4690                 "image_id": "ubuntu20.04",
4691                 "size": "10",
4692                 "keep": False,
4693             }
4694         ]
4695 1         self.ns._add_persistent_root_disk_to_disk_list(
4696             vnfd, target_vdu, persistent_root_disk, disk_list
4697         )
4698 1         self.assertEqual(disk_list, expected_disk_list)
4699 1         mock_select_persistent_root_disk.assert_called_once()
4700 1         mock_volume_keeping_required.assert_called_once()
4701
4702 1     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4703 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4704 1     def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises(
4705         self, mock_volume_keeping_required, mock_select_persistent_root_disk
4706     ):
4707         """Add persistent root disk to disk_list"""
4708 1         root_disk = {
4709             "id": "persistent-root-volume",
4710             "type-of-storage": "persistent-storage:persistent-storage",
4711             "size-of-storage": "10",
4712         }
4713 1         mock_select_persistent_root_disk.side_effect = AttributeError
4714 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
4715 1         vnfd["virtual-storage-desc"][1] = root_disk
4716 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4717 1         persistent_root_disk = {}
4718 1         disk_list = []
4719 1         with self.assertRaises(AttributeError):
4720 1             self.ns._add_persistent_root_disk_to_disk_list(
4721                 vnfd, target_vdu, persistent_root_disk, disk_list
4722             )
4723 1         self.assertEqual(disk_list, [])
4724 1         mock_select_persistent_root_disk.assert_called_once()
4725 1         mock_volume_keeping_required.assert_not_called()
4726
4727 1     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4728 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4729 1     def test_add_persistent_root_disk_to_disk_list_keep_true(
4730         self, mock_volume_keeping_required, mock_select_persistent_root_disk
4731     ):
4732         """Add persistent root disk, keeo volume set to True."""
4733 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
4734 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4735 1         mock_volume_keeping_required.return_value = True
4736 1         root_disk = {
4737             "id": "persistent-root-volume",
4738             "type-of-storage": "persistent-storage:persistent-storage",
4739             "size-of-storage": "10",
4740             "vdu-storage-requirements": [
4741                 {"key": "keep-volume", "value": "true"},
4742             ],
4743         }
4744 1         mock_select_persistent_root_disk.return_value = root_disk
4745 1         persistent_root_disk = {}
4746 1         disk_list = []
4747 1         expected_disk_list = [
4748             {
4749                 "image_id": "ubuntu20.04",
4750                 "size": "10",
4751                 "keep": True,
4752             }
4753         ]
4754 1         self.ns._add_persistent_root_disk_to_disk_list(
4755             vnfd, target_vdu, persistent_root_disk, disk_list
4756         )
4757 1         self.assertEqual(disk_list, expected_disk_list)
4758 1         mock_volume_keeping_required.assert_called_once_with(root_disk)
4759
4760 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4761 1     def test_add_persistent_ordinary_disk_to_disk_list(
4762         self, mock_volume_keeping_required
4763     ):
4764         """Add persistent ordinary disk, keeo volume set to True."""
4765 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4766 1         mock_volume_keeping_required.return_value = False
4767 1         persistent_root_disk = {
4768             "persistent-root-volume": {
4769                 "image_id": "ubuntu20.04",
4770                 "size": "10",
4771                 "keep": True,
4772             }
4773         }
4774 1         ordinary_disk = {
4775             "id": "persistent-volume2",
4776             "type-of-storage": "persistent-storage:persistent-storage",
4777             "size-of-storage": "10",
4778         }
4779 1         persistent_ordinary_disk = {}
4780 1         disk_list = []
4781 1         extra_dict = {}
4782 1         expected_disk_list = [
4783             {
4784                 "size": "10",
4785                 "keep": False,
4786                 "multiattach": False,
4787                 "name": "persistent-volume2",
4788             }
4789         ]
4790 1         self.ns._add_persistent_ordinary_disks_to_disk_list(
4791             target_vdu,
4792             persistent_root_disk,
4793             persistent_ordinary_disk,
4794             disk_list,
4795             extra_dict,
4796         )
4797 1         self.assertEqual(disk_list, expected_disk_list)
4798 1         mock_volume_keeping_required.assert_called_once_with(ordinary_disk)
4799
4800 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4801 1     def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(
4802         self, mock_volume_keeping_required
4803     ):
4804         """Add persistent ordinary disk, vsd id is in root_disk dict."""
4805 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4806 1         mock_volume_keeping_required.return_value = False
4807 1         persistent_root_disk = {
4808             "persistent-root-volume": {
4809                 "image_id": "ubuntu20.04",
4810                 "size": "10",
4811                 "keep": True,
4812             },
4813             "persistent-volume2": {
4814                 "size": "10",
4815             },
4816         }
4817 1         persistent_ordinary_disk = {}
4818 1         disk_list = []
4819 1         extra_dict = {}
4820
4821 1         self.ns._add_persistent_ordinary_disks_to_disk_list(
4822             target_vdu,
4823             persistent_root_disk,
4824             persistent_ordinary_disk,
4825             disk_list,
4826             extra_dict,
4827         )
4828 1         self.assertEqual(disk_list, [])
4829 1         mock_volume_keeping_required.assert_not_called()
4830
4831 1     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4832 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4833 1     def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage(
4834         self, mock_volume_keeping_required, mock_select_persistent_root_disk
4835     ):
4836         """VNFD does not have persistent storage."""
4837 1         vnfd = deepcopy(vnfd_wthout_persistent_storage)
4838 1         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4839 1         mock_select_persistent_root_disk.return_value = None
4840 1         persistent_root_disk = {}
4841 1         disk_list = []
4842 1         self.ns._add_persistent_root_disk_to_disk_list(
4843             vnfd, target_vdu, persistent_root_disk, disk_list
4844         )
4845 1         self.assertEqual(disk_list, [])
4846 1         self.assertEqual(mock_select_persistent_root_disk.call_count, 2)
4847 1         mock_volume_keeping_required.assert_not_called()
4848
4849 1     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4850 1     @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
4851 1     def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk(
4852         self, mock_volume_keeping_required, mock_select_persistent_root_disk
4853     ):
4854         """Persistent_root_disk dict is empty."""
4855 1         vnfd = deepcopy(vnfd_wthout_persistent_storage)
4856 1         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4857 1         mock_select_persistent_root_disk.return_value = None
4858 1         persistent_root_disk = {}
4859 1         disk_list = []
4860 1         self.ns._add_persistent_root_disk_to_disk_list(
4861             vnfd, target_vdu, persistent_root_disk, disk_list
4862         )
4863 1         self.assertEqual(disk_list, [])
4864 1         self.assertEqual(mock_select_persistent_root_disk.call_count, 2)
4865 1         mock_volume_keeping_required.assert_not_called()
4866
4867 1     def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self):
4868         """Invalid extra dict."""
4869 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4870 1         target_vdu["affinity-or-anti-affinity-group-id"] = "sample_affinity-group-id"
4871 1         extra_dict = {}
4872 1         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4873 1         with self.assertRaises(NsException) as err:
4874 1             self.ns._prepare_vdu_affinity_group_list(target_vdu, extra_dict, ns_preffix)
4875 0             self.assertEqual(str(err.exception), "Invalid extra_dict format.")
4876
4877 1     def test_prepare_vdu_affinity_group_list_one_affinity_group(self):
4878         """There is one affinity-group."""
4879 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4880 1         target_vdu["affinity-or-anti-affinity-group-id"] = ["sample_affinity-group-id"]
4881 1         extra_dict = {"depends_on": []}
4882 1         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4883 1         affinity_group_txt = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.sample_affinity-group-id"
4884 1         expected_result = [{"affinity_group_id": "TASK-" + affinity_group_txt}]
4885 1         expected_extra_dict = {"depends_on": [affinity_group_txt]}
4886 1         result = self.ns._prepare_vdu_affinity_group_list(
4887             target_vdu, extra_dict, ns_preffix
4888         )
4889 1         self.assertDictEqual(extra_dict, expected_extra_dict)
4890 1         self.assertEqual(result, expected_result)
4891
4892 1     def test_prepare_vdu_affinity_group_list_several_affinity_groups(self):
4893         """There are two affinity-groups."""
4894 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4895 1         target_vdu["affinity-or-anti-affinity-group-id"] = [
4896             "affinity-group-id1",
4897             "affinity-group-id2",
4898         ]
4899 1         extra_dict = {"depends_on": []}
4900 1         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4901 1         affinity_group_txt1 = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id1"
4902 1         affinity_group_txt2 = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id2"
4903 1         expected_result = [
4904             {"affinity_group_id": "TASK-" + affinity_group_txt1},
4905             {"affinity_group_id": "TASK-" + affinity_group_txt2},
4906         ]
4907 1         expected_extra_dict = {"depends_on": [affinity_group_txt1, affinity_group_txt2]}
4908 1         result = self.ns._prepare_vdu_affinity_group_list(
4909             target_vdu, extra_dict, ns_preffix
4910         )
4911 1         self.assertDictEqual(extra_dict, expected_extra_dict)
4912 1         self.assertEqual(result, expected_result)
4913
4914 1     def test_prepare_vdu_affinity_group_list_no_affinity_group(self):
4915         """There is not any affinity-group."""
4916 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4917 1         extra_dict = {"depends_on": []}
4918 1         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4919 1         result = self.ns._prepare_vdu_affinity_group_list(
4920             target_vdu, extra_dict, ns_preffix
4921         )
4922 1         self.assertDictEqual(extra_dict, {"depends_on": []})
4923 1         self.assertEqual(result, [])
4924
4925 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4926 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4927 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4928 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4929 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4930 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4931 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4932 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4933 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4934 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4935 1     def test_process_vdu_params_with_inst_vol_list(
4936         self,
4937         mock_prepare_vdu_affinity_group_list,
4938         mock_add_persistent_ordinary_disks_to_disk_list,
4939         mock_add_persistent_root_disk_to_disk_list,
4940         mock_find_persistent_volumes,
4941         mock_find_persistent_root_volumes,
4942         mock_prepare_vdu_ssh_keys,
4943         mock_prepare_vdu_cloud_init,
4944         mock_prepare_vdu_interfaces,
4945         mock_locate_vdu_interfaces,
4946         mock_sort_vdu_interfaces,
4947     ):
4948         """Instantiation volume list is empty."""
4949 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4950
4951 1         target_vdu["interfaces"] = interfaces_wth_all_positions
4952
4953 1         vdu_instantiation_vol_list = [
4954             {
4955                 "vim-volume-id": vim_volume_id,
4956                 "name": "persistent-volume2",
4957             }
4958         ]
4959 1         target_vdu["additionalParams"] = {
4960             "OSM": {"vdu_volumes": vdu_instantiation_vol_list}
4961         }
4962 1         mock_prepare_vdu_cloud_init.return_value = {}
4963 1         mock_prepare_vdu_affinity_group_list.return_value = []
4964 1         persistent_root_disk = {
4965             "persistent-root-volume": {
4966                 "image_id": "ubuntu20.04",
4967                 "size": "10",
4968             }
4969         }
4970 1         mock_find_persistent_root_volumes.return_value = persistent_root_disk
4971
4972 1         new_kwargs = deepcopy(kwargs)
4973 1         new_kwargs.update(
4974             {
4975                 "vnfr_id": vnfr_id,
4976                 "nsr_id": nsr_id,
4977                 "tasks_by_target_record_id": {},
4978                 "logger": "logger",
4979             }
4980         )
4981 1         expected_extra_dict_copy = deepcopy(expected_extra_dict)
4982 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
4983 1         db.get_one.return_value = vnfd
4984 1         result = Ns._process_vdu_params(
4985             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
4986         )
4987 1         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
4988 1         mock_locate_vdu_interfaces.assert_not_called()
4989 1         mock_prepare_vdu_cloud_init.assert_called_once()
4990 1         mock_add_persistent_root_disk_to_disk_list.assert_not_called()
4991 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_not_called()
4992 1         mock_prepare_vdu_interfaces.assert_called_once_with(
4993             target_vdu,
4994             expected_extra_dict_copy,
4995             ns_preffix,
4996             vnf_preffix,
4997             "logger",
4998             {},
4999             [],
5000         )
5001 1         self.assertDictEqual(result, expected_extra_dict_copy)
5002 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5003 1         mock_prepare_vdu_affinity_group_list.assert_called_once()
5004 1         mock_find_persistent_volumes.assert_called_once_with(
5005             persistent_root_disk, target_vdu, vdu_instantiation_vol_list, []
5006         )
5007
5008 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5009 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5010 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5011 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5012 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5013 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5014 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5015 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5016 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5017 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5018 1     def test_process_vdu_params_wth_affinity_groups(
5019         self,
5020         mock_prepare_vdu_affinity_group_list,
5021         mock_add_persistent_ordinary_disks_to_disk_list,
5022         mock_add_persistent_root_disk_to_disk_list,
5023         mock_find_persistent_volumes,
5024         mock_find_persistent_root_volumes,
5025         mock_prepare_vdu_ssh_keys,
5026         mock_prepare_vdu_cloud_init,
5027         mock_prepare_vdu_interfaces,
5028         mock_locate_vdu_interfaces,
5029         mock_sort_vdu_interfaces,
5030     ):
5031         """There is cloud-config."""
5032 1         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
5033
5034 1         self.maxDiff = None
5035 1         target_vdu["interfaces"] = interfaces_wth_all_positions
5036 1         mock_prepare_vdu_cloud_init.return_value = {}
5037 1         mock_prepare_vdu_affinity_group_list.return_value = [
5038             "affinity_group_1",
5039             "affinity_group_2",
5040         ]
5041
5042 1         new_kwargs = deepcopy(kwargs)
5043 1         new_kwargs.update(
5044             {
5045                 "vnfr_id": vnfr_id,
5046                 "nsr_id": nsr_id,
5047                 "tasks_by_target_record_id": {},
5048                 "logger": "logger",
5049             }
5050         )
5051 1         expected_extra_dict3 = deepcopy(expected_extra_dict2)
5052 1         expected_extra_dict3["params"]["affinity_group_list"] = [
5053             "affinity_group_1",
5054             "affinity_group_2",
5055         ]
5056 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
5057 1         db.get_one.return_value = vnfd
5058 1         result = Ns._process_vdu_params(
5059             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5060         )
5061 1         self.assertDictEqual(result, expected_extra_dict3)
5062 1         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
5063 1         mock_locate_vdu_interfaces.assert_not_called()
5064 1         mock_prepare_vdu_cloud_init.assert_called_once()
5065 1         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5066 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5067 1         mock_prepare_vdu_interfaces.assert_called_once_with(
5068             target_vdu,
5069             expected_extra_dict3,
5070             ns_preffix,
5071             vnf_preffix,
5072             "logger",
5073             {},
5074             [],
5075         )
5076
5077 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5078 1         mock_prepare_vdu_affinity_group_list.assert_called_once()
5079 1         mock_find_persistent_volumes.assert_not_called()
5080
5081 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5082 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5083 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5084 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5085 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5086 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5087 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5088 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5089 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5090 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5091 1     def test_process_vdu_params_wth_cloud_config(
5092         self,
5093         mock_prepare_vdu_affinity_group_list,
5094         mock_add_persistent_ordinary_disks_to_disk_list,
5095         mock_add_persistent_root_disk_to_disk_list,
5096         mock_find_persistent_volumes,
5097         mock_find_persistent_root_volumes,
5098         mock_prepare_vdu_ssh_keys,
5099         mock_prepare_vdu_cloud_init,
5100         mock_prepare_vdu_interfaces,
5101         mock_locate_vdu_interfaces,
5102         mock_sort_vdu_interfaces,
5103     ):
5104         """There is cloud-config."""
5105 1         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
5106
5107 1         self.maxDiff = None
5108 1         target_vdu["interfaces"] = interfaces_wth_all_positions
5109 1         mock_prepare_vdu_cloud_init.return_value = {
5110             "user-data": user_data,
5111             "boot-data-drive": "vda",
5112         }
5113 1         mock_prepare_vdu_affinity_group_list.return_value = []
5114
5115 1         new_kwargs = deepcopy(kwargs)
5116 1         new_kwargs.update(
5117             {
5118                 "vnfr_id": vnfr_id,
5119                 "nsr_id": nsr_id,
5120                 "tasks_by_target_record_id": {},
5121                 "logger": "logger",
5122             }
5123         )
5124 1         expected_extra_dict3 = deepcopy(expected_extra_dict2)
5125 1         expected_extra_dict3["params"]["cloud_config"] = {
5126             "user-data": user_data,
5127             "boot-data-drive": "vda",
5128         }
5129 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
5130 1         db.get_one.return_value = vnfd
5131 1         result = Ns._process_vdu_params(
5132             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5133         )
5134 1         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
5135 1         mock_locate_vdu_interfaces.assert_not_called()
5136 1         mock_prepare_vdu_cloud_init.assert_called_once()
5137 1         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5138 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5139 1         mock_prepare_vdu_interfaces.assert_called_once_with(
5140             target_vdu,
5141             expected_extra_dict3,
5142             ns_preffix,
5143             vnf_preffix,
5144             "logger",
5145             {},
5146             [],
5147         )
5148 1         self.assertDictEqual(result, expected_extra_dict3)
5149 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(
5150             target_vdu, None, {"user-data": user_data, "boot-data-drive": "vda"}
5151         )
5152 1         mock_prepare_vdu_affinity_group_list.assert_called_once()
5153 1         mock_find_persistent_volumes.assert_not_called()
5154
5155 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5156 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5157 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5158 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5159 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5160 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5161 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5162 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5163 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5164 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5165 1     def test_process_vdu_params_wthout_persistent_storage(
5166         self,
5167         mock_prepare_vdu_affinity_group_list,
5168         mock_add_persistent_ordinary_disks_to_disk_list,
5169         mock_add_persistent_root_disk_to_disk_list,
5170         mock_find_persistent_volumes,
5171         mock_find_persistent_root_volumes,
5172         mock_prepare_vdu_ssh_keys,
5173         mock_prepare_vdu_cloud_init,
5174         mock_prepare_vdu_interfaces,
5175         mock_locate_vdu_interfaces,
5176         mock_sort_vdu_interfaces,
5177     ):
5178         """There is not any persistent storage."""
5179 1         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
5180
5181 1         self.maxDiff = None
5182 1         target_vdu["interfaces"] = interfaces_wth_all_positions
5183 1         mock_prepare_vdu_cloud_init.return_value = {}
5184 1         mock_prepare_vdu_affinity_group_list.return_value = []
5185
5186 1         new_kwargs = deepcopy(kwargs)
5187 1         new_kwargs.update(
5188             {
5189                 "vnfr_id": vnfr_id,
5190                 "nsr_id": nsr_id,
5191                 "tasks_by_target_record_id": {},
5192                 "logger": "logger",
5193             }
5194         )
5195 1         expected_extra_dict_copy = deepcopy(expected_extra_dict2)
5196 1         vnfd = deepcopy(vnfd_wthout_persistent_storage)
5197 1         db.get_one.return_value = vnfd
5198 1         result = Ns._process_vdu_params(
5199             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5200         )
5201 1         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
5202 1         mock_locate_vdu_interfaces.assert_not_called()
5203 1         mock_prepare_vdu_cloud_init.assert_called_once()
5204 1         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5205 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5206 1         mock_prepare_vdu_interfaces.assert_called_once_with(
5207             target_vdu,
5208             expected_extra_dict_copy,
5209             ns_preffix,
5210             vnf_preffix,
5211             "logger",
5212             {},
5213             [],
5214         )
5215 1         self.assertDictEqual(result, expected_extra_dict_copy)
5216 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5217 1         mock_prepare_vdu_affinity_group_list.assert_called_once()
5218 1         mock_find_persistent_volumes.assert_not_called()
5219
5220 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5221 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5222 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5223 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5224 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5225 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5226 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5227 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5228 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5229 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5230 1     def test_process_vdu_params_interfaces_partially_located(
5231         self,
5232         mock_prepare_vdu_affinity_group_list,
5233         mock_add_persistent_ordinary_disks_to_disk_list,
5234         mock_add_persistent_root_disk_to_disk_list,
5235         mock_find_persistent_volumes,
5236         mock_find_persistent_root_volumes,
5237         mock_prepare_vdu_ssh_keys,
5238         mock_prepare_vdu_cloud_init,
5239         mock_prepare_vdu_interfaces,
5240         mock_locate_vdu_interfaces,
5241         mock_sort_vdu_interfaces,
5242     ):
5243         """Some interfaces have position."""
5244 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5245
5246 1         self.maxDiff = None
5247 1         target_vdu["interfaces"] = [
5248             {
5249                 "name": "vdu-eth1",
5250                 "ns-vld-id": "net1",
5251             },
5252             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 2},
5253             {
5254                 "name": "vdu-eth3",
5255                 "ns-vld-id": "mgmtnet",
5256             },
5257         ]
5258 1         mock_prepare_vdu_cloud_init.return_value = {}
5259 1         mock_prepare_vdu_affinity_group_list.return_value = []
5260 1         persistent_root_disk = {
5261             "persistent-root-volume": {
5262                 "image_id": "ubuntu20.04",
5263                 "size": "10",
5264                 "keep": True,
5265             }
5266         }
5267 1         mock_find_persistent_root_volumes.return_value = persistent_root_disk
5268
5269 1         new_kwargs = deepcopy(kwargs)
5270 1         new_kwargs.update(
5271             {
5272                 "vnfr_id": vnfr_id,
5273                 "nsr_id": nsr_id,
5274                 "tasks_by_target_record_id": {},
5275                 "logger": "logger",
5276             }
5277         )
5278
5279 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
5280 1         db.get_one.return_value = vnfd
5281 1         result = Ns._process_vdu_params(
5282             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5283         )
5284 1         expected_extra_dict_copy = deepcopy(expected_extra_dict)
5285 1         mock_sort_vdu_interfaces.assert_not_called()
5286 1         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5287 1         mock_prepare_vdu_cloud_init.assert_called_once()
5288 1         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5289 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5290 1         mock_prepare_vdu_interfaces.assert_called_once_with(
5291             target_vdu,
5292             expected_extra_dict_copy,
5293             ns_preffix,
5294             vnf_preffix,
5295             "logger",
5296             {},
5297             [],
5298         )
5299 1         self.assertDictEqual(result, expected_extra_dict_copy)
5300 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5301 1         mock_prepare_vdu_affinity_group_list.assert_called_once()
5302 1         mock_find_persistent_volumes.assert_not_called()
5303 1         mock_find_persistent_root_volumes.assert_not_called()
5304
5305 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5306 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5307 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5308 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5309 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5310 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5311 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5312 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5313 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5314 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5315 1     def test_process_vdu_params_no_interface_position(
5316         self,
5317         mock_prepare_vdu_affinity_group_list,
5318         mock_add_persistent_ordinary_disks_to_disk_list,
5319         mock_add_persistent_root_disk_to_disk_list,
5320         mock_find_persistent_volumes,
5321         mock_find_persistent_root_volumes,
5322         mock_prepare_vdu_ssh_keys,
5323         mock_prepare_vdu_cloud_init,
5324         mock_prepare_vdu_interfaces,
5325         mock_locate_vdu_interfaces,
5326         mock_sort_vdu_interfaces,
5327     ):
5328         """Interfaces do not have position."""
5329 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5330
5331 1         self.maxDiff = None
5332 1         target_vdu["interfaces"] = interfaces_wthout_positions
5333 1         mock_prepare_vdu_cloud_init.return_value = {}
5334 1         mock_prepare_vdu_affinity_group_list.return_value = []
5335 1         persistent_root_disk = {
5336             "persistent-root-volume": {
5337                 "image_id": "ubuntu20.04",
5338                 "size": "10",
5339                 "keep": True,
5340             }
5341         }
5342 1         mock_find_persistent_root_volumes.return_value = persistent_root_disk
5343 1         new_kwargs = deepcopy(kwargs)
5344 1         new_kwargs.update(
5345             {
5346                 "vnfr_id": vnfr_id,
5347                 "nsr_id": nsr_id,
5348                 "tasks_by_target_record_id": {},
5349                 "logger": "logger",
5350             }
5351         )
5352
5353 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
5354 1         db.get_one.return_value = vnfd
5355 1         result = Ns._process_vdu_params(
5356             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5357         )
5358 1         expected_extra_dict_copy = deepcopy(expected_extra_dict)
5359 1         mock_sort_vdu_interfaces.assert_not_called()
5360 1         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5361 1         mock_prepare_vdu_cloud_init.assert_called_once()
5362 1         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5363 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5364 1         mock_prepare_vdu_interfaces.assert_called_once_with(
5365             target_vdu,
5366             expected_extra_dict_copy,
5367             ns_preffix,
5368             vnf_preffix,
5369             "logger",
5370             {},
5371             [],
5372         )
5373 1         self.assertDictEqual(result, expected_extra_dict_copy)
5374 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5375 1         mock_prepare_vdu_affinity_group_list.assert_called_once()
5376 1         mock_find_persistent_volumes.assert_not_called()
5377 1         mock_find_persistent_root_volumes.assert_not_called()
5378
5379 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5380 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5381 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5382 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5383 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5384 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5385 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5386 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5387 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5388 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5389 1     def test_process_vdu_params_prepare_vdu_interfaces_raises_exception(
5390         self,
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,
5401     ):
5402         """Prepare vdu interfaces method raises exception."""
5403 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5404
5405 1         self.maxDiff = None
5406 1         target_vdu["interfaces"] = interfaces_wthout_positions
5407 1         mock_prepare_vdu_cloud_init.return_value = {}
5408 1         mock_prepare_vdu_affinity_group_list.return_value = []
5409 1         persistent_root_disk = {
5410             "persistent-root-volume": {
5411                 "image_id": "ubuntu20.04",
5412                 "size": "10",
5413                 "keep": True,
5414             }
5415         }
5416 1         mock_find_persistent_root_volumes.return_value = persistent_root_disk
5417 1         new_kwargs = deepcopy(kwargs)
5418 1         new_kwargs.update(
5419             {
5420                 "vnfr_id": vnfr_id,
5421                 "nsr_id": nsr_id,
5422                 "tasks_by_target_record_id": {},
5423                 "logger": "logger",
5424             }
5425         )
5426 1         mock_prepare_vdu_interfaces.side_effect = TypeError
5427
5428 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
5429 1         db.get_one.return_value = vnfd
5430 1         with self.assertRaises(Exception) as err:
5431 1             Ns._process_vdu_params(
5432                 target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5433             )
5434 0             self.assertEqual(type(err), TypeError)
5435 1         mock_sort_vdu_interfaces.assert_not_called()
5436 1         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5437 1         mock_prepare_vdu_cloud_init.assert_not_called()
5438 1         mock_add_persistent_root_disk_to_disk_list.assert_not_called()
5439 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_not_called()
5440 1         mock_prepare_vdu_interfaces.assert_called_once()
5441 1         mock_prepare_vdu_ssh_keys.assert_not_called()
5442 1         mock_prepare_vdu_affinity_group_list.assert_not_called()
5443 1         mock_find_persistent_volumes.assert_not_called()
5444 1         mock_find_persistent_root_volumes.assert_not_called()
5445
5446 1     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5447 1     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5448 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5449 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5450 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5451 1     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5452 1     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5453 1     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5454 1     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5455 1     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5456 1     def test_process_vdu_params_add_persistent_root_disk_raises_exception(
5457         self,
5458         mock_prepare_vdu_affinity_group_list,
5459         mock_add_persistent_ordinary_disks_to_disk_list,
5460         mock_add_persistent_root_disk_to_disk_list,
5461         mock_find_persistent_volumes,
5462         mock_find_persistent_root_volumes,
5463         mock_prepare_vdu_ssh_keys,
5464         mock_prepare_vdu_cloud_init,
5465         mock_prepare_vdu_interfaces,
5466         mock_locate_vdu_interfaces,
5467         mock_sort_vdu_interfaces,
5468     ):
5469         """Add persistent root disk method raises exception."""
5470 1         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5471
5472 1         self.maxDiff = None
5473 1         target_vdu["interfaces"] = interfaces_wthout_positions
5474 1         mock_prepare_vdu_cloud_init.return_value = {}
5475 1         mock_prepare_vdu_affinity_group_list.return_value = []
5476 1         mock_add_persistent_root_disk_to_disk_list.side_effect = KeyError
5477 1         new_kwargs = deepcopy(kwargs)
5478 1         new_kwargs.update(
5479             {
5480                 "vnfr_id": vnfr_id,
5481                 "nsr_id": nsr_id,
5482                 "tasks_by_target_record_id": {},
5483                 "logger": "logger",
5484             }
5485         )
5486
5487 1         vnfd = deepcopy(vnfd_wth_persistent_storage)
5488 1         db.get_one.return_value = vnfd
5489 1         with self.assertRaises(Exception) as err:
5490 1             Ns._process_vdu_params(
5491                 target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5492             )
5493 0             self.assertEqual(type(err), KeyError)
5494 1         mock_sort_vdu_interfaces.assert_not_called()
5495 1         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5496 1         mock_prepare_vdu_cloud_init.assert_called_once()
5497 1         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5498 1         mock_add_persistent_ordinary_disks_to_disk_list.assert_not_called()
5499 1         mock_prepare_vdu_interfaces.assert_called_once_with(
5500             target_vdu,
5501             {
5502                 "depends_on": [
5503                     f"{ns_preffix}:image.0",
5504                     f"{ns_preffix}:flavor.0",
5505                 ]
5506             },
5507             ns_preffix,
5508             vnf_preffix,
5509             "logger",
5510             {},
5511             [],
5512         )
5513
5514 1         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5515 1         mock_prepare_vdu_affinity_group_list.assert_not_called()
5516 1         mock_find_persistent_volumes.assert_not_called()
5517 1         mock_find_persistent_root_volumes.assert_not_called()
5518
5519 1     def test_select_persistent_root_disk(self):
5520 1         vdu = deepcopy(target_vdu_wth_persistent_storage)
5521 1         vdu["virtual-storage-desc"] = [
5522             "persistent-root-volume",
5523             "persistent-volume2",
5524             "ephemeral-volume",
5525         ]
5526 1         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5527 1         expected_result = vsd
5528 1         result = Ns._select_persistent_root_disk(vsd, vdu)
5529 1         self.assertEqual(result, expected_result)
5530
5531 1     def test_select_persistent_root_disk_first_vsd_is_different(self):
5532         """VDU first virtual-storage-desc is different than vsd id."""
5533 1         vdu = deepcopy(target_vdu_wth_persistent_storage)
5534 1         vdu["virtual-storage-desc"] = [
5535             "persistent-volume2",
5536             "persistent-root-volume",
5537             "ephemeral-volume",
5538         ]
5539 1         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5540 1         expected_result = None
5541 1         result = Ns._select_persistent_root_disk(vsd, vdu)
5542 1         self.assertEqual(result, expected_result)
5543
5544 1     def test_select_persistent_root_disk_vsd_is_not_persistent(self):
5545         """vsd type is not persistent."""
5546 1         vdu = deepcopy(target_vdu_wth_persistent_storage)
5547 1         vdu["virtual-storage-desc"] = [
5548             "persistent-volume2",
5549             "persistent-root-volume",
5550             "ephemeral-volume",
5551         ]
5552 1         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5553 1         vsd["type-of-storage"] = "etsi-nfv-descriptors:ephemeral-storage"
5554 1         expected_result = None
5555 1         result = Ns._select_persistent_root_disk(vsd, vdu)
5556 1         self.assertEqual(result, expected_result)
5557
5558 1     def test_select_persistent_root_disk_vsd_does_not_have_size(self):
5559         """vsd size is None."""
5560 1         vdu = deepcopy(target_vdu_wth_persistent_storage)
5561 1         vdu["virtual-storage-desc"] = [
5562             "persistent-volume2",
5563             "persistent-root-volume",
5564             "ephemeral-volume",
5565         ]
5566 1         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5567 1         vsd["size-of-storage"] = None
5568 1         expected_result = None
5569 1         result = Ns._select_persistent_root_disk(vsd, vdu)
5570 1         self.assertEqual(result, expected_result)
5571
5572 1     def test_select_persistent_root_disk_vdu_wthout_vsd(self):
5573         """VDU does not have virtual-storage-desc."""
5574 1         vdu = deepcopy(target_vdu_wth_persistent_storage)
5575 1         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5576 1         expected_result = None
5577 1         result = Ns._select_persistent_root_disk(vsd, vdu)
5578 1         self.assertEqual(result, expected_result)
5579
5580 1     def test_select_persistent_root_disk_invalid_vsd_type(self):
5581         """vsd is list, expected to be a dict."""
5582 1         vdu = deepcopy(target_vdu_wth_persistent_storage)
5583 1         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"]
5584 1         with self.assertRaises(AttributeError):
5585 1             Ns._select_persistent_root_disk(vsd, vdu)