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