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