From 199e6bbabd6af8fd016a9461c12e57ef55beffbc Mon Sep 17 00:00:00 2001 From: Lawrence Lee Date: Wed, 14 Feb 2024 22:13:56 +0000 Subject: [PATCH] add VS test for PL Signed-off-by: Lawrence Lee --- tests/dash_utils.py | 122 ++++++++++++++++++++++++++++++++++++++++ tests/sai_attrs.py | 11 ++++ tests/test_dash_vnet.py | 89 ++++++++++++++--------------- 3 files changed, 174 insertions(+), 48 deletions(-) create mode 100644 tests/dash_utils.py create mode 100644 tests/sai_attrs.py diff --git a/tests/dash_utils.py b/tests/dash_utils.py new file mode 100644 index 00000000000..971ab589d63 --- /dev/null +++ b/tests/dash_utils.py @@ -0,0 +1,122 @@ +from swsscommon import swsscommon +import typing +import pytest + +@pytest.fixture(scope='module') +def dash_db(dvs): + return DashDB(dvs) + +def to_string(value): + if isinstance(value, bool): + return "true" if value else "false" + elif isinstance(value, bytes): + return value + return str(value) + +class ProducerStateTable(swsscommon.ProducerStateTable): + def __setitem__(self, key: str, pairs: typing.Union[dict, list, tuple]): + pairs_str = [] + if isinstance(pairs, dict): + pairs = pairs.items() + for k, v in pairs: + pairs_str.append((to_string(k), to_string(v))) + self.set(key, pairs_str) + + def __delitem__(self, key: str): + self.delete(str(key)) + + +class Table(swsscommon.Table): + def __getitem__(self, key: str): + exists, result = self.get(str(key)) + if not exists: + return None + else: + return dict(result) + + def get_keys(self): + return self.getKeys() + + def get_newly_created_oid(self, old_oids): + new_oids = self.asic_db.wait_for_n_keys(self, len(old_oids) + 1) + oid = [ids for ids in new_oids if ids not in old_oids] + return oid[0] + + +class DashDB(object): + def __init__(self, dvs): + self.dvs = dvs + self.app_dash_routing_type_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_ROUTING_TYPE_TABLE") + self.app_dash_appliance_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_APPLIANCE_TABLE") + self.asic_direction_lookup_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_DIRECTION_LOOKUP_ENTRY") + self.asic_vip_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_VIP_ENTRY") + self.app_dash_vnet_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_VNET_TABLE") + self.asic_dash_vnet_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_VNET") + self.app_dash_eni_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_ENI_TABLE") + self.asic_eni_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_ENI") + self.asic_eni_ether_addr_map_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_ENI_ETHER_ADDRESS_MAP_ENTRY") + self.app_dash_vnet_map_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_VNET_MAPPING_TABLE") + self.asic_dash_outbound_ca_to_pa_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_OUTBOUND_CA_TO_PA_ENTRY") + self.asic_pa_validation_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_PA_VALIDATION_ENTRY") + self.app_dash_route_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_ROUTE_TABLE") + self.app_dash_route_rule_table = ProducerStateTable( + self.dvs.get_app_db().db_connection, "DASH_ROUTE_RULE_TABLE") + self.asic_outbound_routing_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_OUTBOUND_ROUTING_ENTRY") + self.asic_inbound_routing_rule_table = Table( + self.dvs.get_asic_db().db_connection, "ASIC_STATE:SAI_OBJECT_TYPE_INBOUND_ROUTING_ENTRY") + + def create_appliance(self, appliance_id, attr_maps: dict): + self.app_dash_appliance_table[str(appliance_id)] = attr_maps + + def remove_appliance(self, appliance_id): + del self.app_dash_appliance_table[str(appliance_id)] + + def create_vnet(self, vnet, attr_maps: dict): + self.app_dash_vnet_table[str(vnet)] = attr_maps + + def remove_vnet(self, vnet): + del self.app_dash_vnet_table[str(vnet)] + + def create_eni(self, eni, attr_maps: dict): + self.app_dash_eni_table[str(eni)] = attr_maps + + def remove_eni(self, eni): + del self.app_dash_eni_table[str(eni)] + + def create_vnet_map(self, vnet, ip, attr_maps: dict): + self.app_dash_vnet_map_table[str(vnet) + ":" + str(ip)] = attr_maps + + def remove_vnet_map(self, vnet, ip): + del self.app_dash_vnet_map_table[str(vnet) + ":" + str(ip)] + + def create_outbound_routing(self, mac_string, ip, attr_maps: dict): + self.app_dash_route_table[str(mac_string) + ":" + str(ip)] = attr_maps + + def remove_outbound_routing(self, mac_string, ip): + del self.app_dash_route_table[str(mac_string) + ":" + str(ip)] + + def create_inbound_routing(self, mac_string, vni, ip, attr_maps: dict): + self.app_dash_route_rule_table[str(mac_string) + ":" + str(vni) + ":" + str(ip)] = attr_maps + + def remove_inbound_routing(self, mac_string, vni, ip): + del self.app_dash_route_rule_table[str(mac_string) + ":" + str(vni) + ":" + str(ip)] + + def create_routing_type(self, routing_type, attr_maps: dict): + self.app_dash_routing_type_table[str(routing_type)] = attr_maps + + def remove_routing_type(self, routing_type): + del self.app_dash_routing_type_table[str(routing_type)] \ No newline at end of file diff --git a/tests/sai_attrs.py b/tests/sai_attrs.py new file mode 100644 index 00000000000..2f3523883ef --- /dev/null +++ b/tests/sai_attrs.py @@ -0,0 +1,11 @@ +SAI_ENI_ATTR_PL_SIP = 'SAI_ENI_ATTR_PL_SIP' +SAI_ENI_ATTR_PL_SIP_MASK = 'SAI_ENI_ATTR_PL_SIP_MASK' +SAI_ENI_ATTR_PL_UNDERLAY_SIP = 'SAI_ENI_ATTR_PL_UNDERLAY_SIP' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ACTION_SET_TUNNEL_MAPPING = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ACTION_SET_TUNNEL_MAPPING' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ACTION_SET_PRIVATE_LINK_MAPPING = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ACTION_SET_PRIVATE_LINK_MAPPING' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_ACTION = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_ACTION' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_OVERLAY_SIP = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_OVERLAY_SIP' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_OVERLAY_DIP = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_OVERLAY_DIP' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_DASH_ENCAPSULATION = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_DASH_ENCAPSULATION' +SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_TUNNEL_KEY = 'SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_TUNNEL_KEY' +SAI_DASH_ENCAPSULATION_NVGRE = 'SAI_DASH_ENCAPSULATION_NVGRE' \ No newline at end of file diff --git a/tests/test_dash_vnet.py b/tests/test_dash_vnet.py index 950baff9837..20c01e21a66 100644 --- a/tests/test_dash_vnet.py +++ b/tests/test_dash_vnet.py @@ -7,7 +7,7 @@ from dash_api.route_type_pb2 import * from dash_api.types_pb2 import * -from dash_utils import DashDB +from dash_utils import dash_db import time import uuid @@ -18,48 +18,45 @@ NUM_PORTS = 2 class TestDash(object): - def test_appliance(self, dvs): - dashobj = DashDB(dvs) + def test_appliance(self, dash_db): self.appliance_id = "100" self.sip = "10.0.0.1" self.vm_vni = "4321" pb = Appliance() pb.sip.ipv4 = socket.htonl(int(ipaddress.ip_address(self.sip))) pb.vm_vni = int(self.vm_vni) - dashobj.create_appliance(self.appliance_id, {"pb": pb.SerializeToString()}) + dash_db.create_appliance(self.appliance_id, {"pb": pb.SerializeToString()}) time.sleep(3) - direction_entries = dashobj.asic_direction_lookup_table.get_keys() + direction_entries = dash_db.asic_direction_lookup_table.get_keys() assert direction_entries - fvs = dashobj.asic_direction_lookup_table[direction_entries[0]] + fvs = dash_db.asic_direction_lookup_table[direction_entries[0]] for fv in fvs.items(): if fv[0] == "SAI_DIRECTION_LOOKUP_ENTRY_ATTR_ACTION": assert fv[1] == "SAI_DIRECTION_LOOKUP_ENTRY_ACTION_SET_OUTBOUND_DIRECTION" - vip_entries = dashobj.asic_vip_table.get_keys() + vip_entries = dash_db.asic_vip_table.get_keys() assert vip_entries - fvs = dashobj.asic_vip_table[vip_entries[0]] + fvs = dash_db.asic_vip_table[vip_entries[0]] for fv in fvs.items(): if fv[0] == "SAI_VIP_ENTRY_ATTR_ACTION": assert fv[1] == "SAI_VIP_ENTRY_ACTION_ACCEPT" - def test_vnet(self, dvs): - dashobj = DashDB(dvs) + def test_vnet(self, dash_db): self.vnet = "Vnet1" self.vni = "45654" self.guid = "559c6ce8-26ab-4193-b946-ccc6e8f930b2" pb = Vnet() pb.vni = int(self.vni) pb.guid.value = bytes.fromhex(uuid.UUID(self.guid).hex) - dashobj.create_vnet(self.vnet, {"pb": pb.SerializeToString()}) + dash_db.create_vnet(self.vnet, {"pb": pb.SerializeToString()}) time.sleep(3) - vnets = dashobj.asic_dash_vnet_table.get_keys() + vnets = dash_db.asic_dash_vnet_table.get_keys() assert vnets self.vnet_oid = vnets[0] - vnet_attr = dashobj.asic_dash_vnet_table[self.vnet_oid] + vnet_attr = dash_db.asic_dash_vnet_table[self.vnet_oid] assert vnet_attr["SAI_VNET_ATTR_VNI"] == "45654" - def test_eni(self, dvs): - dashobj = DashDB(dvs) + def test_eni(self, dash_db): self.vnet = "Vnet1" self.mac_string = "F4939FEFC47E" self.mac_address = "F4:93:9F:EF:C4:7E" @@ -72,15 +69,15 @@ def test_eni(self, dvs): pb.underlay_ip.ipv4 = socket.htonl(int(ipaddress.ip_address(self.underlay_ip))) pb.admin_state = State.STATE_ENABLED pb.vnet = self.vnet - dashobj.create_eni(self.mac_string, {"pb": pb.SerializeToString()}) + dash_db.create_eni(self.mac_string, {"pb": pb.SerializeToString()}) time.sleep(3) - vnets = dashobj.asic_dash_vnet_table.get_keys() + vnets = dash_db.asic_dash_vnet_table.get_keys() assert vnets self.vnet_oid = vnets[0] - enis = dashobj.asic_eni_table.get_keys() + enis = dash_db.asic_eni_table.get_keys() assert enis self.eni_oid = enis[0]; - fvs = dashobj.asic_eni_table[enis[0]] + fvs = dash_db.asic_eni_table[enis[0]] for fv in fvs.items(): if fv[0] == "SAI_ENI_ATTR_VNET_ID": assert fv[1] == str(self.vnet_oid) @@ -94,15 +91,14 @@ def test_eni(self, dvs): assert fv[1] == "true" time.sleep(3) - eni_addr_maps = dashobj.asic_eni_ether_addr_map_table.get_keys() + eni_addr_maps = dash_db.asic_eni_ether_addr_map_table.get_keys() assert eni_addr_maps - fvs = dashobj.asic_eni_ether_addr_map_table[eni_addr_maps[0]] + fvs = dash_db.asic_eni_ether_addr_map_table[eni_addr_maps[0]] for fv in fvs.items(): if fv[0] == "SAI_ENI_ETHER_ADDRESS_MAP_ENTRY_ATTR_ENI_ID": assert fv[1] == str(self.eni_oid) - def test_vnet_map(self, dvs): - dashobj = DashDB(dvs) + def test_vnet_map(self, dash_db): self.vnet = "Vnet1" self.ip1 = "10.1.1.1" self.ip2 = "10.1.1.2" @@ -114,34 +110,33 @@ def test_vnet_map(self, dvs): route_action.action_name = "action1" route_action.action_type = ACTION_TYPE_MAPROUTING route_type_msg.items.append(route_action) - dashobj.create_routing_type(self.routing_type, {"pb": route_type_msg.SerializeToString()}) + dash_db.create_routing_type(self.routing_type, {"pb": route_type_msg.SerializeToString()}) pb = VnetMapping() pb.mac_address = bytes.fromhex(self.mac_address.replace(":", "")) pb.action_type = RoutingType.ROUTING_TYPE_VNET_ENCAP pb.underlay_ip.ipv4 = socket.htonl(int(ipaddress.ip_address(self.underlay_ip))) - dashobj.create_vnet_map(self.vnet, self.ip1, {"pb": pb.SerializeToString()}) - dashobj.create_vnet_map(self.vnet, self.ip2, {"pb": pb.SerializeToString()}) + dash_db.create_vnet_map(self.vnet, self.ip1, {"pb": pb.SerializeToString()}) + dash_db.create_vnet_map(self.vnet, self.ip2, {"pb": pb.SerializeToString()}) time.sleep(3) - vnet_ca_to_pa_maps = dashobj.asic_dash_outbound_ca_to_pa_table.get_keys() + vnet_ca_to_pa_maps = dash_db.asic_dash_outbound_ca_to_pa_table.get_keys() assert len(vnet_ca_to_pa_maps) >= 2 - fvs = dashobj.asic_dash_outbound_ca_to_pa_table[vnet_ca_to_pa_maps[0]] + fvs = dash_db.asic_dash_outbound_ca_to_pa_table[vnet_ca_to_pa_maps[0]] for fv in fvs.items(): if fv[0] == "SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_UNDERLAY_DIP": assert fv[1] == "101.1.2.3" if fv[0] == "SAI_OUTBOUND_CA_TO_PA_ENTRY_ATTR_OVERLAY_DMAC": assert fv[1] == "F4:93:9F:EF:C4:7E" - vnet_pa_validation_maps = dashobj.asic_pa_validation_table.get_keys() + vnet_pa_validation_maps = dash_db.asic_pa_validation_table.get_keys() assert vnet_pa_validation_maps - fvs = dashobj.asic_pa_validation_table[vnet_pa_validation_maps[0]] + fvs = dash_db.asic_pa_validation_table[vnet_pa_validation_maps[0]] for fv in fvs.items(): if fv[0] == "SAI_PA_VALIDATION_ENTRY_ATTR_ACTION": assert fv[1] == "SAI_PA_VALIDATION_ENTRY_ACTION_PERMIT" - def test_outbound_routing(self, dvs): - dashobj = DashDB(dvs) + def test_outbound_routing(self, dash_db): self.vnet = "Vnet1" self.mac_string = "F4939FEFC47E" self.ip = "10.1.0.0/24" @@ -151,12 +146,12 @@ def test_outbound_routing(self, dvs): pb.action_type = RoutingType.ROUTING_TYPE_VNET_DIRECT pb.vnet_direct.vnet = self.vnet pb.vnet_direct.overlay_ip.ipv4 = socket.htonl(int(ipaddress.ip_address(self.overlay_ip))) - dashobj.create_outbound_routing(self.mac_string, self.ip, {"pb": pb.SerializeToString()}) + dash_db.create_outbound_routing(self.mac_string, self.ip, {"pb": pb.SerializeToString()}) time.sleep(3) - outbound_routing_entries = dashobj.asic_outbound_routing_table.get_keys() + outbound_routing_entries = dash_db.asic_outbound_routing_table.get_keys() assert outbound_routing_entries - fvs = dashobj.asic_outbound_routing_table[outbound_routing_entries[0]] + fvs = dash_db.asic_outbound_routing_table[outbound_routing_entries[0]] for fv in fvs.items(): if fv[0] == "SAI_OUTBOUND_ROUTING_ENTRY_ATTR_ACTION": assert fv[1] == "SAI_OUTBOUND_ROUTING_ENTRY_ACTION_ROUTE_VNET_DIRECT" @@ -164,8 +159,7 @@ def test_outbound_routing(self, dvs): assert fv[1] == "10.0.0.6" assert "SAI_OUTBOUND_ROUTING_ENTRY_ATTR_DST_VNET_ID" in fvs - def test_inbound_routing(self, dvs): - dashobj = DashDB(dvs) + def test_inbound_routing(self, dash_db): self.mac_string = "F4939FEFC47E" self.vnet = "Vnet1" self.vni = "3251" @@ -180,30 +174,29 @@ def test_inbound_routing(self, dvs): pb.protocol = int(self.protocol) pb.vnet = self.vnet - dashobj.create_inbound_routing(self.mac_string, self.vni, self.ip, {"pb": pb.SerializeToString()}) + dash_db.create_inbound_routing(self.mac_string, self.vni, self.ip, {"pb": pb.SerializeToString()}) time.sleep(3) - inbound_routing_entries = dashobj.asic_inbound_routing_rule_table.get_keys() + inbound_routing_entries = dash_db.asic_inbound_routing_rule_table.get_keys() assert inbound_routing_entries - fvs = dashobj.asic_inbound_routing_rule_table[inbound_routing_entries[0]] + fvs = dash_db.asic_inbound_routing_rule_table[inbound_routing_entries[0]] for fv in fvs.items(): if fv[0] == "SAI_INBOUND_ROUTING_ENTRY_ATTR_ACTION": assert fv[1] == "SAI_INBOUND_ROUTING_ENTRY_ACTION_VXLAN_DECAP_PA_VALIDATE" - def test_cleanup(self, dvs): - dashobj = DashDB(dvs) + def test_cleanup(self, dash_db): self.vnet = "Vnet1" self.mac_string = "F4939FEFC47E" self.vni = "3251" self.sip = "10.1.1.1" self.dip = "10.1.0.0/24" self.appliance_id = "100" - dashobj.remove_inbound_routing(self.mac_string, self.vni, self.sip) - dashobj.remove_outbound_routing(self.mac_string, self.dip) - dashobj.remove_eni(self.mac_string) - dashobj.remove_vnet_map(self.vnet, self.sip) - dashobj.remove_vnet(self.vnet) - dashobj.remove_appliance(self.appliance_id) + dash_db.remove_inbound_routing(self.mac_string, self.vni, self.sip) + dash_db.remove_outbound_routing(self.mac_string, self.dip) + dash_db.remove_eni(self.mac_string) + dash_db.remove_vnet_map(self.vnet, self.sip) + dash_db.remove_vnet(self.vnet) + dash_db.remove_appliance(self.appliance_id) # Add Dummy always-pass test at end as workaroud # for issue when Flaky fail on final test it invokes module tear-down