matthew/sonic-buildimage
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[dhcp_server] Add support for smart switch in dhcprelayd (#17779)

Browse Source 
* [dhcp_server] Add support for smart switch in dhcprelayd
This commit is contained in:
Yaqiang Zhu 2024-01-17 01:52:50 +08:00 committed by GitHub
parent 6107b5151f
commit 36e111af80
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 274 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

141
src/sonic-dhcp-utilities/dhcp_utilities/dhcprelayd/dhcprelayd.py
View File

@ -8,21 +8,25 @@ import sys
import syslog
import time
from swsscommon import swsscommon
from dhcp_utilities.common.utils import DhcpDbConnector, terminate_proc, get_target_process_cmds
from dhcp_utilities.common.utils import DhcpDbConnector, terminate_proc, get_target_process_cmds, is_smart_switch
from dhcp_utilities.common.dhcp_db_monitor import DhcpRelaydDbMonitor, DhcpServerTableIntfEnablementEventChecker, \
VlanTableEventChecker, VlanIntfTableEventChecker, DhcpServerFeatureStateChecker
VlanTableEventChecker, VlanIntfTableEventChecker, DhcpServerFeatureStateChecker, MidPlaneTableEventChecker
REDIS_SOCK_PATH = "/var/run/redis/redis.sock"
SUPERVISORD_CONF_PATH = "/etc/supervisor/conf.d/docker-dhcp-relay.supervisord.conf"
DHCP_SERVER_IPV4_SERVER_IP = "DHCP_SERVER_IPV4_SERVER_IP"
DHCP_SERVER_IPV4 = "DHCP_SERVER_IPV4"
VLAN = "VLAN"
MID_PLANE_BRIDGE = "MID_PLANE_BRIDGE"
DEVICE_METADATA = "DEVICE_METADATA"
DEFAULT_SELECT_TIMEOUT = 5000 # millisecond
DHCP_SERVER_INTERFACE = "eth0"
FEATURE_CHECKER = "DhcpServerFeatureStateChecker"
DHCP_SERVER_CHECKER = "DhcpServerTableIntfEnablementEventChecker"
VLAN_CHECKER = "VlanTableEventChecker"
VLAN_INTF_CHECKER = "VlanIntfTableEventChecker"
MID_PLANE_CHECKER = "MidPlaneTableEventChecker"
VLAN_CHECKERS = [VLAN_CHECKER, VLAN_INTF_CHECKER]
KILLED_OLD = 1
NOT_KILLED = 2
NOT_FOUND_PROC = 3
@ -33,8 +37,11 @@ class DhcpRelayd(object):
dhcp_server_feature_enabled = None
dhcp_relay_supervisor_config = {}
supervisord_conf_path = ""
enabled_checkers = set()
smart_switch = False
def __init__(self, db_connector, db_monitor, supervisord_conf_path=SUPERVISORD_CONF_PATH):
def __init__(self, db_connector, db_monitor, supervisord_conf_path=SUPERVISORD_CONF_PATH,
enabled_checkers=[FEATURE_CHECKER]):
"""
Args:
db_connector: db connector obj
@ -46,6 +53,7 @@ class DhcpRelayd(object):
self.enabled_dhcp_interfaces = set()
self.dhcp_server_feature_enabled = None
self.supervisord_conf_path = supervisord_conf_path
self.enabled_checkers = set(enabled_checkers)
def start(self):
"""
@ -53,12 +61,15 @@ class DhcpRelayd(object):
"""
self.dhcp_relay_supervisor_config = self._get_dhcp_relay_config()
self.dhcp_server_feature_enabled = self._is_dhcp_server_enabled()
device_metadata = self.db_connector.get_config_db_table(DEVICE_METADATA)
self.smart_switch = is_smart_switch(device_metadata)
# Sleep to wait dhcrelay process start
time.sleep(5)
if self.dhcp_server_feature_enabled:
# If dhcp_server is enabled, need to stop related relay processes start by supervisord
self._execute_supervisor_dhcp_relay_process("stop")
self.dhcp_relayd_monitor.enable_checkers([DHCP_SERVER_CHECKER, VLAN_CHECKER, VLAN_INTF_CHECKER])
self.enabled_checkers.add(DHCP_SERVER_CHECKER)
self.dhcp_relayd_monitor.enable_checkers(self.enabled_checkers)
def refresh_dhcrelay(self, force_kill=False):
"""
@ -70,9 +81,12 @@ class DhcpRelayd(object):
dhcp_server_ip = self._get_dhcp_server_ip()
dhcp_server_ipv4_table = self.db_connector.get_config_db_table(DHCP_SERVER_IPV4)
vlan_table = self.db_connector.get_config_db_table(VLAN)
mid_plane_table = self.db_connector.get_config_db_table(MID_PLANE_BRIDGE)
mid_plane_bridge_name = mid_plane_table.get("GLOBAL", {}).get("bridge", None)
dhcp_interfaces = set()
self.enabled_dhcp_interfaces = set()
checkers_to_be_enabled = set()
for dhcp_interface, config in dhcp_server_ipv4_table.items():
# Reason for add to enabled_dhcp_interfaces firstly is for below scenario:
# Firstly vlan 1000 is not in vlan table but enabled in dhcp_server table, then add vlan1000 to vlan table
@ -80,9 +94,20 @@ class DhcpRelayd(object):
if config["state"] == "enabled":
dhcp_interfaces.add(dhcp_interface)
self.enabled_dhcp_interfaces.add(dhcp_interface)
if dhcp_interface not in vlan_table:
if dhcp_interface not in vlan_table and dhcp_interface != mid_plane_bridge_name:
dhcp_interfaces.discard(dhcp_interface)
continue
if dhcp_interface in vlan_table:
checkers_to_be_enabled |= set(VLAN_CHECKERS)
elif dhcp_interface == mid_plane_bridge_name and self.smart_switch:
checkers_to_be_enabled |= set([MID_PLANE_CHECKER])
self._enable_checkers(checkers_to_be_enabled - self.enabled_checkers)
# Checkers for FEATURE and DHCP_SERVER_IPV4 table should not be disabled
checkers_to_be_disabled = self.enabled_checkers - checkers_to_be_enabled - \
set([FEATURE_CHECKER, DHCP_SERVER_CHECKER])
self._disable_checkers(checkers_to_be_disabled)
self._start_dhcrelay_process(dhcp_interfaces, dhcp_server_ip, force_kill)
self._start_dhcpmon_process(dhcp_interfaces, force_kill)
@ -96,41 +121,77 @@ class DhcpRelayd(object):
"dhcp_server_feature_enabled": self.dhcp_server_feature_enabled
}
res = (self.dhcp_relayd_monitor.check_db_update(check_param))
dhcp_feature_statue_changed = False
if FEATURE_CHECKER in res:
dhcp_feature_statue_changed = res[FEATURE_CHECKER]
self.dhcp_server_feature_enabled = not self.dhcp_server_feature_enabled if dhcp_feature_statue_changed \
else self.dhcp_server_feature_enabled
# If dhcp_server feature is enabled, dhcprelayd will manage dhcpmon/dhcrelay process
if self.dhcp_server_feature_enabled:
# disabled -> enabled, we need to enable dhcp_server related checkers and do refresh processes
if dhcp_feature_statue_changed:
self.dhcp_relayd_monitor.enable_checkers([DHCP_SERVER_CHECKER, VLAN_CHECKER, VLAN_INTF_CHECKER])
# Stop dhcrelay process
self._execute_supervisor_dhcp_relay_process("stop")
self.refresh_dhcrelay()
# enabled -> enabled, just need to check dhcp_server related tables to see whether need to refresh
else:
# Check vlan_interface table change, if it changed, need to refresh with force kill
if res.get(VLAN_INTF_CHECKER, False):
self.refresh_dhcrelay(True)
elif res.get(VLAN_CHECKER, False) or res.get(DHCP_SERVER_CHECKER, False):
self.refresh_dhcrelay(False)
self._proceed_with_check_res(res, self.dhcp_server_feature_enabled)
# If dhcp_server feature is disabled, dhcprelayd will checke whether dhcpmon/dhcrelay processes,
# if they are not running as expected, dhcprelayd will kill itself to make dhcp_relay container restart.
def _proceed_with_check_res(self, check_res, previous_dhcp_server_status):
"""
Proceed depends on check result
Args:
check_res: result of checker, sample: {
"DhcpServerFeatureStateChecker": True,
"VlanIntfTableEventChecker": False
}
previous_dhcp_server_status: previous enabled/disabled status of dhcp_server feature
"""
dhcp_feature_status_changed = False
if FEATURE_CHECKER in check_res:
dhcp_feature_status_changed = check_res[FEATURE_CHECKER]
self.dhcp_server_feature_enabled = not previous_dhcp_server_status if dhcp_feature_status_changed \
else previous_dhcp_server_status
# If dhcp_server feature is enabled, dhcprelayd will manage dhcpmon/dhcrelay process
if self.dhcp_server_feature_enabled:
# disabled -> enabled, we need to enable dhcp_server checker and do refresh processes
if dhcp_feature_status_changed:
self._enable_checkers([DHCP_SERVER_CHECKER])
# Stop dhcrelay process
self._execute_supervisor_dhcp_relay_process("stop")
self.refresh_dhcrelay()
# enabled -> enabled, just need to check dhcp_server related tables to see whether need to refresh
else:
# enabled -> disabled, we need to disable dhcp_server related checkers and start dhcrelay/dhcpmon
# processes follow supervisord configuration
if dhcp_feature_statue_changed:
self.dhcp_relayd_monitor.disable_checkers([DHCP_SERVER_CHECKER, VLAN_CHECKER, VLAN_INTF_CHECKER])
self._kill_exist_relay_releated_process([], "dhcpmon", True)
self._kill_exist_relay_releated_process([], "dhcrelay", True)
self._execute_supervisor_dhcp_relay_process("start")
# disabled -> disabled, to check whether dhcpmon/dhcrelay running status consistent with supervisord
# configuration
else:
self._check_dhcp_relay_processes()
# Check vlan_interface table change, if it changed, need to refresh with force kill
if (check_res.get(VLAN_INTF_CHECKER, False) or check_res.get(MID_PLANE_CHECKER, False) or
check_res.get(MID_PLANE_CHECKER, False)):
self.refresh_dhcrelay(True)
elif check_res.get(VLAN_CHECKER, False) or check_res.get(DHCP_SERVER_CHECKER, False):
self.refresh_dhcrelay(False)
# If dhcp_server feature is disabled, dhcprelayd will checke whether dhcpmon/dhcrelay processes,
# if they are not running as expected, dhcprelayd will kill itself to make dhcp_relay container restart.
else:
# enabled -> disabled, we need to disable dhcp_server related checkers and start dhcrelay/dhcpmon
# processes follow supervisord configuration
if dhcp_feature_status_changed:
checkers_to_be_disabled = [DHCP_SERVER_CHECKER] + VLAN_CHECKERS
if self.smart_switch:
checkers_to_be_disabled.append(MID_PLANE_CHECKER)
self._disable_checkers(self.enabled_checkers & set(checkers_to_be_disabled))
self._kill_exist_relay_releated_process([], "dhcpmon", True)
self._kill_exist_relay_releated_process([], "dhcrelay", True)
self._execute_supervisor_dhcp_relay_process("start")
# disabled -> disabled, to check whether dhcpmon/dhcrelay running status consistent with supervisord
# configuration
else:
self._check_dhcp_relay_processes()
def _enable_checkers(self, checkers):
"""
Update set of enabled_checkers and enable checkers
Args:
checkers: checkers need to be enabled
"""
new_enabled_checkers = set(checkers) - self.enabled_checkers
self.dhcp_relayd_monitor.enable_checkers(new_enabled_checkers)
self.enabled_checkers = self.enabled_checkers | new_enabled_checkers
def _disable_checkers(self, checkers):
"""
Update set of enabled_checkers and disable checkers
Args:
checkers: checkers need to be disabled
"""
new_disabled_checkers = set(checkers) & self.enabled_checkers
self.dhcp_relayd_monitor.disable_checkers(new_disabled_checkers)
self.enabled_checkers = self.enabled_checkers - new_disabled_checkers
def _is_dhcp_server_enabled(self):
"""
@ -300,10 +361,10 @@ def main():
checkers.append(DhcpServerTableIntfEnablementEventChecker(sel, dhcp_db_connector.config_db))
checkers.append(VlanIntfTableEventChecker(sel, dhcp_db_connector.config_db))
checkers.append(VlanTableEventChecker(sel, dhcp_db_connector.config_db))
checkers.append(MidPlaneTableEventChecker(sel, dhcp_db_connector.config_db))
checkers.append(DhcpServerFeatureStateChecker(sel, dhcp_db_connector.config_db))
db_monitor = DhcpRelaydDbMonitor(dhcp_db_connector, sel, checkers, DEFAULT_SELECT_TIMEOUT)
db_monitor.enable_checkers([FEATURE_CHECKER])
dhcprelayd = DhcpRelayd(dhcp_db_connector, db_monitor)
dhcprelayd = DhcpRelayd(dhcp_db_connector, db_monitor, enabled_checkers=[FEATURE_CHECKER])
dhcprelayd.start()
dhcprelayd.wait()

72
src/sonic-dhcp-utilities/tests/common_utils.py
View File

@ -1,6 +1,10 @@
import heapq
import json
import psutil
from dhcp_utilities.common.dhcp_db_monitor import DhcpRelaydDbMonitor
from dhcp_utilities.common.utils import DhcpDbConnector
from dhcp_utilities.dhcprelayd.dhcprelayd import DhcpRelayd, FEATURE_CHECKER, DHCP_SERVER_CHECKER, VLAN_INTF_CHECKER
from unittest.mock import patch, PropertyMock, call
MOCK_CONFIG_DB_PATH = "tests/test_data/mock_config_db.json"
TEST_DATA_PATH = "tests/test_data/dhcp_db_monitor_test_data.json"
@ -114,3 +118,71 @@ def get_subscribe_table_tested_data(test_name):
class MockSubprocessRes(object):
def __init__(self, returncode):
self.returncode = returncode
def dhcprelayd_refresh_dhcrelay_test(expected_checkers, is_smart_switch, mock_get_config_db_table):
with patch.object(DhcpRelayd, "_get_dhcp_server_ip", return_value="240.127.1.2"), \
patch.object(DhcpDbConnector, "get_config_db_table", side_effect=mock_get_config_db_table), \
patch.object(DhcpRelayd, "_start_dhcrelay_process", return_value=None), \
patch.object(DhcpRelayd, "_start_dhcpmon_process", return_value=None), \
patch.object(DhcpRelayd, "_enable_checkers") as mock_enable_checkers, \
patch.object(DhcpRelayd, "_disable_checkers") as mock_disable_checkers, \
patch.object(DhcpRelayd, "smart_switch", return_value=is_smart_switch, new_callable=PropertyMock):
dhcp_db_connector = DhcpDbConnector()
dhcprelayd = DhcpRelayd(dhcp_db_connector, None)
dhcprelayd.refresh_dhcrelay()
mock_enable_checkers.assert_called_once_with(expected_checkers)
mock_disable_checkers.assert_called_once_with(set())
def dhcprelayd_proceed_with_check_res_test(enabled_checkers, feature_enabled, feature_res, dhcp_server_res,
vlan_intf_res, is_smart_switch, expected_checkers):
with patch.object(DhcpRelayd, "_enable_checkers") as mock_enable_checkers, \
patch.object(DhcpRelayd, "_execute_supervisor_dhcp_relay_process") as mock_execute_process, \
patch.object(DhcpRelayd, "refresh_dhcrelay") as mock_refresh_dhcrelay, \
patch.object(DhcpRelayd, "_disable_checkers") as mock_disable_checkers, \
patch.object(DhcpRelayd, "_kill_exist_relay_releated_process") as mock_kill_process, \
patch.object(DhcpRelayd, "_check_dhcp_relay_processes") as mock_check_process, \
patch.object(DhcpRelayd, "smart_switch", return_value=is_smart_switch,
new_callable=PropertyMock), \
patch.object(DhcpRelayd, "enabled_checkers", return_value=enabled_checkers, new_callable=PropertyMock):
dhcp_db_connector = DhcpDbConnector()
dhcprelayd = DhcpRelayd(dhcp_db_connector, DhcpRelaydDbMonitor(None, None, []))
dhcprelayd.dhcp_server_feature_enabled = True if feature_enabled else False
check_res = {}
if feature_res is not None:
check_res[FEATURE_CHECKER] = feature_res
if dhcp_server_res is not None:
check_res[DHCP_SERVER_CHECKER] = dhcp_server_res
if vlan_intf_res is not None:
check_res[VLAN_INTF_CHECKER] = vlan_intf_res
dhcprelayd._proceed_with_check_res(check_res, feature_enabled)
if feature_res is None or not feature_res:
# Feature status didn't change
# disabled -> disabled
if not feature_enabled:
mock_check_process.assert_called_once_with()
# enabled-> enabled
else:
if vlan_intf_res:
mock_refresh_dhcrelay.assert_called_once_with(True)
elif dhcp_server_res:
mock_refresh_dhcrelay.assert_called_once_with(False)
else:
mock_refresh_dhcrelay.assert_not_called()
else:
# Feature status changed
# enabled -> disabled
if feature_enabled:
mock_disable_checkers.assert_called_once_with(expected_checkers)
mock_kill_process.assert_has_calls([
call([], "dhcpmon", True),
call([], "dhcrelay", True)
])
# disabled-> enabled
else:
mock_enable_checkers.assert_called_once_with([DHCP_SERVER_CHECKER])
mock_execute_process.assert_called_once_with("stop")
mock_refresh_dhcrelay.assert_called_once_with()

27
src/sonic-dhcp-utilities/tests/test_data/mock_config_db_smart_switch.json Normal file
View File

@ -0,0 +1,27 @@
{
"DEVICE_METADATA": {
"localhost": {
"hostname": "sonic-host",
"subtype": "SmartSwitch"
}
},
"MID_PLANE_BRIDGE": {
"GLOBAL": {
"bridge": "bridge_midplane",
"address": "169.254.200.254/24"
}
},
"DHCP_SERVER_IPV4": {
"bridge_midplane": {
"customized_options": [
"option60",
"option223"
],
"gateway": "169.254.200.254",
"lease_time": "900",
"mode": "PORT",
"netmask": "255.255.255.0",
"state": "enabled"
}
}
}

4
src/sonic-dhcp-utilities/tests/test_dhcp_cfggen.py
View File

@ -2,10 +2,10 @@ import copy
import ipaddress
import json
import pytest
from common_utils import MockConfigDb, mock_get_config_db_table, PORT_MODE_CHECKER, SMART_SWITCH_CHECKER
from common_utils import MockConfigDb, mock_get_config_db_table, PORT_MODE_CHECKER
from dhcp_utilities.common.utils import DhcpDbConnector
from dhcp_utilities.dhcpservd.dhcp_cfggen import DhcpServCfgGenerator
from unittest.mock import patch, MagicMock
from unittest.mock import patch
expected_dhcp_config = {
"Dhcp4": {

62
src/sonic-dhcp-utilities/tests/test_dhcprelayd.py
View File

@ -3,12 +3,14 @@ import pytest
import subprocess
import sys
import time
from common_utils import mock_get_config_db_table, MockProc, MockPopen, MockSubprocessRes, mock_exit_func
from common_utils import mock_get_config_db_table, MockProc, MockPopen, MockSubprocessRes, mock_exit_func, \
dhcprelayd_refresh_dhcrelay_test, dhcprelayd_proceed_with_check_res_test
from dhcp_utilities.common.utils import DhcpDbConnector
from dhcp_utilities.common.dhcp_db_monitor import ConfigDbEventChecker, DhcpRelaydDbMonitor
from dhcp_utilities.dhcprelayd.dhcprelayd import DhcpRelayd, KILLED_OLD, NOT_KILLED, NOT_FOUND_PROC
from dhcp_utilities.dhcprelayd.dhcprelayd import DhcpRelayd, KILLED_OLD, NOT_KILLED, NOT_FOUND_PROC, \
DHCP_SERVER_CHECKER, VLAN_CHECKERS
from swsscommon import swsscommon
from unittest.mock import patch, call, ANY, PropertyMock
from unittest.mock import patch, call, PropertyMock
@pytest.mark.parametrize("dhcp_server_enabled", [True, False])
@ -16,30 +18,26 @@ def test_start(mock_swsscommon_dbconnector_init, dhcp_server_enabled):
with patch.object(DhcpRelayd, "_get_dhcp_relay_config") as mock_get_config, \
patch.object(DhcpRelayd, "_is_dhcp_server_enabled", return_value=dhcp_server_enabled) as mock_enabled, \
patch.object(DhcpRelayd, "_execute_supervisor_dhcp_relay_process") as mock_execute, \
patch.object(DhcpRelaydDbMonitor, "enable_checkers") as mock_enable_checkers, \
patch.object(time, "sleep"):
patch.object(time, "sleep"), \
patch.object(DhcpRelaydDbMonitor, "enable_checkers") as mock_enabled_checkers, \
patch.object(DhcpDbConnector, "get_config_db_table", side_effect=mock_get_config_db_table):
dhcp_db_connector = DhcpDbConnector()
dhcprelayd = DhcpRelayd(dhcp_db_connector, DhcpRelaydDbMonitor)
dhcprelayd = DhcpRelayd(dhcp_db_connector, DhcpRelaydDbMonitor(None, None, []))
dhcprelayd.start()
mock_get_config.assert_called_once_with()
mock_enabled.assert_called_once_with()
enabled_checkers = set(["DhcpServerFeatureStateChecker"])
if dhcp_server_enabled:
mock_execute.assert_called_once_with("stop")
mock_enable_checkers.assert_called_once_with([ANY, ANY, ANY])
enabled_checkers.add("DhcpServerTableIntfEnablementEventChecker")
else:
mock_execute.assert_not_called()
mock_enable_checkers.assert_not_called()
mock_enabled_checkers.assert_called_once_with(enabled_checkers)
def test_refresh_dhcrelay(mock_swsscommon_dbconnector_init):
with patch.object(DhcpRelayd, "_get_dhcp_server_ip", return_value="240.127.1.2"), \
patch.object(DhcpDbConnector, "get_config_db_table", side_effect=mock_get_config_db_table), \
patch.object(DhcpRelayd, "_start_dhcrelay_process", return_value=None), \
patch.object(DhcpRelayd, "_start_dhcpmon_process", return_value=None), \
patch.object(ConfigDbEventChecker, "enable"):
dhcp_db_connector = DhcpDbConnector()
dhcprelayd = DhcpRelayd(dhcp_db_connector, None)
dhcprelayd.refresh_dhcrelay()
expected_checkers = set(["VlanIntfTableEventChecker", "VlanTableEventChecker"])
dhcprelayd_refresh_dhcrelay_test(expected_checkers, False, mock_get_config_db_table)
@pytest.mark.parametrize("new_dhcp_interfaces", [[], ["Vlan1000"], ["Vlan1000", "Vlan2000"]])
@ -232,3 +230,35 @@ def test_get_dhcp_relay_config(mock_swsscommon_dbconnector_init, mock_swsscommon
"PortChannel103", "-iu", "PortChannel104", "-im", "eth0"
]
}
@pytest.mark.parametrize("enabled_checkers", [set(["dummy"]), set()])
def test_enable_checkers(mock_swsscommon_dbconnector_init, mock_swsscommon_table_init, enabled_checkers):
with patch.object(DhcpRelayd, "enabled_checkers", return_value=enabled_checkers, new_callable=PropertyMock), \
patch.object(DhcpRelaydDbMonitor, "enable_checkers") as mock_enable_checkers:
dhcp_db_connector = DhcpDbConnector()
dhcprelayd = DhcpRelayd(dhcp_db_connector, DhcpRelaydDbMonitor(None, None, []))
dhcprelayd._enable_checkers(["dummy"])
mock_enable_checkers.assert_called_once_with(set() if "dummy" in enabled_checkers else set(["dummy"]))
@pytest.mark.parametrize("enabled_checkers", [set(["dummy"]), set()])
def test_disable_checkers(mock_swsscommon_dbconnector_init, mock_swsscommon_table_init, enabled_checkers):
with patch.object(DhcpRelayd, "enabled_checkers", return_value=enabled_checkers, new_callable=PropertyMock), \
patch.object(DhcpRelaydDbMonitor, "disable_checkers") as mock_disable_checkers:
dhcp_db_connector = DhcpDbConnector()
dhcprelayd = DhcpRelayd(dhcp_db_connector, DhcpRelaydDbMonitor(None, None, []))
dhcprelayd._disable_checkers(["dummy"])
mock_disable_checkers.assert_called_once_with(set() if "dummy" not in enabled_checkers else set(["dummy"]))
@pytest.mark.parametrize("feature_enabled", [True, False])
@pytest.mark.parametrize("feature_res", [True, False, None])
@pytest.mark.parametrize("dhcp_server_res", [True, False, None])
@pytest.mark.parametrize("vlan_intf_res", [True, False, None])
def test_proceed_with_check_res(mock_swsscommon_dbconnector_init, mock_swsscommon_table_init, feature_enabled,
feature_res, dhcp_server_res, vlan_intf_res):
enabled_checkers = set([DHCP_SERVER_CHECKER] + VLAN_CHECKERS)
expected_checkers = set([DHCP_SERVER_CHECKER] + VLAN_CHECKERS)
dhcprelayd_proceed_with_check_res_test(enabled_checkers, feature_enabled, feature_res, dhcp_server_res,
vlan_intf_res, False, expected_checkers)

26
src/sonic-dhcp-utilities/tests/test_smart_switch.py Normal file
View File

@ -0,0 +1,26 @@
import pytest
from common_utils import MockConfigDb, dhcprelayd_refresh_dhcrelay_test, dhcprelayd_proceed_with_check_res_test
from dhcp_utilities.dhcprelayd.dhcprelayd import DHCP_SERVER_CHECKER, MID_PLANE_CHECKER
MOCK_CONFIG_DB_PATH_SMART_SWITCH = "tests/test_data/mock_config_db_smart_switch.json"
def test_dhcprelayd_refresh_dhcrelay(mock_swsscommon_dbconnector_init):
expected_checkers = set(["MidPlaneTableEventChecker"])
dhcprelayd_refresh_dhcrelay_test(expected_checkers, True, mock_get_config_db_table)
@pytest.mark.parametrize("feature_enabled", [True, False])
@pytest.mark.parametrize("feature_res", [True, False, None])
@pytest.mark.parametrize("dhcp_server_res", [True, False, None])
def test_dhcprelayd_proceed_with_check_res(mock_swsscommon_dbconnector_init, mock_swsscommon_table_init,
feature_enabled, feature_res, dhcp_server_res):
enabled_checkers = set([DHCP_SERVER_CHECKER, MID_PLANE_CHECKER])
expected_checkers = set([DHCP_SERVER_CHECKER, MID_PLANE_CHECKER])
dhcprelayd_proceed_with_check_res_test(enabled_checkers, feature_enabled, feature_res, dhcp_server_res,
None, True, expected_checkers)
def mock_get_config_db_table(table_name):
mock_config_db = MockConfigDb(MOCK_CONFIG_DB_PATH_SMART_SWITCH)
return mock_config_db.get_config_db_table(table_name)