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_threads(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(self):
1511         expected_numa_result = {"cores": 3}
1512         expected_epa_vcpu_set_result = True
1513         guest_epa_quota = {
1514             "cpu-pinning-policy": "DEDICATED",
1515         }
1516         vcpu_count = 3
1517         epa_vcpu_set = False
1518
1519         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1520             guest_epa_quota=guest_epa_quota,
1521             vcpu_count=vcpu_count,
1522             epa_vcpu_set=epa_vcpu_set,
1523         )
1524
1525         self.assertDictEqual(expected_numa_result, numa_result)
1526         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1527
1528     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1529     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1530     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1531     def test__process_guest_epa_params_with_empty_params(
1532         self,
1533         guest_epa_numa_params,
1534         guest_epa_cpu_pinning_params,
1535         guest_epa_quota_params,
1536     ):
1537         expected_result = {}
1538         target_flavor = {}
1539
1540         result = Ns._process_epa_params(
1541             target_flavor=target_flavor,
1542         )
1543
1544         self.assertDictEqual(expected_result, result)
1545         self.assertFalse(guest_epa_numa_params.called)
1546         self.assertFalse(guest_epa_cpu_pinning_params.called)
1547         self.assertFalse(guest_epa_quota_params.called)
1548
1549     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1550     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1551     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1552     def test__process_guest_epa_params_with_wrong_params(
1553         self,
1554         guest_epa_numa_params,
1555         guest_epa_cpu_pinning_params,
1556         guest_epa_quota_params,
1557     ):
1558         expected_result = {}
1559         target_flavor = {
1560             "no-guest-epa": "here",
1561         }
1562
1563         result = Ns._process_epa_params(
1564             target_flavor=target_flavor,
1565         )
1566
1567         self.assertDictEqual(expected_result, result)
1568         self.assertFalse(guest_epa_numa_params.called)
1569         self.assertFalse(guest_epa_cpu_pinning_params.called)
1570         self.assertFalse(guest_epa_quota_params.called)
1571
1572     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1573     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1574     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1575     def test__process_guest_epa_params(
1576         self,
1577         guest_epa_numa_params,
1578         guest_epa_cpu_pinning_params,
1579         guest_epa_quota_params,
1580     ):
1581         expected_result = {
1582             "mem-policy": "STRICT",
1583         }
1584         target_flavor = {
1585             "guest-epa": {
1586                 "vcpu-count": 1,
1587                 "numa-node-policy": {
1588                     "mem-policy": "STRICT",
1589                 },
1590             },
1591         }
1592
1593         guest_epa_numa_params.return_value = ({}, False)
1594         guest_epa_cpu_pinning_params.return_value = ({}, False)
1595         guest_epa_quota_params.return_value = {}
1596
1597         result = Ns._process_epa_params(
1598             target_flavor=target_flavor,
1599         )
1600
1601         self.assertDictEqual(expected_result, result)
1602         self.assertTrue(guest_epa_numa_params.called)
1603         self.assertTrue(guest_epa_cpu_pinning_params.called)
1604         self.assertTrue(guest_epa_quota_params.called)
1605
1606     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1607     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1608     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1609     def test__process_guest_epa_params_with_mempage_size(
1610         self,
1611         guest_epa_numa_params,
1612         guest_epa_cpu_pinning_params,
1613         guest_epa_quota_params,
1614     ):
1615         expected_result = {
1616             "mempage-size": "1G",
1617             "mem-policy": "STRICT",
1618         }
1619         target_flavor = {
1620             "guest-epa": {
1621                 "vcpu-count": 1,
1622                 "mempage-size": "1G",
1623                 "numa-node-policy": {
1624                     "mem-policy": "STRICT",
1625                 },
1626             },
1627         }
1628
1629         guest_epa_numa_params.return_value = ({}, False)
1630         guest_epa_cpu_pinning_params.return_value = ({}, False)
1631         guest_epa_quota_params.return_value = {}
1632
1633         result = Ns._process_epa_params(
1634             target_flavor=target_flavor,
1635         )
1636
1637         self.assertDictEqual(expected_result, result)
1638         self.assertTrue(guest_epa_numa_params.called)
1639         self.assertTrue(guest_epa_cpu_pinning_params.called)
1640         self.assertTrue(guest_epa_quota_params.called)
1641
1642     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1643     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1644     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1645     def test__process_guest_epa_params_with_numa(
1646         self,
1647         guest_epa_numa_params,
1648         guest_epa_cpu_pinning_params,
1649         guest_epa_quota_params,
1650     ):
1651         expected_result = {
1652             "mempage-size": "1G",
1653             "cpu-pinning-policy": "DEDICATED",
1654             "cpu-thread-pinning-policy": "PREFER",
1655             "numas": [
1656                 {
1657                     "cores": 3,
1658                     "memory": 2,
1659                     "paired-threads": 3,
1660                     "paired-threads-id": [("0", "1"), ("4", "5")],
1661                     "threads": 3,
1662                 }
1663             ],
1664             "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1665             "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1666             "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1667             "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1668         }
1669         target_flavor = {
1670             "guest-epa": {
1671                 "vcpu-count": 1,
1672                 "mempage-size": "1G",
1673                 "cpu-pinning-policy": "DEDICATED",
1674                 "cpu-thread-pinning-policy": "PREFER",
1675                 "numa-node-policy": {
1676                     "node": [
1677                         {
1678                             "num-cores": 3,
1679                             "paired-threads": {
1680                                 "num-paired-threads": "3",
1681                                 "paired-thread-ids": [
1682                                     {
1683                                         "thread-a": 0,
1684                                         "thread-b": 1,
1685                                     },
1686                                     {
1687                                         "thread-a": 4,
1688                                         "thread-b": 5,
1689                                     },
1690                                 ],
1691                             },
1692                             "num-threads": "3",
1693                             "memory-mb": 2048,
1694                         },
1695                     ],
1696                 },
1697                 "cpu-quota": {
1698                     "limit": "10",
1699                     "reserve": "20",
1700                     "shares": "30",
1701                 },
1702                 "mem-quota": {
1703                     "limit": "10",
1704                     "reserve": "20",
1705                     "shares": "30",
1706                 },
1707                 "disk-io-quota": {
1708                     "limit": "10",
1709                     "reserve": "20",
1710                     "shares": "30",
1711                 },
1712                 "vif-quota": {
1713                     "limit": "10",
1714                     "reserve": "20",
1715                     "shares": "30",
1716                 },
1717             },
1718         }
1719
1720         guest_epa_numa_params.return_value = (
1721             [
1722                 {
1723                     "cores": 3,
1724                     "paired-threads": 3,
1725                     "paired-threads-id": [("0", "1"), ("4", "5")],
1726                     "threads": 3,
1727                     "memory": 2,
1728                 },
1729             ],
1730             True,
1731         )
1732         guest_epa_cpu_pinning_params.return_value = (
1733             {
1734                 "threads": 3,
1735             },
1736             True,
1737         )
1738         guest_epa_quota_params.return_value = {
1739             "cpu-quota": {
1740                 "limit": 10,
1741                 "reserve": 20,
1742                 "shares": 30,
1743             },
1744             "mem-quota": {
1745                 "limit": 10,
1746                 "reserve": 20,
1747                 "shares": 30,
1748             },
1749             "disk-io-quota": {
1750                 "limit": 10,
1751                 "reserve": 20,
1752                 "shares": 30,
1753             },
1754             "vif-quota": {
1755                 "limit": 10,
1756                 "reserve": 20,
1757                 "shares": 30,
1758             },
1759         }
1760
1761         result = Ns._process_epa_params(
1762             target_flavor=target_flavor,
1763         )
1764         self.assertEqual(expected_result, result)
1765         self.assertTrue(guest_epa_numa_params.called)
1766         self.assertTrue(guest_epa_cpu_pinning_params.called)
1767         self.assertTrue(guest_epa_quota_params.called)
1768
1769     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1770     def test__process_flavor_params_with_empty_target_flavor(
1771         self,
1772         epa_params,
1773     ):
1774
1775         target_flavor = {}
1776         indata = {
1777             "vnf": [
1778                 {
1779                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1780                 },
1781             ],
1782         }
1783         vim_info = {}
1784         target_record_id = ""
1785
1786         with self.assertRaises(KeyError):
1787             Ns._process_flavor_params(
1788                 target_flavor=target_flavor,
1789                 indata=indata,
1790                 vim_info=vim_info,
1791                 target_record_id=target_record_id,
1792             )
1793
1794         self.assertFalse(epa_params.called)
1795
1796     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1797     def test__process_flavor_params_with_wrong_target_flavor(
1798         self,
1799         epa_params,
1800     ):
1801
1802         target_flavor = {
1803             "no-target-flavor": "here",
1804         }
1805         indata = {}
1806         vim_info = {}
1807         target_record_id = ""
1808
1809         with self.assertRaises(KeyError):
1810             Ns._process_flavor_params(
1811                 target_flavor=target_flavor,
1812                 indata=indata,
1813                 vim_info=vim_info,
1814                 target_record_id=target_record_id,
1815             )
1816
1817         self.assertFalse(epa_params.called)
1818
1819     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1820     def test__process_flavor_params_with_empty_indata(
1821         self,
1822         epa_params,
1823     ):
1824
1825         expected_result = {
1826             "find_params": {
1827                 "flavor_data": {
1828                     "disk": 10,
1829                     "ram": 1024,
1830                     "vcpus": 2,
1831                 },
1832             },
1833             "params": {
1834                 "flavor_data": {
1835                     "disk": 10,
1836                     "name": "test",
1837                     "ram": 1024,
1838                     "vcpus": 2,
1839                 },
1840             },
1841         }
1842         target_flavor = {
1843             "name": "test",
1844             "storage-gb": "10",
1845             "memory-mb": "1024",
1846             "vcpu-count": "2",
1847         }
1848         indata = {}
1849         vim_info = {}
1850         target_record_id = ""
1851
1852         epa_params.return_value = {}
1853
1854         result = Ns._process_flavor_params(
1855             target_flavor=target_flavor,
1856             indata=indata,
1857             vim_info=vim_info,
1858             target_record_id=target_record_id,
1859         )
1860
1861         self.assertTrue(epa_params.called)
1862         self.assertDictEqual(result, expected_result)
1863
1864     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1865     def test__process_flavor_params_with_wrong_indata(
1866         self,
1867         epa_params,
1868     ):
1869
1870         expected_result = {
1871             "find_params": {
1872                 "flavor_data": {
1873                     "disk": 10,
1874                     "ram": 1024,
1875                     "vcpus": 2,
1876                 },
1877             },
1878             "params": {
1879                 "flavor_data": {
1880                     "disk": 10,
1881                     "name": "test",
1882                     "ram": 1024,
1883                     "vcpus": 2,
1884                 },
1885             },
1886         }
1887         target_flavor = {
1888             "name": "test",
1889             "storage-gb": "10",
1890             "memory-mb": "1024",
1891             "vcpu-count": "2",
1892         }
1893         indata = {
1894             "no-vnf": "here",
1895         }
1896         vim_info = {}
1897         target_record_id = ""
1898
1899         epa_params.return_value = {}
1900
1901         result = Ns._process_flavor_params(
1902             target_flavor=target_flavor,
1903             indata=indata,
1904             vim_info=vim_info,
1905             target_record_id=target_record_id,
1906         )
1907
1908         self.assertTrue(epa_params.called)
1909         self.assertDictEqual(result, expected_result)
1910
1911     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1912     def test__process_flavor_params_with_ephemeral_disk(
1913         self,
1914         epa_params,
1915     ):
1916         kwargs = {
1917             "db": db,
1918         }
1919
1920         db.get_one.return_value = {
1921             "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
1922             "df": [
1923                 {
1924                     "id": "default-df",
1925                     "vdu-profile": [
1926                         {"id": "without_volumes-VM", "min-number-of-instances": 1}
1927                     ],
1928                 }
1929             ],
1930             "id": "without_volumes-vnf",
1931             "product-name": "without_volumes-vnf",
1932             "vdu": [
1933                 {
1934                     "id": "without_volumes-VM",
1935                     "name": "without_volumes-VM",
1936                     "sw-image-desc": "ubuntu20.04",
1937                     "alternative-sw-image-desc": [
1938                         "ubuntu20.04-aws",
1939                         "ubuntu20.04-azure",
1940                     ],
1941                     "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
1942                 }
1943             ],
1944             "version": "1.0",
1945             "virtual-storage-desc": [
1946                 {"id": "root-volume", "size-of-storage": "10"},
1947                 {
1948                     "id": "ephemeral-volume",
1949                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1950                     "size-of-storage": "1",
1951                 },
1952             ],
1953             "_admin": {
1954                 "storage": {
1955                     "fs": "mongo",
1956                     "path": "/app/storage/",
1957                 },
1958                 "type": "vnfd",
1959             },
1960         }
1961         expected_result = {
1962             "find_params": {
1963                 "flavor_data": {
1964                     "disk": 10,
1965                     "ram": 1024,
1966                     "vcpus": 2,
1967                     "ephemeral": 10,
1968                 },
1969             },
1970             "params": {
1971                 "flavor_data": {
1972                     "disk": 10,
1973                     "name": "test",
1974                     "ram": 1024,
1975                     "vcpus": 2,
1976                     "ephemeral": 10,
1977                 },
1978             },
1979         }
1980         target_flavor = {
1981             "id": "test_id",
1982             "name": "test",
1983             "storage-gb": "10",
1984             "memory-mb": "1024",
1985             "vcpu-count": "2",
1986         }
1987         indata = {
1988             "vnf": [
1989                 {
1990                     "vdur": [
1991                         {
1992                             "ns-flavor-id": "test_id",
1993                             "virtual-storages": [
1994                                 {
1995                                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1996                                     "size-of-storage": "10",
1997                                 },
1998                             ],
1999                         },
2000                     ],
2001                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2002                 },
2003             ],
2004         }
2005         vim_info = {}
2006         target_record_id = ""
2007
2008         epa_params.return_value = {}
2009
2010         result = Ns._process_flavor_params(
2011             target_flavor=target_flavor,
2012             indata=indata,
2013             vim_info=vim_info,
2014             target_record_id=target_record_id,
2015             **kwargs,
2016         )
2017
2018         self.assertTrue(epa_params.called)
2019         self.assertDictEqual(result, expected_result)
2020
2021     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2022     def test__process_flavor_params_with_swap_disk(
2023         self,
2024         epa_params,
2025     ):
2026
2027         expected_result = {
2028             "find_params": {
2029                 "flavor_data": {
2030                     "disk": 10,
2031                     "ram": 1024,
2032                     "vcpus": 2,
2033                     "swap": 20,
2034                 },
2035             },
2036             "params": {
2037                 "flavor_data": {
2038                     "disk": 10,
2039                     "name": "test",
2040                     "ram": 1024,
2041                     "vcpus": 2,
2042                     "swap": 20,
2043                 },
2044             },
2045         }
2046         target_flavor = {
2047             "id": "test_id",
2048             "name": "test",
2049             "storage-gb": "10",
2050             "memory-mb": "1024",
2051             "vcpu-count": "2",
2052         }
2053         indata = {
2054             "vnf": [
2055                 {
2056                     "vdur": [
2057                         {
2058                             "ns-flavor-id": "test_id",
2059                             "virtual-storages": [
2060                                 {
2061                                     "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2062                                     "size-of-storage": "20",
2063                                 },
2064                             ],
2065                         },
2066                     ],
2067                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2068                 },
2069             ],
2070         }
2071         vim_info = {}
2072         target_record_id = ""
2073
2074         epa_params.return_value = {}
2075
2076         result = Ns._process_flavor_params(
2077             target_flavor=target_flavor,
2078             indata=indata,
2079             vim_info=vim_info,
2080             target_record_id=target_record_id,
2081         )
2082
2083         self.assertTrue(epa_params.called)
2084         self.assertDictEqual(result, expected_result)
2085
2086     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2087     def test__process_flavor_params_with_persistent_root_disk(
2088         self,
2089         epa_params,
2090     ):
2091
2092         kwargs = {
2093             "db": db,
2094         }
2095
2096         db.get_one.return_value = {
2097             "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2098             "df": [
2099                 {
2100                     "id": "default-df",
2101                     "vdu-profile": [
2102                         {"id": "several_volumes-VM", "min-number-of-instances": 1}
2103                     ],
2104                 }
2105             ],
2106             "id": "several_volumes-vnf",
2107             "product-name": "several_volumes-vnf",
2108             "vdu": [
2109                 {
2110                     "id": "several_volumes-VM",
2111                     "name": "several_volumes-VM",
2112                     "sw-image-desc": "ubuntu20.04",
2113                     "alternative-sw-image-desc": [
2114                         "ubuntu20.04-aws",
2115                         "ubuntu20.04-azure",
2116                     ],
2117                     "virtual-storage-desc": [
2118                         "persistent-root-volume",
2119                     ],
2120                 }
2121             ],
2122             "version": "1.0",
2123             "virtual-storage-desc": [
2124                 {
2125                     "id": "persistent-root-volume",
2126                     "type-of-storage": "persistent-storage:persistent-storage",
2127                     "size-of-storage": "10",
2128                 },
2129             ],
2130             "_admin": {
2131                 "storage": {
2132                     "fs": "mongo",
2133                     "path": "/app/storage/",
2134                 },
2135                 "type": "vnfd",
2136             },
2137         }
2138         expected_result = {
2139             "find_params": {
2140                 "flavor_data": {
2141                     "disk": 0,
2142                     "ram": 1024,
2143                     "vcpus": 2,
2144                 },
2145             },
2146             "params": {
2147                 "flavor_data": {
2148                     "disk": 0,
2149                     "name": "test",
2150                     "ram": 1024,
2151                     "vcpus": 2,
2152                 },
2153             },
2154         }
2155         target_flavor = {
2156             "id": "test_id",
2157             "name": "test",
2158             "storage-gb": "10",
2159             "memory-mb": "1024",
2160             "vcpu-count": "2",
2161         }
2162         indata = {
2163             "vnf": [
2164                 {
2165                     "vdur": [
2166                         {
2167                             "vdu-name": "several_volumes-VM",
2168                             "ns-flavor-id": "test_id",
2169                             "virtual-storages": [
2170                                 {
2171                                     "type-of-storage": "persistent-storage:persistent-storage",
2172                                     "size-of-storage": "10",
2173                                 },
2174                             ],
2175                         },
2176                     ],
2177                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2178                 },
2179             ],
2180         }
2181         vim_info = {}
2182         target_record_id = ""
2183
2184         epa_params.return_value = {}
2185
2186         result = Ns._process_flavor_params(
2187             target_flavor=target_flavor,
2188             indata=indata,
2189             vim_info=vim_info,
2190             target_record_id=target_record_id,
2191             **kwargs,
2192         )
2193
2194         self.assertTrue(epa_params.called)
2195         self.assertDictEqual(result, expected_result)
2196
2197     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2198     def test__process_flavor_params_with_epa_params(
2199         self,
2200         epa_params,
2201     ):
2202
2203         expected_result = {
2204             "find_params": {
2205                 "flavor_data": {
2206                     "disk": 10,
2207                     "ram": 1024,
2208                     "vcpus": 2,
2209                     "extended": {
2210                         "numa": "there-is-numa-here",
2211                     },
2212                 },
2213             },
2214             "params": {
2215                 "flavor_data": {
2216                     "disk": 10,
2217                     "name": "test",
2218                     "ram": 1024,
2219                     "vcpus": 2,
2220                     "extended": {
2221                         "numa": "there-is-numa-here",
2222                     },
2223                 },
2224             },
2225         }
2226         target_flavor = {
2227             "id": "test_id",
2228             "name": "test",
2229             "storage-gb": "10",
2230             "memory-mb": "1024",
2231             "vcpu-count": "2",
2232         }
2233         indata = {
2234             "vnf": [
2235                 {
2236                     "vdur": [
2237                         {
2238                             "ns-flavor-id": "test_id",
2239                         },
2240                     ],
2241                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2242                 },
2243             ],
2244         }
2245         vim_info = {}
2246         target_record_id = ""
2247
2248         epa_params.return_value = {
2249             "numa": "there-is-numa-here",
2250         }
2251
2252         result = Ns._process_flavor_params(
2253             target_flavor=target_flavor,
2254             indata=indata,
2255             vim_info=vim_info,
2256             target_record_id=target_record_id,
2257         )
2258
2259         self.assertTrue(epa_params.called)
2260         self.assertDictEqual(result, expected_result)
2261
2262     @patch("osm_ng_ro.ns.Ns._process_epa_params")
2263     def test__process_flavor_params(
2264         self,
2265         epa_params,
2266     ):
2267
2268         kwargs = {
2269             "db": db,
2270         }
2271
2272         db.get_one.return_value = {
2273             "_id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2274             "df": [
2275                 {
2276                     "id": "default-df",
2277                     "vdu-profile": [
2278                         {"id": "without_volumes-VM", "min-number-of-instances": 1}
2279                     ],
2280                 }
2281             ],
2282             "id": "without_volumes-vnf",
2283             "product-name": "without_volumes-vnf",
2284             "vdu": [
2285                 {
2286                     "id": "without_volumes-VM",
2287                     "name": "without_volumes-VM",
2288                     "sw-image-desc": "ubuntu20.04",
2289                     "alternative-sw-image-desc": [
2290                         "ubuntu20.04-aws",
2291                         "ubuntu20.04-azure",
2292                     ],
2293                     "virtual-storage-desc": ["root-volume", "ephemeral-volume"],
2294                 }
2295             ],
2296             "version": "1.0",
2297             "virtual-storage-desc": [
2298                 {"id": "root-volume", "size-of-storage": "10"},
2299                 {
2300                     "id": "ephemeral-volume",
2301                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2302                     "size-of-storage": "1",
2303                 },
2304             ],
2305             "_admin": {
2306                 "storage": {
2307                     "fs": "mongo",
2308                     "path": "/app/storage/",
2309                 },
2310                 "type": "vnfd",
2311             },
2312         }
2313
2314         expected_result = {
2315             "find_params": {
2316                 "flavor_data": {
2317                     "disk": 10,
2318                     "ram": 1024,
2319                     "vcpus": 2,
2320                     "ephemeral": 10,
2321                     "swap": 20,
2322                     "extended": {
2323                         "numa": "there-is-numa-here",
2324                     },
2325                 },
2326             },
2327             "params": {
2328                 "flavor_data": {
2329                     "disk": 10,
2330                     "name": "test",
2331                     "ram": 1024,
2332                     "vcpus": 2,
2333                     "ephemeral": 10,
2334                     "swap": 20,
2335                     "extended": {
2336                         "numa": "there-is-numa-here",
2337                     },
2338                 },
2339             },
2340         }
2341         target_flavor = {
2342             "id": "test_id",
2343             "name": "test",
2344             "storage-gb": "10",
2345             "memory-mb": "1024",
2346             "vcpu-count": "2",
2347         }
2348         indata = {
2349             "vnf": [
2350                 {
2351                     "vdur": [
2352                         {
2353                             "ns-flavor-id": "test_id",
2354                             "virtual-storages": [
2355                                 {
2356                                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
2357                                     "size-of-storage": "10",
2358                                 },
2359                                 {
2360                                     "type-of-storage": "etsi-nfv-descriptors:swap-storage",
2361                                     "size-of-storage": "20",
2362                                 },
2363                             ],
2364                         },
2365                     ],
2366                     "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18",
2367                 },
2368             ],
2369         }
2370         vim_info = {}
2371         target_record_id = ""
2372
2373         epa_params.return_value = {
2374             "numa": "there-is-numa-here",
2375         }
2376
2377         result = Ns._process_flavor_params(
2378             target_flavor=target_flavor,
2379             indata=indata,
2380             vim_info=vim_info,
2381             target_record_id=target_record_id,
2382             **kwargs,
2383         )
2384
2385         self.assertTrue(epa_params.called)
2386         self.assertDictEqual(result, expected_result)
2387
2388     def test__ip_profile_to_ro_with_none(self):
2389         ip_profile = None
2390
2391         result = Ns._ip_profile_to_ro(
2392             ip_profile=ip_profile,
2393         )
2394
2395         self.assertIsNone(result)
2396
2397     def test__ip_profile_to_ro_with_empty_profile(self):
2398         ip_profile = {}
2399
2400         result = Ns._ip_profile_to_ro(
2401             ip_profile=ip_profile,
2402         )
2403
2404         self.assertIsNone(result)
2405
2406     def test__ip_profile_to_ro_with_wrong_profile(self):
2407         ip_profile = {
2408             "no-profile": "here",
2409         }
2410         expected_result = {
2411             "ip_version": "IPv4",
2412             "subnet_address": None,
2413             "gateway_address": None,
2414             "dhcp_enabled": False,
2415             "dhcp_start_address": None,
2416             "dhcp_count": None,
2417         }
2418
2419         result = Ns._ip_profile_to_ro(
2420             ip_profile=ip_profile,
2421         )
2422
2423         self.assertDictEqual(expected_result, result)
2424
2425     def test__ip_profile_to_ro_with_ipv4_profile(self):
2426         ip_profile = {
2427             "ip-version": "ipv4",
2428             "subnet-address": "192.168.0.0/24",
2429             "gateway-address": "192.168.0.254",
2430             "dhcp-params": {
2431                 "enabled": True,
2432                 "start-address": "192.168.0.10",
2433                 "count": 25,
2434             },
2435         }
2436         expected_result = {
2437             "ip_version": "IPv4",
2438             "subnet_address": "192.168.0.0/24",
2439             "gateway_address": "192.168.0.254",
2440             "dhcp_enabled": True,
2441             "dhcp_start_address": "192.168.0.10",
2442             "dhcp_count": 25,
2443         }
2444
2445         result = Ns._ip_profile_to_ro(
2446             ip_profile=ip_profile,
2447         )
2448
2449         self.assertDictEqual(expected_result, result)
2450
2451     def test__ip_profile_to_ro_with_ipv6_profile(self):
2452         ip_profile = {
2453             "ip-version": "ipv6",
2454             "subnet-address": "2001:0200:0001::/48",
2455             "gateway-address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
2456             "dhcp-params": {
2457                 "enabled": True,
2458                 "start-address": "2001:0200:0001::0010",
2459                 "count": 25,
2460             },
2461         }
2462         expected_result = {
2463             "ip_version": "IPv6",
2464             "subnet_address": "2001:0200:0001::/48",
2465             "gateway_address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
2466             "dhcp_enabled": True,
2467             "dhcp_start_address": "2001:0200:0001::0010",
2468             "dhcp_count": 25,
2469         }
2470
2471         result = Ns._ip_profile_to_ro(
2472             ip_profile=ip_profile,
2473         )
2474
2475         self.assertDictEqual(expected_result, result)
2476
2477     def test__ip_profile_to_ro_with_dns_server(self):
2478         ip_profile = {
2479             "ip-version": "ipv4",
2480             "subnet-address": "192.168.0.0/24",
2481             "gateway-address": "192.168.0.254",
2482             "dhcp-params": {
2483                 "enabled": True,
2484                 "start-address": "192.168.0.10",
2485                 "count": 25,
2486             },
2487             "dns-server": [
2488                 {
2489                     "address": "8.8.8.8",
2490                 },
2491                 {
2492                     "address": "1.1.1.1",
2493                 },
2494                 {
2495                     "address": "1.0.0.1",
2496                 },
2497             ],
2498         }
2499         expected_result = {
2500             "ip_version": "IPv4",
2501             "subnet_address": "192.168.0.0/24",
2502             "gateway_address": "192.168.0.254",
2503             "dhcp_enabled": True,
2504             "dhcp_start_address": "192.168.0.10",
2505             "dhcp_count": 25,
2506             "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
2507         }
2508
2509         result = Ns._ip_profile_to_ro(
2510             ip_profile=ip_profile,
2511         )
2512
2513         self.assertDictEqual(expected_result, result)
2514
2515     def test__ip_profile_to_ro_with_security_group(self):
2516         ip_profile = {
2517             "ip-version": "ipv4",
2518             "subnet-address": "192.168.0.0/24",
2519             "gateway-address": "192.168.0.254",
2520             "dhcp-params": {
2521                 "enabled": True,
2522                 "start-address": "192.168.0.10",
2523                 "count": 25,
2524             },
2525             "security-group": {
2526                 "some-security-group": "here",
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             "security_group": {
2537                 "some-security-group": "here",
2538             },
2539         }
2540
2541         result = Ns._ip_profile_to_ro(
2542             ip_profile=ip_profile,
2543         )
2544
2545         self.assertDictEqual(expected_result, result)
2546
2547     def test__ip_profile_to_ro(self):
2548         ip_profile = {
2549             "ip-version": "ipv4",
2550             "subnet-address": "192.168.0.0/24",
2551             "gateway-address": "192.168.0.254",
2552             "dhcp-params": {
2553                 "enabled": True,
2554                 "start-address": "192.168.0.10",
2555                 "count": 25,
2556             },
2557             "dns-server": [
2558                 {
2559                     "address": "8.8.8.8",
2560                 },
2561                 {
2562                     "address": "1.1.1.1",
2563                 },
2564                 {
2565                     "address": "1.0.0.1",
2566                 },
2567             ],
2568             "security-group": {
2569                 "some-security-group": "here",
2570             },
2571         }
2572         expected_result = {
2573             "ip_version": "IPv4",
2574             "subnet_address": "192.168.0.0/24",
2575             "gateway_address": "192.168.0.254",
2576             "dhcp_enabled": True,
2577             "dhcp_start_address": "192.168.0.10",
2578             "dhcp_count": 25,
2579             "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
2580             "security_group": {
2581                 "some-security-group": "here",
2582             },
2583         }
2584
2585         result = Ns._ip_profile_to_ro(
2586             ip_profile=ip_profile,
2587         )
2588
2589         self.assertDictEqual(expected_result, result)
2590
2591     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2592     def test__process_net_params_with_empty_params(
2593         self,
2594         ip_profile_to_ro,
2595     ):
2596         target_vld = {
2597             "name": "vld-name",
2598         }
2599         indata = {
2600             "name": "ns-name",
2601         }
2602         vim_info = {
2603             "provider_network": "some-profile-here",
2604         }
2605         target_record_id = ""
2606         expected_result = {
2607             "params": {
2608                 "net_name": "ns-name-vld-name",
2609                 "net_type": "bridge",
2610                 "ip_profile": {
2611                     "some_ip_profile": "here",
2612                 },
2613                 "provider_network_profile": "some-profile-here",
2614             }
2615         }
2616
2617         ip_profile_to_ro.return_value = {
2618             "some_ip_profile": "here",
2619         }
2620
2621         result = Ns._process_net_params(
2622             target_vld=target_vld,
2623             indata=indata,
2624             vim_info=vim_info,
2625             target_record_id=target_record_id,
2626         )
2627
2628         self.assertDictEqual(expected_result, result)
2629         self.assertTrue(ip_profile_to_ro.called)
2630
2631     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2632     def test__process_net_params_with_vim_info_sdn(
2633         self,
2634         ip_profile_to_ro,
2635     ):
2636         target_vld = {
2637             "name": "vld-name",
2638         }
2639         indata = {
2640             "name": "ns-name",
2641         }
2642         vim_info = {
2643             "sdn": "some-sdn",
2644             "sdn-ports": ["some", "ports", "here"],
2645             "vlds": ["some", "vlds", "here"],
2646             "type": "sdn-type",
2647         }
2648         target_record_id = "vld.sdn.something"
2649         expected_result = {
2650             "params": {
2651                 "sdn-ports": ["some", "ports", "here"],
2652                 "vlds": ["some", "vlds", "here"],
2653                 "type": "sdn-type",
2654             }
2655         }
2656
2657         result = Ns._process_net_params(
2658             target_vld=target_vld,
2659             indata=indata,
2660             vim_info=vim_info,
2661             target_record_id=target_record_id,
2662         )
2663
2664         self.assertDictEqual(expected_result, result)
2665         self.assertFalse(ip_profile_to_ro.called)
2666
2667     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2668     def test__process_net_params_with_vim_info_sdn_target_vim(
2669         self,
2670         ip_profile_to_ro,
2671     ):
2672         target_vld = {
2673             "name": "vld-name",
2674         }
2675         indata = {
2676             "name": "ns-name",
2677         }
2678         vim_info = {
2679             "sdn": "some-sdn",
2680             "sdn-ports": ["some", "ports", "here"],
2681             "vlds": ["some", "vlds", "here"],
2682             "target_vim": "some-vim",
2683             "type": "sdn-type",
2684         }
2685         target_record_id = "vld.sdn.something"
2686         expected_result = {
2687             "depends_on": ["some-vim vld.sdn"],
2688             "params": {
2689                 "sdn-ports": ["some", "ports", "here"],
2690                 "vlds": ["some", "vlds", "here"],
2691                 "target_vim": "some-vim",
2692                 "type": "sdn-type",
2693             },
2694         }
2695
2696         result = Ns._process_net_params(
2697             target_vld=target_vld,
2698             indata=indata,
2699             vim_info=vim_info,
2700             target_record_id=target_record_id,
2701         )
2702
2703         self.assertDictEqual(expected_result, result)
2704         self.assertFalse(ip_profile_to_ro.called)
2705
2706     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2707     def test__process_net_params_with_vim_network_name(
2708         self,
2709         ip_profile_to_ro,
2710     ):
2711         target_vld = {
2712             "name": "vld-name",
2713         }
2714         indata = {
2715             "name": "ns-name",
2716         }
2717         vim_info = {
2718             "vim_network_name": "some-network-name",
2719         }
2720         target_record_id = "vld.sdn.something"
2721         expected_result = {
2722             "find_params": {
2723                 "filter_dict": {
2724                     "name": "some-network-name",
2725                 },
2726             },
2727         }
2728
2729         result = Ns._process_net_params(
2730             target_vld=target_vld,
2731             indata=indata,
2732             vim_info=vim_info,
2733             target_record_id=target_record_id,
2734         )
2735
2736         self.assertDictEqual(expected_result, result)
2737         self.assertFalse(ip_profile_to_ro.called)
2738
2739     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2740     def test__process_net_params_with_vim_network_id(
2741         self,
2742         ip_profile_to_ro,
2743     ):
2744         target_vld = {
2745             "name": "vld-name",
2746         }
2747         indata = {
2748             "name": "ns-name",
2749         }
2750         vim_info = {
2751             "vim_network_id": "some-network-id",
2752         }
2753         target_record_id = "vld.sdn.something"
2754         expected_result = {
2755             "find_params": {
2756                 "filter_dict": {
2757                     "id": "some-network-id",
2758                 },
2759             },
2760         }
2761
2762         result = Ns._process_net_params(
2763             target_vld=target_vld,
2764             indata=indata,
2765             vim_info=vim_info,
2766             target_record_id=target_record_id,
2767         )
2768
2769         self.assertDictEqual(expected_result, result)
2770         self.assertFalse(ip_profile_to_ro.called)
2771
2772     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2773     def test__process_net_params_with_mgmt_network(
2774         self,
2775         ip_profile_to_ro,
2776     ):
2777         target_vld = {
2778             "id": "vld-id",
2779             "name": "vld-name",
2780             "mgmt-network": "some-mgmt-network",
2781         }
2782         indata = {
2783             "name": "ns-name",
2784         }
2785         vim_info = {}
2786         target_record_id = "vld.sdn.something"
2787         expected_result = {
2788             "find_params": {
2789                 "mgmt": True,
2790                 "name": "vld-id",
2791             },
2792         }
2793
2794         result = Ns._process_net_params(
2795             target_vld=target_vld,
2796             indata=indata,
2797             vim_info=vim_info,
2798             target_record_id=target_record_id,
2799         )
2800
2801         self.assertDictEqual(expected_result, result)
2802         self.assertFalse(ip_profile_to_ro.called)
2803
2804     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2805     def test__process_net_params_with_underlay_eline(
2806         self,
2807         ip_profile_to_ro,
2808     ):
2809         target_vld = {
2810             "name": "vld-name",
2811             "underlay": "some-underlay-here",
2812             "type": "ELINE",
2813         }
2814         indata = {
2815             "name": "ns-name",
2816         }
2817         vim_info = {
2818             "provider_network": "some-profile-here",
2819         }
2820         target_record_id = ""
2821         expected_result = {
2822             "params": {
2823                 "ip_profile": {
2824                     "some_ip_profile": "here",
2825                 },
2826                 "net_name": "ns-name-vld-name",
2827                 "net_type": "ptp",
2828                 "provider_network_profile": "some-profile-here",
2829             }
2830         }
2831
2832         ip_profile_to_ro.return_value = {
2833             "some_ip_profile": "here",
2834         }
2835
2836         result = Ns._process_net_params(
2837             target_vld=target_vld,
2838             indata=indata,
2839             vim_info=vim_info,
2840             target_record_id=target_record_id,
2841         )
2842
2843         self.assertDictEqual(expected_result, result)
2844         self.assertTrue(ip_profile_to_ro.called)
2845
2846     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2847     def test__process_net_params_with_underlay_elan(
2848         self,
2849         ip_profile_to_ro,
2850     ):
2851         target_vld = {
2852             "name": "vld-name",
2853             "underlay": "some-underlay-here",
2854             "type": "ELAN",
2855         }
2856         indata = {
2857             "name": "ns-name",
2858         }
2859         vim_info = {
2860             "provider_network": "some-profile-here",
2861         }
2862         target_record_id = ""
2863         expected_result = {
2864             "params": {
2865                 "ip_profile": {
2866                     "some_ip_profile": "here",
2867                 },
2868                 "net_name": "ns-name-vld-name",
2869                 "net_type": "data",
2870                 "provider_network_profile": "some-profile-here",
2871             }
2872         }
2873
2874         ip_profile_to_ro.return_value = {
2875             "some_ip_profile": "here",
2876         }
2877
2878         result = Ns._process_net_params(
2879             target_vld=target_vld,
2880             indata=indata,
2881             vim_info=vim_info,
2882             target_record_id=target_record_id,
2883         )
2884
2885         self.assertDictEqual(expected_result, result)
2886         self.assertTrue(ip_profile_to_ro.called)
2887
2888     def test__get_cloud_init_exception(self):
2889         db_mock = MagicMock(name="database mock")
2890         fs_mock = None
2891
2892         location = ""
2893
2894         with self.assertRaises(NsException):
2895             Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2896
2897     def test__get_cloud_init_file_fs_exception(self):
2898         db_mock = MagicMock(name="database mock")
2899         fs_mock = None
2900
2901         location = "vnfr_id_123456:file:test_file"
2902         db_mock.get_one.return_value = {
2903             "_admin": {
2904                 "storage": {
2905                     "folder": "/home/osm",
2906                     "pkg-dir": "vnfr_test_dir",
2907                 },
2908             },
2909         }
2910
2911         with self.assertRaises(NsException):
2912             Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2913
2914     def test__get_cloud_init_file(self):
2915         db_mock = MagicMock(name="database mock")
2916         fs_mock = MagicMock(name="filesystem mock")
2917         file_mock = MagicMock(name="file mock")
2918
2919         location = "vnfr_id_123456:file:test_file"
2920         cloud_init_content = "this is a cloud init file content"
2921
2922         db_mock.get_one.return_value = {
2923             "_admin": {
2924                 "storage": {
2925                     "folder": "/home/osm",
2926                     "pkg-dir": "vnfr_test_dir",
2927                 },
2928             },
2929         }
2930         fs_mock.file_open.return_value = file_mock
2931         file_mock.__enter__.return_value.read.return_value = cloud_init_content
2932
2933         result = Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2934
2935         self.assertEqual(cloud_init_content, result)
2936
2937     def test__get_cloud_init_vdu(self):
2938         db_mock = MagicMock(name="database mock")
2939         fs_mock = None
2940
2941         location = "vnfr_id_123456:vdu:0"
2942         cloud_init_content = "this is a cloud init file content"
2943
2944         db_mock.get_one.return_value = {
2945             "vdu": {
2946                 0: {
2947                     "cloud-init": cloud_init_content,
2948                 },
2949             },
2950         }
2951
2952         result = Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2953
2954         self.assertEqual(cloud_init_content, result)
2955
2956     @patch("jinja2.Environment.__init__")
2957     def test__parse_jinja2_undefined_error(self, env_mock: Mock):
2958         cloud_init_content = None
2959         params = None
2960         context = None
2961
2962         env_mock.side_effect = UndefinedError("UndefinedError occurred.")
2963
2964         with self.assertRaises(NsException):
2965             Ns._parse_jinja2(
2966                 cloud_init_content=cloud_init_content, params=params, context=context
2967             )
2968
2969     @patch("jinja2.Environment.__init__")
2970     def test__parse_jinja2_template_error(self, env_mock: Mock):
2971         cloud_init_content = None
2972         params = None
2973         context = None
2974
2975         env_mock.side_effect = TemplateError("TemplateError occurred.")
2976
2977         with self.assertRaises(NsException):
2978             Ns._parse_jinja2(
2979                 cloud_init_content=cloud_init_content, params=params, context=context
2980             )
2981
2982     @patch("jinja2.Environment.__init__")
2983     def test__parse_jinja2_template_not_found(self, env_mock: Mock):
2984         cloud_init_content = None
2985         params = None
2986         context = None
2987
2988         env_mock.side_effect = TemplateNotFound("TemplateNotFound occurred.")
2989
2990         with self.assertRaises(NsException):
2991             Ns._parse_jinja2(
2992                 cloud_init_content=cloud_init_content, params=params, context=context
2993             )
2994
2995     def test_rendering_jinja2_temp_without_special_characters(self):
2996         cloud_init_content = """
2997         disk_setup:
2998             ephemeral0:
2999                 table_type: {{type}}
3000                 layout: True
3001                 overwrite: {{is_override}}
3002         runcmd:
3003              - [ ls, -l, / ]
3004              - [ sh, -xc, "echo $(date) '{{command}}'" ]
3005         """
3006         params = {
3007             "type": "mbr",
3008             "is_override": "False",
3009             "command": "; mkdir abc",
3010         }
3011         context = "cloud-init for VM"
3012         expected_result = """
3013         disk_setup:
3014             ephemeral0:
3015                 table_type: mbr
3016                 layout: True
3017                 overwrite: False
3018         runcmd:
3019              - [ ls, -l, / ]
3020              - [ sh, -xc, "echo $(date) '; mkdir abc'" ]
3021         """
3022         result = Ns._parse_jinja2(
3023             cloud_init_content=cloud_init_content, params=params, context=context
3024         )
3025         self.assertEqual(result, expected_result)
3026
3027     def test_rendering_jinja2_temp_with_special_characters(self):
3028         cloud_init_content = """
3029         disk_setup:
3030             ephemeral0:
3031                 table_type: {{type}}
3032                 layout: True
3033                 overwrite: {{is_override}}
3034         runcmd:
3035              - [ ls, -l, / ]
3036              - [ sh, -xc, "echo $(date) '{{command}}'" ]
3037         """
3038         params = {
3039             "type": "mbr",
3040             "is_override": "False",
3041             "command": "& rm -rf",
3042         }
3043         context = "cloud-init for VM"
3044         expected_result = """
3045         disk_setup:
3046             ephemeral0:
3047                 table_type: mbr
3048                 layout: True
3049                 overwrite: False
3050         runcmd:
3051              - [ ls, -l, / ]
3052              - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
3053         """
3054         result = Ns._parse_jinja2(
3055             cloud_init_content=cloud_init_content, params=params, context=context
3056         )
3057         self.assertNotEqual(result, expected_result)
3058
3059     def test_rendering_jinja2_temp_with_special_characters_autoescape_is_false(self):
3060         with patch("osm_ng_ro.ns.Environment") as mock_environment:
3061             mock_environment.return_value = Environment(
3062                 undefined=StrictUndefined,
3063                 autoescape=select_autoescape(default_for_string=False, default=False),
3064             )
3065             cloud_init_content = """
3066                 disk_setup:
3067                     ephemeral0:
3068                         table_type: {{type}}
3069                         layout: True
3070                         overwrite: {{is_override}}
3071                 runcmd:
3072                      - [ ls, -l, / ]
3073                      - [ sh, -xc, "echo $(date) '{{command}}'" ]
3074                 """
3075             params = {
3076                 "type": "mbr",
3077                 "is_override": "False",
3078                 "command": "& rm -rf /",
3079             }
3080             context = "cloud-init for VM"
3081             expected_result = """
3082                 disk_setup:
3083                     ephemeral0:
3084                         table_type: mbr
3085                         layout: True
3086                         overwrite: False
3087                 runcmd:
3088                      - [ ls, -l, / ]
3089                      - [ sh, -xc, "echo $(date) '& rm -rf /'" ]
3090                 """
3091             result = Ns._parse_jinja2(
3092                 cloud_init_content=cloud_init_content,
3093                 params=params,
3094                 context=context,
3095             )
3096             self.assertEqual(result, expected_result)
3097
3098     @patch("osm_ng_ro.ns.Ns._assign_vim")
3099     def test__rebuild_start_stop_task(self, assign_vim):
3100         self.ns = Ns()
3101         extra_dict = {}
3102         actions = ["start", "stop", "rebuild"]
3103         vdu_id = "bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3104         vnf_id = "665b4165-ce24-4320-bf19-b9a45bade49f"
3105         vdu_index = "0"
3106         action_id = "bb937f49-3870-4169-b758-9732e1ff40f3"
3107         nsr_id = "993166fe-723e-4680-ac4b-b1af2541ae31"
3108         task_index = 0
3109         target_vim = "vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
3110         t = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3111         for action in actions:
3112             expected_result = {
3113                 "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3114                 "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3115                 "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3116                 "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:0",
3117                 "status": "SCHEDULED",
3118                 "action": "EXEC",
3119                 "item": "update",
3120                 "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0",
3121                 "target_record_id": t,
3122                 "params": {
3123                     "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3124                     "action": action,
3125                 },
3126             }
3127             extra_dict["params"] = {
3128                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3129                 "action": action,
3130             }
3131             task = self.ns.rebuild_start_stop_task(
3132                 vdu_id,
3133                 vnf_id,
3134                 vdu_index,
3135                 action_id,
3136                 nsr_id,
3137                 task_index,
3138                 target_vim,
3139                 extra_dict,
3140             )
3141             self.assertEqual(task.get("action_id"), action_id)
3142             self.assertEqual(task.get("nsr_id"), nsr_id)
3143             self.assertEqual(task.get("target_id"), target_vim)
3144             self.assertDictEqual(task, expected_result)
3145
3146     @patch("osm_ng_ro.ns.Ns._assign_vim")
3147     def test_verticalscale_task(self, assign_vim):
3148         self.ns = Ns()
3149         extra_dict = {}
3150         vdu_index = "1"
3151         action_id = "bb937f49-3870-4169-b758-9732e1ff40f3"
3152         nsr_id = "993166fe-723e-4680-ac4b-b1af2541ae31"
3153         task_index = 1
3154         target_record_id = (
3155             "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
3156             "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3157         )
3158
3159         expected_result = {
3160             "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3161             "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3162             "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3163             "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
3164             "status": "SCHEDULED",
3165             "action": "EXEC",
3166             "item": "verticalscale",
3167             "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1",
3168             "target_record_id": target_record_id,
3169             "params": {
3170                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3171                 "flavor_dict": "flavor_dict",
3172             },
3173         }
3174         vdu = {
3175             "id": "bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
3176             "vim_info": {
3177                 "vim:f9f370ac-0d44-41a7-9000-457f2332bc35": {"interfaces": []}
3178             },
3179         }
3180         vnf = {"_id": "665b4165-ce24-4320-bf19-b9a45bade49f"}
3181         extra_dict["params"] = {
3182             "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3183             "flavor_dict": "flavor_dict",
3184         }
3185         task = self.ns.verticalscale_task(
3186             vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3187         )
3188
3189         self.assertDictEqual(task, expected_result)
3190
3191     @patch("osm_ng_ro.ns.Ns._assign_vim")
3192     def test_migrate_task(self, assign_vim):
3193         self.ns = Ns()
3194         extra_dict = {}
3195         vdu_index = "1"
3196         action_id = "bb937f49-3870-4169-b758-9732e1ff40f3"
3197         nsr_id = "993166fe-723e-4680-ac4b-b1af2541ae31"
3198         task_index = 1
3199         target_record_id = (
3200             "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:"
3201             "vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1"
3202         )
3203
3204         expected_result = {
3205             "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35",
3206             "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3",
3207             "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31",
3208             "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1",
3209             "status": "SCHEDULED",
3210             "action": "EXEC",
3211             "item": "migrate",
3212             "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1",
3213             "target_record_id": target_record_id,
3214             "params": {
3215                 "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3216                 "migrate_host": "migrateToHost",
3217             },
3218         }
3219         vdu = {
3220             "id": "bb9c43f9-10a2-4569-a8a8-957c3528b6d1",
3221             "vim_info": {
3222                 "vim:f9f370ac-0d44-41a7-9000-457f2332bc35": {"interfaces": []}
3223             },
3224         }
3225         vnf = {"_id": "665b4165-ce24-4320-bf19-b9a45bade49f"}
3226         extra_dict["params"] = {
3227             "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396",
3228             "migrate_host": "migrateToHost",
3229         }
3230         task = self.ns.migrate_task(
3231             vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict
3232         )
3233
3234         self.assertDictEqual(task, expected_result)
3235
3236
3237 class TestProcessVduParams(unittest.TestCase):
3238     def setUp(self):
3239         self.ns = Ns()
3240         self.logger = CopyingMock(autospec=True)
3241
3242     def test_find_persistent_root_volumes_empty_instantiation_vol_list(self):
3243         """Find persistent root volume, instantiation_vol_list is empty."""
3244         vnfd = deepcopy(vnfd_wth_persistent_storage)
3245         target_vdu = target_vdu_wth_persistent_storage
3246         vdu_instantiation_volumes_list = []
3247         disk_list = []
3248         expected_persist_root_disk = {
3249             "persistent-root-volume": {
3250                 "image_id": "ubuntu20.04",
3251                 "size": "10",
3252             }
3253         }
3254         expected_disk_list = [
3255             {
3256                 "image_id": "ubuntu20.04",
3257                 "size": "10",
3258             },
3259         ]
3260         persist_root_disk = self.ns.find_persistent_root_volumes(
3261             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3262         )
3263         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3264         self.assertEqual(disk_list, expected_disk_list)
3265         self.assertEqual(len(disk_list), 1)
3266
3267     def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(self):
3268         """Find persistent root volume, always selects the first vsd as root volume."""
3269         vnfd = deepcopy(vnfd_wth_persistent_storage)
3270         vnfd["vdu"][0]["virtual-storage-desc"] = [
3271             "persistent-volume2",
3272             "persistent-root-volume",
3273             "ephemeral-volume",
3274         ]
3275         target_vdu = target_vdu_wth_persistent_storage
3276         vdu_instantiation_volumes_list = []
3277         disk_list = []
3278         expected_persist_root_disk = {
3279             "persistent-volume2": {
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_empty_size_of_storage(self):
3298         """Find persistent root volume, size of storage is empty."""
3299         vnfd = deepcopy(vnfd_wth_persistent_storage)
3300         vnfd["virtual-storage-desc"][0]["size-of-storage"] = ""
3301         vnfd["vdu"][0]["virtual-storage-desc"] = [
3302             "persistent-volume2",
3303             "persistent-root-volume",
3304             "ephemeral-volume",
3305         ]
3306         target_vdu = target_vdu_wth_persistent_storage
3307         vdu_instantiation_volumes_list = []
3308         disk_list = []
3309         persist_root_disk = self.ns.find_persistent_root_volumes(
3310             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3311         )
3312         self.assertEqual(persist_root_disk, None)
3313         self.assertEqual(disk_list, [])
3314
3315     def test_find_persistent_root_empty_disk_list(self):
3316         """Find persistent root volume, empty disk list."""
3317         vnfd = deepcopy(vnfd_wth_persistent_storage)
3318         target_vdu = target_vdu_wth_persistent_storage
3319         vdu_instantiation_volumes_list = []
3320         disk_list = []
3321         expected_persist_root_disk = {
3322             "persistent-root-volume": {
3323                 "image_id": "ubuntu20.04",
3324                 "size": "10",
3325             }
3326         }
3327         expected_disk_list = [
3328             {
3329                 "image_id": "ubuntu20.04",
3330                 "size": "10",
3331             },
3332         ]
3333         persist_root_disk = self.ns.find_persistent_root_volumes(
3334             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3335         )
3336         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3337         self.assertEqual(disk_list, expected_disk_list)
3338         self.assertEqual(len(disk_list), 1)
3339
3340     def test_find_persistent_root_volumes_target_vdu_mismatch(self):
3341         """Find persistent root volume, target vdu name is not matching."""
3342         vnfd = deepcopy(vnfd_wth_persistent_storage)
3343         vnfd["vdu"][0]["name"] = "Several_Volumes-VM"
3344         target_vdu = target_vdu_wth_persistent_storage
3345         vdu_instantiation_volumes_list = []
3346         disk_list = []
3347         result = self.ns.find_persistent_root_volumes(
3348             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3349         )
3350         self.assertEqual(result, None)
3351         self.assertEqual(disk_list, [])
3352         self.assertEqual(len(disk_list), 0)
3353
3354     def test_find_persistent_root_volumes_with_instantiation_vol_list(self):
3355         """Find persistent root volume, existing volume needs to be used."""
3356         vnfd = deepcopy(vnfd_wth_persistent_storage)
3357         target_vdu = target_vdu_wth_persistent_storage
3358         vdu_instantiation_volumes_list = [
3359             {
3360                 "vim-volume-id": vim_volume_id,
3361                 "name": "persistent-root-volume",
3362             }
3363         ]
3364         disk_list = []
3365         expected_persist_root_disk = {
3366             "persistent-root-volume": {
3367                 "vim_volume_id": vim_volume_id,
3368                 "image_id": "ubuntu20.04",
3369             },
3370         }
3371         expected_disk_list = [
3372             {
3373                 "vim_volume_id": vim_volume_id,
3374                 "image_id": "ubuntu20.04",
3375             },
3376         ]
3377         persist_root_disk = self.ns.find_persistent_root_volumes(
3378             vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3379         )
3380         self.assertEqual(persist_root_disk, expected_persist_root_disk)
3381         self.assertEqual(disk_list, expected_disk_list)
3382         self.assertEqual(len(disk_list), 1)
3383
3384     def test_find_persistent_root_volumes_invalid_instantiation_params(self):
3385         """Find persistent root volume, existing volume id keyword is invalid."""
3386         vnfd = deepcopy(vnfd_wth_persistent_storage)
3387         target_vdu = target_vdu_wth_persistent_storage
3388         vdu_instantiation_volumes_list = [
3389             {
3390                 "volume-id": vim_volume_id,
3391                 "name": "persistent-root-volume",
3392             }
3393         ]
3394         disk_list = []
3395         with self.assertRaises(KeyError):
3396             self.ns.find_persistent_root_volumes(
3397                 vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
3398             )
3399
3400         self.assertEqual(disk_list, [])
3401         self.assertEqual(len(disk_list), 0)
3402
3403     def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list(
3404         self,
3405     ):
3406         """Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty."""
3407         persistent_root_disk = {
3408             "persistent-root-volume": {
3409                 "image_id": "ubuntu20.04",
3410                 "size": "10",
3411             }
3412         }
3413
3414         target_vdu = target_vdu_wth_persistent_storage
3415         vdu_instantiation_volumes_list = []
3416         disk_list = [
3417             {
3418                 "image_id": "ubuntu20.04",
3419                 "size": "10",
3420             },
3421         ]
3422
3423         expected_disk_list = [
3424             {
3425                 "image_id": "ubuntu20.04",
3426                 "size": "10",
3427             },
3428             {
3429                 "size": "10",
3430             },
3431         ]
3432         self.ns.find_persistent_volumes(
3433             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3434         )
3435         self.assertEqual(disk_list, expected_disk_list)
3436
3437     def test_find_persistent_volumes_vdu_wth_inst_vol_list(self):
3438         """Find persistent ordinary volume, vim-volume-id is given as instantiation parameter."""
3439         persistent_root_disk = {
3440             "persistent-root-volume": {
3441                 "image_id": "ubuntu20.04",
3442                 "size": "10",
3443             }
3444         }
3445         vdu_instantiation_volumes_list = [
3446             {
3447                 "vim-volume-id": vim_volume_id,
3448                 "name": "persistent-volume2",
3449             }
3450         ]
3451         target_vdu = target_vdu_wth_persistent_storage
3452         disk_list = [
3453             {
3454                 "image_id": "ubuntu20.04",
3455                 "size": "10",
3456             },
3457         ]
3458         expected_disk_list = [
3459             {
3460                 "image_id": "ubuntu20.04",
3461                 "size": "10",
3462             },
3463             {
3464                 "vim_volume_id": vim_volume_id,
3465             },
3466         ]
3467         self.ns.find_persistent_volumes(
3468             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3469         )
3470         self.assertEqual(disk_list, expected_disk_list)
3471
3472     def test_find_persistent_volumes_vdu_wthout_persistent_storage(self):
3473         """Find persistent ordinary volume, there is not any persistent disk."""
3474         persistent_root_disk = {}
3475         vdu_instantiation_volumes_list = []
3476         target_vdu = target_vdu_wthout_persistent_storage
3477         disk_list = []
3478         self.ns.find_persistent_volumes(
3479             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3480         )
3481         self.assertEqual(disk_list, disk_list)
3482
3483     def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk(
3484         self,
3485     ):
3486         """There is persistent root disk, but there is not ordinary persistent disk."""
3487         persistent_root_disk = {
3488             "persistent-root-volume": {
3489                 "image_id": "ubuntu20.04",
3490                 "size": "10",
3491             }
3492         }
3493         vdu_instantiation_volumes_list = []
3494         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3495         target_vdu["virtual-storages"] = [
3496             {
3497                 "id": "persistent-root-volume",
3498                 "size-of-storage": "10",
3499                 "type-of-storage": "persistent-storage:persistent-storage",
3500             },
3501             {
3502                 "id": "ephemeral-volume",
3503                 "size-of-storage": "1",
3504                 "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
3505             },
3506         ]
3507         disk_list = [
3508             {
3509                 "image_id": "ubuntu20.04",
3510                 "size": "10",
3511             },
3512         ]
3513         self.ns.find_persistent_volumes(
3514             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3515         )
3516         self.assertEqual(disk_list, disk_list)
3517
3518     def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(self):
3519         """Find persistent ordinary volume, volume id is not persistent_root_disk dict,
3520         vim-volume-id is given as instantiation parameter but disk id is not matching."""
3521         vdu_instantiation_volumes_list = [
3522             {
3523                 "vim-volume-id": vim_volume_id,
3524                 "name": "persistent-volume3",
3525             }
3526         ]
3527         persistent_root_disk = {
3528             "persistent-root-volume": {
3529                 "image_id": "ubuntu20.04",
3530                 "size": "10",
3531             }
3532         }
3533         disk_list = [
3534             {
3535                 "image_id": "ubuntu20.04",
3536                 "size": "10",
3537             },
3538         ]
3539         expected_disk_list = [
3540             {
3541                 "image_id": "ubuntu20.04",
3542                 "size": "10",
3543             },
3544             {
3545                 "size": "10",
3546             },
3547         ]
3548
3549         target_vdu = target_vdu_wth_persistent_storage
3550         self.ns.find_persistent_volumes(
3551             persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
3552         )
3553         self.assertEqual(disk_list, expected_disk_list)
3554
3555     def test_sort_vdu_interfaces_position_all_wth_positions(self):
3556         """Interfaces are sorted according to position, all have positions."""
3557         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3558         target_vdu["interfaces"] = [
3559             {
3560                 "name": "vdu-eth1",
3561                 "ns-vld-id": "datanet",
3562                 "position": 2,
3563             },
3564             {
3565                 "name": "vdu-eth0",
3566                 "ns-vld-id": "mgmtnet",
3567                 "position": 1,
3568             },
3569         ]
3570         sorted_interfaces = [
3571             {
3572                 "name": "vdu-eth0",
3573                 "ns-vld-id": "mgmtnet",
3574                 "position": 1,
3575             },
3576             {
3577                 "name": "vdu-eth1",
3578                 "ns-vld-id": "datanet",
3579                 "position": 2,
3580             },
3581         ]
3582         self.ns._sort_vdu_interfaces(target_vdu)
3583         self.assertEqual(target_vdu["interfaces"], sorted_interfaces)
3584
3585     def test_sort_vdu_interfaces_position_some_wth_position(self):
3586         """Interfaces are sorted according to position, some of them have positions."""
3587         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3588         target_vdu["interfaces"] = [
3589             {
3590                 "name": "vdu-eth0",
3591                 "ns-vld-id": "mgmtnet",
3592             },
3593             {
3594                 "name": "vdu-eth1",
3595                 "ns-vld-id": "datanet",
3596                 "position": 1,
3597             },
3598         ]
3599         sorted_interfaces = [
3600             {
3601                 "name": "vdu-eth1",
3602                 "ns-vld-id": "datanet",
3603                 "position": 1,
3604             },
3605             {
3606                 "name": "vdu-eth0",
3607                 "ns-vld-id": "mgmtnet",
3608             },
3609         ]
3610         self.ns._sort_vdu_interfaces(target_vdu)
3611         self.assertEqual(target_vdu["interfaces"], sorted_interfaces)
3612
3613     def test_sort_vdu_interfaces_position_empty_interface_list(self):
3614         """Interface list is empty."""
3615         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3616         target_vdu["interfaces"] = []
3617         sorted_interfaces = []
3618         self.ns._sort_vdu_interfaces(target_vdu)
3619         self.assertEqual(target_vdu["interfaces"], sorted_interfaces)
3620
3621     def test_partially_locate_vdu_interfaces(self):
3622         """Some interfaces have positions."""
3623         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3624         target_vdu["interfaces"] = [
3625             {
3626                 "name": "vdu-eth1",
3627                 "ns-vld-id": "net1",
3628             },
3629             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3630             {
3631                 "name": "vdu-eth3",
3632                 "ns-vld-id": "mgmtnet",
3633             },
3634             {
3635                 "name": "vdu-eth1",
3636                 "ns-vld-id": "datanet",
3637                 "position": 1,
3638             },
3639         ]
3640         self.ns._partially_locate_vdu_interfaces(target_vdu)
3641         self.assertDictEqual(
3642             target_vdu["interfaces"][0],
3643             {
3644                 "name": "vdu-eth1",
3645                 "ns-vld-id": "datanet",
3646                 "position": 1,
3647             },
3648         )
3649         self.assertDictEqual(
3650             target_vdu["interfaces"][2],
3651             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3652         )
3653
3654     def test_partially_locate_vdu_interfaces_position_start_from_0(self):
3655         """Some interfaces have positions, position start from 0."""
3656         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3657         target_vdu["interfaces"] = [
3658             {
3659                 "name": "vdu-eth1",
3660                 "ns-vld-id": "net1",
3661             },
3662             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3663             {
3664                 "name": "vdu-eth3",
3665                 "ns-vld-id": "mgmtnet",
3666             },
3667             {
3668                 "name": "vdu-eth1",
3669                 "ns-vld-id": "datanet",
3670                 "position": 0,
3671             },
3672         ]
3673         self.ns._partially_locate_vdu_interfaces(target_vdu)
3674         self.assertDictEqual(
3675             target_vdu["interfaces"][0],
3676             {
3677                 "name": "vdu-eth1",
3678                 "ns-vld-id": "datanet",
3679                 "position": 0,
3680             },
3681         )
3682         self.assertDictEqual(
3683             target_vdu["interfaces"][3],
3684             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 3},
3685         )
3686
3687     def test_partially_locate_vdu_interfaces_wthout_position(self):
3688         """Interfaces do not have positions."""
3689         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3690         target_vdu["interfaces"] = interfaces_wthout_positions
3691         expected_result = deepcopy(target_vdu["interfaces"])
3692         self.ns._partially_locate_vdu_interfaces(target_vdu)
3693         self.assertEqual(target_vdu["interfaces"], expected_result)
3694
3695     def test_partially_locate_vdu_interfaces_all_has_position(self):
3696         """All interfaces have position."""
3697         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3698         target_vdu["interfaces"] = interfaces_wth_all_positions
3699         expected_interfaces = [
3700             {
3701                 "name": "vdu-eth2",
3702                 "ns-vld-id": "net2",
3703                 "position": 0,
3704             },
3705             {
3706                 "name": "vdu-eth3",
3707                 "ns-vld-id": "mgmtnet",
3708                 "position": 1,
3709             },
3710             {
3711                 "name": "vdu-eth1",
3712                 "ns-vld-id": "net1",
3713                 "position": 2,
3714             },
3715         ]
3716         self.ns._partially_locate_vdu_interfaces(target_vdu)
3717         self.assertEqual(target_vdu["interfaces"], expected_interfaces)
3718
3719     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3720     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3721     def test_prepare_vdu_cloud_init(self, mock_parse_jinja2, mock_get_cloud_init):
3722         """Target_vdu has cloud-init and boot-data-drive."""
3723         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3724         target_vdu["cloud-init"] = "sample-cloud-init-path"
3725         target_vdu["boot-data-drive"] = "vda"
3726         vdu2cloud_init = {}
3727         mock_get_cloud_init.return_value = cloud_init_content
3728         mock_parse_jinja2.return_value = user_data
3729         expected_result = {
3730             "user-data": user_data,
3731             "boot-data-drive": "vda",
3732         }
3733         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3734         self.assertDictEqual(result, expected_result)
3735         mock_get_cloud_init.assert_called_once_with(
3736             db=db, fs=fs, location="sample-cloud-init-path"
3737         )
3738         mock_parse_jinja2.assert_called_once_with(
3739             cloud_init_content=cloud_init_content,
3740             params=None,
3741             context="sample-cloud-init-path",
3742         )
3743
3744     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3745     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3746     def test_prepare_vdu_cloud_init_get_cloud_init_raise_exception(
3747         self, mock_parse_jinja2, mock_get_cloud_init
3748     ):
3749         """Target_vdu has cloud-init and boot-data-drive, get_cloud_init method raises exception."""
3750         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3751         target_vdu["cloud-init"] = "sample-cloud-init-path"
3752         target_vdu["boot-data-drive"] = "vda"
3753         vdu2cloud_init = {}
3754         mock_get_cloud_init.side_effect = NsException(
3755             "Mismatch descriptor for cloud init."
3756         )
3757
3758         with self.assertRaises(NsException) as err:
3759             self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3760             self.assertEqual(str(err.exception), "Mismatch descriptor for cloud init.")
3761
3762         mock_get_cloud_init.assert_called_once_with(
3763             db=db, fs=fs, location="sample-cloud-init-path"
3764         )
3765         mock_parse_jinja2.assert_not_called()
3766
3767     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3768     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3769     def test_prepare_vdu_cloud_init_parse_jinja2_raise_exception(
3770         self, mock_parse_jinja2, mock_get_cloud_init
3771     ):
3772         """Target_vdu has cloud-init and boot-data-drive, parse_jinja2 method raises exception."""
3773         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3774         target_vdu["cloud-init"] = "sample-cloud-init-path"
3775         target_vdu["boot-data-drive"] = "vda"
3776         vdu2cloud_init = {}
3777         mock_get_cloud_init.return_value = cloud_init_content
3778         mock_parse_jinja2.side_effect = NsException("Error parsing cloud-init content.")
3779
3780         with self.assertRaises(NsException) as err:
3781             self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3782             self.assertEqual(str(err.exception), "Error parsing cloud-init content.")
3783         mock_get_cloud_init.assert_called_once_with(
3784             db=db, fs=fs, location="sample-cloud-init-path"
3785         )
3786         mock_parse_jinja2.assert_called_once_with(
3787             cloud_init_content=cloud_init_content,
3788             params=None,
3789             context="sample-cloud-init-path",
3790         )
3791
3792     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3793     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3794     def test_prepare_vdu_cloud_init_vdu_wthout_boot_data_drive(
3795         self, mock_parse_jinja2, mock_get_cloud_init
3796     ):
3797         """Target_vdu has cloud-init but do not have boot-data-drive."""
3798         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3799         target_vdu["cloud-init"] = "sample-cloud-init-path"
3800         vdu2cloud_init = {}
3801         mock_get_cloud_init.return_value = cloud_init_content
3802         mock_parse_jinja2.return_value = user_data
3803         expected_result = {
3804             "user-data": user_data,
3805         }
3806         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3807         self.assertDictEqual(result, expected_result)
3808         mock_get_cloud_init.assert_called_once_with(
3809             db=db, fs=fs, location="sample-cloud-init-path"
3810         )
3811         mock_parse_jinja2.assert_called_once_with(
3812             cloud_init_content=cloud_init_content,
3813             params=None,
3814             context="sample-cloud-init-path",
3815         )
3816
3817     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3818     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3819     def test_prepare_vdu_cloud_init_exists_in_vdu2cloud_init(
3820         self, mock_parse_jinja2, mock_get_cloud_init
3821     ):
3822         """Target_vdu has cloud-init, vdu2cloud_init dict has cloud-init_content."""
3823         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3824         target_vdu["cloud-init"] = "sample-cloud-init-path"
3825         target_vdu["boot-data-drive"] = "vda"
3826         vdu2cloud_init = {"sample-cloud-init-path": cloud_init_content}
3827         mock_parse_jinja2.return_value = user_data
3828         expected_result = {
3829             "user-data": user_data,
3830             "boot-data-drive": "vda",
3831         }
3832         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3833         self.assertDictEqual(result, expected_result)
3834         mock_get_cloud_init.assert_not_called()
3835         mock_parse_jinja2.assert_called_once_with(
3836             cloud_init_content=cloud_init_content,
3837             params=None,
3838             context="sample-cloud-init-path",
3839         )
3840
3841     @patch("osm_ng_ro.ns.Ns._get_cloud_init")
3842     @patch("osm_ng_ro.ns.Ns._parse_jinja2")
3843     def test_prepare_vdu_cloud_init_no_cloud_init(
3844         self, mock_parse_jinja2, mock_get_cloud_init
3845     ):
3846         """Target_vdu do not have cloud-init."""
3847         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
3848         target_vdu["boot-data-drive"] = "vda"
3849         vdu2cloud_init = {}
3850         expected_result = {
3851             "boot-data-drive": "vda",
3852         }
3853         result = self.ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs)
3854         self.assertDictEqual(result, expected_result)
3855         mock_get_cloud_init.assert_not_called()
3856         mock_parse_jinja2.assert_not_called()
3857
3858     def test_check_vld_information_of_interfaces_ns_vld_vnf_vld_both_exist(self):
3859         """ns_vld and vnf_vld both exist."""
3860         interface = {
3861             "name": "vdu-eth0",
3862             "ns-vld-id": "mgmtnet",
3863             "vnf-vld-id": "mgmt_cp_int",
3864         }
3865         expected_result = f"{ns_preffix}:vld.mgmtnet"
3866         result = self.ns._check_vld_information_of_interfaces(
3867             interface, ns_preffix, vnf_preffix
3868         )
3869         self.assertEqual(result, expected_result)
3870
3871     def test_check_vld_information_of_interfaces_empty_interfaces(self):
3872         """Interface dict is empty."""
3873         interface = {}
3874         result = self.ns._check_vld_information_of_interfaces(
3875             interface, ns_preffix, vnf_preffix
3876         )
3877         self.assertEqual(result, "")
3878
3879     def test_check_vld_information_of_interfaces_has_only_vnf_vld(self):
3880         """Interface dict has only vnf_vld."""
3881         interface = {
3882             "name": "vdu-eth0",
3883             "vnf-vld-id": "mgmt_cp_int",
3884         }
3885         expected_result = f"{vnf_preffix}:vld.mgmt_cp_int"
3886         result = self.ns._check_vld_information_of_interfaces(
3887             interface, ns_preffix, vnf_preffix
3888         )
3889         self.assertEqual(result, expected_result)
3890
3891     def test_check_vld_information_of_interfaces_has_vnf_vld_wthout_vnf_prefix(
3892         self,
3893     ):
3894         """Interface dict has only vnf_vld but vnf_preffix does not exist."""
3895         interface = {
3896             "name": "vdu-eth0",
3897             "vnf-vld-id": "mgmt_cp_int",
3898         }
3899         vnf_preffix = None
3900         with self.assertRaises(Exception) as err:
3901             self.ns._check_vld_information_of_interfaces(
3902                 interface, ns_preffix, vnf_preffix
3903             )
3904             self.assertEqual(type(err), TypeError)
3905
3906     def test_prepare_interface_port_security_has_security_details(self):
3907         """Interface dict has port security details."""
3908         interface = {
3909             "name": "vdu-eth0",
3910             "ns-vld-id": "mgmtnet",
3911             "vnf-vld-id": "mgmt_cp_int",
3912             "port-security-enabled": True,
3913             "port-security-disable-strategy": "allow-address-pairs",
3914         }
3915         expected_interface = {
3916             "name": "vdu-eth0",
3917             "ns-vld-id": "mgmtnet",
3918             "vnf-vld-id": "mgmt_cp_int",
3919             "port_security": True,
3920             "port_security_disable_strategy": "allow-address-pairs",
3921         }
3922         self.ns._prepare_interface_port_security(interface)
3923         self.assertDictEqual(interface, expected_interface)
3924
3925     def test_prepare_interface_port_security_empty_interfaces(self):
3926         """Interface dict is empty."""
3927         interface = {}
3928         expected_interface = {}
3929         self.ns._prepare_interface_port_security(interface)
3930         self.assertDictEqual(interface, expected_interface)
3931
3932     def test_prepare_interface_port_security_wthout_port_security(self):
3933         """Interface dict does not have port security details."""
3934         interface = {
3935             "name": "vdu-eth0",
3936             "ns-vld-id": "mgmtnet",
3937             "vnf-vld-id": "mgmt_cp_int",
3938         }
3939         expected_interface = {
3940             "name": "vdu-eth0",
3941             "ns-vld-id": "mgmtnet",
3942             "vnf-vld-id": "mgmt_cp_int",
3943         }
3944         self.ns._prepare_interface_port_security(interface)
3945         self.assertDictEqual(interface, expected_interface)
3946
3947     def test_create_net_item_of_interface_floating_ip_port_security(self):
3948         """Interface dict has floating ip, port-security details."""
3949         interface = {
3950             "name": "vdu-eth0",
3951             "vcpi": "sample_vcpi",
3952             "port_security": True,
3953             "port_security_disable_strategy": "allow-address-pairs",
3954             "floating_ip": "10.1.1.12",
3955             "ns-vld-id": "mgmtnet",
3956             "vnf-vld-id": "mgmt_cp_int",
3957         }
3958         net_text = f"{ns_preffix}"
3959         expected_net_item = {
3960             "name": "vdu-eth0",
3961             "port_security": True,
3962             "port_security_disable_strategy": "allow-address-pairs",
3963             "floating_ip": "10.1.1.12",
3964             "net_id": f"TASK-{ns_preffix}",
3965             "type": "virtual",
3966         }
3967         result = self.ns._create_net_item_of_interface(interface, net_text)
3968         self.assertDictEqual(result, expected_net_item)
3969
3970     def test_create_net_item_of_interface_invalid_net_text(self):
3971         """net-text is invalid."""
3972         interface = {
3973             "name": "vdu-eth0",
3974             "vcpi": "sample_vcpi",
3975             "port_security": True,
3976             "port_security_disable_strategy": "allow-address-pairs",
3977             "floating_ip": "10.1.1.12",
3978             "ns-vld-id": "mgmtnet",
3979             "vnf-vld-id": "mgmt_cp_int",
3980         }
3981         net_text = None
3982         with self.assertRaises(TypeError):
3983             self.ns._create_net_item_of_interface(interface, net_text)
3984
3985     def test_create_net_item_of_interface_empty_interface(self):
3986         """Interface dict is empty."""
3987         interface = {}
3988         net_text = ns_preffix
3989         expected_net_item = {
3990             "net_id": f"TASK-{ns_preffix}",
3991             "type": "virtual",
3992         }
3993         result = self.ns._create_net_item_of_interface(interface, net_text)
3994         self.assertDictEqual(result, expected_net_item)
3995
3996     @patch("osm_ng_ro.ns.deep_get")
3997     def test_prepare_type_of_interface_type_sriov(self, mock_deep_get):
3998         """Interface type is SR-IOV."""
3999         interface = {
4000             "name": "vdu-eth0",
4001             "vcpi": "sample_vcpi",
4002             "port_security": True,
4003             "port_security_disable_strategy": "allow-address-pairs",
4004             "floating_ip": "10.1.1.12",
4005             "ns-vld-id": "mgmtnet",
4006             "vnf-vld-id": "mgmt_cp_int",
4007             "type": "SR-IOV",
4008         }
4009         mock_deep_get.return_value = "SR-IOV"
4010         net_text = ns_preffix
4011         net_item = {}
4012         expected_net_item = {
4013             "use": "data",
4014             "model": "SR-IOV",
4015             "type": "SR-IOV",
4016         }
4017         self.ns._prepare_type_of_interface(
4018             interface, tasks_by_target_record_id, net_text, net_item
4019         )
4020         self.assertDictEqual(net_item, expected_net_item)
4021         self.assertEqual(
4022             "data",
4023             tasks_by_target_record_id[net_text]["extra_dict"]["params"]["net_type"],
4024         )
4025         mock_deep_get.assert_called_once_with(
4026             tasks_by_target_record_id, net_text, "extra_dict", "params", "net_type"
4027         )
4028
4029     @patch("osm_ng_ro.ns.deep_get")
4030     def test_prepare_type_of_interface_type_pic_passthrough_deep_get_return_empty_dict(
4031         self, mock_deep_get
4032     ):
4033         """Interface type is PCI-PASSTHROUGH, deep_get method return empty dict."""
4034         interface = {
4035             "name": "vdu-eth0",
4036             "vcpi": "sample_vcpi",
4037             "port_security": True,
4038             "port_security_disable_strategy": "allow-address-pairs",
4039             "floating_ip": "10.1.1.12",
4040             "ns-vld-id": "mgmtnet",
4041             "vnf-vld-id": "mgmt_cp_int",
4042             "type": "PCI-PASSTHROUGH",
4043         }
4044         mock_deep_get.return_value = {}
4045         tasks_by_target_record_id = {}
4046         net_text = ns_preffix
4047         net_item = {}
4048         expected_net_item = {
4049             "use": "data",
4050             "model": "PCI-PASSTHROUGH",
4051             "type": "PCI-PASSTHROUGH",
4052         }
4053         self.ns._prepare_type_of_interface(
4054             interface, tasks_by_target_record_id, net_text, net_item
4055         )
4056         self.assertDictEqual(net_item, expected_net_item)
4057         mock_deep_get.assert_called_once_with(
4058             tasks_by_target_record_id, net_text, "extra_dict", "params", "net_type"
4059         )
4060
4061     @patch("osm_ng_ro.ns.deep_get")
4062     def test_prepare_type_of_interface_type_mgmt(self, mock_deep_get):
4063         """Interface type is mgmt."""
4064         interface = {
4065             "name": "vdu-eth0",
4066             "vcpi": "sample_vcpi",
4067             "port_security": True,
4068             "port_security_disable_strategy": "allow-address-pairs",
4069             "floating_ip": "10.1.1.12",
4070             "ns-vld-id": "mgmtnet",
4071             "vnf-vld-id": "mgmt_cp_int",
4072             "type": "OM-MGMT",
4073         }
4074         tasks_by_target_record_id = {}
4075         net_text = ns_preffix
4076         net_item = {}
4077         expected_net_item = {
4078             "use": "mgmt",
4079         }
4080         self.ns._prepare_type_of_interface(
4081             interface, tasks_by_target_record_id, net_text, net_item
4082         )
4083         self.assertDictEqual(net_item, expected_net_item)
4084         mock_deep_get.assert_not_called()
4085
4086     @patch("osm_ng_ro.ns.deep_get")
4087     def test_prepare_type_of_interface_type_bridge(self, mock_deep_get):
4088         """Interface type is bridge."""
4089         interface = {
4090             "name": "vdu-eth0",
4091             "vcpi": "sample_vcpi",
4092             "port_security": True,
4093             "port_security_disable_strategy": "allow-address-pairs",
4094             "floating_ip": "10.1.1.12",
4095             "ns-vld-id": "mgmtnet",
4096             "vnf-vld-id": "mgmt_cp_int",
4097         }
4098         tasks_by_target_record_id = {}
4099         net_text = ns_preffix
4100         net_item = {}
4101         expected_net_item = {
4102             "use": "bridge",
4103             "model": None,
4104         }
4105         self.ns._prepare_type_of_interface(
4106             interface, tasks_by_target_record_id, net_text, net_item
4107         )
4108         self.assertDictEqual(net_item, expected_net_item)
4109         mock_deep_get.assert_not_called()
4110
4111     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4112     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4113     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4114     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4115     def test_prepare_vdu_interfaces(
4116         self,
4117         mock_type_of_interface,
4118         mock_item_of_interface,
4119         mock_port_security,
4120         mock_vld_information_of_interface,
4121     ):
4122         """Prepare vdu interfaces successfully."""
4123         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4124         interface_1 = {
4125             "name": "vdu-eth1",
4126             "ns-vld-id": "net1",
4127             "ip-address": "13.2.12.31",
4128             "mgmt-interface": True,
4129         }
4130         interface_2 = {
4131             "name": "vdu-eth2",
4132             "vnf-vld-id": "net2",
4133             "mac-address": "d0:94:66:ed:fc:e2",
4134         }
4135         interface_3 = {
4136             "name": "vdu-eth3",
4137             "ns-vld-id": "mgmtnet",
4138         }
4139         target_vdu["interfaces"] = [interface_1, interface_2, interface_3]
4140         extra_dict = {
4141             "params": "test_params",
4142             "find_params": "test_find_params",
4143             "depends_on": [],
4144         }
4145
4146         net_text_1 = f"{ns_preffix}:net1"
4147         net_text_2 = f"{vnf_preffix}:net2"
4148         net_text_3 = f"{ns_preffix}:mgmtnet"
4149         net_item_1 = {
4150             "name": "vdu-eth1",
4151             "net_id": f"TASK-{ns_preffix}",
4152             "type": "virtual",
4153         }
4154         net_item_2 = {
4155             "name": "vdu-eth2",
4156             "net_id": f"TASK-{ns_preffix}",
4157             "type": "virtual",
4158         }
4159         net_item_3 = {
4160             "name": "vdu-eth3",
4161             "net_id": f"TASK-{ns_preffix}",
4162             "type": "virtual",
4163         }
4164         mock_item_of_interface.side_effect = [net_item_1, net_item_2, net_item_3]
4165         mock_vld_information_of_interface.side_effect = [
4166             net_text_1,
4167             net_text_2,
4168             net_text_3,
4169         ]
4170         net_list = []
4171         expected_extra_dict = {
4172             "params": "test_params",
4173             "find_params": "test_find_params",
4174             "depends_on": [net_text_1, net_text_2, net_text_3],
4175             "mgmt_vdu_interface": 0,
4176         }
4177         updated_net_item1 = deepcopy(net_item_1)
4178         updated_net_item1.update({"ip_address": "13.2.12.31"})
4179         updated_net_item2 = deepcopy(net_item_2)
4180         updated_net_item2.update({"mac_address": "d0:94:66:ed:fc:e2"})
4181         expected_net_list = [updated_net_item1, updated_net_item2, net_item_3]
4182         self.ns._prepare_vdu_interfaces(
4183             target_vdu,
4184             extra_dict,
4185             ns_preffix,
4186             vnf_preffix,
4187             self.logger,
4188             tasks_by_target_record_id,
4189             net_list,
4190         )
4191         _call_mock_vld_information_of_interface = (
4192             mock_vld_information_of_interface.call_args_list
4193         )
4194         self.assertEqual(
4195             _call_mock_vld_information_of_interface[0][0],
4196             (interface_1, ns_preffix, vnf_preffix),
4197         )
4198         self.assertEqual(
4199             _call_mock_vld_information_of_interface[1][0],
4200             (interface_2, ns_preffix, vnf_preffix),
4201         )
4202         self.assertEqual(
4203             _call_mock_vld_information_of_interface[2][0],
4204             (interface_3, ns_preffix, vnf_preffix),
4205         )
4206
4207         _call_mock_port_security = mock_port_security.call_args_list
4208         self.assertEqual(_call_mock_port_security[0].args[0], interface_1)
4209         self.assertEqual(_call_mock_port_security[1].args[0], interface_2)
4210         self.assertEqual(_call_mock_port_security[2].args[0], interface_3)
4211
4212         _call_mock_item_of_interface = mock_item_of_interface.call_args_list
4213         self.assertEqual(_call_mock_item_of_interface[0][0], (interface_1, net_text_1))
4214         self.assertEqual(_call_mock_item_of_interface[1][0], (interface_2, net_text_2))
4215         self.assertEqual(_call_mock_item_of_interface[2][0], (interface_3, net_text_3))
4216
4217         _call_mock_type_of_interface = mock_type_of_interface.call_args_list
4218         self.assertEqual(
4219             _call_mock_type_of_interface[0][0],
4220             (interface_1, tasks_by_target_record_id, net_text_1, net_item_1),
4221         )
4222         self.assertEqual(
4223             _call_mock_type_of_interface[1][0],
4224             (interface_2, tasks_by_target_record_id, net_text_2, net_item_2),
4225         )
4226         self.assertEqual(
4227             _call_mock_type_of_interface[2][0],
4228             (interface_3, tasks_by_target_record_id, net_text_3, net_item_3),
4229         )
4230         self.assertEqual(net_list, expected_net_list)
4231         self.assertEqual(extra_dict, expected_extra_dict)
4232         self.logger.error.assert_not_called()
4233
4234     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4235     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4236     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4237     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4238     def test_prepare_vdu_interfaces_create_net_item_raise_exception(
4239         self,
4240         mock_type_of_interface,
4241         mock_item_of_interface,
4242         mock_port_security,
4243         mock_vld_information_of_interface,
4244     ):
4245         """Prepare vdu interfaces, create_net_item_of_interface method raise exception."""
4246         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4247         interface_1 = {
4248             "name": "vdu-eth1",
4249             "ns-vld-id": "net1",
4250             "ip-address": "13.2.12.31",
4251             "mgmt-interface": True,
4252         }
4253         interface_2 = {
4254             "name": "vdu-eth2",
4255             "vnf-vld-id": "net2",
4256             "mac-address": "d0:94:66:ed:fc:e2",
4257         }
4258         interface_3 = {
4259             "name": "vdu-eth3",
4260             "ns-vld-id": "mgmtnet",
4261         }
4262         target_vdu["interfaces"] = [interface_1, interface_2, interface_3]
4263         extra_dict = {
4264             "params": "test_params",
4265             "find_params": "test_find_params",
4266             "depends_on": [],
4267         }
4268         net_text_1 = f"{ns_preffix}:net1"
4269         mock_item_of_interface.side_effect = [TypeError, TypeError, TypeError]
4270
4271         mock_vld_information_of_interface.side_effect = [net_text_1]
4272         net_list = []
4273         expected_extra_dict = {
4274             "params": "test_params",
4275             "find_params": "test_find_params",
4276             "depends_on": [net_text_1],
4277         }
4278         with self.assertRaises(TypeError):
4279             self.ns._prepare_vdu_interfaces(
4280                 target_vdu,
4281                 extra_dict,
4282                 ns_preffix,
4283                 vnf_preffix,
4284                 self.logger,
4285                 tasks_by_target_record_id,
4286                 net_list,
4287             )
4288
4289         _call_mock_vld_information_of_interface = (
4290             mock_vld_information_of_interface.call_args_list
4291         )
4292         self.assertEqual(
4293             _call_mock_vld_information_of_interface[0][0],
4294             (interface_1, ns_preffix, vnf_preffix),
4295         )
4296
4297         _call_mock_port_security = mock_port_security.call_args_list
4298         self.assertEqual(_call_mock_port_security[0].args[0], interface_1)
4299
4300         _call_mock_item_of_interface = mock_item_of_interface.call_args_list
4301         self.assertEqual(_call_mock_item_of_interface[0][0], (interface_1, net_text_1))
4302
4303         mock_type_of_interface.assert_not_called()
4304         self.logger.error.assert_not_called()
4305         self.assertEqual(net_list, [])
4306         self.assertEqual(extra_dict, expected_extra_dict)
4307
4308     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4309     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4310     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4311     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4312     def test_prepare_vdu_interfaces_vld_information_is_empty(
4313         self,
4314         mock_type_of_interface,
4315         mock_item_of_interface,
4316         mock_port_security,
4317         mock_vld_information_of_interface,
4318     ):
4319         """Prepare vdu interfaces, check_vld_information_of_interface method returns empty result."""
4320         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4321         interface_1 = {
4322             "name": "vdu-eth1",
4323             "ns-vld-id": "net1",
4324             "ip-address": "13.2.12.31",
4325             "mgmt-interface": True,
4326         }
4327         interface_2 = {
4328             "name": "vdu-eth2",
4329             "vnf-vld-id": "net2",
4330             "mac-address": "d0:94:66:ed:fc:e2",
4331         }
4332         interface_3 = {
4333             "name": "vdu-eth3",
4334             "ns-vld-id": "mgmtnet",
4335         }
4336         target_vdu["interfaces"] = [interface_1, interface_2, interface_3]
4337         extra_dict = {
4338             "params": "test_params",
4339             "find_params": "test_find_params",
4340             "depends_on": [],
4341         }
4342         mock_vld_information_of_interface.side_effect = ["", "", ""]
4343         net_list = []
4344         self.ns._prepare_vdu_interfaces(
4345             target_vdu,
4346             extra_dict,
4347             ns_preffix,
4348             vnf_preffix,
4349             self.logger,
4350             tasks_by_target_record_id,
4351             net_list,
4352         )
4353
4354         _call_mock_vld_information_of_interface = (
4355             mock_vld_information_of_interface.call_args_list
4356         )
4357         self.assertEqual(
4358             _call_mock_vld_information_of_interface[0][0],
4359             (interface_1, ns_preffix, vnf_preffix),
4360         )
4361         self.assertEqual(
4362             _call_mock_vld_information_of_interface[1][0],
4363             (interface_2, ns_preffix, vnf_preffix),
4364         )
4365         self.assertEqual(
4366             _call_mock_vld_information_of_interface[2][0],
4367             (interface_3, ns_preffix, vnf_preffix),
4368         )
4369
4370         _call_logger = self.logger.error.call_args_list
4371         self.assertEqual(
4372             _call_logger[0][0],
4373             ("Interface 0 from vdu several_volumes-VM not connected to any vld",),
4374         )
4375         self.assertEqual(
4376             _call_logger[1][0],
4377             ("Interface 1 from vdu several_volumes-VM not connected to any vld",),
4378         )
4379         self.assertEqual(
4380             _call_logger[2][0],
4381             ("Interface 2 from vdu several_volumes-VM not connected to any vld",),
4382         )
4383         self.assertEqual(net_list, [])
4384         self.assertEqual(
4385             extra_dict,
4386             {
4387                 "params": "test_params",
4388                 "find_params": "test_find_params",
4389                 "depends_on": [],
4390             },
4391         )
4392
4393         mock_item_of_interface.assert_not_called()
4394         mock_port_security.assert_not_called()
4395         mock_type_of_interface.assert_not_called()
4396
4397     @patch("osm_ng_ro.ns.Ns._check_vld_information_of_interfaces")
4398     @patch("osm_ng_ro.ns.Ns._prepare_interface_port_security")
4399     @patch("osm_ng_ro.ns.Ns._create_net_item_of_interface")
4400     @patch("osm_ng_ro.ns.Ns._prepare_type_of_interface")
4401     def test_prepare_vdu_interfaces_empty_interface_list(
4402         self,
4403         mock_type_of_interface,
4404         mock_item_of_interface,
4405         mock_port_security,
4406         mock_vld_information_of_interface,
4407     ):
4408         """Prepare vdu interfaces, interface list is empty."""
4409         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4410         target_vdu["interfaces"] = []
4411         extra_dict = {}
4412         net_list = []
4413         self.ns._prepare_vdu_interfaces(
4414             target_vdu,
4415             extra_dict,
4416             ns_preffix,
4417             vnf_preffix,
4418             self.logger,
4419             tasks_by_target_record_id,
4420             net_list,
4421         )
4422         mock_type_of_interface.assert_not_called()
4423         mock_vld_information_of_interface.assert_not_called()
4424         mock_item_of_interface.assert_not_called()
4425         mock_port_security.assert_not_called()
4426
4427     def test_prepare_vdu_ssh_keys(self):
4428         """Target_vdu has ssh-keys and ro_nsr_public_key exists."""
4429         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4430         target_vdu["ssh-keys"] = ["sample-ssh-key"]
4431         ro_nsr_public_key = {"public_key": "path_of_public_key"}
4432         target_vdu["ssh-access-required"] = True
4433         cloud_config = {}
4434         expected_cloud_config = {
4435             "key-pairs": ["sample-ssh-key", {"public_key": "path_of_public_key"}]
4436         }
4437         self.ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
4438         self.assertDictEqual(cloud_config, expected_cloud_config)
4439
4440     def test_prepare_vdu_ssh_keys_target_vdu_wthout_ssh_keys(self):
4441         """Target_vdu does not have ssh-keys."""
4442         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4443         ro_nsr_public_key = {"public_key": "path_of_public_key"}
4444         target_vdu["ssh-access-required"] = True
4445         cloud_config = {}
4446         expected_cloud_config = {"key-pairs": [{"public_key": "path_of_public_key"}]}
4447         self.ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
4448         self.assertDictEqual(cloud_config, expected_cloud_config)
4449
4450     def test_prepare_vdu_ssh_keys_ssh_access_is_not_required(self):
4451         """Target_vdu has ssh-keys, ssh-access is not required."""
4452         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4453         target_vdu["ssh-keys"] = ["sample-ssh-key"]
4454         ro_nsr_public_key = {"public_key": "path_of_public_key"}
4455         target_vdu["ssh-access-required"] = False
4456         cloud_config = {}
4457         expected_cloud_config = {"key-pairs": ["sample-ssh-key"]}
4458         self.ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config)
4459         self.assertDictEqual(cloud_config, expected_cloud_config)
4460
4461     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4462     def test_add_persistent_root_disk_to_disk_list(
4463         self, mock_select_persistent_root_disk
4464     ):
4465         """Add persistent root disk to disk_list"""
4466         root_disk = {
4467             "id": "persistent-root-volume",
4468             "type-of-storage": "persistent-storage:persistent-storage",
4469             "size-of-storage": "10",
4470         }
4471         mock_select_persistent_root_disk.return_value = root_disk
4472         vnfd = deepcopy(vnfd_wth_persistent_storage)
4473         vnfd["virtual-storage-desc"][1] = root_disk
4474         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4475         persistent_root_disk = {}
4476         disk_list = []
4477         expected_disk_list = [
4478             {
4479                 "image_id": "ubuntu20.04",
4480                 "size": "10",
4481             }
4482         ]
4483         self.ns._add_persistent_root_disk_to_disk_list(
4484             vnfd, target_vdu, persistent_root_disk, disk_list
4485         )
4486         self.assertEqual(disk_list, expected_disk_list)
4487         mock_select_persistent_root_disk.assert_called_once()
4488
4489     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4490     def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises(
4491         self, mock_select_persistent_root_disk
4492     ):
4493         """Add persistent root disk to disk_list"""
4494         root_disk = {
4495             "id": "persistent-root-volume",
4496             "type-of-storage": "persistent-storage:persistent-storage",
4497             "size-of-storage": "10",
4498         }
4499         mock_select_persistent_root_disk.side_effect = AttributeError
4500         vnfd = deepcopy(vnfd_wth_persistent_storage)
4501         vnfd["virtual-storage-desc"][1] = root_disk
4502         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4503         persistent_root_disk = {}
4504         disk_list = []
4505         with self.assertRaises(AttributeError):
4506             self.ns._add_persistent_root_disk_to_disk_list(
4507                 vnfd, target_vdu, persistent_root_disk, disk_list
4508             )
4509         self.assertEqual(disk_list, [])
4510         mock_select_persistent_root_disk.assert_called_once()
4511
4512     def test_add_persistent_ordinary_disk_to_disk_list(self):
4513         """Add persistent ordinary disk to disk_list"""
4514         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4515         persistent_root_disk = {
4516             "persistent-root-volume": {
4517                 "image_id": "ubuntu20.04",
4518                 "size": "10",
4519             }
4520         }
4521         persistent_ordinary_disk = {}
4522         disk_list = []
4523         expected_disk_list = [
4524             {
4525                 "size": "10",
4526             }
4527         ]
4528         self.ns._add_persistent_ordinary_disks_to_disk_list(
4529             target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list
4530         )
4531         self.assertEqual(disk_list, expected_disk_list)
4532
4533     def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(self):
4534         """Add persistent ordinary disk, vsd id is in root_disk dict."""
4535         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4536         persistent_root_disk = {
4537             "persistent-root-volume": {
4538                 "image_id": "ubuntu20.04",
4539                 "size": "10",
4540             },
4541             "persistent-volume2": {
4542                 "size": "10",
4543             },
4544         }
4545         persistent_ordinary_disk = {}
4546         disk_list = []
4547
4548         self.ns._add_persistent_ordinary_disks_to_disk_list(
4549             target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list
4550         )
4551         self.assertEqual(disk_list, [])
4552
4553     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4554     def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage(
4555         self, mock_select_persistent_root_disk
4556     ):
4557         """VNFD does not have persistent storage."""
4558         vnfd = deepcopy(vnfd_wthout_persistent_storage)
4559         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4560         mock_select_persistent_root_disk.return_value = None
4561         persistent_root_disk = {}
4562         disk_list = []
4563         self.ns._add_persistent_root_disk_to_disk_list(
4564             vnfd, target_vdu, persistent_root_disk, disk_list
4565         )
4566         self.assertEqual(disk_list, [])
4567         self.assertEqual(mock_select_persistent_root_disk.call_count, 2)
4568
4569     @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
4570     def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk(
4571         self, mock_select_persistent_root_disk
4572     ):
4573         """Persistent_root_disk dict is empty."""
4574         vnfd = deepcopy(vnfd_wthout_persistent_storage)
4575         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4576         mock_select_persistent_root_disk.return_value = None
4577         persistent_root_disk = {}
4578         disk_list = []
4579         self.ns._add_persistent_root_disk_to_disk_list(
4580             vnfd, target_vdu, persistent_root_disk, disk_list
4581         )
4582         self.assertEqual(disk_list, [])
4583         self.assertEqual(mock_select_persistent_root_disk.call_count, 2)
4584
4585     def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self):
4586         """Invalid extra dict."""
4587         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4588         target_vdu["affinity-or-anti-affinity-group-id"] = "sample_affinity-group-id"
4589         extra_dict = {}
4590         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4591         with self.assertRaises(NsException) as err:
4592             self.ns._prepare_vdu_affinity_group_list(target_vdu, extra_dict, ns_preffix)
4593             self.assertEqual(str(err.exception), "Invalid extra_dict format.")
4594
4595     def test_prepare_vdu_affinity_group_list_one_affinity_group(self):
4596         """There is one affinity-group."""
4597         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4598         target_vdu["affinity-or-anti-affinity-group-id"] = ["sample_affinity-group-id"]
4599         extra_dict = {"depends_on": []}
4600         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4601         affinity_group_txt = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.sample_affinity-group-id"
4602         expected_result = [{"affinity_group_id": "TASK-" + affinity_group_txt}]
4603         expected_extra_dict = {"depends_on": [affinity_group_txt]}
4604         result = self.ns._prepare_vdu_affinity_group_list(
4605             target_vdu, extra_dict, ns_preffix
4606         )
4607         self.assertDictEqual(extra_dict, expected_extra_dict)
4608         self.assertEqual(result, expected_result)
4609
4610     def test_prepare_vdu_affinity_group_list_several_affinity_groups(self):
4611         """There are two affinity-groups."""
4612         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4613         target_vdu["affinity-or-anti-affinity-group-id"] = [
4614             "affinity-group-id1",
4615             "affinity-group-id2",
4616         ]
4617         extra_dict = {"depends_on": []}
4618         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4619         affinity_group_txt1 = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id1"
4620         affinity_group_txt2 = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3:affinity-or-anti-affinity-group.affinity-group-id2"
4621         expected_result = [
4622             {"affinity_group_id": "TASK-" + affinity_group_txt1},
4623             {"affinity_group_id": "TASK-" + affinity_group_txt2},
4624         ]
4625         expected_extra_dict = {"depends_on": [affinity_group_txt1, affinity_group_txt2]}
4626         result = self.ns._prepare_vdu_affinity_group_list(
4627             target_vdu, extra_dict, ns_preffix
4628         )
4629         self.assertDictEqual(extra_dict, expected_extra_dict)
4630         self.assertEqual(result, expected_result)
4631
4632     def test_prepare_vdu_affinity_group_list_no_affinity_group(self):
4633         """There is not any affinity-group."""
4634         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4635         extra_dict = {"depends_on": []}
4636         ns_preffix = "nsrs:th47f48-9870-4169-b758-9732e1ff40f3"
4637         result = self.ns._prepare_vdu_affinity_group_list(
4638             target_vdu, extra_dict, ns_preffix
4639         )
4640         self.assertDictEqual(extra_dict, {"depends_on": []})
4641         self.assertEqual(result, [])
4642
4643     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4644     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4645     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4646     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4647     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4648     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4649     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4650     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4651     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4652     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4653     def test_process_vdu_params_with_inst_vol_list(
4654         self,
4655         mock_prepare_vdu_affinity_group_list,
4656         mock_add_persistent_ordinary_disks_to_disk_list,
4657         mock_add_persistent_root_disk_to_disk_list,
4658         mock_find_persistent_volumes,
4659         mock_find_persistent_root_volumes,
4660         mock_prepare_vdu_ssh_keys,
4661         mock_prepare_vdu_cloud_init,
4662         mock_prepare_vdu_interfaces,
4663         mock_locate_vdu_interfaces,
4664         mock_sort_vdu_interfaces,
4665     ):
4666         """Instantiation volume list is empty."""
4667         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4668
4669         target_vdu["interfaces"] = interfaces_wth_all_positions
4670
4671         vdu_instantiation_vol_list = [
4672             {
4673                 "vim-volume-id": vim_volume_id,
4674                 "name": "persistent-volume2",
4675             }
4676         ]
4677         target_vdu["additionalParams"] = {
4678             "OSM": {"vdu_volumes": vdu_instantiation_vol_list}
4679         }
4680         mock_prepare_vdu_cloud_init.return_value = {}
4681         mock_prepare_vdu_affinity_group_list.return_value = []
4682         persistent_root_disk = {
4683             "persistent-root-volume": {
4684                 "image_id": "ubuntu20.04",
4685                 "size": "10",
4686             }
4687         }
4688         mock_find_persistent_root_volumes.return_value = persistent_root_disk
4689
4690         new_kwargs = deepcopy(kwargs)
4691         new_kwargs.update(
4692             {
4693                 "vnfr_id": vnfr_id,
4694                 "nsr_id": nsr_id,
4695                 "tasks_by_target_record_id": {},
4696                 "logger": "logger",
4697             }
4698         )
4699         expected_extra_dict_copy = deepcopy(expected_extra_dict)
4700         vnfd = deepcopy(vnfd_wth_persistent_storage)
4701         db.get_one.return_value = vnfd
4702         result = Ns._process_vdu_params(
4703             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
4704         )
4705         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
4706         mock_locate_vdu_interfaces.assert_not_called()
4707         mock_prepare_vdu_cloud_init.assert_called_once()
4708         mock_add_persistent_root_disk_to_disk_list.assert_not_called()
4709         mock_add_persistent_ordinary_disks_to_disk_list.assert_not_called()
4710         mock_prepare_vdu_interfaces.assert_called_once_with(
4711             target_vdu,
4712             expected_extra_dict_copy,
4713             ns_preffix,
4714             vnf_preffix,
4715             "logger",
4716             {},
4717             [],
4718         )
4719         self.assertDictEqual(result, expected_extra_dict_copy)
4720         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
4721         mock_prepare_vdu_affinity_group_list.assert_called_once()
4722         mock_find_persistent_volumes.assert_called_once_with(
4723             persistent_root_disk, target_vdu, vdu_instantiation_vol_list, []
4724         )
4725
4726     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4727     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4728     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4729     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4730     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4731     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4732     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4733     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4734     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4735     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4736     def test_process_vdu_params_wth_affinity_groups(
4737         self,
4738         mock_prepare_vdu_affinity_group_list,
4739         mock_add_persistent_ordinary_disks_to_disk_list,
4740         mock_add_persistent_root_disk_to_disk_list,
4741         mock_find_persistent_volumes,
4742         mock_find_persistent_root_volumes,
4743         mock_prepare_vdu_ssh_keys,
4744         mock_prepare_vdu_cloud_init,
4745         mock_prepare_vdu_interfaces,
4746         mock_locate_vdu_interfaces,
4747         mock_sort_vdu_interfaces,
4748     ):
4749         """There is cloud-config."""
4750         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4751
4752         self.maxDiff = None
4753         target_vdu["interfaces"] = interfaces_wth_all_positions
4754         mock_prepare_vdu_cloud_init.return_value = {}
4755         mock_prepare_vdu_affinity_group_list.return_value = [
4756             "affinity_group_1",
4757             "affinity_group_2",
4758         ]
4759
4760         new_kwargs = deepcopy(kwargs)
4761         new_kwargs.update(
4762             {
4763                 "vnfr_id": vnfr_id,
4764                 "nsr_id": nsr_id,
4765                 "tasks_by_target_record_id": {},
4766                 "logger": "logger",
4767             }
4768         )
4769         expected_extra_dict3 = deepcopy(expected_extra_dict2)
4770         expected_extra_dict3["params"]["affinity_group_list"] = [
4771             "affinity_group_1",
4772             "affinity_group_2",
4773         ]
4774         vnfd = deepcopy(vnfd_wth_persistent_storage)
4775         db.get_one.return_value = vnfd
4776         result = Ns._process_vdu_params(
4777             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
4778         )
4779         self.assertDictEqual(result, expected_extra_dict3)
4780         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
4781         mock_locate_vdu_interfaces.assert_not_called()
4782         mock_prepare_vdu_cloud_init.assert_called_once()
4783         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
4784         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
4785         mock_prepare_vdu_interfaces.assert_called_once_with(
4786             target_vdu,
4787             expected_extra_dict3,
4788             ns_preffix,
4789             vnf_preffix,
4790             "logger",
4791             {},
4792             [],
4793         )
4794
4795         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
4796         mock_prepare_vdu_affinity_group_list.assert_called_once()
4797         mock_find_persistent_volumes.assert_not_called()
4798
4799     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4800     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4801     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4802     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4803     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4804     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4805     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4806     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4807     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4808     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4809     def test_process_vdu_params_wth_cloud_config(
4810         self,
4811         mock_prepare_vdu_affinity_group_list,
4812         mock_add_persistent_ordinary_disks_to_disk_list,
4813         mock_add_persistent_root_disk_to_disk_list,
4814         mock_find_persistent_volumes,
4815         mock_find_persistent_root_volumes,
4816         mock_prepare_vdu_ssh_keys,
4817         mock_prepare_vdu_cloud_init,
4818         mock_prepare_vdu_interfaces,
4819         mock_locate_vdu_interfaces,
4820         mock_sort_vdu_interfaces,
4821     ):
4822         """There is cloud-config."""
4823         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4824
4825         self.maxDiff = None
4826         target_vdu["interfaces"] = interfaces_wth_all_positions
4827         mock_prepare_vdu_cloud_init.return_value = {
4828             "user-data": user_data,
4829             "boot-data-drive": "vda",
4830         }
4831         mock_prepare_vdu_affinity_group_list.return_value = []
4832
4833         new_kwargs = deepcopy(kwargs)
4834         new_kwargs.update(
4835             {
4836                 "vnfr_id": vnfr_id,
4837                 "nsr_id": nsr_id,
4838                 "tasks_by_target_record_id": {},
4839                 "logger": "logger",
4840             }
4841         )
4842         expected_extra_dict3 = deepcopy(expected_extra_dict2)
4843         expected_extra_dict3["params"]["cloud_config"] = {
4844             "user-data": user_data,
4845             "boot-data-drive": "vda",
4846         }
4847         vnfd = deepcopy(vnfd_wth_persistent_storage)
4848         db.get_one.return_value = vnfd
4849         result = Ns._process_vdu_params(
4850             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
4851         )
4852         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
4853         mock_locate_vdu_interfaces.assert_not_called()
4854         mock_prepare_vdu_cloud_init.assert_called_once()
4855         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
4856         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
4857         mock_prepare_vdu_interfaces.assert_called_once_with(
4858             target_vdu,
4859             expected_extra_dict3,
4860             ns_preffix,
4861             vnf_preffix,
4862             "logger",
4863             {},
4864             [],
4865         )
4866         self.assertDictEqual(result, expected_extra_dict3)
4867         mock_prepare_vdu_ssh_keys.assert_called_once_with(
4868             target_vdu, None, {"user-data": user_data, "boot-data-drive": "vda"}
4869         )
4870         mock_prepare_vdu_affinity_group_list.assert_called_once()
4871         mock_find_persistent_volumes.assert_not_called()
4872
4873     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4874     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4875     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4876     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4877     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4878     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4879     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4880     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4881     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4882     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4883     def test_process_vdu_params_wthout_persistent_storage(
4884         self,
4885         mock_prepare_vdu_affinity_group_list,
4886         mock_add_persistent_ordinary_disks_to_disk_list,
4887         mock_add_persistent_root_disk_to_disk_list,
4888         mock_find_persistent_volumes,
4889         mock_find_persistent_root_volumes,
4890         mock_prepare_vdu_ssh_keys,
4891         mock_prepare_vdu_cloud_init,
4892         mock_prepare_vdu_interfaces,
4893         mock_locate_vdu_interfaces,
4894         mock_sort_vdu_interfaces,
4895     ):
4896         """There is not any persistent storage."""
4897         target_vdu = deepcopy(target_vdu_wthout_persistent_storage)
4898
4899         self.maxDiff = None
4900         target_vdu["interfaces"] = interfaces_wth_all_positions
4901         mock_prepare_vdu_cloud_init.return_value = {}
4902         mock_prepare_vdu_affinity_group_list.return_value = []
4903
4904         new_kwargs = deepcopy(kwargs)
4905         new_kwargs.update(
4906             {
4907                 "vnfr_id": vnfr_id,
4908                 "nsr_id": nsr_id,
4909                 "tasks_by_target_record_id": {},
4910                 "logger": "logger",
4911             }
4912         )
4913         expected_extra_dict_copy = deepcopy(expected_extra_dict2)
4914         vnfd = deepcopy(vnfd_wthout_persistent_storage)
4915         db.get_one.return_value = vnfd
4916         result = Ns._process_vdu_params(
4917             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
4918         )
4919         mock_sort_vdu_interfaces.assert_called_once_with(target_vdu)
4920         mock_locate_vdu_interfaces.assert_not_called()
4921         mock_prepare_vdu_cloud_init.assert_called_once()
4922         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
4923         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
4924         mock_prepare_vdu_interfaces.assert_called_once_with(
4925             target_vdu,
4926             expected_extra_dict_copy,
4927             ns_preffix,
4928             vnf_preffix,
4929             "logger",
4930             {},
4931             [],
4932         )
4933         self.assertDictEqual(result, expected_extra_dict_copy)
4934         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
4935         mock_prepare_vdu_affinity_group_list.assert_called_once()
4936         mock_find_persistent_volumes.assert_not_called()
4937
4938     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
4939     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
4940     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
4941     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
4942     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
4943     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
4944     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
4945     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
4946     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
4947     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
4948     def test_process_vdu_params_interfaces_partially_located(
4949         self,
4950         mock_prepare_vdu_affinity_group_list,
4951         mock_add_persistent_ordinary_disks_to_disk_list,
4952         mock_add_persistent_root_disk_to_disk_list,
4953         mock_find_persistent_volumes,
4954         mock_find_persistent_root_volumes,
4955         mock_prepare_vdu_ssh_keys,
4956         mock_prepare_vdu_cloud_init,
4957         mock_prepare_vdu_interfaces,
4958         mock_locate_vdu_interfaces,
4959         mock_sort_vdu_interfaces,
4960     ):
4961         """Some interfaces have position."""
4962         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
4963
4964         self.maxDiff = None
4965         target_vdu["interfaces"] = [
4966             {
4967                 "name": "vdu-eth1",
4968                 "ns-vld-id": "net1",
4969             },
4970             {"name": "vdu-eth2", "ns-vld-id": "net2", "position": 2},
4971             {
4972                 "name": "vdu-eth3",
4973                 "ns-vld-id": "mgmtnet",
4974             },
4975         ]
4976         mock_prepare_vdu_cloud_init.return_value = {}
4977         mock_prepare_vdu_affinity_group_list.return_value = []
4978         persistent_root_disk = {
4979             "persistent-root-volume": {
4980                 "image_id": "ubuntu20.04",
4981                 "size": "10",
4982             }
4983         }
4984         mock_find_persistent_root_volumes.return_value = persistent_root_disk
4985
4986         new_kwargs = deepcopy(kwargs)
4987         new_kwargs.update(
4988             {
4989                 "vnfr_id": vnfr_id,
4990                 "nsr_id": nsr_id,
4991                 "tasks_by_target_record_id": {},
4992                 "logger": "logger",
4993             }
4994         )
4995
4996         vnfd = deepcopy(vnfd_wth_persistent_storage)
4997         db.get_one.return_value = vnfd
4998         result = Ns._process_vdu_params(
4999             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5000         )
5001         expected_extra_dict_copy = deepcopy(expected_extra_dict)
5002         mock_sort_vdu_interfaces.assert_not_called()
5003         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5004         mock_prepare_vdu_cloud_init.assert_called_once()
5005         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5006         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5007         mock_prepare_vdu_interfaces.assert_called_once_with(
5008             target_vdu,
5009             expected_extra_dict_copy,
5010             ns_preffix,
5011             vnf_preffix,
5012             "logger",
5013             {},
5014             [],
5015         )
5016         self.assertDictEqual(result, expected_extra_dict_copy)
5017         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5018         mock_prepare_vdu_affinity_group_list.assert_called_once()
5019         mock_find_persistent_volumes.assert_not_called()
5020         mock_find_persistent_root_volumes.assert_not_called()
5021
5022     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5023     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5024     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5025     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5026     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5027     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5028     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5029     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5030     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5031     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5032     def test_process_vdu_params_no_interface_position(
5033         self,
5034         mock_prepare_vdu_affinity_group_list,
5035         mock_add_persistent_ordinary_disks_to_disk_list,
5036         mock_add_persistent_root_disk_to_disk_list,
5037         mock_find_persistent_volumes,
5038         mock_find_persistent_root_volumes,
5039         mock_prepare_vdu_ssh_keys,
5040         mock_prepare_vdu_cloud_init,
5041         mock_prepare_vdu_interfaces,
5042         mock_locate_vdu_interfaces,
5043         mock_sort_vdu_interfaces,
5044     ):
5045         """Interfaces do not have position."""
5046         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5047
5048         self.maxDiff = None
5049         target_vdu["interfaces"] = interfaces_wthout_positions
5050         mock_prepare_vdu_cloud_init.return_value = {}
5051         mock_prepare_vdu_affinity_group_list.return_value = []
5052         persistent_root_disk = {
5053             "persistent-root-volume": {
5054                 "image_id": "ubuntu20.04",
5055                 "size": "10",
5056             }
5057         }
5058         mock_find_persistent_root_volumes.return_value = persistent_root_disk
5059         new_kwargs = deepcopy(kwargs)
5060         new_kwargs.update(
5061             {
5062                 "vnfr_id": vnfr_id,
5063                 "nsr_id": nsr_id,
5064                 "tasks_by_target_record_id": {},
5065                 "logger": "logger",
5066             }
5067         )
5068
5069         vnfd = deepcopy(vnfd_wth_persistent_storage)
5070         db.get_one.return_value = vnfd
5071         result = Ns._process_vdu_params(
5072             target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5073         )
5074         expected_extra_dict_copy = deepcopy(expected_extra_dict)
5075         mock_sort_vdu_interfaces.assert_not_called()
5076         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5077         mock_prepare_vdu_cloud_init.assert_called_once()
5078         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5079         mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once()
5080         mock_prepare_vdu_interfaces.assert_called_once_with(
5081             target_vdu,
5082             expected_extra_dict_copy,
5083             ns_preffix,
5084             vnf_preffix,
5085             "logger",
5086             {},
5087             [],
5088         )
5089         self.assertDictEqual(result, expected_extra_dict_copy)
5090         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5091         mock_prepare_vdu_affinity_group_list.assert_called_once()
5092         mock_find_persistent_volumes.assert_not_called()
5093         mock_find_persistent_root_volumes.assert_not_called()
5094
5095     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5096     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5097     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5098     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5099     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5100     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5101     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5102     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5103     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5104     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5105     def test_process_vdu_params_prepare_vdu_interfaces_raises_exception(
5106         self,
5107         mock_prepare_vdu_affinity_group_list,
5108         mock_add_persistent_ordinary_disks_to_disk_list,
5109         mock_add_persistent_root_disk_to_disk_list,
5110         mock_find_persistent_volumes,
5111         mock_find_persistent_root_volumes,
5112         mock_prepare_vdu_ssh_keys,
5113         mock_prepare_vdu_cloud_init,
5114         mock_prepare_vdu_interfaces,
5115         mock_locate_vdu_interfaces,
5116         mock_sort_vdu_interfaces,
5117     ):
5118         """Prepare vdu interfaces method raises exception."""
5119         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5120
5121         self.maxDiff = None
5122         target_vdu["interfaces"] = interfaces_wthout_positions
5123         mock_prepare_vdu_cloud_init.return_value = {}
5124         mock_prepare_vdu_affinity_group_list.return_value = []
5125         persistent_root_disk = {
5126             "persistent-root-volume": {
5127                 "image_id": "ubuntu20.04",
5128                 "size": "10",
5129             }
5130         }
5131         mock_find_persistent_root_volumes.return_value = persistent_root_disk
5132         new_kwargs = deepcopy(kwargs)
5133         new_kwargs.update(
5134             {
5135                 "vnfr_id": vnfr_id,
5136                 "nsr_id": nsr_id,
5137                 "tasks_by_target_record_id": {},
5138                 "logger": "logger",
5139             }
5140         )
5141         mock_prepare_vdu_interfaces.side_effect = TypeError
5142
5143         vnfd = deepcopy(vnfd_wth_persistent_storage)
5144         db.get_one.return_value = vnfd
5145         with self.assertRaises(Exception) as err:
5146             Ns._process_vdu_params(
5147                 target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5148             )
5149             self.assertEqual(type(err), TypeError)
5150         mock_sort_vdu_interfaces.assert_not_called()
5151         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5152         mock_prepare_vdu_cloud_init.assert_not_called()
5153         mock_add_persistent_root_disk_to_disk_list.assert_not_called()
5154         mock_add_persistent_ordinary_disks_to_disk_list.assert_not_called()
5155         mock_prepare_vdu_interfaces.assert_called_once()
5156         mock_prepare_vdu_ssh_keys.assert_not_called()
5157         mock_prepare_vdu_affinity_group_list.assert_not_called()
5158         mock_find_persistent_volumes.assert_not_called()
5159         mock_find_persistent_root_volumes.assert_not_called()
5160
5161     @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces")
5162     @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces")
5163     @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces")
5164     @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init")
5165     @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys")
5166     @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes")
5167     @patch("osm_ng_ro.ns.Ns.find_persistent_volumes")
5168     @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list")
5169     @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list")
5170     @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list")
5171     def test_process_vdu_params_add_persistent_root_disk_raises_exception(
5172         self,
5173         mock_prepare_vdu_affinity_group_list,
5174         mock_add_persistent_ordinary_disks_to_disk_list,
5175         mock_add_persistent_root_disk_to_disk_list,
5176         mock_find_persistent_volumes,
5177         mock_find_persistent_root_volumes,
5178         mock_prepare_vdu_ssh_keys,
5179         mock_prepare_vdu_cloud_init,
5180         mock_prepare_vdu_interfaces,
5181         mock_locate_vdu_interfaces,
5182         mock_sort_vdu_interfaces,
5183     ):
5184         """Add persistent root disk method raises exception."""
5185         target_vdu = deepcopy(target_vdu_wth_persistent_storage)
5186
5187         self.maxDiff = None
5188         target_vdu["interfaces"] = interfaces_wthout_positions
5189         mock_prepare_vdu_cloud_init.return_value = {}
5190         mock_prepare_vdu_affinity_group_list.return_value = []
5191         mock_add_persistent_root_disk_to_disk_list.side_effect = KeyError
5192         new_kwargs = deepcopy(kwargs)
5193         new_kwargs.update(
5194             {
5195                 "vnfr_id": vnfr_id,
5196                 "nsr_id": nsr_id,
5197                 "tasks_by_target_record_id": {},
5198                 "logger": "logger",
5199             }
5200         )
5201
5202         vnfd = deepcopy(vnfd_wth_persistent_storage)
5203         db.get_one.return_value = vnfd
5204         with self.assertRaises(Exception) as err:
5205             Ns._process_vdu_params(
5206                 target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs
5207             )
5208             self.assertEqual(type(err), KeyError)
5209         mock_sort_vdu_interfaces.assert_not_called()
5210         mock_locate_vdu_interfaces.assert_called_once_with(target_vdu)
5211         mock_prepare_vdu_cloud_init.assert_called_once()
5212         mock_add_persistent_root_disk_to_disk_list.assert_called_once()
5213         mock_add_persistent_ordinary_disks_to_disk_list.assert_not_called()
5214         mock_prepare_vdu_interfaces.assert_called_once_with(
5215             target_vdu,
5216             {
5217                 "depends_on": [
5218                     f"{ns_preffix}:image.0",
5219                     f"{ns_preffix}:flavor.0",
5220                 ]
5221             },
5222             ns_preffix,
5223             vnf_preffix,
5224             "logger",
5225             {},
5226             [],
5227         )
5228
5229         mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {})
5230         mock_prepare_vdu_affinity_group_list.assert_not_called()
5231         mock_find_persistent_volumes.assert_not_called()
5232         mock_find_persistent_root_volumes.assert_not_called()
5233
5234     def test_select_persistent_root_disk(self):
5235         vdu = deepcopy(target_vdu_wth_persistent_storage)
5236         vdu["virtual-storage-desc"] = [
5237             "persistent-root-volume",
5238             "persistent-volume2",
5239             "ephemeral-volume",
5240         ]
5241         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5242         expected_result = vsd
5243         result = Ns._select_persistent_root_disk(vsd, vdu)
5244         self.assertEqual(result, expected_result)
5245
5246     def test_select_persistent_root_disk_first_vsd_is_different(self):
5247         """VDU first virtual-storage-desc is different than vsd id."""
5248         vdu = deepcopy(target_vdu_wth_persistent_storage)
5249         vdu["virtual-storage-desc"] = [
5250             "persistent-volume2",
5251             "persistent-root-volume",
5252             "ephemeral-volume",
5253         ]
5254         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5255         expected_result = None
5256         result = Ns._select_persistent_root_disk(vsd, vdu)
5257         self.assertEqual(result, expected_result)
5258
5259     def test_select_persistent_root_disk_vsd_is_not_persistent(self):
5260         """vsd type is not persistent."""
5261         vdu = deepcopy(target_vdu_wth_persistent_storage)
5262         vdu["virtual-storage-desc"] = [
5263             "persistent-volume2",
5264             "persistent-root-volume",
5265             "ephemeral-volume",
5266         ]
5267         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5268         vsd["type-of-storage"] = "etsi-nfv-descriptors:ephemeral-storage"
5269         expected_result = None
5270         result = Ns._select_persistent_root_disk(vsd, vdu)
5271         self.assertEqual(result, expected_result)
5272
5273     def test_select_persistent_root_disk_vsd_does_not_have_size(self):
5274         """vsd size is None."""
5275         vdu = deepcopy(target_vdu_wth_persistent_storage)
5276         vdu["virtual-storage-desc"] = [
5277             "persistent-volume2",
5278             "persistent-root-volume",
5279             "ephemeral-volume",
5280         ]
5281         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5282         vsd["size-of-storage"] = None
5283         expected_result = None
5284         result = Ns._select_persistent_root_disk(vsd, vdu)
5285         self.assertEqual(result, expected_result)
5286
5287     def test_select_persistent_root_disk_vdu_wthout_vsd(self):
5288         """VDU does not have virtual-storage-desc."""
5289         vdu = deepcopy(target_vdu_wth_persistent_storage)
5290         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"][1]
5291         expected_result = None
5292         result = Ns._select_persistent_root_disk(vsd, vdu)
5293         self.assertEqual(result, expected_result)
5294
5295     def test_select_persistent_root_disk_invalid_vsd_type(self):
5296         """vsd is list, expected to be a dict."""
5297         vdu = deepcopy(target_vdu_wth_persistent_storage)
5298         vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"]
5299         with self.assertRaises(AttributeError):
5300             Ns._select_persistent_root_disk(vsd, vdu)