X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=NG-RO%2Fosm_ng_ro%2Ftests%2Ftest_ns.py;h=a7f4409c06a4d826b0f77bf76a963ca623e64a95;hb=df48655ed39c3b5202289bcc84cd1618ce5464af;hp=fcfe897f6169334bfb8d5666585bc6e4f2bbd607;hpb=17852167b8cb19a82c07cf763a58e9e0261b3cc6;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 fcfe897f..a7f4409c 100644 --- a/NG-RO/osm_ng_ro/tests/test_ns.py +++ b/NG-RO/osm_ng_ro/tests/test_ns.py @@ -207,24 +207,6 @@ expected_extra_dict2 = { }, } -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": { @@ -349,11 +331,14 @@ expected_result_vertical_scale = { "params": { "vim_vm_id": "f37b18ef-3caa-4dc9-ab91-15c669b16396", "flavor_dict": "flavor_dict", + "flavor_id": "TASK-nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0", }, + "depends_on": ["nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0"], } vdu = { "id": vdu_id, "vim_info": {target_vim: {"interfaces": []}}, + "ns-flavor-id": "0", } vnf = {"_id": vnf_id} extra_dict_vertical_scale = { @@ -2386,6 +2371,53 @@ class TestNs(unittest.TestCase): self.assertTrue(epa_params.called) self.assertDictEqual(result, expected_result) + @patch("osm_ng_ro.ns.Ns._process_epa_params") + def test__process_flavor_params_with_vim_flavor_id( + self, + epa_params, + ): + expected_result = { + "find_params": { + "vim_flavor_id": "test.flavor", + }, + } + target_flavor = { + "id": "test_id", + "name": "test", + "storage-gb": "10", + "memory-mb": "1024", + "vcpu-count": "2", + } + indata = { + "vnf": [ + { + "vdur": [ + { + "ns-flavor-id": "test_id", + "additionalParams": { + "OSM": {"vim_flavor_id": "test.flavor"} + }, + }, + ], + "vnfd-id": "ad6356e3-698c-43bf-9901-3aae9e9b9d18", + }, + ], + } + vim_info = {} + target_record_id = "" + + epa_params.return_value = {} + + result = Ns._process_flavor_params( + target_flavor=target_flavor, + indata=indata, + vim_info=vim_info, + target_record_id=target_record_id, + ) + + self.assertFalse(epa_params.called) + self.assertDictEqual(result, expected_result) + @patch("osm_ng_ro.ns.Ns._process_epa_params") def test__process_flavor_params( self, @@ -3181,11 +3213,25 @@ class TestNs(unittest.TestCase): self, assign_vim ): self.ns = Ns() - extra_dict = {} + extra_dict = {"params": {}} vdu_index = "1" task_index = 1 - expected_result = deepcopy(expected_result_vertical_scale) - expected_result.pop("params") + expected_result = { + "target_id": "vim:f9f370ac-0d44-41a7-9000-457f2332bc35", + "action_id": "bb937f49-3870-4169-b758-9732e1ff40f3", + "nsr_id": "993166fe-723e-4680-ac4b-b1af2541ae31", + "task_id": "bb937f49-3870-4169-b758-9732e1ff40f3:1", + "status": "SCHEDULED", + "action": "EXEC", + "item": "verticalscale", + "target_record": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.1.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35", + "target_record_id": "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.bb9c43f9-10a2-4569-a8a8-957c3528b6d1", + "params": { + "flavor_id": "TASK-nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0" + }, + "depends_on": ["nsrs:993166fe-723e-4680-ac4b-b1af2541ae31:flavor.0"], + } + task = self.ns.verticalscale_task( vdu, vnf, vdu_index, action_id, nsr_id_2, task_index, extra_dict ) @@ -3368,7 +3414,7 @@ class TestProcessVduParams(unittest.TestCase): persist_root_disk = self.ns.find_persistent_root_volumes( vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list ) - self.assertEqual(persist_root_disk, None) + self.assertEqual(persist_root_disk, {}) mock_volume_keeping_required.assert_not_called() self.assertEqual(disk_list, []) @@ -4976,76 +5022,6 @@ 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") @@ -5624,3 +5600,234 @@ class TestProcessVduParams(unittest.TestCase): vsd = deepcopy(vnfd_wth_persistent_storage)["virtual-storage-desc"] with self.assertRaises(AttributeError): Ns._select_persistent_root_disk(vsd, vdu) + + +class TestSFC(unittest.TestCase): + def setUp(self): + self.ns = Ns() + self.logger = CopyingMock(autospec=True) + + @patch("osm_ng_ro.ns.Ns._prefix_ip_address") + @patch("osm_ng_ro.ns.Ns._process_ip_proto") + @patch("osm_ng_ro.ns.Ns._get_vnfr_vdur_text") + def test_process_classification_params( + self, mock_get_vnfr_vdur_text, mock_process_ip_proto, mock_prefix_ip_address + ): + db = Mock() + mock_prefix_ip_address.side_effect = ["10.10.10.10/32", "20.20.20.20/32"] + mock_process_ip_proto.return_value = "tcp" + mock_get_vnfr_vdur_text.return_value = "vdur_text" + vim_info, indata, target_record_id = {}, {}, "" + target_classification = { + "vnfr_id": "1234", + "source-ip-address": "10.10.10.10", + "destination-ip-address": "20.20.20.20", + "ip-proto": "6", + "id": "rule1", + "source-port": "0", + "destination-port": 5555, + "vdur_id": "5678", + "ingress_port_index": 0, + "vim_info": vim_info, + } + kwargs = {"db": db} + + expected_result = { + "depends_on": ["vdur_text"], + "params": { + "destination_ip_prefix": "20.20.20.20/32", + "destination_port_range_max": 5555, + "destination_port_range_min": 5555, + "logical_source_port": "TASK-vdur_text", + "logical_source_port_index": 0, + "name": "rule1", + "protocol": "tcp", + "source_ip_prefix": "10.10.10.10/32", + "source_port_range_max": "0", + "source_port_range_min": "0", + }, + } + + result = self.ns._process_classification_params( + target_classification, indata, vim_info, target_record_id, **kwargs + ) + self.assertEqual(expected_result, result) + + def test_process_sfp_params(self): + sf_text = "nsrs:1234:sf.sf1" + classi_text = "nsrs:1234:classification.rule1" + vim_info, indata, target_record_id = {}, {}, "" + target_sfp = { + "id": "sfp1", + "sfs": ["sf1"], + "classifications": ["rule1"], + "vim_info": vim_info, + } + + kwargs = {"nsr_id": "1234"} + + expected_result = { + "depends_on": [sf_text, classi_text], + "params": { + "name": "sfp1", + "sfs": ["TASK-" + sf_text], + "classifications": ["TASK-" + classi_text], + }, + } + + result = self.ns._process_sfp_params( + target_sfp, indata, vim_info, target_record_id, **kwargs + ) + self.assertEqual(expected_result, result) + + def test_process_sf_params(self): + sfi_text = "nsrs::sfi.sfi1" + vim_info, indata, target_record_id = {}, {}, "" + target_sf = {"id": "sf1", "sfis": ["sfi1"], "vim_info": vim_info} + + kwargs = {"ns_id": "1234"} + + expected_result = { + "depends_on": [sfi_text], + "params": { + "name": "sf1", + "sfis": ["TASK-" + sfi_text], + }, + } + + result = self.ns._process_sf_params( + target_sf, indata, vim_info, target_record_id, **kwargs + ) + self.assertEqual(expected_result, result) + + @patch("osm_ng_ro.ns.Ns._get_vnfr_vdur_text") + def test_process_sfi_params(self, mock_get_vnfr_vdur_text): + db = Mock() + mock_get_vnfr_vdur_text.return_value = "vdur_text" + vim_info, indata, target_record_id = {}, {}, "" + target_sfi = { + "id": "sfi1", + "ingress_port": "vnf-cp0-ext", + "egress_port": "vnf-cp0-ext", + "vnfr_id": "1234", + "vdur_id": "5678", + "ingress_port_index": 0, + "egress_port_index": 0, + "vim_info": {}, + } + kwargs = {"db": db} + + expected_result = { + "depends_on": ["vdur_text"], + "params": { + "name": "sfi1", + "ingress_port": "TASK-vdur_text", + "egress_port": "TASK-vdur_text", + "ingress_port_index": 0, + "egress_port_index": 0, + }, + } + + result = self.ns._process_sfi_params( + target_sfi, indata, vim_info, target_record_id, **kwargs + ) + self.assertEqual(expected_result, result) + + def test_process_vnfgd_sfp(self): + sfp = { + "id": "sfp1", + "position-desc-id": [ + { + "id": "position1", + "cp-profile-id": [{"id": "sf1"}], + "match-attributes": [{"id": "rule1"}], + } + ], + } + expected_result = {"id": "sfp1", "sfs": ["sf1"], "classifications": ["rule1"]} + + result = self.ns._process_vnfgd_sfp(sfp) + self.assertEqual(expected_result, result) + + def test_process_vnfgd_sf(self): + sf = {"id": "sf1", "constituent-profile-elements": [{"id": "sfi1", "order": 0}]} + expected_result = {"id": "sf1", "sfis": ["sfi1"]} + + result = self.ns._process_vnfgd_sf(sf) + self.assertEqual(expected_result, result) + + def test_process_vnfgd_sfi(self): + sfi = { + "id": "sfi1", + "constituent-base-element-id": "vnf", + "order": 0, + "ingress-constituent-cpd-id": "vnf-cp0-ext", + "egress-constituent-cpd-id": "vnf-cp0-ext", + } + db_vnfrs = { + "1234": { + "id": "1234", + "member-vnf-index-ref": "vnf", + "connection-point": [ + { + "name": "vnf-cp0-ext", + "connection-point-id": "vdu-eth0-int", + "connection-point-vdu-id": "5678", + "id": "vnf-cp0-ext", + } + ], + } + } + expected_result = { + "id": "sfi1", + "ingress_port": "vnf-cp0-ext", + "egress_port": "vnf-cp0-ext", + "vnfr_id": "1234", + "vdur_id": "5678", + "ingress_port_index": 0, + "egress_port_index": 0, + } + + result = self.ns._process_vnfgd_sfi(sfi, db_vnfrs) + self.assertEqual(expected_result, result) + + def test_process_vnfgd_classification(self): + classification = { + "id": "rule1", + "ip-proto": 6, + "source-ip-address": "10.10.10.10", + "destination-ip-address": "20.20.20.20", + "constituent-base-element-id": "vnf", + "constituent-cpd-id": "vnf-cp0-ext", + "destination-port": 5555, + } + db_vnfrs = { + "1234": { + "id": "1234", + "member-vnf-index-ref": "vnf", + "connection-point": [ + { + "name": "vnf-cp0-ext", + "connection-point-id": "vdu-eth0-int", + "connection-point-vdu-id": "5678", + "id": "vnf-cp0-ext", + } + ], + } + } + + expected_result = { + "id": "rule1", + "ip-proto": 6, + "source-ip-address": "10.10.10.10", + "destination-ip-address": "20.20.20.20", + "destination-port": 5555, + "vnfr_id": "1234", + "vdur_id": "5678", + "ingress_port_index": 0, + "constituent-base-element-id": "vnf", + "constituent-cpd-id": "vnf-cp0-ext", + } + + result = self.ns._process_vnfgd_classification(classification, db_vnfrs) + self.assertEqual(expected_result, result)