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
  18 import unittest
  19 from unittest.mock import MagicMock, Mock, patch
  20
  21 from jinja2 import TemplateError, TemplateNotFound, UndefinedError
  22 from osm_ng_ro.ns import Ns, NsException
  23
  24
  25 __author__ = "Eduardo Sousa"
  26 __date__ = "$19-NOV-2021 00:00:00$"
  27
  28
  29 class TestNs(unittest.TestCase):
  30     def setUp(self):
  31         pass
  32
  33     def test__create_task_without_extra_dict(self):
  34         expected_result = {
  35             "target_id": "vim_openstack_1",
  36             "action_id": "123456",
  37             "nsr_id": "654321",
  38             "task_id": "123456:1",
  39             "status": "SCHEDULED",
  40             "action": "CREATE",
  41             "item": "test_item",
  42             "target_record": "test_target_record",
  43             "target_record_id": "test_target_record_id",
  44         }
  45         deployment_info = {
  46             "action_id": "123456",
  47             "nsr_id": "654321",
  48             "task_index": 1,
  49         }
  50
  51         task = Ns._create_task(
  52             deployment_info=deployment_info,
  53             target_id="vim_openstack_1",
  54             item="test_item",
  55             action="CREATE",
  56             target_record="test_target_record",
  57             target_record_id="test_target_record_id",
  58         )
  59
  60         self.assertEqual(deployment_info.get("task_index"), 2)
  61         self.assertDictEqual(task, expected_result)
  62
  63     def test__create_task(self):
  64         expected_result = {
  65             "target_id": "vim_openstack_1",
  66             "action_id": "123456",
  67             "nsr_id": "654321",
  68             "task_id": "123456:1",
  69             "status": "SCHEDULED",
  70             "action": "CREATE",
  71             "item": "test_item",
  72             "target_record": "test_target_record",
  73             "target_record_id": "test_target_record_id",
  74             # values coming from extra_dict
  75             "params": "test_params",
  76             "find_params": "test_find_params",
  77             "depends_on": "test_depends_on",
  78         }
  79         deployment_info = {
  80             "action_id": "123456",
  81             "nsr_id": "654321",
  82             "task_index": 1,
  83         }
  84
  85         task = Ns._create_task(
  86             deployment_info=deployment_info,
  87             target_id="vim_openstack_1",
  88             item="test_item",
  89             action="CREATE",
  90             target_record="test_target_record",
  91             target_record_id="test_target_record_id",
  92             extra_dict={
  93                 "params": "test_params",
  94                 "find_params": "test_find_params",
  95                 "depends_on": "test_depends_on",
  96             },
  97         )
  98
  99         self.assertEqual(deployment_info.get("task_index"), 2)
 100         self.assertDictEqual(task, expected_result)
 101
 102     @patch("osm_ng_ro.ns.time")
 103     def test__create_ro_task(self, mock_time: Mock):
 104         now = 1637324838.994551
 105         mock_time.return_value = now
 106         task = {
 107             "target_id": "vim_openstack_1",
 108             "action_id": "123456",
 109             "nsr_id": "654321",
 110             "task_id": "123456:1",
 111             "status": "SCHEDULED",
 112             "action": "CREATE",
 113             "item": "test_item",
 114             "target_record": "test_target_record",
 115             "target_record_id": "test_target_record_id",
 116             # values coming from extra_dict
 117             "params": "test_params",
 118             "find_params": "test_find_params",
 119             "depends_on": "test_depends_on",
 120         }
 121         expected_result = {
 122             "_id": "123456:1",
 123             "locked_by": None,
 124             "locked_at": 0.0,
 125             "target_id": "vim_openstack_1",
 126             "vim_info": {
 127                 "created": False,
 128                 "created_items": None,
 129                 "vim_id": None,
 130                 "vim_name": None,
 131                 "vim_status": None,
 132                 "vim_details": None,
 133                 "refresh_at": None,
 134             },
 135             "modified_at": now,
 136             "created_at": now,
 137             "to_check_at": now,
 138             "tasks": [task],
 139         }
 140
 141         ro_task = Ns._create_ro_task(
 142             target_id="vim_openstack_1",
 143             task=task,
 144         )
 145
 146         self.assertDictEqual(ro_task, expected_result)
 147
 148     def test__process_image_params_with_empty_target_image(self):
 149         expected_result = {
 150             "find_params": {},
 151         }
 152         target_image = {}
 153
 154         result = Ns._process_image_params(
 155             target_image=target_image,
 156             indata=None,
 157             vim_info=None,
 158             target_record_id=None,
 159         )
 160
 161         self.assertDictEqual(expected_result, result)
 162
 163     def test__process_image_params_with_wrong_target_image(self):
 164         expected_result = {
 165             "find_params": {},
 166         }
 167         target_image = {
 168             "no_image": "to_see_here",
 169         }
 170
 171         result = Ns._process_image_params(
 172             target_image=target_image,
 173             indata=None,
 174             vim_info=None,
 175             target_record_id=None,
 176         )
 177
 178         self.assertDictEqual(expected_result, result)
 179
 180     def test__process_image_params_with_image(self):
 181         expected_result = {
 182             "find_params": {
 183                 "filter_dict": {
 184                     "name": "cirros",
 185                 },
 186             },
 187         }
 188         target_image = {
 189             "image": "cirros",
 190         }
 191
 192         result = Ns._process_image_params(
 193             target_image=target_image,
 194             indata=None,
 195             vim_info=None,
 196             target_record_id=None,
 197         )
 198
 199         self.assertDictEqual(expected_result, result)
 200
 201     def test__process_image_params_with_vim_image_id(self):
 202         expected_result = {
 203             "find_params": {
 204                 "filter_dict": {
 205                     "id": "123456",
 206                 },
 207             },
 208         }
 209         target_image = {
 210             "vim_image_id": "123456",
 211         }
 212
 213         result = Ns._process_image_params(
 214             target_image=target_image,
 215             indata=None,
 216             vim_info=None,
 217             target_record_id=None,
 218         )
 219
 220         self.assertDictEqual(expected_result, result)
 221
 222     def test__process_image_params_with_image_checksum(self):
 223         expected_result = {
 224             "find_params": {
 225                 "filter_dict": {
 226                     "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
 227                 },
 228             },
 229         }
 230         target_image = {
 231             "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
 232         }
 233
 234         result = Ns._process_image_params(
 235             target_image=target_image,
 236             indata=None,
 237             vim_info=None,
 238             target_record_id=None,
 239         )
 240
 241         self.assertDictEqual(expected_result, result)
 242
 243     def test__get_resource_allocation_params_with_empty_target_image(self):
 244         expected_result = {}
 245         quota_descriptor = {}
 246
 247         result = Ns._get_resource_allocation_params(
 248             quota_descriptor=quota_descriptor,
 249         )
 250
 251         self.assertDictEqual(expected_result, result)
 252
 253     def test__get_resource_allocation_params_with_wrong_target_image(self):
 254         expected_result = {}
 255         quota_descriptor = {
 256             "no_quota": "present_here",
 257         }
 258
 259         result = Ns._get_resource_allocation_params(
 260             quota_descriptor=quota_descriptor,
 261         )
 262
 263         self.assertDictEqual(expected_result, result)
 264
 265     def test__get_resource_allocation_params_with_limit(self):
 266         expected_result = {
 267             "limit": 10,
 268         }
 269         quota_descriptor = {
 270             "limit": "10",
 271         }
 272
 273         result = Ns._get_resource_allocation_params(
 274             quota_descriptor=quota_descriptor,
 275         )
 276
 277         self.assertDictEqual(expected_result, result)
 278
 279     def test__get_resource_allocation_params_with_reserve(self):
 280         expected_result = {
 281             "reserve": 20,
 282         }
 283         quota_descriptor = {
 284             "reserve": "20",
 285         }
 286
 287         result = Ns._get_resource_allocation_params(
 288             quota_descriptor=quota_descriptor,
 289         )
 290
 291         self.assertDictEqual(expected_result, result)
 292
 293     def test__get_resource_allocation_params_with_shares(self):
 294         expected_result = {
 295             "shares": 30,
 296         }
 297         quota_descriptor = {
 298             "shares": "30",
 299         }
 300
 301         result = Ns._get_resource_allocation_params(
 302             quota_descriptor=quota_descriptor,
 303         )
 304
 305         self.assertDictEqual(expected_result, result)
 306
 307     def test__get_resource_allocation_params(self):
 308         expected_result = {
 309             "limit": 10,
 310             "reserve": 20,
 311             "shares": 30,
 312         }
 313         quota_descriptor = {
 314             "limit": "10",
 315             "reserve": "20",
 316             "shares": "30",
 317         }
 318
 319         result = Ns._get_resource_allocation_params(
 320             quota_descriptor=quota_descriptor,
 321         )
 322
 323         self.assertDictEqual(expected_result, result)
 324
 325     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 326     def test__process_guest_epa_quota_params_with_empty_quota_epa_cpu(
 327         self,
 328         resource_allocation,
 329     ):
 330         expected_result = {}
 331         guest_epa_quota = {}
 332         epa_vcpu_set = True
 333
 334         result = Ns._process_guest_epa_quota_params(
 335             guest_epa_quota=guest_epa_quota,
 336             epa_vcpu_set=epa_vcpu_set,
 337         )
 338
 339         self.assertDictEqual(expected_result, result)
 340         self.assertFalse(resource_allocation.called)
 341
 342     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 343     def test__process_guest_epa_quota_params_with_empty_quota_false_epa_cpu(
 344         self,
 345         resource_allocation,
 346     ):
 347         expected_result = {}
 348         guest_epa_quota = {}
 349         epa_vcpu_set = False
 350
 351         result = Ns._process_guest_epa_quota_params(
 352             guest_epa_quota=guest_epa_quota,
 353             epa_vcpu_set=epa_vcpu_set,
 354         )
 355
 356         self.assertDictEqual(expected_result, result)
 357         self.assertFalse(resource_allocation.called)
 358
 359     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 360     def test__process_guest_epa_quota_params_with_wrong_quota_epa_cpu(
 361         self,
 362         resource_allocation,
 363     ):
 364         expected_result = {}
 365         guest_epa_quota = {
 366             "no-quota": "nothing",
 367         }
 368         epa_vcpu_set = True
 369
 370         result = Ns._process_guest_epa_quota_params(
 371             guest_epa_quota=guest_epa_quota,
 372             epa_vcpu_set=epa_vcpu_set,
 373         )
 374
 375         self.assertDictEqual(expected_result, result)
 376         self.assertFalse(resource_allocation.called)
 377
 378     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 379     def test__process_guest_epa_quota_params_with_wrong_quota_false_epa_cpu(
 380         self,
 381         resource_allocation,
 382     ):
 383         expected_result = {}
 384         guest_epa_quota = {
 385             "no-quota": "nothing",
 386         }
 387         epa_vcpu_set = False
 388
 389         result = Ns._process_guest_epa_quota_params(
 390             guest_epa_quota=guest_epa_quota,
 391             epa_vcpu_set=epa_vcpu_set,
 392         )
 393
 394         self.assertDictEqual(expected_result, result)
 395         self.assertFalse(resource_allocation.called)
 396
 397     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 398     def test__process_guest_epa_quota_params_with_cpu_quota_epa_cpu(
 399         self,
 400         resource_allocation,
 401     ):
 402         expected_result = {}
 403         guest_epa_quota = {
 404             "cpu-quota": {
 405                 "limit": "10",
 406                 "reserve": "20",
 407                 "shares": "30",
 408             },
 409         }
 410         epa_vcpu_set = True
 411
 412         result = Ns._process_guest_epa_quota_params(
 413             guest_epa_quota=guest_epa_quota,
 414             epa_vcpu_set=epa_vcpu_set,
 415         )
 416
 417         self.assertDictEqual(expected_result, result)
 418         self.assertFalse(resource_allocation.called)
 419
 420     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 421     def test__process_guest_epa_quota_params_with_cpu_quota_false_epa_cpu(
 422         self,
 423         resource_allocation,
 424     ):
 425         expected_result = {
 426             "cpu-quota": {
 427                 "limit": 10,
 428                 "reserve": 20,
 429                 "shares": 30,
 430             },
 431         }
 432         guest_epa_quota = {
 433             "cpu-quota": {
 434                 "limit": "10",
 435                 "reserve": "20",
 436                 "shares": "30",
 437             },
 438         }
 439         epa_vcpu_set = False
 440
 441         resource_allocation_param = {
 442             "limit": "10",
 443             "reserve": "20",
 444             "shares": "30",
 445         }
 446         resource_allocation.return_value = {
 447             "limit": 10,
 448             "reserve": 20,
 449             "shares": 30,
 450         }
 451
 452         result = Ns._process_guest_epa_quota_params(
 453             guest_epa_quota=guest_epa_quota,
 454             epa_vcpu_set=epa_vcpu_set,
 455         )
 456
 457         resource_allocation.assert_called_once_with(resource_allocation_param)
 458         self.assertDictEqual(expected_result, result)
 459
 460     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 461     def test__process_guest_epa_quota_params_with_mem_quota_epa_cpu(
 462         self,
 463         resource_allocation,
 464     ):
 465         expected_result = {
 466             "mem-quota": {
 467                 "limit": 10,
 468                 "reserve": 20,
 469                 "shares": 30,
 470             },
 471         }
 472         guest_epa_quota = {
 473             "mem-quota": {
 474                 "limit": "10",
 475                 "reserve": "20",
 476                 "shares": "30",
 477             },
 478         }
 479         epa_vcpu_set = True
 480
 481         resource_allocation_param = {
 482             "limit": "10",
 483             "reserve": "20",
 484             "shares": "30",
 485         }
 486         resource_allocation.return_value = {
 487             "limit": 10,
 488             "reserve": 20,
 489             "shares": 30,
 490         }
 491
 492         result = Ns._process_guest_epa_quota_params(
 493             guest_epa_quota=guest_epa_quota,
 494             epa_vcpu_set=epa_vcpu_set,
 495         )
 496
 497         resource_allocation.assert_called_once_with(resource_allocation_param)
 498         self.assertDictEqual(expected_result, result)
 499
 500     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 501     def test__process_guest_epa_quota_params_with_mem_quota_false_epa_cpu(
 502         self,
 503         resource_allocation,
 504     ):
 505         expected_result = {
 506             "mem-quota": {
 507                 "limit": 10,
 508                 "reserve": 20,
 509                 "shares": 30,
 510             },
 511         }
 512         guest_epa_quota = {
 513             "mem-quota": {
 514                 "limit": "10",
 515                 "reserve": "20",
 516                 "shares": "30",
 517             },
 518         }
 519         epa_vcpu_set = False
 520
 521         resource_allocation_param = {
 522             "limit": "10",
 523             "reserve": "20",
 524             "shares": "30",
 525         }
 526         resource_allocation.return_value = {
 527             "limit": 10,
 528             "reserve": 20,
 529             "shares": 30,
 530         }
 531
 532         result = Ns._process_guest_epa_quota_params(
 533             guest_epa_quota=guest_epa_quota,
 534             epa_vcpu_set=epa_vcpu_set,
 535         )
 536
 537         resource_allocation.assert_called_once_with(resource_allocation_param)
 538         self.assertDictEqual(expected_result, result)
 539
 540     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 541     def test__process_guest_epa_quota_params_with_disk_io_quota_epa_cpu(
 542         self,
 543         resource_allocation,
 544     ):
 545         expected_result = {
 546             "disk-io-quota": {
 547                 "limit": 10,
 548                 "reserve": 20,
 549                 "shares": 30,
 550             },
 551         }
 552         guest_epa_quota = {
 553             "disk-io-quota": {
 554                 "limit": "10",
 555                 "reserve": "20",
 556                 "shares": "30",
 557             },
 558         }
 559         epa_vcpu_set = True
 560
 561         resource_allocation_param = {
 562             "limit": "10",
 563             "reserve": "20",
 564             "shares": "30",
 565         }
 566         resource_allocation.return_value = {
 567             "limit": 10,
 568             "reserve": 20,
 569             "shares": 30,
 570         }
 571
 572         result = Ns._process_guest_epa_quota_params(
 573             guest_epa_quota=guest_epa_quota,
 574             epa_vcpu_set=epa_vcpu_set,
 575         )
 576
 577         resource_allocation.assert_called_once_with(resource_allocation_param)
 578         self.assertDictEqual(expected_result, result)
 579
 580     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 581     def test__process_guest_epa_quota_params_with_disk_io_quota_false_epa_cpu(
 582         self,
 583         resource_allocation,
 584     ):
 585         expected_result = {
 586             "disk-io-quota": {
 587                 "limit": 10,
 588                 "reserve": 20,
 589                 "shares": 30,
 590             },
 591         }
 592         guest_epa_quota = {
 593             "disk-io-quota": {
 594                 "limit": "10",
 595                 "reserve": "20",
 596                 "shares": "30",
 597             },
 598         }
 599         epa_vcpu_set = False
 600
 601         resource_allocation_param = {
 602             "limit": "10",
 603             "reserve": "20",
 604             "shares": "30",
 605         }
 606         resource_allocation.return_value = {
 607             "limit": 10,
 608             "reserve": 20,
 609             "shares": 30,
 610         }
 611
 612         result = Ns._process_guest_epa_quota_params(
 613             guest_epa_quota=guest_epa_quota,
 614             epa_vcpu_set=epa_vcpu_set,
 615         )
 616
 617         resource_allocation.assert_called_once_with(resource_allocation_param)
 618         self.assertDictEqual(expected_result, result)
 619
 620     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 621     def test__process_guest_epa_quota_params_with_vif_quota_epa_cpu(
 622         self,
 623         resource_allocation,
 624     ):
 625         expected_result = {
 626             "vif-quota": {
 627                 "limit": 10,
 628                 "reserve": 20,
 629                 "shares": 30,
 630             },
 631         }
 632         guest_epa_quota = {
 633             "vif-quota": {
 634                 "limit": "10",
 635                 "reserve": "20",
 636                 "shares": "30",
 637             },
 638         }
 639         epa_vcpu_set = True
 640
 641         resource_allocation_param = {
 642             "limit": "10",
 643             "reserve": "20",
 644             "shares": "30",
 645         }
 646         resource_allocation.return_value = {
 647             "limit": 10,
 648             "reserve": 20,
 649             "shares": 30,
 650         }
 651
 652         result = Ns._process_guest_epa_quota_params(
 653             guest_epa_quota=guest_epa_quota,
 654             epa_vcpu_set=epa_vcpu_set,
 655         )
 656
 657         resource_allocation.assert_called_once_with(resource_allocation_param)
 658         self.assertDictEqual(expected_result, result)
 659
 660     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 661     def test__process_guest_epa_quota_params_with_vif_quota_false_epa_cpu(
 662         self,
 663         resource_allocation,
 664     ):
 665         expected_result = {
 666             "vif-quota": {
 667                 "limit": 10,
 668                 "reserve": 20,
 669                 "shares": 30,
 670             },
 671         }
 672         guest_epa_quota = {
 673             "vif-quota": {
 674                 "limit": "10",
 675                 "reserve": "20",
 676                 "shares": "30",
 677             },
 678         }
 679         epa_vcpu_set = False
 680
 681         resource_allocation_param = {
 682             "limit": "10",
 683             "reserve": "20",
 684             "shares": "30",
 685         }
 686         resource_allocation.return_value = {
 687             "limit": 10,
 688             "reserve": 20,
 689             "shares": 30,
 690         }
 691
 692         result = Ns._process_guest_epa_quota_params(
 693             guest_epa_quota=guest_epa_quota,
 694             epa_vcpu_set=epa_vcpu_set,
 695         )
 696
 697         resource_allocation.assert_called_once_with(resource_allocation_param)
 698         self.assertDictEqual(expected_result, result)
 699
 700     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 701     def test__process_guest_epa_quota_params_with_quota_epa_cpu(
 702         self,
 703         resource_allocation,
 704     ):
 705         expected_result = {
 706             "mem-quota": {
 707                 "limit": 10,
 708                 "reserve": 20,
 709                 "shares": 30,
 710             },
 711             "disk-io-quota": {
 712                 "limit": 10,
 713                 "reserve": 20,
 714                 "shares": 30,
 715             },
 716             "vif-quota": {
 717                 "limit": 10,
 718                 "reserve": 20,
 719                 "shares": 30,
 720             },
 721         }
 722         guest_epa_quota = {
 723             "cpu-quota": {
 724                 "limit": "10",
 725                 "reserve": "20",
 726                 "shares": "30",
 727             },
 728             "mem-quota": {
 729                 "limit": "10",
 730                 "reserve": "20",
 731                 "shares": "30",
 732             },
 733             "disk-io-quota": {
 734                 "limit": "10",
 735                 "reserve": "20",
 736                 "shares": "30",
 737             },
 738             "vif-quota": {
 739                 "limit": "10",
 740                 "reserve": "20",
 741                 "shares": "30",
 742             },
 743         }
 744         epa_vcpu_set = True
 745
 746         resource_allocation.return_value = {
 747             "limit": 10,
 748             "reserve": 20,
 749             "shares": 30,
 750         }
 751
 752         result = Ns._process_guest_epa_quota_params(
 753             guest_epa_quota=guest_epa_quota,
 754             epa_vcpu_set=epa_vcpu_set,
 755         )
 756
 757         self.assertTrue(resource_allocation.called)
 758         self.assertDictEqual(expected_result, result)
 759
 760     @patch("osm_ng_ro.ns.Ns._get_resource_allocation_params")
 761     def test__process_guest_epa_quota_params_with_quota_epa_cpu_no_set(
 762         self,
 763         resource_allocation,
 764     ):
 765         expected_result = {
 766             "cpu-quota": {
 767                 "limit": 10,
 768                 "reserve": 20,
 769                 "shares": 30,
 770             },
 771             "mem-quota": {
 772                 "limit": 10,
 773                 "reserve": 20,
 774                 "shares": 30,
 775             },
 776             "disk-io-quota": {
 777                 "limit": 10,
 778                 "reserve": 20,
 779                 "shares": 30,
 780             },
 781             "vif-quota": {
 782                 "limit": 10,
 783                 "reserve": 20,
 784                 "shares": 30,
 785             },
 786         }
 787         guest_epa_quota = {
 788             "cpu-quota": {
 789                 "limit": "10",
 790                 "reserve": "20",
 791                 "shares": "30",
 792             },
 793             "mem-quota": {
 794                 "limit": "10",
 795                 "reserve": "20",
 796                 "shares": "30",
 797             },
 798             "disk-io-quota": {
 799                 "limit": "10",
 800                 "reserve": "20",
 801                 "shares": "30",
 802             },
 803             "vif-quota": {
 804                 "limit": "10",
 805                 "reserve": "20",
 806                 "shares": "30",
 807             },
 808         }
 809         epa_vcpu_set = False
 810
 811         resource_allocation.return_value = {
 812             "limit": 10,
 813             "reserve": 20,
 814             "shares": 30,
 815         }
 816
 817         result = Ns._process_guest_epa_quota_params(
 818             guest_epa_quota=guest_epa_quota,
 819             epa_vcpu_set=epa_vcpu_set,
 820         )
 821
 822         self.assertTrue(resource_allocation.called)
 823         self.assertDictEqual(expected_result, result)
 824
 825     def test__process_guest_epa_numa_params_with_empty_numa_params(self):
 826         expected_numa_result = {}
 827         expected_epa_vcpu_set_result = False
 828         guest_epa_quota = {}
 829
 830         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 831             guest_epa_quota=guest_epa_quota,
 832         )
 833
 834         self.assertDictEqual(expected_numa_result, numa_result)
 835         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 836
 837     def test__process_guest_epa_numa_params_with_wrong_numa_params(self):
 838         expected_numa_result = {}
 839         expected_epa_vcpu_set_result = False
 840         guest_epa_quota = {"no_nume": "here"}
 841
 842         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 843             guest_epa_quota=guest_epa_quota,
 844         )
 845
 846         self.assertDictEqual(expected_numa_result, numa_result)
 847         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 848
 849     def test__process_guest_epa_numa_params_with_numa_node_policy(self):
 850         expected_numa_result = {}
 851         expected_epa_vcpu_set_result = False
 852         guest_epa_quota = {"numa-node-policy": {}}
 853
 854         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 855             guest_epa_quota=guest_epa_quota,
 856         )
 857
 858         self.assertDictEqual(expected_numa_result, numa_result)
 859         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 860
 861     def test__process_guest_epa_numa_params_with_no_node(self):
 862         expected_numa_result = {}
 863         expected_epa_vcpu_set_result = False
 864         guest_epa_quota = {
 865             "numa-node-policy": {
 866                 "node": [],
 867             },
 868         }
 869
 870         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 871             guest_epa_quota=guest_epa_quota,
 872         )
 873
 874         self.assertDictEqual(expected_numa_result, numa_result)
 875         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 876
 877     def test__process_guest_epa_numa_params_with_1_node_num_cores(self):
 878         expected_numa_result = {"cores": 3}
 879         expected_epa_vcpu_set_result = True
 880         guest_epa_quota = {
 881             "numa-node-policy": {
 882                 "node": [
 883                     {
 884                         "num-cores": 3,
 885                     },
 886                 ],
 887             },
 888         }
 889
 890         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 891             guest_epa_quota=guest_epa_quota,
 892         )
 893
 894         self.assertDictEqual(expected_numa_result, numa_result)
 895         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 896
 897     def test__process_guest_epa_numa_params_with_1_node_paired_threads(self):
 898         expected_numa_result = {"paired-threads": 3}
 899         expected_epa_vcpu_set_result = True
 900         guest_epa_quota = {
 901             "numa-node-policy": {
 902                 "node": [
 903                     {
 904                         "paired-threads": {"num-paired-threads": "3"},
 905                     },
 906                 ],
 907             },
 908         }
 909
 910         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 911             guest_epa_quota=guest_epa_quota,
 912         )
 913
 914         self.assertDictEqual(expected_numa_result, numa_result)
 915         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 916
 917     def test__process_guest_epa_numa_params_with_1_node_paired_threads_ids(self):
 918         expected_numa_result = {
 919             "paired-threads-id": [("0", "1"), ("4", "5")],
 920         }
 921         expected_epa_vcpu_set_result = False
 922         guest_epa_quota = {
 923             "numa-node-policy": {
 924                 "node": [
 925                     {
 926                         "paired-threads": {
 927                             "paired-thread-ids": [
 928                                 {
 929                                     "thread-a": 0,
 930                                     "thread-b": 1,
 931                                 },
 932                                 {
 933                                     "thread-a": 4,
 934                                     "thread-b": 5,
 935                                 },
 936                             ],
 937                         },
 938                     },
 939                 ],
 940             },
 941         }
 942
 943         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 944             guest_epa_quota=guest_epa_quota,
 945         )
 946
 947         self.assertDictEqual(expected_numa_result, numa_result)
 948         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 949
 950     def test__process_guest_epa_numa_params_with_1_node_num_threads(self):
 951         expected_numa_result = {"threads": 3}
 952         expected_epa_vcpu_set_result = True
 953         guest_epa_quota = {
 954             "numa-node-policy": {
 955                 "node": [
 956                     {
 957                         "num-threads": "3",
 958                     },
 959                 ],
 960             },
 961         }
 962
 963         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 964             guest_epa_quota=guest_epa_quota,
 965         )
 966
 967         self.assertDictEqual(expected_numa_result, numa_result)
 968         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 969
 970     def test__process_guest_epa_numa_params_with_1_node_memory_mb(self):
 971         expected_numa_result = {"memory": 2}
 972         expected_epa_vcpu_set_result = False
 973         guest_epa_quota = {
 974             "numa-node-policy": {
 975                 "node": [
 976                     {
 977                         "memory-mb": 2048,
 978                     },
 979                 ],
 980             },
 981         }
 982
 983         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
 984             guest_epa_quota=guest_epa_quota,
 985         )
 986
 987         self.assertDictEqual(expected_numa_result, numa_result)
 988         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
 989
 990     def test__process_guest_epa_numa_params_with_1_node(self):
 991         expected_numa_result = {
 992             "cores": 3,
 993             "paired-threads": 3,
 994             "paired-threads-id": [("0", "1"), ("4", "5")],
 995             "threads": 3,
 996             "memory": 2,
 997         }
 998         expected_epa_vcpu_set_result = True
 999         guest_epa_quota = {
1000             "numa-node-policy": {
1001                 "node": [
1002                     {
1003                         "num-cores": 3,
1004                         "paired-threads": {
1005                             "num-paired-threads": "3",
1006                             "paired-thread-ids": [
1007                                 {
1008                                     "thread-a": 0,
1009                                     "thread-b": 1,
1010                                 },
1011                                 {
1012                                     "thread-a": 4,
1013                                     "thread-b": 5,
1014                                 },
1015                             ],
1016                         },
1017                         "num-threads": "3",
1018                         "memory-mb": 2048,
1019                     },
1020                 ],
1021             },
1022         }
1023
1024         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1025             guest_epa_quota=guest_epa_quota,
1026         )
1027
1028         self.assertDictEqual(expected_numa_result, numa_result)
1029         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1030
1031     def test__process_guest_epa_numa_params_with_2_nodes(self):
1032         expected_numa_result = {
1033             "cores": 3,
1034             "paired-threads": 3,
1035             "paired-threads-id": [("0", "1"), ("4", "5")],
1036             "threads": 3,
1037             "memory": 2,
1038         }
1039         expected_epa_vcpu_set_result = True
1040         guest_epa_quota = {
1041             "numa-node-policy": {
1042                 "node": [
1043                     {
1044                         "num-cores": 3,
1045                         "paired-threads": {
1046                             "num-paired-threads": "3",
1047                             "paired-thread-ids": [
1048                                 {
1049                                     "thread-a": 0,
1050                                     "thread-b": 1,
1051                                 },
1052                                 {
1053                                     "thread-a": 4,
1054                                     "thread-b": 5,
1055                                 },
1056                             ],
1057                         },
1058                         "num-threads": "3",
1059                         "memory-mb": 2048,
1060                     },
1061                     {
1062                         "num-cores": 7,
1063                         "paired-threads": {
1064                             "num-paired-threads": "7",
1065                             "paired-thread-ids": [
1066                                 {
1067                                     "thread-a": 2,
1068                                     "thread-b": 3,
1069                                 },
1070                                 {
1071                                     "thread-a": 5,
1072                                     "thread-b": 6,
1073                                 },
1074                             ],
1075                         },
1076                         "num-threads": "4",
1077                         "memory-mb": 4096,
1078                     },
1079                 ],
1080             },
1081         }
1082
1083         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_numa_params(
1084             guest_epa_quota=guest_epa_quota,
1085         )
1086
1087         self.assertDictEqual(expected_numa_result, numa_result)
1088         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1089
1090     def test__process_guest_epa_cpu_pinning_params_with_empty_params(self):
1091         expected_numa_result = {}
1092         expected_epa_vcpu_set_result = False
1093         guest_epa_quota = {}
1094         vcpu_count = 0
1095         epa_vcpu_set = False
1096
1097         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1098             guest_epa_quota=guest_epa_quota,
1099             vcpu_count=vcpu_count,
1100             epa_vcpu_set=epa_vcpu_set,
1101         )
1102
1103         self.assertDictEqual(expected_numa_result, numa_result)
1104         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1105
1106     def test__process_guest_epa_cpu_pinning_params_with_wrong_params(self):
1107         expected_numa_result = {}
1108         expected_epa_vcpu_set_result = False
1109         guest_epa_quota = {
1110             "no-cpu-pinning-policy": "here",
1111         }
1112         vcpu_count = 0
1113         epa_vcpu_set = False
1114
1115         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1116             guest_epa_quota=guest_epa_quota,
1117             vcpu_count=vcpu_count,
1118             epa_vcpu_set=epa_vcpu_set,
1119         )
1120
1121         self.assertDictEqual(expected_numa_result, numa_result)
1122         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1123
1124     def test__process_guest_epa_cpu_pinning_params_with_epa_vcpu_set(self):
1125         expected_numa_result = {}
1126         expected_epa_vcpu_set_result = True
1127         guest_epa_quota = {
1128             "cpu-pinning-policy": "DEDICATED",
1129         }
1130         vcpu_count = 0
1131         epa_vcpu_set = True
1132
1133         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1134             guest_epa_quota=guest_epa_quota,
1135             vcpu_count=vcpu_count,
1136             epa_vcpu_set=epa_vcpu_set,
1137         )
1138
1139         self.assertDictEqual(expected_numa_result, numa_result)
1140         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1141
1142     def test__process_guest_epa_cpu_pinning_params_with_threads(self):
1143         expected_numa_result = {"threads": 3}
1144         expected_epa_vcpu_set_result = True
1145         guest_epa_quota = {
1146             "cpu-pinning-policy": "DEDICATED",
1147             "cpu-thread-pinning-policy": "PREFER",
1148         }
1149         vcpu_count = 3
1150         epa_vcpu_set = False
1151
1152         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1153             guest_epa_quota=guest_epa_quota,
1154             vcpu_count=vcpu_count,
1155             epa_vcpu_set=epa_vcpu_set,
1156         )
1157
1158         self.assertDictEqual(expected_numa_result, numa_result)
1159         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1160
1161     def test__process_guest_epa_cpu_pinning_params(self):
1162         expected_numa_result = {"cores": 3}
1163         expected_epa_vcpu_set_result = True
1164         guest_epa_quota = {
1165             "cpu-pinning-policy": "DEDICATED",
1166         }
1167         vcpu_count = 3
1168         epa_vcpu_set = False
1169
1170         numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params(
1171             guest_epa_quota=guest_epa_quota,
1172             vcpu_count=vcpu_count,
1173             epa_vcpu_set=epa_vcpu_set,
1174         )
1175
1176         self.assertDictEqual(expected_numa_result, numa_result)
1177         self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result)
1178
1179     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1180     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1181     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1182     def test__process_guest_epa_params_with_empty_params(
1183         self,
1184         guest_epa_numa_params,
1185         guest_epa_cpu_pinning_params,
1186         guest_epa_quota_params,
1187     ):
1188         expected_result = {}
1189         target_flavor = {}
1190
1191         result = Ns._process_epa_params(
1192             target_flavor=target_flavor,
1193         )
1194
1195         self.assertDictEqual(expected_result, result)
1196         self.assertFalse(guest_epa_numa_params.called)
1197         self.assertFalse(guest_epa_cpu_pinning_params.called)
1198         self.assertFalse(guest_epa_quota_params.called)
1199
1200     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1201     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1202     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1203     def test__process_guest_epa_params_with_wrong_params(
1204         self,
1205         guest_epa_numa_params,
1206         guest_epa_cpu_pinning_params,
1207         guest_epa_quota_params,
1208     ):
1209         expected_result = {}
1210         target_flavor = {
1211             "no-guest-epa": "here",
1212         }
1213
1214         result = Ns._process_epa_params(
1215             target_flavor=target_flavor,
1216         )
1217
1218         self.assertDictEqual(expected_result, result)
1219         self.assertFalse(guest_epa_numa_params.called)
1220         self.assertFalse(guest_epa_cpu_pinning_params.called)
1221         self.assertFalse(guest_epa_quota_params.called)
1222
1223     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1224     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1225     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1226     def test__process_guest_epa_params(
1227         self,
1228         guest_epa_numa_params,
1229         guest_epa_cpu_pinning_params,
1230         guest_epa_quota_params,
1231     ):
1232         expected_result = {}
1233         target_flavor = {
1234             "guest-epa": {
1235                 "vcpu-count": 1,
1236             },
1237         }
1238
1239         guest_epa_numa_params.return_value = ({}, False)
1240         guest_epa_cpu_pinning_params.return_value = ({}, False)
1241         guest_epa_quota_params.return_value = {}
1242
1243         result = Ns._process_epa_params(
1244             target_flavor=target_flavor,
1245         )
1246
1247         self.assertDictEqual(expected_result, result)
1248         self.assertTrue(guest_epa_numa_params.called)
1249         self.assertTrue(guest_epa_cpu_pinning_params.called)
1250         self.assertTrue(guest_epa_quota_params.called)
1251
1252     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1253     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1254     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1255     def test__process_guest_epa_params_with_mempage_size(
1256         self,
1257         guest_epa_numa_params,
1258         guest_epa_cpu_pinning_params,
1259         guest_epa_quota_params,
1260     ):
1261         expected_result = {
1262             "mempage-size": "1G",
1263         }
1264         target_flavor = {
1265             "guest-epa": {"vcpu-count": 1, "mempage-size": "1G"},
1266         }
1267
1268         guest_epa_numa_params.return_value = ({}, False)
1269         guest_epa_cpu_pinning_params.return_value = ({}, False)
1270         guest_epa_quota_params.return_value = {}
1271
1272         result = Ns._process_epa_params(
1273             target_flavor=target_flavor,
1274         )
1275
1276         self.assertDictEqual(expected_result, result)
1277         self.assertTrue(guest_epa_numa_params.called)
1278         self.assertTrue(guest_epa_cpu_pinning_params.called)
1279         self.assertTrue(guest_epa_quota_params.called)
1280
1281     @patch("osm_ng_ro.ns.Ns._process_guest_epa_quota_params")
1282     @patch("osm_ng_ro.ns.Ns._process_guest_epa_cpu_pinning_params")
1283     @patch("osm_ng_ro.ns.Ns._process_guest_epa_numa_params")
1284     def test__process_guest_epa_params_with_numa(
1285         self,
1286         guest_epa_numa_params,
1287         guest_epa_cpu_pinning_params,
1288         guest_epa_quota_params,
1289     ):
1290         expected_result = {
1291             "mempage-size": "1G",
1292             "numas": [
1293                 {
1294                     "cores": 3,
1295                     "memory": 2,
1296                     "paired-threads": 3,
1297                     "paired-threads-id": [("0", "1"), ("4", "5")],
1298                     "threads": 3,
1299                 }
1300             ],
1301             "cpu-quota": {"limit": 10, "reserve": 20, "shares": 30},
1302             "disk-io-quota": {"limit": 10, "reserve": 20, "shares": 30},
1303             "mem-quota": {"limit": 10, "reserve": 20, "shares": 30},
1304             "vif-quota": {"limit": 10, "reserve": 20, "shares": 30},
1305         }
1306         target_flavor = {
1307             "guest-epa": {
1308                 "vcpu-count": 1,
1309                 "mempage-size": "1G",
1310                 "cpu-pinning-policy": "DEDICATED",
1311                 "cpu-thread-pinning-policy": "PREFER",
1312                 "numa-node-policy": {
1313                     "node": [
1314                         {
1315                             "num-cores": 3,
1316                             "paired-threads": {
1317                                 "num-paired-threads": "3",
1318                                 "paired-thread-ids": [
1319                                     {
1320                                         "thread-a": 0,
1321                                         "thread-b": 1,
1322                                     },
1323                                     {
1324                                         "thread-a": 4,
1325                                         "thread-b": 5,
1326                                     },
1327                                 ],
1328                             },
1329                             "num-threads": "3",
1330                             "memory-mb": 2048,
1331                         },
1332                     ],
1333                 },
1334                 "cpu-quota": {
1335                     "limit": "10",
1336                     "reserve": "20",
1337                     "shares": "30",
1338                 },
1339                 "mem-quota": {
1340                     "limit": "10",
1341                     "reserve": "20",
1342                     "shares": "30",
1343                 },
1344                 "disk-io-quota": {
1345                     "limit": "10",
1346                     "reserve": "20",
1347                     "shares": "30",
1348                 },
1349                 "vif-quota": {
1350                     "limit": "10",
1351                     "reserve": "20",
1352                     "shares": "30",
1353                 },
1354             },
1355         }
1356
1357         guest_epa_numa_params.return_value = (
1358             {
1359                 "cores": 3,
1360                 "paired-threads": 3,
1361                 "paired-threads-id": [("0", "1"), ("4", "5")],
1362                 "threads": 3,
1363                 "memory": 2,
1364             },
1365             True,
1366         )
1367         guest_epa_cpu_pinning_params.return_value = (
1368             {
1369                 "threads": 3,
1370             },
1371             True,
1372         )
1373         guest_epa_quota_params.return_value = {
1374             "cpu-quota": {
1375                 "limit": 10,
1376                 "reserve": 20,
1377                 "shares": 30,
1378             },
1379             "mem-quota": {
1380                 "limit": 10,
1381                 "reserve": 20,
1382                 "shares": 30,
1383             },
1384             "disk-io-quota": {
1385                 "limit": 10,
1386                 "reserve": 20,
1387                 "shares": 30,
1388             },
1389             "vif-quota": {
1390                 "limit": 10,
1391                 "reserve": 20,
1392                 "shares": 30,
1393             },
1394         }
1395
1396         result = Ns._process_epa_params(
1397             target_flavor=target_flavor,
1398         )
1399
1400         self.assertDictEqual(expected_result, result)
1401         self.assertTrue(guest_epa_numa_params.called)
1402         self.assertTrue(guest_epa_cpu_pinning_params.called)
1403         self.assertTrue(guest_epa_quota_params.called)
1404
1405     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1406     def test__process_flavor_params_with_empty_target_flavor(
1407         self,
1408         epa_params,
1409     ):
1410         target_flavor = {}
1411         indata = {}
1412         vim_info = {}
1413         target_record_id = ""
1414
1415         with self.assertRaises(KeyError):
1416             Ns._process_flavor_params(
1417                 target_flavor=target_flavor,
1418                 indata=indata,
1419                 vim_info=vim_info,
1420                 target_record_id=target_record_id,
1421             )
1422
1423         self.assertFalse(epa_params.called)
1424
1425     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1426     def test__process_flavor_params_with_wrong_target_flavor(
1427         self,
1428         epa_params,
1429     ):
1430         target_flavor = {
1431             "no-target-flavor": "here",
1432         }
1433         indata = {}
1434         vim_info = {}
1435         target_record_id = ""
1436
1437         with self.assertRaises(KeyError):
1438             Ns._process_flavor_params(
1439                 target_flavor=target_flavor,
1440                 indata=indata,
1441                 vim_info=vim_info,
1442                 target_record_id=target_record_id,
1443             )
1444
1445         self.assertFalse(epa_params.called)
1446
1447     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1448     def test__process_flavor_params_with_empty_indata(
1449         self,
1450         epa_params,
1451     ):
1452         expected_result = {
1453             "find_params": {
1454                 "flavor_data": {
1455                     "disk": 10,
1456                     "ram": 1024,
1457                     "vcpus": 2,
1458                 },
1459             },
1460             "params": {
1461                 "flavor_data": {
1462                     "disk": 10,
1463                     "name": "test",
1464                     "ram": 1024,
1465                     "vcpus": 2,
1466                 },
1467             },
1468         }
1469         target_flavor = {
1470             "name": "test",
1471             "storage-gb": "10",
1472             "memory-mb": "1024",
1473             "vcpu-count": "2",
1474         }
1475         indata = {}
1476         vim_info = {}
1477         target_record_id = ""
1478
1479         epa_params.return_value = {}
1480
1481         result = Ns._process_flavor_params(
1482             target_flavor=target_flavor,
1483             indata=indata,
1484             vim_info=vim_info,
1485             target_record_id=target_record_id,
1486         )
1487
1488         self.assertTrue(epa_params.called)
1489         self.assertDictEqual(result, expected_result)
1490
1491     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1492     def test__process_flavor_params_with_wrong_indata(
1493         self,
1494         epa_params,
1495     ):
1496         expected_result = {
1497             "find_params": {
1498                 "flavor_data": {
1499                     "disk": 10,
1500                     "ram": 1024,
1501                     "vcpus": 2,
1502                 },
1503             },
1504             "params": {
1505                 "flavor_data": {
1506                     "disk": 10,
1507                     "name": "test",
1508                     "ram": 1024,
1509                     "vcpus": 2,
1510                 },
1511             },
1512         }
1513         target_flavor = {
1514             "name": "test",
1515             "storage-gb": "10",
1516             "memory-mb": "1024",
1517             "vcpu-count": "2",
1518         }
1519         indata = {
1520             "no-vnf": "here",
1521         }
1522         vim_info = {}
1523         target_record_id = ""
1524
1525         epa_params.return_value = {}
1526
1527         result = Ns._process_flavor_params(
1528             target_flavor=target_flavor,
1529             indata=indata,
1530             vim_info=vim_info,
1531             target_record_id=target_record_id,
1532         )
1533
1534         self.assertTrue(epa_params.called)
1535         self.assertDictEqual(result, expected_result)
1536
1537     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1538     def test__process_flavor_params_with_ephemeral_disk(
1539         self,
1540         epa_params,
1541     ):
1542         expected_result = {
1543             "find_params": {
1544                 "flavor_data": {
1545                     "disk": 10,
1546                     "ram": 1024,
1547                     "vcpus": 2,
1548                     "ephemeral": 10,
1549                 },
1550             },
1551             "params": {
1552                 "flavor_data": {
1553                     "disk": 10,
1554                     "name": "test",
1555                     "ram": 1024,
1556                     "vcpus": 2,
1557                     "ephemeral": 10,
1558                 },
1559             },
1560         }
1561         target_flavor = {
1562             "id": "test_id",
1563             "name": "test",
1564             "storage-gb": "10",
1565             "memory-mb": "1024",
1566             "vcpu-count": "2",
1567         }
1568         indata = {
1569             "vnf": [
1570                 {
1571                     "vdur": [
1572                         {
1573                             "ns-flavor-id": "test_id",
1574                             "virtual-storages": [
1575                                 {
1576                                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1577                                     "size-of-storage": "10",
1578                                 },
1579                             ],
1580                         },
1581                     ],
1582                 },
1583             ],
1584         }
1585         vim_info = {}
1586         target_record_id = ""
1587
1588         epa_params.return_value = {}
1589
1590         result = Ns._process_flavor_params(
1591             target_flavor=target_flavor,
1592             indata=indata,
1593             vim_info=vim_info,
1594             target_record_id=target_record_id,
1595         )
1596
1597         self.assertTrue(epa_params.called)
1598         self.assertDictEqual(result, expected_result)
1599
1600     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1601     def test__process_flavor_params_with_swap_disk(
1602         self,
1603         epa_params,
1604     ):
1605         expected_result = {
1606             "find_params": {
1607                 "flavor_data": {
1608                     "disk": 10,
1609                     "ram": 1024,
1610                     "vcpus": 2,
1611                     "swap": 20,
1612                 },
1613             },
1614             "params": {
1615                 "flavor_data": {
1616                     "disk": 10,
1617                     "name": "test",
1618                     "ram": 1024,
1619                     "vcpus": 2,
1620                     "swap": 20,
1621                 },
1622             },
1623         }
1624         target_flavor = {
1625             "id": "test_id",
1626             "name": "test",
1627             "storage-gb": "10",
1628             "memory-mb": "1024",
1629             "vcpu-count": "2",
1630         }
1631         indata = {
1632             "vnf": [
1633                 {
1634                     "vdur": [
1635                         {
1636                             "ns-flavor-id": "test_id",
1637                             "virtual-storages": [
1638                                 {
1639                                     "type-of-storage": "etsi-nfv-descriptors:swap-storage",
1640                                     "size-of-storage": "20",
1641                                 },
1642                             ],
1643                         },
1644                     ],
1645                 },
1646             ],
1647         }
1648         vim_info = {}
1649         target_record_id = ""
1650
1651         epa_params.return_value = {}
1652
1653         result = Ns._process_flavor_params(
1654             target_flavor=target_flavor,
1655             indata=indata,
1656             vim_info=vim_info,
1657             target_record_id=target_record_id,
1658         )
1659
1660         self.assertTrue(epa_params.called)
1661         self.assertDictEqual(result, expected_result)
1662
1663     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1664     def test__process_flavor_params_with_epa_params(
1665         self,
1666         epa_params,
1667     ):
1668         expected_result = {
1669             "find_params": {
1670                 "flavor_data": {
1671                     "disk": 10,
1672                     "ram": 1024,
1673                     "vcpus": 2,
1674                     "extended": {
1675                         "numa": "there-is-numa-here",
1676                     },
1677                 },
1678             },
1679             "params": {
1680                 "flavor_data": {
1681                     "disk": 10,
1682                     "name": "test",
1683                     "ram": 1024,
1684                     "vcpus": 2,
1685                     "extended": {
1686                         "numa": "there-is-numa-here",
1687                     },
1688                 },
1689             },
1690         }
1691         target_flavor = {
1692             "id": "test_id",
1693             "name": "test",
1694             "storage-gb": "10",
1695             "memory-mb": "1024",
1696             "vcpu-count": "2",
1697         }
1698         indata = {}
1699         vim_info = {}
1700         target_record_id = ""
1701
1702         epa_params.return_value = {
1703             "numa": "there-is-numa-here",
1704         }
1705
1706         result = Ns._process_flavor_params(
1707             target_flavor=target_flavor,
1708             indata=indata,
1709             vim_info=vim_info,
1710             target_record_id=target_record_id,
1711         )
1712
1713         self.assertTrue(epa_params.called)
1714         self.assertDictEqual(result, expected_result)
1715
1716     @patch("osm_ng_ro.ns.Ns._process_epa_params")
1717     def test__process_flavor_params(
1718         self,
1719         epa_params,
1720     ):
1721         expected_result = {
1722             "find_params": {
1723                 "flavor_data": {
1724                     "disk": 10,
1725                     "ram": 1024,
1726                     "vcpus": 2,
1727                     "ephemeral": 10,
1728                     "swap": 20,
1729                     "extended": {
1730                         "numa": "there-is-numa-here",
1731                     },
1732                 },
1733             },
1734             "params": {
1735                 "flavor_data": {
1736                     "disk": 10,
1737                     "name": "test",
1738                     "ram": 1024,
1739                     "vcpus": 2,
1740                     "ephemeral": 10,
1741                     "swap": 20,
1742                     "extended": {
1743                         "numa": "there-is-numa-here",
1744                     },
1745                 },
1746             },
1747         }
1748         target_flavor = {
1749             "id": "test_id",
1750             "name": "test",
1751             "storage-gb": "10",
1752             "memory-mb": "1024",
1753             "vcpu-count": "2",
1754         }
1755         indata = {
1756             "vnf": [
1757                 {
1758                     "vdur": [
1759                         {
1760                             "ns-flavor-id": "test_id",
1761                             "virtual-storages": [
1762                                 {
1763                                     "type-of-storage": "etsi-nfv-descriptors:ephemeral-storage",
1764                                     "size-of-storage": "10",
1765                                 },
1766                                 {
1767                                     "type-of-storage": "etsi-nfv-descriptors:swap-storage",
1768                                     "size-of-storage": "20",
1769                                 },
1770                             ],
1771                         },
1772                     ],
1773                 },
1774             ],
1775         }
1776         vim_info = {}
1777         target_record_id = ""
1778
1779         epa_params.return_value = {
1780             "numa": "there-is-numa-here",
1781         }
1782
1783         result = Ns._process_flavor_params(
1784             target_flavor=target_flavor,
1785             indata=indata,
1786             vim_info=vim_info,
1787             target_record_id=target_record_id,
1788         )
1789
1790         self.assertTrue(epa_params.called)
1791         self.assertDictEqual(result, expected_result)
1792
1793     def test__ip_profile_to_ro_with_none(self):
1794         ip_profile = None
1795
1796         result = Ns._ip_profile_to_ro(
1797             ip_profile=ip_profile,
1798         )
1799
1800         self.assertIsNone(result)
1801
1802     def test__ip_profile_to_ro_with_empty_profile(self):
1803         ip_profile = {}
1804
1805         result = Ns._ip_profile_to_ro(
1806             ip_profile=ip_profile,
1807         )
1808
1809         self.assertIsNone(result)
1810
1811     def test__ip_profile_to_ro_with_wrong_profile(self):
1812         ip_profile = {
1813             "no-profile": "here",
1814         }
1815         expected_result = {
1816             "ip_version": "IPv4",
1817             "subnet_address": None,
1818             "gateway_address": None,
1819             "dhcp_enabled": False,
1820             "dhcp_start_address": None,
1821             "dhcp_count": None,
1822         }
1823
1824         result = Ns._ip_profile_to_ro(
1825             ip_profile=ip_profile,
1826         )
1827
1828         self.assertDictEqual(expected_result, result)
1829
1830     def test__ip_profile_to_ro_with_ipv4_profile(self):
1831         ip_profile = {
1832             "ip-version": "ipv4",
1833             "subnet-address": "192.168.0.0/24",
1834             "gateway-address": "192.168.0.254",
1835             "dhcp-params": {
1836                 "enabled": True,
1837                 "start-address": "192.168.0.10",
1838                 "count": 25,
1839             },
1840         }
1841         expected_result = {
1842             "ip_version": "IPv4",
1843             "subnet_address": "192.168.0.0/24",
1844             "gateway_address": "192.168.0.254",
1845             "dhcp_enabled": True,
1846             "dhcp_start_address": "192.168.0.10",
1847             "dhcp_count": 25,
1848         }
1849
1850         result = Ns._ip_profile_to_ro(
1851             ip_profile=ip_profile,
1852         )
1853
1854         self.assertDictEqual(expected_result, result)
1855
1856     def test__ip_profile_to_ro_with_ipv6_profile(self):
1857         ip_profile = {
1858             "ip-version": "ipv6",
1859             "subnet-address": "2001:0200:0001::/48",
1860             "gateway-address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
1861             "dhcp-params": {
1862                 "enabled": True,
1863                 "start-address": "2001:0200:0001::0010",
1864                 "count": 25,
1865             },
1866         }
1867         expected_result = {
1868             "ip_version": "IPv6",
1869             "subnet_address": "2001:0200:0001::/48",
1870             "gateway_address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe",
1871             "dhcp_enabled": True,
1872             "dhcp_start_address": "2001:0200:0001::0010",
1873             "dhcp_count": 25,
1874         }
1875
1876         result = Ns._ip_profile_to_ro(
1877             ip_profile=ip_profile,
1878         )
1879
1880         self.assertDictEqual(expected_result, result)
1881
1882     def test__ip_profile_to_ro_with_dns_server(self):
1883         ip_profile = {
1884             "ip-version": "ipv4",
1885             "subnet-address": "192.168.0.0/24",
1886             "gateway-address": "192.168.0.254",
1887             "dhcp-params": {
1888                 "enabled": True,
1889                 "start-address": "192.168.0.10",
1890                 "count": 25,
1891             },
1892             "dns-server": [
1893                 {
1894                     "address": "8.8.8.8",
1895                 },
1896                 {
1897                     "address": "1.1.1.1",
1898                 },
1899                 {
1900                     "address": "1.0.0.1",
1901                 },
1902             ],
1903         }
1904         expected_result = {
1905             "ip_version": "IPv4",
1906             "subnet_address": "192.168.0.0/24",
1907             "gateway_address": "192.168.0.254",
1908             "dhcp_enabled": True,
1909             "dhcp_start_address": "192.168.0.10",
1910             "dhcp_count": 25,
1911             "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
1912         }
1913
1914         result = Ns._ip_profile_to_ro(
1915             ip_profile=ip_profile,
1916         )
1917
1918         self.assertDictEqual(expected_result, result)
1919
1920     def test__ip_profile_to_ro_with_security_group(self):
1921         ip_profile = {
1922             "ip-version": "ipv4",
1923             "subnet-address": "192.168.0.0/24",
1924             "gateway-address": "192.168.0.254",
1925             "dhcp-params": {
1926                 "enabled": True,
1927                 "start-address": "192.168.0.10",
1928                 "count": 25,
1929             },
1930             "security-group": {
1931                 "some-security-group": "here",
1932             },
1933         }
1934         expected_result = {
1935             "ip_version": "IPv4",
1936             "subnet_address": "192.168.0.0/24",
1937             "gateway_address": "192.168.0.254",
1938             "dhcp_enabled": True,
1939             "dhcp_start_address": "192.168.0.10",
1940             "dhcp_count": 25,
1941             "security_group": {
1942                 "some-security-group": "here",
1943             },
1944         }
1945
1946         result = Ns._ip_profile_to_ro(
1947             ip_profile=ip_profile,
1948         )
1949
1950         self.assertDictEqual(expected_result, result)
1951
1952     def test__ip_profile_to_ro(self):
1953         ip_profile = {
1954             "ip-version": "ipv4",
1955             "subnet-address": "192.168.0.0/24",
1956             "gateway-address": "192.168.0.254",
1957             "dhcp-params": {
1958                 "enabled": True,
1959                 "start-address": "192.168.0.10",
1960                 "count": 25,
1961             },
1962             "dns-server": [
1963                 {
1964                     "address": "8.8.8.8",
1965                 },
1966                 {
1967                     "address": "1.1.1.1",
1968                 },
1969                 {
1970                     "address": "1.0.0.1",
1971                 },
1972             ],
1973             "security-group": {
1974                 "some-security-group": "here",
1975             },
1976         }
1977         expected_result = {
1978             "ip_version": "IPv4",
1979             "subnet_address": "192.168.0.0/24",
1980             "gateway_address": "192.168.0.254",
1981             "dhcp_enabled": True,
1982             "dhcp_start_address": "192.168.0.10",
1983             "dhcp_count": 25,
1984             "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1",
1985             "security_group": {
1986                 "some-security-group": "here",
1987             },
1988         }
1989
1990         result = Ns._ip_profile_to_ro(
1991             ip_profile=ip_profile,
1992         )
1993
1994         self.assertDictEqual(expected_result, result)
1995
1996     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
1997     def test__process_net_params_with_empty_params(
1998         self,
1999         ip_profile_to_ro,
2000     ):
2001         target_vld = {
2002             "name": "vld-name",
2003         }
2004         indata = {
2005             "name": "ns-name",
2006         }
2007         vim_info = {
2008             "provider_network": "some-profile-here",
2009         }
2010         target_record_id = ""
2011         expected_result = {
2012             "params": {
2013                 "net_name": "ns-name-vld-name",
2014                 "net_type": "bridge",
2015                 "ip_profile": {
2016                     "some_ip_profile": "here",
2017                 },
2018                 "provider_network_profile": "some-profile-here",
2019             }
2020         }
2021
2022         ip_profile_to_ro.return_value = {
2023             "some_ip_profile": "here",
2024         }
2025
2026         result = Ns._process_net_params(
2027             target_vld=target_vld,
2028             indata=indata,
2029             vim_info=vim_info,
2030             target_record_id=target_record_id,
2031         )
2032
2033         self.assertDictEqual(expected_result, result)
2034         self.assertTrue(ip_profile_to_ro.called)
2035
2036     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2037     def test__process_net_params_with_vim_info_sdn(
2038         self,
2039         ip_profile_to_ro,
2040     ):
2041         target_vld = {
2042             "name": "vld-name",
2043         }
2044         indata = {
2045             "name": "ns-name",
2046         }
2047         vim_info = {
2048             "sdn": "some-sdn",
2049             "sdn-ports": ["some", "ports", "here"],
2050             "vlds": ["some", "vlds", "here"],
2051             "type": "sdn-type",
2052         }
2053         target_record_id = "vld.sdn.something"
2054         expected_result = {
2055             "params": {
2056                 "sdn-ports": ["some", "ports", "here"],
2057                 "vlds": ["some", "vlds", "here"],
2058                 "type": "sdn-type",
2059             }
2060         }
2061
2062         result = Ns._process_net_params(
2063             target_vld=target_vld,
2064             indata=indata,
2065             vim_info=vim_info,
2066             target_record_id=target_record_id,
2067         )
2068
2069         self.assertDictEqual(expected_result, result)
2070         self.assertFalse(ip_profile_to_ro.called)
2071
2072     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2073     def test__process_net_params_with_vim_info_sdn_target_vim(
2074         self,
2075         ip_profile_to_ro,
2076     ):
2077         target_vld = {
2078             "name": "vld-name",
2079         }
2080         indata = {
2081             "name": "ns-name",
2082         }
2083         vim_info = {
2084             "sdn": "some-sdn",
2085             "sdn-ports": ["some", "ports", "here"],
2086             "vlds": ["some", "vlds", "here"],
2087             "target_vim": "some-vim",
2088             "type": "sdn-type",
2089         }
2090         target_record_id = "vld.sdn.something"
2091         expected_result = {
2092             "depends_on": ["some-vim vld.sdn"],
2093             "params": {
2094                 "sdn-ports": ["some", "ports", "here"],
2095                 "vlds": ["some", "vlds", "here"],
2096                 "target_vim": "some-vim",
2097                 "type": "sdn-type",
2098             },
2099         }
2100
2101         result = Ns._process_net_params(
2102             target_vld=target_vld,
2103             indata=indata,
2104             vim_info=vim_info,
2105             target_record_id=target_record_id,
2106         )
2107
2108         self.assertDictEqual(expected_result, result)
2109         self.assertFalse(ip_profile_to_ro.called)
2110
2111     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2112     def test__process_net_params_with_vim_network_name(
2113         self,
2114         ip_profile_to_ro,
2115     ):
2116         target_vld = {
2117             "name": "vld-name",
2118         }
2119         indata = {
2120             "name": "ns-name",
2121         }
2122         vim_info = {
2123             "vim_network_name": "some-network-name",
2124         }
2125         target_record_id = "vld.sdn.something"
2126         expected_result = {
2127             "find_params": {
2128                 "filter_dict": {
2129                     "name": "some-network-name",
2130                 },
2131             },
2132         }
2133
2134         result = Ns._process_net_params(
2135             target_vld=target_vld,
2136             indata=indata,
2137             vim_info=vim_info,
2138             target_record_id=target_record_id,
2139         )
2140
2141         self.assertDictEqual(expected_result, result)
2142         self.assertFalse(ip_profile_to_ro.called)
2143
2144     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2145     def test__process_net_params_with_vim_network_id(
2146         self,
2147         ip_profile_to_ro,
2148     ):
2149         target_vld = {
2150             "name": "vld-name",
2151         }
2152         indata = {
2153             "name": "ns-name",
2154         }
2155         vim_info = {
2156             "vim_network_id": "some-network-id",
2157         }
2158         target_record_id = "vld.sdn.something"
2159         expected_result = {
2160             "find_params": {
2161                 "filter_dict": {
2162                     "id": "some-network-id",
2163                 },
2164             },
2165         }
2166
2167         result = Ns._process_net_params(
2168             target_vld=target_vld,
2169             indata=indata,
2170             vim_info=vim_info,
2171             target_record_id=target_record_id,
2172         )
2173
2174         self.assertDictEqual(expected_result, result)
2175         self.assertFalse(ip_profile_to_ro.called)
2176
2177     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2178     def test__process_net_params_with_mgmt_network(
2179         self,
2180         ip_profile_to_ro,
2181     ):
2182         target_vld = {
2183             "id": "vld-id",
2184             "name": "vld-name",
2185             "mgmt-network": "some-mgmt-network",
2186         }
2187         indata = {
2188             "name": "ns-name",
2189         }
2190         vim_info = {}
2191         target_record_id = "vld.sdn.something"
2192         expected_result = {
2193             "find_params": {
2194                 "mgmt": True,
2195                 "name": "vld-id",
2196             },
2197         }
2198
2199         result = Ns._process_net_params(
2200             target_vld=target_vld,
2201             indata=indata,
2202             vim_info=vim_info,
2203             target_record_id=target_record_id,
2204         )
2205
2206         self.assertDictEqual(expected_result, result)
2207         self.assertFalse(ip_profile_to_ro.called)
2208
2209     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2210     def test__process_net_params_with_underlay_eline(
2211         self,
2212         ip_profile_to_ro,
2213     ):
2214         target_vld = {
2215             "name": "vld-name",
2216             "underlay": "some-underlay-here",
2217             "type": "ELINE",
2218         }
2219         indata = {
2220             "name": "ns-name",
2221         }
2222         vim_info = {
2223             "provider_network": "some-profile-here",
2224         }
2225         target_record_id = ""
2226         expected_result = {
2227             "params": {
2228                 "ip_profile": {
2229                     "some_ip_profile": "here",
2230                 },
2231                 "net_name": "ns-name-vld-name",
2232                 "net_type": "ptp",
2233                 "provider_network_profile": "some-profile-here",
2234             }
2235         }
2236
2237         ip_profile_to_ro.return_value = {
2238             "some_ip_profile": "here",
2239         }
2240
2241         result = Ns._process_net_params(
2242             target_vld=target_vld,
2243             indata=indata,
2244             vim_info=vim_info,
2245             target_record_id=target_record_id,
2246         )
2247
2248         self.assertDictEqual(expected_result, result)
2249         self.assertTrue(ip_profile_to_ro.called)
2250
2251     @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
2252     def test__process_net_params_with_underlay_elan(
2253         self,
2254         ip_profile_to_ro,
2255     ):
2256         target_vld = {
2257             "name": "vld-name",
2258             "underlay": "some-underlay-here",
2259             "type": "ELAN",
2260         }
2261         indata = {
2262             "name": "ns-name",
2263         }
2264         vim_info = {
2265             "provider_network": "some-profile-here",
2266         }
2267         target_record_id = ""
2268         expected_result = {
2269             "params": {
2270                 "ip_profile": {
2271                     "some_ip_profile": "here",
2272                 },
2273                 "net_name": "ns-name-vld-name",
2274                 "net_type": "data",
2275                 "provider_network_profile": "some-profile-here",
2276             }
2277         }
2278
2279         ip_profile_to_ro.return_value = {
2280             "some_ip_profile": "here",
2281         }
2282
2283         result = Ns._process_net_params(
2284             target_vld=target_vld,
2285             indata=indata,
2286             vim_info=vim_info,
2287             target_record_id=target_record_id,
2288         )
2289
2290         self.assertDictEqual(expected_result, result)
2291         self.assertTrue(ip_profile_to_ro.called)
2292
2293     def test__get_cloud_init_exception(self):
2294         db_mock = MagicMock(name="database mock")
2295         fs_mock = None
2296
2297         location = ""
2298
2299         with self.assertRaises(NsException):
2300             Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2301
2302     def test__get_cloud_init_file_fs_exception(self):
2303         db_mock = MagicMock(name="database mock")
2304         fs_mock = None
2305
2306         location = "vnfr_id_123456:file:test_file"
2307         db_mock.get_one.return_value = {
2308             "_admin": {
2309                 "storage": {
2310                     "folder": "/home/osm",
2311                     "pkg-dir": "vnfr_test_dir",
2312                 },
2313             },
2314         }
2315
2316         with self.assertRaises(NsException):
2317             Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2318
2319     def test__get_cloud_init_file(self):
2320         db_mock = MagicMock(name="database mock")
2321         fs_mock = MagicMock(name="filesystem mock")
2322         file_mock = MagicMock(name="file mock")
2323
2324         location = "vnfr_id_123456:file:test_file"
2325         cloud_init_content = "this is a cloud init file content"
2326
2327         db_mock.get_one.return_value = {
2328             "_admin": {
2329                 "storage": {
2330                     "folder": "/home/osm",
2331                     "pkg-dir": "vnfr_test_dir",
2332                 },
2333             },
2334         }
2335         fs_mock.file_open.return_value = file_mock
2336         file_mock.__enter__.return_value.read.return_value = cloud_init_content
2337
2338         result = Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2339
2340         self.assertEqual(cloud_init_content, result)
2341
2342     def test__get_cloud_init_vdu(self):
2343         db_mock = MagicMock(name="database mock")
2344         fs_mock = None
2345
2346         location = "vnfr_id_123456:vdu:0"
2347         cloud_init_content = "this is a cloud init file content"
2348
2349         db_mock.get_one.return_value = {
2350             "vdu": {
2351                 0: {
2352                     "cloud-init": cloud_init_content,
2353                 },
2354             },
2355         }
2356
2357         result = Ns._get_cloud_init(db=db_mock, fs=fs_mock, location=location)
2358
2359         self.assertEqual(cloud_init_content, result)
2360
2361     @patch("jinja2.Environment.__init__")
2362     def test__parse_jinja2_undefined_error(self, env_mock: Mock):
2363         cloud_init_content = None
2364         params = None
2365         context = None
2366
2367         env_mock.side_effect = UndefinedError("UndefinedError occurred.")
2368
2369         with self.assertRaises(NsException):
2370             Ns._parse_jinja2(
2371                 cloud_init_content=cloud_init_content, params=params, context=context
2372             )
2373
2374     @patch("jinja2.Environment.__init__")
2375     def test__parse_jinja2_template_error(self, env_mock: Mock):
2376         cloud_init_content = None
2377         params = None
2378         context = None
2379
2380         env_mock.side_effect = TemplateError("TemplateError occurred.")
2381
2382         with self.assertRaises(NsException):
2383             Ns._parse_jinja2(
2384                 cloud_init_content=cloud_init_content, params=params, context=context
2385             )
2386
2387     @patch("jinja2.Environment.__init__")
2388     def test__parse_jinja2_template_not_found(self, env_mock: Mock):
2389         cloud_init_content = None
2390         params = None
2391         context = None
2392
2393         env_mock.side_effect = TemplateNotFound("TemplateNotFound occurred.")
2394
2395         with self.assertRaises(NsException):
2396             Ns._parse_jinja2(
2397                 cloud_init_content=cloud_init_content, params=params, context=context
2398             )
2399
2400     def test__parse_jinja2(self):
2401         pass
2402
2403     def test__process_vdu_params_empty_kargs(self):
2404         pass
2405
2406     def test__process_vdu_params_interface_ns_vld_id(self):
2407         pass
2408
2409     def test__process_vdu_params_interface_vnf_vld_id(self):
2410         pass
2411
2412     def test__process_vdu_params_interface_unknown(self):
2413         pass
2414
2415     def test__process_vdu_params_interface_port_security_enabled(self):
2416         pass
2417
2418     def test__process_vdu_params_interface_port_security_disable_strategy(self):
2419         pass
2420
2421     def test__process_vdu_params_interface_sriov(self):
2422         pass
2423
2424     def test__process_vdu_params_interface_pci_passthrough(self):
2425         pass
2426
2427     def test__process_vdu_params_interface_om_mgmt(self):
2428         pass
2429
2430     def test__process_vdu_params_interface_mgmt_interface(self):
2431         pass
2432
2433     def test__process_vdu_params_interface_mgmt_vnf(self):
2434         pass
2435
2436     def test__process_vdu_params_interface_bridge(self):
2437         pass
2438
2439     def test__process_vdu_params_interface_ip_address(self):
2440         pass
2441
2442     def test__process_vdu_params_interface_mac_address(self):
2443         pass
2444
2445     def test__process_vdu_params_vdu_cloud_init_missing(self):
2446         pass
2447
2448     def test__process_vdu_params_vdu_cloud_init_present(self):
2449         pass
2450
2451     def test__process_vdu_params_vdu_boot_data_drive(self):
2452         pass
2453
2454     def test__process_vdu_params_vdu_ssh_keys(self):
2455         pass
2456
2457     def test__process_vdu_params_vdu_ssh_access_required(self):
2458         pass
2459
2460     def test__process_vdu_params_vdu_virtual_storages(self):
2461         pass
2462
2463     def test__process_vdu_params(self):
2464         pass