}
extra_dict["params"] = params
expected_result = deepcopy(expected_result_rebuild_start_stop)
- expected_result[
- "target_record"
- ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ expected_result["target_record"] = (
+ "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ )
expected_result["params"] = params
task = self.ns.rebuild_start_stop_task(
vdu_id,
vdu_index = "0"
task_index = 0
expected_result = deepcopy(expected_result_rebuild_start_stop)
- expected_result[
- "target_record"
- ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ expected_result["target_record"] = (
+ "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ )
for _ in actions:
task = self.ns.rebuild_start_stop_task(
vdu_id,
}
extra_dict["params"] = params
expected_result = deepcopy(expected_result_rebuild_start_stop)
- expected_result[
- "target_record"
- ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ expected_result["target_record"] = (
+ "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ )
expected_result["params"] = params
task = self.ns.rebuild_start_stop_task(
vdu_id,
}
extra_dict["params"] = params
expected_result = deepcopy(expected_result_rebuild_start_stop)
- expected_result[
- "target_record"
- ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ expected_result["target_record"] = (
+ "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.0.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ )
expected_result["params"] = params
expected_result["task_id"] = "bb937f49-3870-4169-b758-9732e1ff40f3:3"
task = self.ns.rebuild_start_stop_task(
vdu_index = "4"
task_index = 1
expected_result = deepcopy(expected_result_migrate)
- expected_result[
- "target_record"
- ] = "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ expected_result["target_record"] = (
+ "vnfrs:665b4165-ce24-4320-bf19-b9a45bade49f:vdur.4.vim_info.vim:f9f370ac-0d44-41a7-9000-457f2332bc35"
+ )
task = self.ns.migrate_task(
vdu, vnf, vdu_index, action_id, nsr_id_2, task_index, extra_dict_migrate
)
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