NG-RO/osm_ng_ro/tests/test_ns.py

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