Add unittests

sonic-net · Jul 6, 2024 · 1b764b5 · 1b764b5
1 parent e872d52
commit 1b764b5
Show file tree

Hide file tree

Showing 5 changed files with 490 additions and 0 deletions.
diff --git a/tests/mock_tests/Makefile.am b/tests/mock_tests/Makefile.am
@@ -118,6 +118,9 @@ tests_SOURCES = aclorch_ut.cpp \
                 $(top_srcdir)/orchagent/muxorch.cpp \
                 $(top_srcdir)/orchagent/mlagorch.cpp \
                 $(top_srcdir)/orchagent/isolationgrouporch.cpp \
+                $(top_srcdir)/orchagent/shlorch.cpp \
+                $(top_srcdir)/orchagent/l2nhgorch.cpp \
+                $(top_srcdir)/orchagent/evpnmhorch.cpp \
                 $(top_srcdir)/orchagent/macsecorch.cpp \
                 $(top_srcdir)/orchagent/lagid.cpp \
                 $(top_srcdir)/orchagent/bfdorch.cpp \

diff --git a/tests/mock_tests/fdborch/flush_syncd_notif_ut.cpp b/tests/mock_tests/fdborch/flush_syncd_notif_ut.cpp
@@ -523,6 +523,7 @@ namespace fdb_syncd_flush_test
         fdbData.type = "dynamic";
         fdbData.origin = FDB_ORIGIN_VXLAN_ADVERTIZED;
         fdbData.remote_ip = "1.1.1.1";
+        fdbData.nhg_id = 0;
         fdbData.esi = "";
         fdbData.vni = 100;
         FdbEntry entry;

