From 28ce7eb9fc1439f6d76c14e374afcceb30435679 Mon Sep 17 00:00:00 2001 From: Aldo Ortega Date: Wed, 8 Mar 2023 12:18:55 -0500 Subject: [PATCH 1/3] Added tests for mef_eline changes --- tests/test_e2e_10_mef_eline.py | 577 +++++++++++++++++++++++++++++++++ 1 file changed, 577 insertions(+) diff --git a/tests/test_e2e_10_mef_eline.py b/tests/test_e2e_10_mef_eline.py index 98dd371..459a74e 100644 --- a/tests/test_e2e_10_mef_eline.py +++ b/tests/test_e2e_10_mef_eline.py @@ -4,6 +4,7 @@ from datetime import datetime, timedelta import pytest +from random import randrange import requests from tests.helpers import NetworkTest @@ -1369,3 +1370,579 @@ def test_155_removing_evc_metadata_persistent(self): data = response.json() assert my_key not in data['metadata'] assert len(data['metadata']) > 0 + + def test_160_create_untagged_evc(self): + """Test create an EVC with untagged in both uni""" + payload = { + "name": "evc1", + "dynamic_backup_path": True, + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:01:1" + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:02:1" + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, data=json.dumps(payload), headers={'Content-type': 'application/json'}) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"][2] + + expected = { + "match": { + "in_port": 1, + "dl_vlan": 0 + }, + "actions": [ + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 3} + ], + "priority": 20000 + } + assert data["match"] == expected["match"] + assert data["priority"] == expected["priority"] + assert data["instructions"][0]["actions"] == expected["actions"] + + h11, h2 = self.net.net.get('h11', 'h2') + h11.cmd('ip addr add 100.0.0.11/24 dev %s' % (h11.intfNames()[0])) + h2.cmd('ip addr add 100.0.0.2/24 dev %s' % (h2.intfNames()[0])) + result = h11.cmd('ping -c1 100.0.0.2') + assert ', 0% packet loss,' in result + + # Clean up + h11.cmd('ip addr del 100.0.0.11/24 dev %s' % (h11.intfNames()[0])) + h2.cmd('ip addr del 100.0.0.2/24 dev %s' % (h2.intfNames()[0])) + self.net.restart_kytos_clean() + + def test_165_create_any_evc(self): + """Test create an EVC with any in both uni""" + payload = { + "name": "evc1", + "dynamic_backup_path": True, + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:1" + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:02:1" + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, data=json.dumps(payload), headers={'Content-type': 'application/json'}) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"][4] + + expected = { + "match": { + "in_port": 1, + "dl_vlan": "4096/4096" + }, + "actions": [ + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 3} + ], + "priority": 15000 + } + assert data["match"] == expected["match"] + assert data["priority"] == expected["priority"] + assert data["instructions"][0]["actions"] == expected["actions"] + + ra_vlan = randrange(1, 4096) + h11, h2 = self.net.net.get('h11', 'h2') + h11.cmd('ip link add link %s name vlan_ra type vlan id %s' % (h11.intfNames()[0], ra_vlan)) + h11.cmd('ip link set up vlan_ra') + h11.cmd('ip addr add 100.0.0.11/24 dev vlan_ra') + h2.cmd('ip link add link %s name vlan_ra type vlan id %s' % (h2.intfNames()[0], ra_vlan)) + h2.cmd('ip link set up vlan_ra') + h2.cmd('ip addr add 100.0.0.2/24 dev vlan_ra') + result = h11.cmd('ping -c1 100.0.0.2') + assert ', 0% packet loss,' in result + + # Clean up + h11.cmd('ip link del vlan_ra') + h2.cmd('ip link del vlan_ra') + self.net.restart_kytos_clean() + + def test_170_create_any_100_evc(self): + """Test create an EVC with any and 100 as uni.tag.value""" + payload = { + "name": "evc1", + "dynamic_backup_path": True, + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:1" + }, + "uni_z": { + "tag": {"tag_type": 1, "value": 100}, + "interface_id": "00:00:00:00:00:00:00:02:1" + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, data=json.dumps(payload), headers={'Content-type': 'application/json'}) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": "4096/4096"}, + "actions": [ + {"action_type": "set_vlan", "vlan_id": 100}, + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 3} + ], + "priority": 15000}, + {"match": {"in_port": 1, "dl_vlan": 100}, + "actions": [ + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 2} + ], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"][4] + assert data["match"] == expected[0]["match"] + assert data["priority"] == expected[0]["priority"] + assert data["instructions"][0]["actions"] == expected[0]["actions"] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:02' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:02"]["flows"][2] + assert data["match"] == expected[1]["match"] + assert data["priority"] == expected[1]["priority"] + assert data["instructions"][0]["actions"] == expected[1]["actions"] + + h11, h2 = self.net.net.get('h11', 'h2') + h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) + h11.cmd('ip link set up vlan100') + h11.cmd('ip addr add 100.0.0.11/24 dev vlan100') + h2.cmd('ip link add link %s name vlan100 type vlan id 100' % (h2.intfNames()[0])) + h2.cmd('ip link set up vlan100') + h2.cmd('ip addr add 100.0.0.2/24 dev vlan100') + result = h11.cmd('ping -c1 100.0.0.2') + assert ', 0% packet loss,' in result + + # Clean up + h11.cmd('ip link del vlan100') + h2.cmd('ip link del vlan100') + self.net.restart_kytos_clean() + + def test_175_create_100_untagged_evc(self): + """Test create an EVC with 100 and untagged as uni.tag.value""" + payload = { + "name": "evc1", + "dynamic_backup_path": True, + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": 100}, + "interface_id": "00:00:00:00:00:00:00:01:1" + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:02:1" + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, data=json.dumps(payload), headers={'Content-type': 'application/json'}) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": 100}, + "actions": [ + {"action_type": "pop_vlan"}, + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 3} + ], + "priority": 20000}, + {"match": {"in_port": 1, "dl_vlan": 0}, + "actions": [ + {"action_type": "push_vlan", "tag_type": "c"}, + {"action_type": "set_vlan", "vlan_id": 100}, + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 2} + ], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"][2] + assert data["match"] == expected[0]["match"] + assert data["priority"] == expected[0]["priority"] + assert data["instructions"][0]["actions"] == expected[0]["actions"] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:02' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:02"]["flows"][2] + assert data["match"] == expected[1]["match"] + assert data["priority"] == expected[1]["priority"] + assert data["instructions"][0]["actions"] == expected[1]["actions"] + + h11, h2 = self.net.net.get('h11', 'h2') + h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) + h11.cmd('ip link set up vlan100') + h11.cmd('ip addr add 100.0.0.11/24 dev vlan100') + h2.cmd('ip addr add 100.0.0.2/24 dev %s' % (h2.intfNames()[0])) + result = h11.cmd('ping -c1 100.0.0.2') + assert ', 0% packet loss,' in result + + # Clean up + h11.cmd('ip link del vlan100') + h2.cmd('ip addr del 100.0.0.2/24 dev %s' % (h2.intfNames()[0])) + self.net.restart_kytos_clean() + + def test_180_create_any_untagged_evc(self): + """Test create an EVC with any and untagged as uni.tag.value""" + payload = { + "name": "evc1", + "dynamic_backup_path": True, + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:1" + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:02:1" + } + } + + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, data=json.dumps(payload), headers={'Content-type': 'application/json'}) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": "4096/4096"}, + "actions": [ + {"action_type": "pop_vlan"}, + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 3} + ], + "priority": 15000}, + {"match": {"in_port": 1, "dl_vlan": 0}, + "actions": [ + {"action_type": "push_vlan", "tag_type": "s"}, + {"action_type": "set_vlan", "vlan_id": 1}, + {"action_type": "output", "port": 2} + ], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"][4] + assert data["match"] == expected[0]["match"] + assert data["priority"] == expected[0]["priority"] + assert data["instructions"][0]["actions"] == expected[0]["actions"] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:02' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:02"]["flows"][2] + assert data["match"] == expected[1]["match"] + assert data["priority"] == expected[1]["priority"] + assert data["instructions"][0]["actions"] == expected[1]["actions"] + + # Clean up + self.net.restart_kytos_clean() + + def test_185_create_any_intra_evc(self): + """Test create an intra-switch EVC with any as uni.tag.value""" + payload = { + "name": "my evc1", + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:1", + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:2", + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, json=payload) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": "4096/4096"}, + "actions": [{"action_type": "output", "port": 2}], + "priority": 15000}, + {"match": {"in_port": 2, "dl_vlan": "4096/4096"}, + "actions": [{"action_type": "output", "port": 1}], + "priority": 15000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + assert data[2]["match"] == expected[0]["match"] + assert data[2]["priority"] == expected[0]["priority"] + assert data[2]["instructions"][0]["actions"] == expected[0]["actions"] + assert data[3]["match"] == expected[1]["match"] + assert data[3]["priority"] == expected[1]["priority"] + assert data[3]["instructions"][0]["actions"] == expected[1]["actions"] + + ra_vlan = randrange(1, 4096) + h11, h12 = self.net.net.get('h11', 'h12') + h11.cmd('ip link add link %s name ra_vlan type vlan id %s' % (h11.intfNames()[0], ra_vlan)) + h11.cmd('ip link set up ra_vlan') + h11.cmd('ip addr add 10.1.1.11/24 dev ra_vlan') + h12.cmd('ip link add link %s name ra_vlan type vlan id %s' % (h12.intfNames()[0], ra_vlan)) + h12.cmd('ip link set up ra_vlan') + h12.cmd('ip addr add 10.1.1.12/24 dev ra_vlan') + result = h11.cmd('ping -c1 10.1.1.12') + assert ', 0% packet loss,' in result + + # clean up + h11.cmd('ip link del ra_vlan') + h12.cmd('ip link del ra_vlan') + self.net.restart_kytos_clean() + + def test_190_create_untagged_intra_evc(self): + """Test create an intra-switch EVC with untagged as uni.tag.value""" + payload = { + "name": "my evc1", + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:01:1", + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:01:2", + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, json=payload) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": 0}, + "actions": [{"action_type": "output", "port": 2}], + "priority": 20000}, + {"match": {"in_port": 2, "dl_vlan": 0}, + "actions": [{"action_type": "output", "port": 1}], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + assert data[2]["match"] == expected[0]["match"] + assert data[2]["priority"] == expected[0]["priority"] + assert data[2]["instructions"][0]["actions"] == expected[0]["actions"] + assert data[3]["match"] == expected[1]["match"] + assert data[3]["priority"] == expected[1]["priority"] + assert data[3]["instructions"][0]["actions"] == expected[1]["actions"] + + h11, h12 = self.net.net.get('h11', 'h12') + h11.cmd('ip addr add 100.1.1.11/24 dev %s' % (h11.intfNames()[0])) + h12.cmd('ip addr add 100.1.1.12/24 dev %s' % (h12.intfNames()[0])) + result = h11.cmd('ping -c1 100.1.1.12') + assert ', 0% packet loss,' in result + + # clean up + h11.cmd('ip addr del 100.1.1.11/24 dev %s' % (h11.intfNames()[0])) + h12.cmd('ip addr del 100.1.1.12/24 dev %s' % (h12.intfNames()[0])) + self.net.restart_kytos_clean() + + def test_195_create_any_100_intra_evc(self): + """Test create an intra-switch EVC with any and 100 as uni.tag.value""" + payload = { + "name": "my evc1", + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:1", + }, + "uni_z": { + "tag": {"tag_type": 1, "value": 100}, + "interface_id": "00:00:00:00:00:00:00:01:2", + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, json=payload) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": "4096/4096"}, + "actions": [{"action_type": "set_vlan", "vlan_id": 100}, + {"action_type": "output", "port": 2}], + "priority": 15000}, + {"match": {"in_port": 2, "dl_vlan": 100}, + "actions": [{"action_type": "output", "port": 1}], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + assert data[3]["match"] == expected[0]["match"] + assert data[3]["priority"] == expected[0]["priority"] + assert data[3]["instructions"][0]["actions"] == expected[0]["actions"] + assert data[2]["match"] == expected[1]["match"] + assert data[2]["priority"] == expected[1]["priority"] + assert data[2]["instructions"][0]["actions"] == expected[1]["actions"] + + h11, h12 = self.net.net.get('h11', 'h12') + h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) + h11.cmd('ip link set up vlan100') + h11.cmd('ip addr add 10.1.1.11/24 dev vlan100') + h12.cmd('ip link add link %s name vlan100 type vlan id 100' % (h12.intfNames()[0])) + h12.cmd('ip link set up vlan100') + h12.cmd('ip addr add 10.1.1.12/24 dev vlan100') + result = h11.cmd('ping -c1 10.1.1.12') + assert ', 0% packet loss,' in result + + # clean up + h11.cmd('ip link del vlan100') + h12.cmd('ip link del vlan100') + self.net.restart_kytos_clean() + + def test_200_create_100_untagged_intra_evc(self): + """Test create an intra-switch EVC with 100 and untagged as + uni.tag.value""" + payload = { + "name": "my evc1", + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": 100}, + "interface_id": "00:00:00:00:00:00:00:01:1", + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:01:2", + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, json=payload) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": 100}, + "actions": [ + {"action_type": "pop_vlan"}, + {"action_type": "output", "port": 2} + ], + "priority": 20000}, + {"match": {"in_port": 2, "dl_vlan": 0}, + "actions": [ + {"action_type": "push_vlan", "tag_type": 'c'}, + {"action_type": "set_vlan", "vlan_id": 100}, + {"action_type": "output", "port": 1} + ], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + assert data[2]["match"] == expected[0]["match"] + assert data[2]["priority"] == expected[0]["priority"] + assert data[2]["instructions"][0]["actions"] == expected[0]["actions"] + assert data[3]["match"] == expected[1]["match"] + assert data[3]["priority"] == expected[1]["priority"] + assert data[3]["instructions"][0]["actions"] == expected[1]["actions"] + + h11, h12 = self.net.net.get('h11', 'h12') + h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) + h11.cmd('ip link set up vlan100') + h11.cmd('ip addr add 100.1.1.11/24 dev vlan100') + h12.cmd('ip addr add 100.1.1.12/24 dev %s' % (h12.intfNames()[0])) + result = h11.cmd('ping -c1 100.1.1.12') + assert ', 0% packet loss,' in result + + # clean up + h11.cmd('ip link del vlan100') + h12.cmd('ip addr del 100.1.1.12/24 dev %s' % (h12.intfNames()[0])) + self.net.restart_kytos_clean() + + def test_205_create_any_untagged_intra_evc(self): + """Test create an intra-switch EVC with any and untagged as + uni.tag.value""" + payload = { + "name": "my evc1", + "enabled": True, + "uni_a": { + "tag": {"tag_type": 1, "value": "any"}, + "interface_id": "00:00:00:00:00:00:00:01:1", + }, + "uni_z": { + "tag": {"tag_type": 1, "value": "untagged"}, + "interface_id": "00:00:00:00:00:00:00:01:2", + } + } + api_url = KYTOS_API + '/mef_eline/v2/evc/' + response = requests.post(api_url, json=payload) + assert response.status_code == 201, response.text + data = response.json() + assert 'circuit_id' in data + time.sleep(10) + + expected = [ + {"match": {"in_port": 1, "dl_vlan": "4096/4096"}, + "actions": [{"action_type": "pop_vlan"}, + {"action_type": "output", "port": 2}], + "priority": 15000}, + {"match": {"in_port": 2, "dl_vlan": 0}, + "actions": [{"action_type": "output", "port": 1}], + "priority": 20000} + ] + + api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' + response = requests.get(api_url) + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + assert data[3]["match"] == expected[0]["match"] + assert data[3]["priority"] == expected[0]["priority"] + assert data[3]["instructions"][0]["actions"] == expected[0]["actions"] + assert data[2]["match"] == expected[1]["match"] + assert data[2]["priority"] == expected[1]["priority"] + assert data[2]["instructions"][0]["actions"] == expected[1]["actions"] + + # Clean up + self.net.restart_kytos_clean() From 17e310a3e89cba1527c8d3dbaab521a44cc2a8b3 Mon Sep 17 00:00:00 2001 From: Aldo Ortega Date: Fri, 10 Mar 2023 16:09:24 -0500 Subject: [PATCH 2/3] Automatized search for flow --- tests/test_e2e_10_mef_eline.py | 169 +++++++++++++++++---------------- 1 file changed, 89 insertions(+), 80 deletions(-) diff --git a/tests/test_e2e_10_mef_eline.py b/tests/test_e2e_10_mef_eline.py index 459a74e..5a2893e 100644 --- a/tests/test_e2e_10_mef_eline.py +++ b/tests/test_e2e_10_mef_eline.py @@ -1371,6 +1371,14 @@ def test_155_removing_evc_metadata_persistent(self): assert my_key not in data['metadata'] assert len(data['metadata']) > 0 + @staticmethod + def get_flow_by_vlan_match(data, dl_vlan): + """Get flow from data matching with dl_vlan""" + for flow in data: + if flow["match"].get("dl_vlan", None) == dl_vlan: + return flow + return None + def test_160_create_untagged_evc(self): """Test create an EVC with untagged in both uni""" payload = { @@ -1395,13 +1403,11 @@ def test_160_create_untagged_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:01"]["flows"][2] + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + untagged_flow = self.get_flow_by_vlan_match(data, 0) expected = { - "match": { - "in_port": 1, - "dl_vlan": 0 - }, + "match": {"in_port": 1, "dl_vlan": 0}, "actions": [ {"action_type": "push_vlan", "tag_type": "s"}, {"action_type": "set_vlan", "vlan_id": 1}, @@ -1409,9 +1415,9 @@ def test_160_create_untagged_evc(self): ], "priority": 20000 } - assert data["match"] == expected["match"] - assert data["priority"] == expected["priority"] - assert data["instructions"][0]["actions"] == expected["actions"] + assert untagged_flow["match"] == expected["match"] + assert untagged_flow["priority"] == expected["priority"] + assert untagged_flow["instructions"][0]["actions"] == expected["actions"] h11, h2 = self.net.net.get('h11', 'h2') h11.cmd('ip addr add 100.0.0.11/24 dev %s' % (h11.intfNames()[0])) @@ -1448,13 +1454,11 @@ def test_165_create_any_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:01"]["flows"][4] + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + any_flow = self.get_flow_by_vlan_match(data, "4096/4096") expected = { - "match": { - "in_port": 1, - "dl_vlan": "4096/4096" - }, + "match": {"in_port": 1, "dl_vlan": "4096/4096"}, "actions": [ {"action_type": "push_vlan", "tag_type": "s"}, {"action_type": "set_vlan", "vlan_id": 1}, @@ -1462,9 +1466,9 @@ def test_165_create_any_evc(self): ], "priority": 15000 } - assert data["match"] == expected["match"] - assert data["priority"] == expected["priority"] - assert data["instructions"][0]["actions"] == expected["actions"] + assert any_flow["match"] == expected["match"] + assert any_flow["priority"] == expected["priority"] + assert any_flow["instructions"][0]["actions"] == expected["actions"] ra_vlan = randrange(1, 4096) h11, h2 = self.net.net.get('h11', 'h2') @@ -1524,17 +1528,19 @@ def test_170_create_any_100_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:01"]["flows"][4] - assert data["match"] == expected[0]["match"] - assert data["priority"] == expected[0]["priority"] - assert data["instructions"][0]["actions"] == expected[0]["actions"] + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + any_flow = self.get_flow_by_vlan_match(data, "4096/4096") + assert any_flow["match"] == expected[0]["match"] + assert any_flow["priority"] == expected[0]["priority"] + assert any_flow["instructions"][0]["actions"] == expected[0]["actions"] api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:02' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:02"]["flows"][2] - assert data["match"] == expected[1]["match"] - assert data["priority"] == expected[1]["priority"] - assert data["instructions"][0]["actions"] == expected[1]["actions"] + data = response.json()["00:00:00:00:00:00:00:02"]["flows"] + common_flow = self.get_flow_by_vlan_match(data, 100) + assert common_flow["match"] == expected[1]["match"] + assert common_flow["priority"] == expected[1]["priority"] + assert common_flow["instructions"][0]["actions"] == expected[1]["actions"] h11, h2 = self.net.net.get('h11', 'h2') h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) @@ -1595,17 +1601,19 @@ def test_175_create_100_untagged_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:01"]["flows"][2] - assert data["match"] == expected[0]["match"] - assert data["priority"] == expected[0]["priority"] - assert data["instructions"][0]["actions"] == expected[0]["actions"] + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + common_flow = self.get_flow_by_vlan_match(data, 100) + assert common_flow["match"] == expected[0]["match"] + assert common_flow["priority"] == expected[0]["priority"] + assert common_flow["instructions"][0]["actions"] == expected[0]["actions"] api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:02' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:02"]["flows"][2] - assert data["match"] == expected[1]["match"] - assert data["priority"] == expected[1]["priority"] - assert data["instructions"][0]["actions"] == expected[1]["actions"] + data = response.json()["00:00:00:00:00:00:00:02"]["flows"] + untagged_flow = self.get_flow_by_vlan_match(data, 0) + assert untagged_flow["match"] == expected[1]["match"] + assert untagged_flow["priority"] == expected[1]["priority"] + assert untagged_flow["instructions"][0]["actions"] == expected[1]["actions"] h11, h2 = self.net.net.get('h11', 'h2') h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) @@ -1663,17 +1671,19 @@ def test_180_create_any_untagged_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:01"]["flows"][4] - assert data["match"] == expected[0]["match"] - assert data["priority"] == expected[0]["priority"] - assert data["instructions"][0]["actions"] == expected[0]["actions"] + data = response.json()["00:00:00:00:00:00:00:01"]["flows"] + any_flow = self.get_flow_by_vlan_match(data, "4096/4096") + assert any_flow["match"] == expected[0]["match"] + assert any_flow["priority"] == expected[0]["priority"] + assert any_flow["instructions"][0]["actions"] == expected[0]["actions"] api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:02' response = requests.get(api_url) - data = response.json()["00:00:00:00:00:00:00:02"]["flows"][2] - assert data["match"] == expected[1]["match"] - assert data["priority"] == expected[1]["priority"] - assert data["instructions"][0]["actions"] == expected[1]["actions"] + data = response.json()["00:00:00:00:00:00:00:02"]["flows"] + untagged_flow = self.get_flow_by_vlan_match(data, 0) + assert untagged_flow["match"] == expected[1]["match"] + assert untagged_flow["priority"] == expected[1]["priority"] + assert untagged_flow["instructions"][0]["actions"] == expected[1]["actions"] # Clean up self.net.restart_kytos_clean() @@ -1703,20 +1713,15 @@ def test_185_create_any_intra_evc(self): {"match": {"in_port": 1, "dl_vlan": "4096/4096"}, "actions": [{"action_type": "output", "port": 2}], "priority": 15000}, - {"match": {"in_port": 2, "dl_vlan": "4096/4096"}, - "actions": [{"action_type": "output", "port": 1}], - "priority": 15000} ] api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) data = response.json()["00:00:00:00:00:00:00:01"]["flows"] - assert data[2]["match"] == expected[0]["match"] - assert data[2]["priority"] == expected[0]["priority"] - assert data[2]["instructions"][0]["actions"] == expected[0]["actions"] - assert data[3]["match"] == expected[1]["match"] - assert data[3]["priority"] == expected[1]["priority"] - assert data[3]["instructions"][0]["actions"] == expected[1]["actions"] + any_flow = self.get_flow_by_vlan_match(data, "4096/4096") + assert any_flow["match"] == expected[0]["match"] + assert any_flow["priority"] == expected[0]["priority"] + assert any_flow["instructions"][0]["actions"] == expected[0]["actions"] ra_vlan = randrange(1, 4096) h11, h12 = self.net.net.get('h11', 'h12') @@ -1755,24 +1760,19 @@ def test_190_create_untagged_intra_evc(self): assert 'circuit_id' in data time.sleep(10) - expected = [ - {"match": {"in_port": 1, "dl_vlan": 0}, + expected = { + "match": {"in_port": 1, "dl_vlan": 0}, "actions": [{"action_type": "output", "port": 2}], - "priority": 20000}, - {"match": {"in_port": 2, "dl_vlan": 0}, - "actions": [{"action_type": "output", "port": 1}], - "priority": 20000} - ] + "priority": 20000 + } api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) data = response.json()["00:00:00:00:00:00:00:01"]["flows"] - assert data[2]["match"] == expected[0]["match"] - assert data[2]["priority"] == expected[0]["priority"] - assert data[2]["instructions"][0]["actions"] == expected[0]["actions"] - assert data[3]["match"] == expected[1]["match"] - assert data[3]["priority"] == expected[1]["priority"] - assert data[3]["instructions"][0]["actions"] == expected[1]["actions"] + untagged_flow = self.get_flow_by_vlan_match(data, 0) + assert untagged_flow["match"] == expected["match"] + assert untagged_flow["priority"] == expected["priority"] + assert untagged_flow["instructions"][0]["actions"] == expected["actions"] h11, h12 = self.net.net.get('h11', 'h12') h11.cmd('ip addr add 100.1.1.11/24 dev %s' % (h11.intfNames()[0])) @@ -1819,12 +1819,15 @@ def test_195_create_any_100_intra_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) data = response.json()["00:00:00:00:00:00:00:01"]["flows"] - assert data[3]["match"] == expected[0]["match"] - assert data[3]["priority"] == expected[0]["priority"] - assert data[3]["instructions"][0]["actions"] == expected[0]["actions"] - assert data[2]["match"] == expected[1]["match"] - assert data[2]["priority"] == expected[1]["priority"] - assert data[2]["instructions"][0]["actions"] == expected[1]["actions"] + any_flow = self.get_flow_by_vlan_match(data, "4096/4096") + assert any_flow["match"] == expected[0]["match"] + assert any_flow["priority"] == expected[0]["priority"] + assert any_flow["instructions"][0]["actions"] == expected[0]["actions"] + + commom_flow = self.get_flow_by_vlan_match(data, 100) + assert commom_flow["match"] == expected[1]["match"] + assert commom_flow["priority"] == expected[1]["priority"] + assert commom_flow["instructions"][0]["actions"] == expected[1]["actions"] h11, h12 = self.net.net.get('h11', 'h12') h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) @@ -1882,12 +1885,15 @@ def test_200_create_100_untagged_intra_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) data = response.json()["00:00:00:00:00:00:00:01"]["flows"] - assert data[2]["match"] == expected[0]["match"] - assert data[2]["priority"] == expected[0]["priority"] - assert data[2]["instructions"][0]["actions"] == expected[0]["actions"] - assert data[3]["match"] == expected[1]["match"] - assert data[3]["priority"] == expected[1]["priority"] - assert data[3]["instructions"][0]["actions"] == expected[1]["actions"] + commom_flow = self.get_flow_by_vlan_match(data, 100) + assert commom_flow["match"] == expected[0]["match"] + assert commom_flow["priority"] == expected[0]["priority"] + assert commom_flow["instructions"][0]["actions"] == expected[0]["actions"] + + untagged_flow = self.get_flow_by_vlan_match(data, 0) + assert untagged_flow["match"] == expected[1]["match"] + assert untagged_flow["priority"] == expected[1]["priority"] + assert untagged_flow["instructions"][0]["actions"] == expected[1]["actions"] h11, h12 = self.net.net.get('h11', 'h12') h11.cmd('ip link add link %s name vlan100 type vlan id 100' % (h11.intfNames()[0])) @@ -1937,12 +1943,15 @@ def test_205_create_any_untagged_intra_evc(self): api_url = KYTOS_API + '/flow_manager/v2/flows/00:00:00:00:00:00:00:01' response = requests.get(api_url) data = response.json()["00:00:00:00:00:00:00:01"]["flows"] - assert data[3]["match"] == expected[0]["match"] - assert data[3]["priority"] == expected[0]["priority"] - assert data[3]["instructions"][0]["actions"] == expected[0]["actions"] - assert data[2]["match"] == expected[1]["match"] - assert data[2]["priority"] == expected[1]["priority"] - assert data[2]["instructions"][0]["actions"] == expected[1]["actions"] + any_flow = self.get_flow_by_vlan_match(data, "4096/4096") + assert any_flow["match"] == expected[0]["match"] + assert any_flow["priority"] == expected[0]["priority"] + assert any_flow["instructions"][0]["actions"] == expected[0]["actions"] + + untagged_flow = self.get_flow_by_vlan_match(data, 0) + assert untagged_flow["match"] == expected[1]["match"] + assert untagged_flow["priority"] == expected[1]["priority"] + assert untagged_flow["instructions"][0]["actions"] == expected[1]["actions"] # Clean up self.net.restart_kytos_clean() From 89d4f76e090d3feb6275dfb07494de4b37869739 Mon Sep 17 00:00:00 2001 From: Aldo Ortega Date: Tue, 14 Mar 2023 12:18:39 -0400 Subject: [PATCH 3/3] Deleting unnecessary restart function --- tests/test_e2e_10_mef_eline.py | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/tests/test_e2e_10_mef_eline.py b/tests/test_e2e_10_mef_eline.py index 5a2893e..bb74666 100644 --- a/tests/test_e2e_10_mef_eline.py +++ b/tests/test_e2e_10_mef_eline.py @@ -1428,7 +1428,6 @@ def test_160_create_untagged_evc(self): # Clean up h11.cmd('ip addr del 100.0.0.11/24 dev %s' % (h11.intfNames()[0])) h2.cmd('ip addr del 100.0.0.2/24 dev %s' % (h2.intfNames()[0])) - self.net.restart_kytos_clean() def test_165_create_any_evc(self): """Test create an EVC with any in both uni""" @@ -1484,7 +1483,6 @@ def test_165_create_any_evc(self): # Clean up h11.cmd('ip link del vlan_ra') h2.cmd('ip link del vlan_ra') - self.net.restart_kytos_clean() def test_170_create_any_100_evc(self): """Test create an EVC with any and 100 as uni.tag.value""" @@ -1555,7 +1553,6 @@ def test_170_create_any_100_evc(self): # Clean up h11.cmd('ip link del vlan100') h2.cmd('ip link del vlan100') - self.net.restart_kytos_clean() def test_175_create_100_untagged_evc(self): """Test create an EVC with 100 and untagged as uni.tag.value""" @@ -1626,7 +1623,6 @@ def test_175_create_100_untagged_evc(self): # Clean up h11.cmd('ip link del vlan100') h2.cmd('ip addr del 100.0.0.2/24 dev %s' % (h2.intfNames()[0])) - self.net.restart_kytos_clean() def test_180_create_any_untagged_evc(self): """Test create an EVC with any and untagged as uni.tag.value""" @@ -1685,9 +1681,6 @@ def test_180_create_any_untagged_evc(self): assert untagged_flow["priority"] == expected[1]["priority"] assert untagged_flow["instructions"][0]["actions"] == expected[1]["actions"] - # Clean up - self.net.restart_kytos_clean() - def test_185_create_any_intra_evc(self): """Test create an intra-switch EVC with any as uni.tag.value""" payload = { @@ -1737,7 +1730,6 @@ def test_185_create_any_intra_evc(self): # clean up h11.cmd('ip link del ra_vlan') h12.cmd('ip link del ra_vlan') - self.net.restart_kytos_clean() def test_190_create_untagged_intra_evc(self): """Test create an intra-switch EVC with untagged as uni.tag.value""" @@ -1783,7 +1775,6 @@ def test_190_create_untagged_intra_evc(self): # clean up h11.cmd('ip addr del 100.1.1.11/24 dev %s' % (h11.intfNames()[0])) h12.cmd('ip addr del 100.1.1.12/24 dev %s' % (h12.intfNames()[0])) - self.net.restart_kytos_clean() def test_195_create_any_100_intra_evc(self): """Test create an intra-switch EVC with any and 100 as uni.tag.value""" @@ -1842,7 +1833,6 @@ def test_195_create_any_100_intra_evc(self): # clean up h11.cmd('ip link del vlan100') h12.cmd('ip link del vlan100') - self.net.restart_kytos_clean() def test_200_create_100_untagged_intra_evc(self): """Test create an intra-switch EVC with 100 and untagged as @@ -1906,7 +1896,6 @@ def test_200_create_100_untagged_intra_evc(self): # clean up h11.cmd('ip link del vlan100') h12.cmd('ip addr del 100.1.1.12/24 dev %s' % (h12.intfNames()[0])) - self.net.restart_kytos_clean() def test_205_create_any_untagged_intra_evc(self): """Test create an intra-switch EVC with any and untagged as @@ -1952,6 +1941,3 @@ def test_205_create_any_untagged_intra_evc(self): assert untagged_flow["match"] == expected[1]["match"] assert untagged_flow["priority"] == expected[1]["priority"] assert untagged_flow["instructions"][0]["actions"] == expected[1]["actions"] - - # Clean up - self.net.restart_kytos_clean()