[smart_switch][dhcp_server] Add related checker for smart_switch in dhcp_db_monitor (#17338)

src/sonic-dhcp-utilities/dhcp_utilities/common/dhcp_db_monitor.py

@@ -13,6 +13,8 @@ VLAN = "VLAN"
 VLAN_MEMBER = "VLAN_MEMBER"
 VLAN_INTERFACE = "VLAN_INTERFACE"
 FEATURE = "FEATURE"
+MID_PLANE_BRIDGE = "MID_PLANE_BRIDGE"
+DPUS = "DPUS"
 
 
 class ConfigDbEventChecker(object):
@@ -344,6 +346,45 @@ class VlanMemberTableEventChecker(ConfigDbEventChecker):
         return False
 
 
+class MidPlaneTableEventChecker(ConfigDbEventChecker):
+    """
+    This event checker interested in changes in MID_PLANE_BRIDGE table
+    """
+    table_name = MID_PLANE_BRIDGE
+
+    def __init__(self, sel, db):
+        self.table_name = MID_PLANE_BRIDGE
+        ConfigDbEventChecker.__init__(self, sel, db)
+
+    def _get_parameter(self, db_snapshot):
+        return ConfigDbEventChecker.get_parameter_by_name(db_snapshot, "enabled_dhcp_interfaces")
+
+    def _process_check(self, key, op, entry, enabled_dhcp_interfaces):
+        if op == "DEL":
+            return True
+        for field, value in entry:
+            if field == "bridge" and value in enabled_dhcp_interfaces:
+                return True
+        return False
+
+
+class DpusTableEventChecker(ConfigDbEventChecker):
+    """
+    This event checker interested in changes in DPUS table
+    """
+    table_name = DPUS
+
+    def __init__(self, sel, db):
+        self.table_name = DPUS
+        ConfigDbEventChecker.__init__(self, sel, db)
+
+    def _get_parameter(self, db_snapshot):
+        return True, None
+
+    def _process_check(self, key, op, entry, param):
+        return True
+
+
 class DhcpServerFeatureStateChecker(ConfigDbEventChecker):
     """
     This event checker interested in dhcp_server feature state change in FEATURE table

src/sonic-dhcp-utilities/tests/test_data/dhcp_db_monitor_test_data.json

@@ -289,5 +289,51 @@
                 "pre_disabled": false
             }
         }
+    ],
+    "test_mid_plane_update": [
+        {
+            "table": [
+                ["GLOBAL", "SET", [["bridge", "bridge_midplane"], ["ip_prefix", "169.254.200.254/24"]]]
+            ],
+            "exp_res": true
+        },
+        {
+            "table": [
+                ["GLOBAL", "SET", [["bridge", "bridge_midplane2"], ["ip_prefix", "169.254.200.254/24"]]]
+            ],
+            "exp_res": false
+        },
+        {
+            "table": [
+                ["GLOBAL", "DEL", []]
+            ],
+            "exp_res": true
+        }
+    ],
+    "test_dpus_update": [
+        {
+            "table": [
+                ["dpu0", "SET", [["midplane_interface", "dpu0"]]]
+            ],
+            "exp_res": true
+        },
+        {
+            "table": [
+                ["dpu1", "SET", [["midplane_interface", "dpu1"]]]
+            ],
+            "exp_res": true
+        },
+        {
+            "table": [
+                ["dpu0", "DEL", []]
+            ],
+            "exp_res": true
+        },
+        {
+            "table": [
+                ["dpu1", "DEL", []]
+            ],
+            "exp_res": true
+        }
     ]
 }

src/sonic-dhcp-utilities/tests/test_dhcp_db_monitor.py

@@ -5,7 +5,8 @@ from common_utils import MockSubscribeTable, get_subscribe_table_tested_data, \
 from dhcp_utilities.common.dhcp_db_monitor import DhcpRelaydDbMonitor, DhcpServdDbMonitor, ConfigDbEventChecker, \
     DhcpServerTableIntfEnablementEventChecker, DhcpServerTableCfgChangeEventChecker, \
     DhcpPortTableEventChecker, DhcpRangeTableEventChecker, DhcpOptionTableEventChecker, \
-    VlanTableEventChecker, VlanMemberTableEventChecker, VlanIntfTableEventChecker, DhcpServerFeatureStateChecker
+    VlanTableEventChecker, VlanMemberTableEventChecker, VlanIntfTableEventChecker, DhcpServerFeatureStateChecker, \
+    MidPlaneTableEventChecker, DpusTableEventChecker
 from dhcp_utilities.common.utils import DhcpDbConnector
 from swsscommon import swsscommon
 from unittest.mock import patch, ANY, PropertyMock, MagicMock
@@ -366,3 +367,37 @@ def test_feature_table_checker(mock_swsscommon_dbconnector_init, tested_data, te
             expected_res = tested_data["exp_res"]["pre_enabled"] if tested_db_snapshot["dhcp_server_feature_enabled"] \
                 else tested_data["exp_res"]["pre_disabled"]
             assert expected_res == check_res
+
+
+@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"bridge_midplane": ["dpu0"]}}, {}])
+@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_mid_plane_update"))
+def test_mid_plane_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot):
+    with patch.object(ConfigDbEventChecker, "enable"), \
+         patch.object(ConfigDbEventChecker, "subscriber_state_table",
+                      return_value=MockSubscribeTable(tested_data["table"]), new_callable=PropertyMock), \
+         patch.object(sys, "exit"):
+        sel = swsscommon.Select()
+        db_event_checker = MidPlaneTableEventChecker(sel, MagicMock())
+        expected_res = tested_data["exp_res"]
+        check_res = db_event_checker.check_update_event(tested_db_snapshot)
+        if "enabled_dhcp_interfaces" not in tested_db_snapshot:
+            assert check_res
+        else:
+            assert expected_res == check_res
+
+
+@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"bridge_midplane": ["dpu0"]}}, {}])
+@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_dpus_update"))
+def test_dpus_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot):
+    with patch.object(ConfigDbEventChecker, "enable"), \
+         patch.object(ConfigDbEventChecker, "subscriber_state_table",
+                      return_value=MockSubscribeTable(tested_data["table"]), new_callable=PropertyMock), \
+         patch.object(sys, "exit"):
+        sel = swsscommon.Select()
+        db_event_checker = DpusTableEventChecker(sel, MagicMock())
+        expected_res = tested_data["exp_res"]
+        check_res = db_event_checker.check_update_event(tested_db_snapshot)
+        if "enabled_dhcp_interfaces" not in tested_db_snapshot:
+            assert check_res
+        else:
+            assert expected_res == check_res