X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=inline;f=NG-RO%2Fosm_ng_ro%2Ftests%2Ftest_ns.py;h=d966a861ce82cb7747318213ec41cb1c2bf1b9d1;hb=730cfaff466fb3c9b1446ecef5213916195ff861;hp=f4a1c87a85ade201c3b04a0b66e50b47a2e4c96d;hpb=4bc8eb99dbadadf5013de93fcb424038149e1576;p=osm%2FRO.git diff --git a/NG-RO/osm_ng_ro/tests/test_ns.py b/NG-RO/osm_ng_ro/tests/test_ns.py index f4a1c87a..d966a861 100644 --- a/NG-RO/osm_ng_ro/tests/test_ns.py +++ b/NG-RO/osm_ng_ro/tests/test_ns.py @@ -71,6 +71,9 @@ vnfd_wth_persistent_storage = { "id": "persistent-root-volume", "type-of-storage": "persistent-storage:persistent-storage", "size-of-storage": "10", + "vdu-storage-requirements": [ + {"key": "keep-volume", "value": "true"}, + ], }, { "id": "ephemeral-volume", @@ -144,6 +147,9 @@ target_vdu_wth_persistent_storage = { "id": "persistent-root-volume", "size-of-storage": "10", "type-of-storage": "persistent-storage:persistent-storage", + "vdu-storage-requirements": [ + {"key": "keep-volume", "value": "true"}, + ], }, { "id": "ephemeral-volume", @@ -200,6 +206,25 @@ expected_extra_dict2 = { "start": True, }, } + +expected_extra_dict3 = { + "depends_on": [ + f"{ns_preffix}:image.0", + ], + "params": { + "affinity_group_list": [], + "availability_zone_index": None, + "availability_zone_list": None, + "cloud_config": None, + "description": "without_volumes-VM", + "disk_list": [], + "flavor_id": "flavor_test", + "image_id": f"TASK-{ns_preffix}:image.0", + "name": "sample_name-vnf-several-volu-without_volumes-VM-0", + "net_list": [], + "start": True, + }, +} tasks_by_target_record_id = { "nsrs:th47f48-9870-4169-b758-9732e1ff40f3": { "extra_dict": { @@ -1488,7 +1513,7 @@ class TestNs(unittest.TestCase): self.assertDictEqual(expected_numa_result, numa_result) self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result) - def test__process_guest_epa_cpu_pinning_params_with_threads(self): + def test__process_guest_epa_cpu_pinning_params_with_policy_prefer(self): expected_numa_result = {"threads": 3} expected_epa_vcpu_set_result = True guest_epa_quota = { @@ -1507,9 +1532,47 @@ class TestNs(unittest.TestCase): self.assertDictEqual(expected_numa_result, numa_result) self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result) - def test__process_guest_epa_cpu_pinning_params(self): + def test__process_guest_epa_cpu_pinning_params_with_policy_isolate(self): expected_numa_result = {"cores": 3} expected_epa_vcpu_set_result = True + guest_epa_quota = { + "cpu-pinning-policy": "DEDICATED", + "cpu-thread-pinning-policy": "ISOLATE", + } + vcpu_count = 3 + epa_vcpu_set = False + + numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params( + guest_epa_quota=guest_epa_quota, + vcpu_count=vcpu_count, + epa_vcpu_set=epa_vcpu_set, + ) + + self.assertDictEqual(expected_numa_result, numa_result) + self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result) + + def test__process_guest_epa_cpu_pinning_params_with_policy_require(self): + expected_numa_result = {"threads": 3} + expected_epa_vcpu_set_result = True + guest_epa_quota = { + "cpu-pinning-policy": "DEDICATED", + "cpu-thread-pinning-policy": "REQUIRE", + } + vcpu_count = 3 + epa_vcpu_set = False + + numa_result, epa_vcpu_set_result = Ns._process_guest_epa_cpu_pinning_params( + guest_epa_quota=guest_epa_quota, + vcpu_count=vcpu_count, + epa_vcpu_set=epa_vcpu_set, + ) + + self.assertDictEqual(expected_numa_result, numa_result) + self.assertEqual(expected_epa_vcpu_set_result, epa_vcpu_set_result) + + def test__process_guest_epa_cpu_pinning_params(self): + expected_numa_result = {"threads": 3} + expected_epa_vcpu_set_result = True guest_epa_quota = { "cpu-pinning-policy": "DEDICATED", } @@ -1771,7 +1834,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - target_flavor = {} indata = { "vnf": [ @@ -1798,7 +1860,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - target_flavor = { "no-target-flavor": "here", } @@ -1821,7 +1882,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - expected_result = { "find_params": { "flavor_data": { @@ -1866,7 +1926,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - expected_result = { "find_params": { "flavor_data": { @@ -2023,7 +2082,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - expected_result = { "find_params": { "flavor_data": { @@ -2088,7 +2146,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - kwargs = { "db": db, } @@ -2199,7 +2256,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - expected_result = { "find_params": { "flavor_data": { @@ -2264,7 +2320,6 @@ class TestNs(unittest.TestCase): self, epa_params, ): - kwargs = { "db": db, } @@ -2385,213 +2440,8 @@ class TestNs(unittest.TestCase): self.assertTrue(epa_params.called) self.assertDictEqual(result, expected_result) - def test__ip_profile_to_ro_with_none(self): - ip_profile = None - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertIsNone(result) - - def test__ip_profile_to_ro_with_empty_profile(self): - ip_profile = {} - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertIsNone(result) - - def test__ip_profile_to_ro_with_wrong_profile(self): - ip_profile = { - "no-profile": "here", - } - expected_result = { - "ip_version": "IPv4", - "subnet_address": None, - "gateway_address": None, - "dhcp_enabled": False, - "dhcp_start_address": None, - "dhcp_count": None, - } - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertDictEqual(expected_result, result) - - def test__ip_profile_to_ro_with_ipv4_profile(self): - ip_profile = { - "ip-version": "ipv4", - "subnet-address": "192.168.0.0/24", - "gateway-address": "192.168.0.254", - "dhcp-params": { - "enabled": True, - "start-address": "192.168.0.10", - "count": 25, - }, - } - expected_result = { - "ip_version": "IPv4", - "subnet_address": "192.168.0.0/24", - "gateway_address": "192.168.0.254", - "dhcp_enabled": True, - "dhcp_start_address": "192.168.0.10", - "dhcp_count": 25, - } - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertDictEqual(expected_result, result) - - def test__ip_profile_to_ro_with_ipv6_profile(self): - ip_profile = { - "ip-version": "ipv6", - "subnet-address": "2001:0200:0001::/48", - "gateway-address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe", - "dhcp-params": { - "enabled": True, - "start-address": "2001:0200:0001::0010", - "count": 25, - }, - } - expected_result = { - "ip_version": "IPv6", - "subnet_address": "2001:0200:0001::/48", - "gateway_address": "2001:0200:0001:ffff:ffff:ffff:ffff:fffe", - "dhcp_enabled": True, - "dhcp_start_address": "2001:0200:0001::0010", - "dhcp_count": 25, - } - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertDictEqual(expected_result, result) - - def test__ip_profile_to_ro_with_dns_server(self): - ip_profile = { - "ip-version": "ipv4", - "subnet-address": "192.168.0.0/24", - "gateway-address": "192.168.0.254", - "dhcp-params": { - "enabled": True, - "start-address": "192.168.0.10", - "count": 25, - }, - "dns-server": [ - { - "address": "8.8.8.8", - }, - { - "address": "1.1.1.1", - }, - { - "address": "1.0.0.1", - }, - ], - } - expected_result = { - "ip_version": "IPv4", - "subnet_address": "192.168.0.0/24", - "gateway_address": "192.168.0.254", - "dhcp_enabled": True, - "dhcp_start_address": "192.168.0.10", - "dhcp_count": 25, - "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1", - } - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertDictEqual(expected_result, result) - - def test__ip_profile_to_ro_with_security_group(self): - ip_profile = { - "ip-version": "ipv4", - "subnet-address": "192.168.0.0/24", - "gateway-address": "192.168.0.254", - "dhcp-params": { - "enabled": True, - "start-address": "192.168.0.10", - "count": 25, - }, - "security-group": { - "some-security-group": "here", - }, - } - expected_result = { - "ip_version": "IPv4", - "subnet_address": "192.168.0.0/24", - "gateway_address": "192.168.0.254", - "dhcp_enabled": True, - "dhcp_start_address": "192.168.0.10", - "dhcp_count": 25, - "security_group": { - "some-security-group": "here", - }, - } - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertDictEqual(expected_result, result) - - def test__ip_profile_to_ro(self): - ip_profile = { - "ip-version": "ipv4", - "subnet-address": "192.168.0.0/24", - "gateway-address": "192.168.0.254", - "dhcp-params": { - "enabled": True, - "start-address": "192.168.0.10", - "count": 25, - }, - "dns-server": [ - { - "address": "8.8.8.8", - }, - { - "address": "1.1.1.1", - }, - { - "address": "1.0.0.1", - }, - ], - "security-group": { - "some-security-group": "here", - }, - } - expected_result = { - "ip_version": "IPv4", - "subnet_address": "192.168.0.0/24", - "gateway_address": "192.168.0.254", - "dhcp_enabled": True, - "dhcp_start_address": "192.168.0.10", - "dhcp_count": 25, - "dns_address": "8.8.8.8;1.1.1.1;1.0.0.1", - "security_group": { - "some-security-group": "here", - }, - } - - result = Ns._ip_profile_to_ro( - ip_profile=ip_profile, - ) - - self.assertDictEqual(expected_result, result) - - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_empty_params( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2601,6 +2451,9 @@ class TestNs(unittest.TestCase): } vim_info = { "provider_network": "some-profile-here", + "ip_profile": { + "some_ip_profile": "here", + }, } target_record_id = "" expected_result = { @@ -2614,10 +2467,6 @@ class TestNs(unittest.TestCase): } } - ip_profile_to_ro.return_value = { - "some_ip_profile": "here", - } - result = Ns._process_net_params( target_vld=target_vld, indata=indata, @@ -2626,12 +2475,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertTrue(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_vim_info_sdn( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2662,12 +2508,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertFalse(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_vim_info_sdn_target_vim( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2701,12 +2544,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertFalse(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_vim_network_name( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2734,12 +2574,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertFalse(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_vim_network_id( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2767,12 +2604,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertFalse(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_mgmt_network( self, - ip_profile_to_ro, ): target_vld = { "id": "vld-id", @@ -2799,12 +2633,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertFalse(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_underlay_eline( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2816,6 +2647,9 @@ class TestNs(unittest.TestCase): } vim_info = { "provider_network": "some-profile-here", + "ip_profile": { + "some_ip_profile": "here", + }, } target_record_id = "" expected_result = { @@ -2829,10 +2663,6 @@ class TestNs(unittest.TestCase): } } - ip_profile_to_ro.return_value = { - "some_ip_profile": "here", - } - result = Ns._process_net_params( target_vld=target_vld, indata=indata, @@ -2841,12 +2671,9 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertTrue(ip_profile_to_ro.called) - @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro") def test__process_net_params_with_underlay_elan( self, - ip_profile_to_ro, ): target_vld = { "name": "vld-name", @@ -2858,6 +2685,9 @@ class TestNs(unittest.TestCase): } vim_info = { "provider_network": "some-profile-here", + "ip_profile": { + "some_ip_profile": "here", + }, } target_record_id = "" expected_result = { @@ -2871,10 +2701,6 @@ class TestNs(unittest.TestCase): } } - ip_profile_to_ro.return_value = { - "some_ip_profile": "here", - } - result = Ns._process_net_params( target_vld=target_vld, indata=indata, @@ -2883,7 +2709,6 @@ class TestNs(unittest.TestCase): ) self.assertDictEqual(expected_result, result) - self.assertTrue(ip_profile_to_ro.called) def test__get_cloud_init_exception(self): db_mock = MagicMock(name="database mock") @@ -3239,32 +3064,48 @@ class TestProcessVduParams(unittest.TestCase): self.ns = Ns() self.logger = CopyingMock(autospec=True) - def test_find_persistent_root_volumes_empty_instantiation_vol_list(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_empty_instantiation_vol_list( + self, mock_volume_keeping_required + ): """Find persistent root volume, instantiation_vol_list is empty.""" vnfd = deepcopy(vnfd_wth_persistent_storage) target_vdu = target_vdu_wth_persistent_storage vdu_instantiation_volumes_list = [] disk_list = [] + mock_volume_keeping_required.return_value = True + expected_root_disk = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + "vdu-storage-requirements": [{"key": "keep-volume", "value": "true"}], + } expected_persist_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, } } expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": True, }, ] persist_root_disk = self.ns.find_persistent_root_volumes( vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(persist_root_disk, expected_persist_root_disk) + mock_volume_keeping_required.assert_called_once_with(expected_root_disk) self.assertEqual(disk_list, expected_disk_list) self.assertEqual(len(disk_list), 1) - def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_always_selects_first_vsd_as_root( + self, mock_volume_keeping_required + ): """Find persistent root volume, always selects the first vsd as root volume.""" vnfd = deepcopy(vnfd_wth_persistent_storage) vnfd["vdu"][0]["virtual-storage-desc"] = [ @@ -3275,26 +3116,38 @@ class TestProcessVduParams(unittest.TestCase): target_vdu = target_vdu_wth_persistent_storage vdu_instantiation_volumes_list = [] disk_list = [] + mock_volume_keeping_required.return_value = True + expected_root_disk = { + "id": "persistent-volume2", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + } expected_persist_root_disk = { "persistent-volume2": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, } } expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": True, }, ] persist_root_disk = self.ns.find_persistent_root_volumes( vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(persist_root_disk, expected_persist_root_disk) + mock_volume_keeping_required.assert_called_once_with(expected_root_disk) self.assertEqual(disk_list, expected_disk_list) self.assertEqual(len(disk_list), 1) - def test_find_persistent_root_volumes_empty_size_of_storage(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_empty_size_of_storage( + self, mock_volume_keeping_required + ): """Find persistent root volume, size of storage is empty.""" vnfd = deepcopy(vnfd_wth_persistent_storage) vnfd["virtual-storage-desc"][0]["size-of-storage"] = "" @@ -3310,34 +3163,54 @@ class TestProcessVduParams(unittest.TestCase): vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(persist_root_disk, None) + mock_volume_keeping_required.assert_not_called() self.assertEqual(disk_list, []) - def test_find_persistent_root_empty_disk_list(self): - """Find persistent root volume, empty disk list.""" + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_keeping_is_not_required( + self, mock_volume_keeping_required + ): + """Find persistent root volume, volume keeping is not required.""" vnfd = deepcopy(vnfd_wth_persistent_storage) + vnfd["virtual-storage-desc"][1]["vdu-storage-requirements"] = [ + {"key": "keep-volume", "value": "false"}, + ] target_vdu = target_vdu_wth_persistent_storage vdu_instantiation_volumes_list = [] disk_list = [] + mock_volume_keeping_required.return_value = False + expected_root_disk = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + "vdu-storage-requirements": [{"key": "keep-volume", "value": "false"}], + } expected_persist_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": False, } } expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, ] persist_root_disk = self.ns.find_persistent_root_volumes( vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(persist_root_disk, expected_persist_root_disk) + mock_volume_keeping_required.assert_called_once_with(expected_root_disk) self.assertEqual(disk_list, expected_disk_list) self.assertEqual(len(disk_list), 1) - def test_find_persistent_root_volumes_target_vdu_mismatch(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_target_vdu_mismatch( + self, mock_volume_keeping_required + ): """Find persistent root volume, target vdu name is not matching.""" vnfd = deepcopy(vnfd_wth_persistent_storage) vnfd["vdu"][0]["name"] = "Several_Volumes-VM" @@ -3348,10 +3221,14 @@ class TestProcessVduParams(unittest.TestCase): vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(result, None) + mock_volume_keeping_required.assert_not_called() self.assertEqual(disk_list, []) self.assertEqual(len(disk_list), 0) - def test_find_persistent_root_volumes_with_instantiation_vol_list(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_with_instantiation_vol_list( + self, mock_volume_keeping_required + ): """Find persistent root volume, existing volume needs to be used.""" vnfd = deepcopy(vnfd_wth_persistent_storage) target_vdu = target_vdu_wth_persistent_storage @@ -3378,10 +3255,14 @@ class TestProcessVduParams(unittest.TestCase): vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(persist_root_disk, expected_persist_root_disk) + mock_volume_keeping_required.assert_not_called() self.assertEqual(disk_list, expected_disk_list) self.assertEqual(len(disk_list), 1) - def test_find_persistent_root_volumes_invalid_instantiation_params(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_root_volumes_invalid_instantiation_params( + self, mock_volume_keeping_required + ): """Find persistent root volume, existing volume id keyword is invalid.""" vnfd = deepcopy(vnfd_wth_persistent_storage) target_vdu = target_vdu_wth_persistent_storage @@ -3396,50 +3277,64 @@ class TestProcessVduParams(unittest.TestCase): self.ns.find_persistent_root_volumes( vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) - + mock_volume_keeping_required.assert_not_called() self.assertEqual(disk_list, []) self.assertEqual(len(disk_list), 0) + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list( - self, + self, mock_volume_keeping_required ): """Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty.""" persistent_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": False, } } - + mock_volume_keeping_required.return_value = False target_vdu = target_vdu_wth_persistent_storage vdu_instantiation_volumes_list = [] disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, ] - + expected_disk = { + "id": "persistent-volume2", + "size-of-storage": "10", + "type-of-storage": "persistent-storage:persistent-storage", + } expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, { "size": "10", + "keep": False, }, ] self.ns.find_persistent_volumes( persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(disk_list, expected_disk_list) + mock_volume_keeping_required.assert_called_once_with(expected_disk) - def test_find_persistent_volumes_vdu_wth_inst_vol_list(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_volumes_vdu_wth_inst_vol_list( + self, mock_volume_keeping_required + ): """Find persistent ordinary volume, vim-volume-id is given as instantiation parameter.""" persistent_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": False, } } vdu_instantiation_volumes_list = [ @@ -3453,12 +3348,14 @@ class TestProcessVduParams(unittest.TestCase): { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, ] expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, { "vim_volume_id": vim_volume_id, @@ -3468,35 +3365,47 @@ class TestProcessVduParams(unittest.TestCase): persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(disk_list, expected_disk_list) + mock_volume_keeping_required.assert_not_called() - def test_find_persistent_volumes_vdu_wthout_persistent_storage(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_volumes_vdu_wthout_persistent_storage( + self, mock_volume_keeping_required + ): """Find persistent ordinary volume, there is not any persistent disk.""" persistent_root_disk = {} vdu_instantiation_volumes_list = [] + mock_volume_keeping_required.return_value = False target_vdu = target_vdu_wthout_persistent_storage disk_list = [] self.ns.find_persistent_volumes( persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(disk_list, disk_list) + mock_volume_keeping_required.assert_not_called() + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk( - self, + self, mock_volume_keeping_required ): """There is persistent root disk, but there is not ordinary persistent disk.""" persistent_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": False, } } vdu_instantiation_volumes_list = [] + mock_volume_keeping_required.return_value = False target_vdu = deepcopy(target_vdu_wth_persistent_storage) target_vdu["virtual-storages"] = [ { "id": "persistent-root-volume", "size-of-storage": "10", "type-of-storage": "persistent-storage:persistent-storage", + "vdu-storage-requirements": [ + {"key": "keep-volume", "value": "true"}, + ], }, { "id": "ephemeral-volume", @@ -3508,16 +3417,23 @@ class TestProcessVduParams(unittest.TestCase): { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, ] self.ns.find_persistent_volumes( persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(disk_list, disk_list) + mock_volume_keeping_required.assert_not_called() - def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch( + self, mock_volume_keeping_required + ): """Find persistent ordinary volume, volume id is not persistent_root_disk dict, - vim-volume-id is given as instantiation parameter but disk id is not matching.""" + vim-volume-id is given as instantiation parameter but disk id is not matching. + """ + mock_volume_keeping_required.return_value = True vdu_instantiation_volumes_list = [ { "vim-volume-id": vim_volume_id, @@ -3528,29 +3444,87 @@ class TestProcessVduParams(unittest.TestCase): "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": False, } } disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, ] expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, }, { "size": "10", + "keep": True, }, ] - + expected_disk = { + "id": "persistent-volume2", + "size-of-storage": "10", + "type-of-storage": "persistent-storage:persistent-storage", + } target_vdu = target_vdu_wth_persistent_storage self.ns.find_persistent_volumes( persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list ) self.assertEqual(disk_list, expected_disk_list) + mock_volume_keeping_required.assert_called_once_with(expected_disk) + + def test_is_volume_keeping_required_true(self): + """Volume keeping is required.""" + virtual_storage_descriptor = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + "vdu-storage-requirements": [ + {"key": "keep-volume", "value": "true"}, + ], + } + result = self.ns.is_volume_keeping_required(virtual_storage_descriptor) + self.assertEqual(result, True) + + def test_is_volume_keeping_required_false(self): + """Volume keeping is not required.""" + virtual_storage_descriptor = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + "vdu-storage-requirements": [ + {"key": "keep-volume", "value": "false"}, + ], + } + result = self.ns.is_volume_keeping_required(virtual_storage_descriptor) + self.assertEqual(result, False) + + def test_is_volume_keeping_required_wthout_vdu_storage_reqirement(self): + """Volume keeping is not required, vdu-storage-requirements key does not exist.""" + virtual_storage_descriptor = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + } + result = self.ns.is_volume_keeping_required(virtual_storage_descriptor) + self.assertEqual(result, False) + + def test_is_volume_keeping_required_wrong_keyword(self): + """vdu-storage-requirements key to indicate keeping-volume is wrong.""" + virtual_storage_descriptor = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + "vdu-storage-requirements": [ + {"key": "hold-volume", "value": "true"}, + ], + } + result = self.ns.is_volume_keeping_required(virtual_storage_descriptor) + self.assertEqual(result, False) def test_sort_vdu_interfaces_position_all_wth_positions(self): """Interfaces are sorted according to position, all have positions.""" @@ -4459,10 +4433,11 @@ class TestProcessVduParams(unittest.TestCase): self.assertDictEqual(cloud_config, expected_cloud_config) @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk") - def test_add_persistent_root_disk_to_disk_list( - self, mock_select_persistent_root_disk + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_add_persistent_root_disk_to_disk_list_keep_false( + self, mock_volume_keeping_required, mock_select_persistent_root_disk ): - """Add persistent root disk to disk_list""" + """Add persistent root disk to disk_list, keep volume set to False.""" root_disk = { "id": "persistent-root-volume", "type-of-storage": "persistent-storage:persistent-storage", @@ -4474,10 +4449,12 @@ class TestProcessVduParams(unittest.TestCase): target_vdu = deepcopy(target_vdu_wth_persistent_storage) persistent_root_disk = {} disk_list = [] + mock_volume_keeping_required.return_value = False expected_disk_list = [ { "image_id": "ubuntu20.04", "size": "10", + "keep": False, } ] self.ns._add_persistent_root_disk_to_disk_list( @@ -4485,10 +4462,12 @@ class TestProcessVduParams(unittest.TestCase): ) self.assertEqual(disk_list, expected_disk_list) mock_select_persistent_root_disk.assert_called_once() + mock_volume_keeping_required.assert_called_once() @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk") + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises( - self, mock_select_persistent_root_disk + self, mock_volume_keeping_required, mock_select_persistent_root_disk ): """Add persistent root disk to disk_list""" root_disk = { @@ -4508,35 +4487,86 @@ class TestProcessVduParams(unittest.TestCase): ) self.assertEqual(disk_list, []) mock_select_persistent_root_disk.assert_called_once() + mock_volume_keeping_required.assert_not_called() + + @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk") + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_add_persistent_root_disk_to_disk_list_keep_true( + self, mock_volume_keeping_required, mock_select_persistent_root_disk + ): + """Add persistent root disk, keeo volume set to True.""" + vnfd = deepcopy(vnfd_wth_persistent_storage) + target_vdu = deepcopy(target_vdu_wth_persistent_storage) + mock_volume_keeping_required.return_value = True + root_disk = { + "id": "persistent-root-volume", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + "vdu-storage-requirements": [ + {"key": "keep-volume", "value": "true"}, + ], + } + mock_select_persistent_root_disk.return_value = root_disk + persistent_root_disk = {} + disk_list = [] + expected_disk_list = [ + { + "image_id": "ubuntu20.04", + "size": "10", + "keep": True, + } + ] + self.ns._add_persistent_root_disk_to_disk_list( + vnfd, target_vdu, persistent_root_disk, disk_list + ) + self.assertEqual(disk_list, expected_disk_list) + mock_volume_keeping_required.assert_called_once_with(root_disk) - def test_add_persistent_ordinary_disk_to_disk_list(self): - """Add persistent ordinary disk to disk_list""" + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_add_persistent_ordinary_disk_to_disk_list( + self, mock_volume_keeping_required + ): + """Add persistent ordinary disk, keeo volume set to True.""" target_vdu = deepcopy(target_vdu_wth_persistent_storage) + mock_volume_keeping_required.return_value = False persistent_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, } } + ordinary_disk = { + "id": "persistent-volume2", + "type-of-storage": "persistent-storage:persistent-storage", + "size-of-storage": "10", + } persistent_ordinary_disk = {} disk_list = [] expected_disk_list = [ { "size": "10", + "keep": False, } ] self.ns._add_persistent_ordinary_disks_to_disk_list( target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list ) self.assertEqual(disk_list, expected_disk_list) + mock_volume_keeping_required.assert_called_once_with(ordinary_disk) - def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(self): + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") + def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict( + self, mock_volume_keeping_required + ): """Add persistent ordinary disk, vsd id is in root_disk dict.""" target_vdu = deepcopy(target_vdu_wth_persistent_storage) + mock_volume_keeping_required.return_value = False persistent_root_disk = { "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, }, "persistent-volume2": { "size": "10", @@ -4549,10 +4579,12 @@ class TestProcessVduParams(unittest.TestCase): target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list ) self.assertEqual(disk_list, []) + mock_volume_keeping_required.assert_not_called() @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk") + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage( - self, mock_select_persistent_root_disk + self, mock_volume_keeping_required, mock_select_persistent_root_disk ): """VNFD does not have persistent storage.""" vnfd = deepcopy(vnfd_wthout_persistent_storage) @@ -4565,10 +4597,12 @@ class TestProcessVduParams(unittest.TestCase): ) self.assertEqual(disk_list, []) self.assertEqual(mock_select_persistent_root_disk.call_count, 2) + mock_volume_keeping_required.assert_not_called() @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk") + @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required") def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk( - self, mock_select_persistent_root_disk + self, mock_volume_keeping_required, mock_select_persistent_root_disk ): """Persistent_root_disk dict is empty.""" vnfd = deepcopy(vnfd_wthout_persistent_storage) @@ -4581,6 +4615,7 @@ class TestProcessVduParams(unittest.TestCase): ) self.assertEqual(disk_list, []) self.assertEqual(mock_select_persistent_root_disk.call_count, 2) + mock_volume_keeping_required.assert_not_called() def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self): """Invalid extra dict.""" @@ -4723,6 +4758,76 @@ class TestProcessVduParams(unittest.TestCase): persistent_root_disk, target_vdu, vdu_instantiation_vol_list, [] ) + @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces") + @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces") + @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces") + @patch("osm_ng_ro.ns.Ns._prepare_vdu_cloud_init") + @patch("osm_ng_ro.ns.Ns._prepare_vdu_ssh_keys") + @patch("osm_ng_ro.ns.Ns.find_persistent_root_volumes") + @patch("osm_ng_ro.ns.Ns.find_persistent_volumes") + @patch("osm_ng_ro.ns.Ns._add_persistent_root_disk_to_disk_list") + @patch("osm_ng_ro.ns.Ns._add_persistent_ordinary_disks_to_disk_list") + @patch("osm_ng_ro.ns.Ns._prepare_vdu_affinity_group_list") + def test_process_vdu_params_with_inst_flavor_id( + self, + mock_prepare_vdu_affinity_group_list, + mock_add_persistent_ordinary_disks_to_disk_list, + mock_add_persistent_root_disk_to_disk_list, + mock_find_persistent_volumes, + mock_find_persistent_root_volumes, + mock_prepare_vdu_ssh_keys, + mock_prepare_vdu_cloud_init, + mock_prepare_vdu_interfaces, + mock_locate_vdu_interfaces, + mock_sort_vdu_interfaces, + ): + """Instantiation volume list is empty.""" + target_vdu = deepcopy(target_vdu_wthout_persistent_storage) + + target_vdu["interfaces"] = interfaces_wth_all_positions + + vdu_instantiation_flavor_id = "flavor_test" + + target_vdu["additionalParams"] = { + "OSM": {"vim_flavor_id": vdu_instantiation_flavor_id} + } + mock_prepare_vdu_cloud_init.return_value = {} + mock_prepare_vdu_affinity_group_list.return_value = [] + + new_kwargs = deepcopy(kwargs) + new_kwargs.update( + { + "vnfr_id": vnfr_id, + "nsr_id": nsr_id, + "tasks_by_target_record_id": {}, + "logger": "logger", + } + ) + expected_extra_dict_copy = deepcopy(expected_extra_dict3) + vnfd = deepcopy(vnfd_wth_persistent_storage) + db.get_one.return_value = vnfd + result = Ns._process_vdu_params( + target_vdu, indata, vim_info=None, target_record_id=None, **new_kwargs + ) + mock_sort_vdu_interfaces.assert_called_once_with(target_vdu) + mock_locate_vdu_interfaces.assert_not_called() + mock_prepare_vdu_cloud_init.assert_called_once() + mock_add_persistent_root_disk_to_disk_list.assert_called_once() + mock_add_persistent_ordinary_disks_to_disk_list.assert_called_once() + mock_prepare_vdu_interfaces.assert_called_once_with( + target_vdu, + expected_extra_dict_copy, + ns_preffix, + vnf_preffix, + "logger", + {}, + [], + ) + self.assertDictEqual(result, expected_extra_dict_copy) + mock_prepare_vdu_ssh_keys.assert_called_once_with(target_vdu, None, {}) + mock_prepare_vdu_affinity_group_list.assert_called_once() + mock_find_persistent_volumes.assert_not_called() + @patch("osm_ng_ro.ns.Ns._sort_vdu_interfaces") @patch("osm_ng_ro.ns.Ns._partially_locate_vdu_interfaces") @patch("osm_ng_ro.ns.Ns._prepare_vdu_interfaces") @@ -4979,6 +5084,7 @@ class TestProcessVduParams(unittest.TestCase): "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, } } mock_find_persistent_root_volumes.return_value = persistent_root_disk @@ -5053,6 +5159,7 @@ class TestProcessVduParams(unittest.TestCase): "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, } } mock_find_persistent_root_volumes.return_value = persistent_root_disk @@ -5126,6 +5233,7 @@ class TestProcessVduParams(unittest.TestCase): "persistent-root-volume": { "image_id": "ubuntu20.04", "size": "10", + "keep": True, } } mock_find_persistent_root_volumes.return_value = persistent_root_disk