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%
1897/1909
100%
0/0

Coverage Breakdown by Class

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