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