},
}
-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": {
"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 = {
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,
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
)
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")
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)
projects / osm / RO.git / blobdiff