This commit is contained in:
parent
af08f29d4d
commit
27c1e9bb42
@ -211,15 +211,12 @@ def test_db_event_checker_subscribe_table(mock_swsscommon_dbconnector_init, enab
|
||||
mock_sub.assert_called_once_with(ANY, "")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": "Vlan1000"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"Vlan1000"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_dhcp_server_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_dhcp_server_table_cfg_change_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot,
|
||||
enabled):
|
||||
def test_dhcp_server_table_cfg_change_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = DhcpServerTableCfgChangeEventChecker(sel, MagicMock())
|
||||
@ -232,15 +229,12 @@ def test_dhcp_server_table_cfg_change_checker(mock_swsscommon_dbconnector_init,
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": "Vlan1000"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"Vlan1000"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_dhcp_server_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_dhcp_server_table_enablement_change_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot,
|
||||
enabled):
|
||||
def test_dhcp_server_table_enablement_change_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = DhcpServerTableIntfEnablementEventChecker(sel, MagicMock())
|
||||
@ -253,14 +247,12 @@ def test_dhcp_server_table_enablement_change_checker(mock_swsscommon_dbconnector
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": "Vlan1000"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"Vlan1000"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_port_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_dhcp_port_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_dhcp_port_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = DhcpPortTableEventChecker(sel, MagicMock())
|
||||
@ -272,14 +264,12 @@ def test_dhcp_port_table_checker(mock_swsscommon_dbconnector_init, tested_data,
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"used_range": "range1"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"used_range": {"range1"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_range_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_dhcp_range_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_dhcp_range_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = DhcpRangeTableEventChecker(sel, MagicMock())
|
||||
@ -291,14 +281,12 @@ def test_dhcp_range_table_checker(mock_swsscommon_dbconnector_init, tested_data,
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"used_options": "option223"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"used_options": {"option223"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_option_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_dhcp_option_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_dhcp_option_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = DhcpOptionTableEventChecker(sel, MagicMock())
|
||||
@ -310,14 +298,12 @@ def test_dhcp_option_table_checker(mock_swsscommon_dbconnector_init, tested_data
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": "Vlan1000"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"Vlan1000"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_vlan_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_vlan_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_vlan_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = VlanTableEventChecker(sel, MagicMock())
|
||||
@ -329,14 +315,12 @@ def test_vlan_table_checker(mock_swsscommon_dbconnector_init, tested_data, teste
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": "Vlan1000"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"Vlan1000"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_vlan_intf_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_vlan_intf_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_vlan_intf_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = VlanIntfTableEventChecker(sel, MagicMock())
|
||||
@ -348,14 +332,12 @@ def test_vlan_intf_table_checker(mock_swsscommon_dbconnector_init, tested_data,
|
||||
assert expected_res == check_res
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": "Vlan1000"}, {}])
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"enabled_dhcp_interfaces": {"Vlan1000"}}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_vlan_member_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_vlan_member_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_vlan_member_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = VlanMemberTableEventChecker(sel, MagicMock())
|
||||
@ -370,12 +352,10 @@ def test_vlan_member_table_checker(mock_swsscommon_dbconnector_init, tested_data
|
||||
@pytest.mark.parametrize("tested_db_snapshot", [{"dhcp_server_feature_enabled": True},
|
||||
{"dhcp_server_feature_enabled": False}, {}])
|
||||
@pytest.mark.parametrize("tested_data", get_subscribe_table_tested_data("test_feature_update"))
|
||||
@pytest.mark.parametrize("enabled", [True, False])
|
||||
def test_feature_table_checker(mock_swsscommon_dbconnector_init, tested_data, tested_db_snapshot, enabled):
|
||||
def test_feature_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(ConfigDbEventChecker, "enabled", return_value=enabled, new_callable=PropertyMock), \
|
||||
patch.object(sys, "exit"):
|
||||
sel = swsscommon.Select()
|
||||
db_event_checker = DhcpServerFeatureStateChecker(sel, MagicMock())
|
||||
|
@ -142,5 +142,13 @@ def test_validate_ttr_type(test_data):
|
||||
|
||||
|
||||
def test_get_target_process_cmds():
|
||||
with patch.object(psutil, "process_iter", return_value=[MockProc("dhcrelay", 1), MockProc("dhcpmon", 2)], new_callable=PropertyMock):
|
||||
with patch.object(psutil, "process_iter", return_value=[MockProc("dhcrelay", 1), MockProc("dhcpmon", 2)],
|
||||
new_callable=PropertyMock):
|
||||
res = utils.get_target_process_cmds("dhcrelay")
|
||||
expected_res = [
|
||||
[
|
||||
"/usr/sbin/dhcrelay", "-d", "-m", "discard", "-a", "%h:%p", "%P", "--name-alias-map-file",
|
||||
"/tmp/port-name-alias-map.txt", "-id", "Vlan1000", "-iu", "docker0", "240.127.1.2"
|
||||
]
|
||||
]
|
||||
assert res == expected_res
|
||||
|
Reference in New Issue
Block a user