diff --git a/tests/test_evpn_multihoming.py b/tests/test_evpn_multihoming.py
@@ -0,0 +1,199 @@
+from swsscommon import swsscommon
+from evpn_tunnel import VxlanTunnel,VxlanEvpnHelper
+
+import time
+import pytest
+
+app_df_name = "DF_ELECTION_TABLE"
+app_shl_name = "SPLIT_HORIZON_LIST_TABLE"
+asic_isolation_name = "ASIC_STATE:SAI_OBJECT_TYPE_ISOLATION_GROUP"
+asic_brport_name = "ASIC_STATE:SAI_OBJECT_TYPE_BRIDGE_PORT"
+asic_tunnel_name = "ASIC_STATE:SAI_OBJECT_TYPE_TUNNEL"
+asic_vlan_member_name = "ASIC_STATE:SAI_OBJECT_TYPE_VLAN_MEMBER"
+tunnel_src_ip = '10.0.0.2'
+portchannel = 'PortChannel0'
+vtep = '2.2.2.2'
+tunnel_name = 'tunnel_2'
+remote_ip_6 = "2.2.2.2"
+map_name = 'map_100_100'
+vlan_name = "Vlan100"
+vni = "100"
+nvo_name = "nvo1"
+
+
+class TestEvpnMultihoming(object):
+    def setup_db(self, dvs):
+        self.app_db = swsscommon.DBConnector(swsscommon.APPL_DB, dvs.redis_sock, 0)
+        self.asic_db = swsscommon.DBConnector(swsscommon.ASIC_DB, dvs.redis_sock, 0)
+        self.conf_db = swsscommon.DBConnector(swsscommon.CONFIG_DB, dvs.redis_sock, 0)
+        self.helper = VxlanEvpnHelper()
+        self.vxlan_obj = VxlanTunnel()
+
+    def create_portchannel(self, portchannel_name):
+        self.helper.create_entry_tbl(
+            self.conf_db,
+            "PORTCHANNEL", portchannel_name,
+            [
+                ("admin_status", "up"),
+                ("fast_rate", "false"),
+                ("lacp_key", "auto"),
+                ("min_links", "1"),
+                ("mtu", "9100"),
+            ],
+        )
+        time.sleep(2)
+
+    def create_shl_appl_table(self, portchannel_name, vteps):
+        self.helper.create_entry_pst(
+            self.app_db,
+            app_shl_name, portchannel_name,
+            [
+                ("vteps", vteps),
+            ],
+        )
+        time.sleep(2)
+
+    def create_df_appl_table(self, portchannel_name, non_df):
+        non_df_str = "true" if non_df else "false"
+        self.helper.create_entry_pst(
+            self.app_db,
+            app_df_name, portchannel_name,
+            [
+                ("non_df", non_df_str),
+            ],
+        )
+        time.sleep(2)
+
+    def create_vlan_member(self, vlan, interface):
+        self.helper.create_entry_tbl(
+            self.conf_db,
+            "VLAN_MEMBER", vlan + "|" + interface,
+            [
+                ("tagging_mode", "untagged")
+            ],
+        )
+        time.sleep(1)
+
+    def check_shl_appl_table(self, portchannel_name, vteps):
+        exp_attr = {
+            "vteps": vteps
+        }
+        self.helper.check_object(self.app_db, app_shl_name, portchannel_name, exp_attr)
+
+    def check_df_appl_table(self, portchannel_name, non_df):
+        non_df_str = "true" if non_df else "false"
+
+        exp_attr = {
+            "non_df": non_df_str,
+        }
+        self.helper.check_object(self.app_db, app_df_name, portchannel_name, exp_attr)
+
+    def check_tunnel_ip(self, oid, ip):
+        tbl = swsscommon.Table(self.asic_db, asic_tunnel_name)
+        (status, fvs) = tbl.get(oid)
+        assert status
+        values = dict(fvs)
+        assert "SAI_TUNNEL_ATTR_ENCAP_DST_IP" in values
+        assert values["SAI_TUNNEL_ATTR_ENCAP_DST_IP"] == ip
+
+    def check_df_asic_table(self, non_df=None):
+        entry_created = False
+
+        if non_df is True:
+            df_mode = "SAI_VLAN_DF_MODE_BLOCKED"
+        elif non_df is False:
+            df_mode = "SAI_VLAN_DF_MODE_UNBLOCKED"
+        else:
+            df_mode = "SAI_VLAN_DF_MODE_IGNORE"
+
+        tbl = swsscommon.Table(self.asic_db, asic_vlan_member_name)
+        for key in tbl.getKeys():
+            (status, fvs) = tbl.get(key)
+            assert status
+            values = dict(fvs)
+            if "SAI_VLAN_MEMBER_ATTR_VLAN_DF_MODE" not in values:
+                continue
+            assert values["SAI_VLAN_MEMBER_ATTR_VLAN_DF_MODE"] == df_mode
+            entry_created = True
+
+        assert entry_created, f"{asic_vlan_member_name}: DF-election entry not present"
+
+    def remove_shf_appl_table(self, portchannel_name):
+        self.helper.delete_entry_pst(
+            self.app_db,
+            app_shl_name, portchannel_name,
+        )
+        time.sleep(2)
+
+    def remove_df_appl_table(self, portchannel_name):
+        self.helper.delete_entry_pst(
+            self.app_db,
+            app_df_name, portchannel_name,
+        )
+        time.sleep(2)
+
+    def check_shf_asic_db_entries_deleted(self):
+        tbl = swsscommon.Table(self.asic_db, asic_isolation_name)
+        entries = tbl.getKeys()
+        assert len(entries) == 0, f"{asic_isolation_name} entries not deleted from ASIC_DB"
+
+    #    Test 1 - DF-election
+    @pytest.mark.parametrize("non_df", [True, False])
+    def test_df_election(self, dvs, testlog, non_df):
+        self.setup_db(dvs)
+
+        self.create_portchannel(portchannel)
+        self.vxlan_obj.create_vlan1(dvs, vlan_name)
+        self.vxlan_obj.check_vlan_obj(dvs, vni)
+        self.create_vlan_member(vlan_name, portchannel)
+
+        self.create_df_appl_table(portchannel, non_df)
+        self.check_df_appl_table(portchannel, non_df)
+
+        self.check_df_asic_table(non_df)
+
+        self.remove_df_appl_table(portchannel)
+        self.check_df_asic_table()
+
+    #    Test 2 - Split Horizon Filtering
+    def test_split_horizon_filtering(self, dvs, testlog):
+        self.setup_db(dvs)
+
+        self.create_portchannel(portchannel)
+        self.vxlan_obj.create_vlan1(dvs, vlan_name)
+        self.vxlan_obj.check_vlan_obj(dvs, vni)
+        self.vxlan_obj.create_vxlan_tunnel(dvs, tunnel_name, tunnel_src_ip)
+        self.vxlan_obj.create_evpn_nvo(dvs, nvo_name, tunnel_name)
+        self.vxlan_obj.create_vxlan_tunnel_map(dvs, tunnel_name, map_name, vni, vlan_name)
+        self.vxlan_obj.create_evpn_remote_vni(dvs, vlan_name, remote_ip_6, vni)
+        time.sleep(1)
+
+        self.create_shl_appl_table(portchannel, vtep)
+        self.check_shl_appl_table(portchannel, vtep)
+
+        tbl = swsscommon.Table(self.asic_db, asic_isolation_name)
+        entries = tbl.getKeys()
+        assert len(entries) == 1
+        (status, fvs) = tbl.get(entries[0])
+        assert status
+        for fv in fvs:
+            if fv[0] == "SAI_ISOLATION_GROUP_ATTR_TYPE":
+                assert fv[1] == "SAI_ISOLATION_GROUP_TYPE_BRIDGE_PORT"
+
+        tbl = swsscommon.Table(self.asic_db, asic_brport_name)
+        for key in tbl.getKeys():
+            (status, fvs) = tbl.get(key)
+            assert status
+            values = dict(fvs)
+            if "SAI_BRIDGE_PORT_ATTR_ISOLATION_GROUP" not in values:
+                continue
+            assert "SAI_BRIDGE_PORT_ATTR_TUNNEL_ID" in values
+            oid = values["SAI_BRIDGE_PORT_ATTR_TUNNEL_ID"]
+            self.check_tunnel_ip(oid, vtep)
+
+        self.remove_shf_appl_table(portchannel)
+        self.check_shf_asic_db_entries_deleted()
+
+
+
+
diff --git a/tests/test_fdbnhg.py b/tests/test_fdbnhg.py
@@ -0,0 +1,114 @@
+import pytest
+from swsscommon import swsscommon
+from typing import Dict, List
+
+tunnel_nh_id1 = '5'
+tunnel_nh_id2 = '6'
+tunnel_nh_ip1 = '10.5.5.5'
+tunnel_nh_ip2 = '10.6.6.6'
+tunnel_nhg_id = '536870913'
+
+app_fdb_name = 'VXLAN_FDB_TABLE:'
+tunnel_device = 'vtep-100'
+tunnel_vlan = 'Vlan100'
+tunnel_remote_fdb = '12:34:55:12:34:56'
+tunnel_remote_fdb_type = 'extern_learn'
+tunnel_vni = '1000'
+tunnel_src_ip = '1.1.1.1'
+dstport = '4789'
+tunnel_remote_fdb_type_static = 'static'
+
+
+def create_fdb_nexthop(dvs, id: str, ip: str) -> Dict[str, str]:
+    dvs.runcmd(f"ip nexthop add id {id} via {ip} fdb")
+    fvs = {"remote_vtep": ip}
+    dvs.get_app_db().wait_for_exact_match(swsscommon.APP_L2_NEXTHOP_GROUP_TABLE_NAME, id, fvs)
+
+
+def create_fdb_nexthop_group(dvs, id: str, nh_ids: List[str]) -> Dict[str, str]:
+    dvs.runcmd(f"ip nexthop add id {id} group {'/'.join(nh_ids)} fdb")
+    fvs = {"nexthop_group": ','.join(nh_ids)}
+    dvs.get_app_db().wait_for_exact_match(swsscommon.APP_L2_NEXTHOP_GROUP_TABLE_NAME, id, fvs)
+
+
+def remove_l2_nexthop(dvs, id: str):
+    dvs.runcmd(f"ip nexthop del id {id}")
+    return dvs.get_app_db().wait_for_deleted_entry(swsscommon.APP_L2_NEXTHOP_GROUP_TABLE_NAME, id)
+
+
+class TestFdbNhg(object):
+    @pytest.fixture
+    def setup_teardown_test(self, dvs):
+        dvs.setup_db()
+        yield
+        # ip nexthop objects are not disposed of otherwise upon test failure
+        dvs.runcmd("ip nexthop flush")
+
+    def test_add_nexthop(self, dvs, testlog, setup_teardown_test):
+        create_fdb_nexthop(dvs, tunnel_nh_id1, tunnel_nh_ip1)
+        remove_l2_nexthop(dvs, tunnel_nh_id1)
+
+    def test_add_nexthop_group_1_nh(self, dvs, testlog, setup_teardown_test):
+        create_fdb_nexthop(dvs, tunnel_nh_id1, tunnel_nh_ip1)
+        create_fdb_nexthop_group(dvs, tunnel_nhg_id, [tunnel_nh_id1])
+
+        remove_l2_nexthop(dvs, tunnel_nhg_id)
+        remove_l2_nexthop(dvs, tunnel_nh_id1)
+
+    def test_add_nexthop_group_2_nh(self, dvs, testlog, setup_teardown_test):
+        create_fdb_nexthop(dvs, tunnel_nh_id1, tunnel_nh_ip1)
+        create_fdb_nexthop(dvs, tunnel_nh_id2, tunnel_nh_ip2)
+        create_fdb_nexthop_group(dvs, tunnel_nhg_id, [tunnel_nh_id1, tunnel_nh_id2])
+
+        remove_l2_nexthop(dvs, tunnel_nhg_id)
+        remove_l2_nexthop(dvs, tunnel_nh_id1)
+        remove_l2_nexthop(dvs, tunnel_nh_id2)
+
+    def test_add_fdb_nexthop_group(self, dvs, testlog, setup_teardown_test):
+        create_fdb_nexthop(dvs, tunnel_nh_id1, tunnel_nh_ip1)
+        create_fdb_nexthop(dvs, tunnel_nh_id2, tunnel_nh_ip2)
+        create_fdb_nexthop_group(dvs, tunnel_nhg_id, [tunnel_nh_id1, tunnel_nh_id2])
+
+        dvs.runcmd(f"ip link add {tunnel_device} type vxlan id {tunnel_vni} local {tunnel_src_ip} dstport {dstport}")
+        dvs.runcmd(f"ip link set up {tunnel_device}")
+        dvs.runcmd(f"bridge fdb add {tunnel_remote_fdb} dev {tunnel_device} nhid {tunnel_nhg_id} self {tunnel_remote_fdb_type}")
+
+        # Check in the APP DB for the FDB entry to be present APP_VXLAN_FDB_TABLE_NAME "APP_VXLAN_FDB_TABLE_NAME"
+        # check application database
+        fvs = dvs.get_app_db().wait_for_entry(app_fdb_name+tunnel_vlan, tunnel_remote_fdb)
+        assert len(fvs) == 3
+        assert fvs.get("nexthop_group") == tunnel_nhg_id
+        assert fvs.get("type") == tunnel_remote_fdb_type_static
+        assert fvs.get("vni") == tunnel_vni
+
+        # Remove the fdb entry, and check the APP_DB
+        dvs.runcmd(f"bridge fdb del {tunnel_remote_fdb} dev {tunnel_device} nhid {tunnel_nhg_id} self {tunnel_remote_fdb_type}")
+        intf_entries = dvs.get_app_db().wait_for_deleted_entry(app_fdb_name+tunnel_vlan, tunnel_remote_fdb)
+
+        remove_l2_nexthop(dvs, tunnel_nhg_id)
+        remove_l2_nexthop(dvs, tunnel_nh_id1)
+        remove_l2_nexthop(dvs, tunnel_nh_id2)
+
+    def test_del_fdb_nhg_via_nhg_del(self, dvs, testlog, setup_teardown_test):
+        dvs.runcmd("swssloglevel -l INFO -c fpmsyncd")
+        dvs.runcmd("swssloglevel -l INFO -c fdbsyncd")
+        create_fdb_nexthop(dvs, tunnel_nh_id1, tunnel_nh_ip1)
+        create_fdb_nexthop(dvs, tunnel_nh_id2, tunnel_nh_ip2)
+        create_fdb_nexthop_group(dvs, tunnel_nhg_id, [tunnel_nh_id1, tunnel_nh_id2])
+
+        dvs.runcmd(f"ip link add {tunnel_device} type vxlan id {tunnel_vni} local {tunnel_src_ip} dstport {dstport}")
+        dvs.runcmd(f"ip link set up {tunnel_device}")
+        dvs.runcmd(f"bridge fdb add {tunnel_remote_fdb} dev {tunnel_device} nhid {tunnel_nhg_id} self {tunnel_remote_fdb_type}")
+
+        # Check in the APP DB for the FDB entry to be present APP_VXLAN_FDB_TABLE_NAME "APP_VXLAN_FDB_TABLE_NAME"
+        # check application database
+        fvs = dvs.get_app_db().wait_for_entry(app_fdb_name+tunnel_vlan, tunnel_remote_fdb)
+        assert len(fvs) == 3
+        assert fvs.get("nexthop_group") == tunnel_nhg_id
+        assert fvs.get("type") == tunnel_remote_fdb_type_static
+        assert fvs.get("vni") == tunnel_vni
+
+        remove_l2_nexthop(dvs, tunnel_nh_id1)
+        remove_l2_nexthop(dvs, tunnel_nh_id2)
+        dvs.get_app_db().wait_for_deleted_entry(swsscommon.APP_L2_NEXTHOP_GROUP_TABLE_NAME, tunnel_nhg_id)
+        dvs.get_app_db().wait_for_deleted_entry(app_fdb_name+tunnel_vlan, tunnel_remote_fdb)