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