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

[dhcp_server] add config dhcp_server update (#17736)

Browse Source 
* add config dhcp_server update
This commit is contained in:
Xichen96 2024-01-11 02:33:05 +08:00 committed by GitHub
parent 2e27b11621
commit 0ecab6cfbc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 126 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
dockers/docker-dhcp-server/cli-plugin-tests/test_config_dhcp_server.py
View File

@ -158,3 +158,85 @@ class TestConfigDHCPServer(object):
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["disable"], ["Vlan200"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_does_not_exist(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan200"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_already_exist(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_wrong_mode(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--mode=XX"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_lease_time(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--lease_time=1800"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "lease_time") == "1800"
def test_config_dhcp_server_ipv4_update_wrong_lease_time(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--lease_time=-1800"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_dup_gw_nm(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--dup_gw_nm"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "gateway") == "100.1.1.1"
assert mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "netmask") == "255.255.255.0"
def test_config_dhcp_server_ipv4_update_dup_gw_nm_no_data(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan300", "--dup_gw_nm"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_gateway(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--gateway=100.1.1.9"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "gateway") == "100.1.1.9"
def test_config_dhcp_server_ipv4_update_wrong_gateway(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--gateway=100.1.9"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_update_netmask(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--netmask=255.255.254.0"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "netmask") == "255.255.254.0"
def test_config_dhcp_server_ipv4_update_wrong_netmask(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--netmask=255.255.254"], obj=db)
assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)

44
dockers/docker-dhcp-server/cli/config/plugins/dhcp_server.py
View File

@ -116,6 +116,50 @@ def dhcp_server_ipv4_del(db, dhcp_interface):
ctx.fail("Dhcp interface {} does not exist in config db".format(dhcp_interface))
@dhcp_server_ipv4.command(name="update")
@click.argument("dhcp_interface", required=True)
@click.option("--mode", required=False)
@click.option("--lease_time", required=False)
@click.option("--dup_gw_nm", required=False, default=False, is_flag=True)
@click.option("--gateway", required=False)
@click.option("--netmask", required=False)
@clicommon.pass_db
def dhcp_server_ipv4_update(db, mode, lease_time, dup_gw_nm, gateway, netmask, dhcp_interface):
ctx = click.get_current_context()
dbconn = db.db
key = "DHCP_SERVER_IPV4|" + dhcp_interface
if not dbconn.exists("CONFIG_DB", key):
ctx.fail("Dhcp interface {} does not exist in config db".format(dhcp_interface))
if mode:
if mode != "PORT":
ctx.fail("Only mode PORT is supported")
else:
dbconn.set("CONFIG_DB", key, "mode", mode)
if lease_time:
if not validate_str_type("uint32", lease_time):
ctx.fail("lease_time is required and must be nonnegative integer")
else:
dbconn.set("CONFIG_DB", key, "lease_time", lease_time)
if dup_gw_nm:
dup_success = False
for key in dbconn.keys("CONFIG_DB", "VLAN_INTERFACE|" + dhcp_interface + "|*"):
intf = ipaddress.ip_interface(key.split("|")[2])
if intf.version != 4:
continue
dup_success = True
gateway, netmask = str(intf.ip), str(intf.netmask)
if not dup_success:
ctx.fail("failed to found gateway and netmask for Vlan interface {}".format(dhcp_interface))
elif gateway and not validate_str_type("ipv4-address", gateway):
ctx.fail("gateway must be valid ipv4 string")
elif netmask and not validate_str_type("ipv4-address", netmask):
ctx.fail("netmask must be valid ipv4 string")
if gateway:
dbconn.set("CONFIG_DB", key, "gateway", gateway)
if netmask:
dbconn.set("CONFIG_DB", key, "netmask", netmask)
@dhcp_server_ipv4.command(name="enable")
@click.argument("dhcp_interface", required=True)
@clicommon.pass_db