NG-RO/osm_ng_ro/tests/test_ns.py

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