[dhcp_server] add config dhcp server range (#17741)

* add range related function and ut
This commit is contained in:
Xichen96 2024-01-17 11:24:57 +08:00 committed by GitHub
parent 00fa56760f
commit a100f15ba2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 217 additions and 15 deletions

View File

@ -49,10 +49,10 @@
"ips": "100.1.1.10,10.1.1.11" "ips": "100.1.1.10,10.1.1.11"
}, },
"DHCP_SERVER_IPV4_PORT|Vlan100|Ethernet7": { "DHCP_SERVER_IPV4_PORT|Vlan100|Ethernet7": {
"ranges": "range1,range2" "ranges": "range1,range3"
}, },
"DHCP_SERVER_IPV4_PORT|Vlan200|Ethernet8": { "DHCP_SERVER_IPV4_PORT|Vlan200|Ethernet8": {
"ranges": "range3,range4" "ranges": "range1,range4"
}, },
"DHCP_SERVER_IPV4_PORT|Ethernet9": { "DHCP_SERVER_IPV4_PORT|Ethernet9": {
"ranges": "range5,range6" "ranges": "range5,range6"

View File

@ -240,3 +240,133 @@ class TestConfigDHCPServer(object):
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["update"], ["Vlan100", "--netmask=255.255.254"], obj=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) assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
def test_config_dhcp_server_ipv4_range_add(self, mock_db):
expected_value = {
"range": "10.10.10.10,10.10.10.11"
}
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["add"], \
["range4", "10.10.10.10", "10.10.10.11"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get_all("CONFIG_DB", "DHCP_SERVER_IPV4_RANGE|range4") == expected_value
def test_config_dhcp_server_ipv4_range_add_existing(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["add"], \
["range1", "10.10.10.10", "10.10.10.11"], 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_range_add_single_ip(self, mock_db):
expected_value = {
"range": "10.10.10.10,10.10.10.10"
}
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["add"], \
["range4", "10.10.10.10"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get_all("CONFIG_DB", "DHCP_SERVER_IPV4_RANGE|range4") == expected_value
def test_config_dhcp_server_ipv4_range_add_wrong_ip(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["add"], \
["range4", "10.10.10"], 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_range_add_wrong_order(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["add"], \
["range4", "10.10.10.10", "10.10.10.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_range_update(self, mock_db):
expected_value = {
"range": "10.10.10.10,10.10.10.11"
}
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["update"], \
["range1", "10.10.10.10", "10.10.10.11"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get_all("CONFIG_DB", "DHCP_SERVER_IPV4_RANGE|range1") == expected_value
def test_config_dhcp_server_ipv4_range_update_nonexisting(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["update"], \
["range4", "10.10.10.10", "10.10.10.11"], 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_range_update_single_ip(self, mock_db):
expected_value = {
"range": "10.10.10.10,10.10.10.10"
}
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["update"], \
["range1", "10.10.10.10"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.get_all("CONFIG_DB", "DHCP_SERVER_IPV4_RANGE|range1") == expected_value
def test_config_dhcp_server_ipv4_range_update_wrong_ip(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["update"], \
["range1", "10.10.10"], 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_range_add_wrong_order(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["update"], \
["range1", "10.10.10.10", "10.10.10.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_range_delete(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["del"], \
["range2"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.exists("CONFIG_DB", "DHCP_SERVER_IPV4_RANGE|range2") == False
def test_config_dhcp_server_ipv4_range_delete_nonexisting(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["del"], \
["range4"], 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_range_delete_referenced(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["del"], \
["range1"], 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_range_delete_referenced_force(self, mock_db):
runner = CliRunner()
db = clicommon.Db()
db.db = mock_db
result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["range"].commands["del"], \
["range1", "--force"], obj=db)
assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
assert mock_db.exists("CONFIG_DB", "DHCP_SERVER_IPV4_RANGE|range1") == False

View File

@ -215,9 +215,9 @@ class TestShowDHCPServer(object):
| | 10.1.1.11 | | | 10.1.1.11 |
+-------------------+------------+ +-------------------+------------+
| Vlan100|Ethernet7 | range1 | | Vlan100|Ethernet7 | range1 |
| | range2 | | | range3 |
+-------------------+------------+ +-------------------+------------+
| Vlan200|Ethernet8 | range3 | | Vlan200|Ethernet8 | range1 |
| | range4 | | | range4 |
+-------------------+------------+ +-------------------+------------+
| Ethernet9 | range5 | | Ethernet9 | range5 |
@ -237,7 +237,7 @@ class TestShowDHCPServer(object):
| Interface | Bind | | Interface | Bind |
+===================+========+ +===================+========+
| Vlan100|Ethernet7 | range1 | | Vlan100|Ethernet7 | range1 |
| | range2 | | | range3 |
+-------------------+--------+ +-------------------+--------+
""" """
runner = CliRunner() runner = CliRunner()
@ -256,7 +256,7 @@ class TestShowDHCPServer(object):
| | 10.1.1.11 | | | 10.1.1.11 |
+-------------------+------------+ +-------------------+------------+
| Vlan100|Ethernet7 | range1 | | Vlan100|Ethernet7 | range1 |
| | range2 | | | range3 |
+-------------------+------------+ +-------------------+------------+
""" """
runner = CliRunner() runner = CliRunner()
@ -271,7 +271,7 @@ class TestShowDHCPServer(object):
+-------------------+--------+ +-------------------+--------+
| Interface | Bind | | Interface | Bind |
+===================+========+ +===================+========+
| Vlan200|Ethernet8 | range3 | | Vlan200|Ethernet8 | range1 |
| | range4 | | | range4 |
+-------------------+--------+ +-------------------+--------+
""" """

View File

@ -8,7 +8,7 @@ import string
SUPPORT_TYPE = ["binary", "boolean", "ipv4-address", "string", "uint8", "uint16", "uint32"] SUPPORT_TYPE = ["binary", "boolean", "ipv4-address", "string", "uint8", "uint16", "uint32"]
def validate_str_type(type, value): def validate_str_type(type_, value):
""" """
To validate whether type is consistent with string value To validate whether type is consistent with string value
Args: Args:
@ -20,27 +20,27 @@ def validate_str_type(type, value):
""" """
if not isinstance(value, str): if not isinstance(value, str):
return False return False
if type not in SUPPORT_TYPE: if type_ not in SUPPORT_TYPE:
return False return False
if type == "string": if type_ == "string":
return True return True
if type == "binary": if type_ == "binary":
if len(value) == 0 or len(value) % 2 != 0: if len(value) == 0 or len(value) % 2 != 0:
return False return False
return all(c in set(string.hexdigits) for c in value) return all(c in set(string.hexdigits) for c in value)
if type == "boolean": if type_ == "boolean":
return value in ["true", "false"] return value in ["true", "false"]
if type == "ipv4-address": if type_ == "ipv4-address":
try: try:
if len(value.split(".")) != 4: if len(value.split(".")) != 4:
return False return False
return ipaddress.ip_address(value).version == 4 return ipaddress.ip_address(value).version == 4
except ValueError: except ValueError:
return False return False
if type.startswith("uint"): if type_.startswith("uint"):
if not value.isdigit(): if not value.isdigit():
return False return False
length = int("".join([c for c in type if c.isdigit()])) length = int("".join([c for c in type_ if c.isdigit()]))
return 0 <= int(value) <= int(pow(2, length)) - 1 return 0 <= int(value) <= int(pow(2, length)) - 1
return False return False
@ -186,6 +186,78 @@ def dhcp_server_ipv4_disable(db, dhcp_interface):
ctx.fail("Failed to disable, dhcp interface {} does not exist".format(dhcp_interface)) ctx.fail("Failed to disable, dhcp interface {} does not exist".format(dhcp_interface))
@dhcp_server_ipv4.group(cls=clicommon.AliasedGroup, name="range")
def dhcp_server_ipv4_range():
pass
def count_ipv4(start, end):
ip1 = int(ipaddress.IPv4Address(start))
ip2 = int(ipaddress.IPv4Address(end))
return ip2 - ip1 + 1
@dhcp_server_ipv4_range.command(name="add")
@click.argument("range_name", required=True)
@click.argument("ip_start", required=True)
@click.argument("ip_end", required=False)
@clicommon.pass_db
def dhcp_server_ipv4_range_add(db, range_name, ip_start, ip_end):
ctx = click.get_current_context()
if not ip_end:
ip_end = ip_start
if not validate_str_type("ipv4-address", ip_start) or not validate_str_type("ipv4-address", ip_end):
ctx.fail("ip_start or ip_end is not valid ipv4 address")
if count_ipv4(ip_start, ip_end) < 1:
ctx.fail("range value is illegal")
dbconn = db.db
key = "DHCP_SERVER_IPV4_RANGE|" + range_name
if dbconn.exists("CONFIG_DB", key):
ctx.fail("Range {} already exist".format(range_name))
else:
dbconn.hmset("CONFIG_DB", key, {"range": ip_start + "," + ip_end})
@dhcp_server_ipv4_range.command(name="update")
@click.argument("range_name", required=True)
@click.argument("ip_start", required=True)
@click.argument("ip_end", required=False)
@clicommon.pass_db
def dhcp_server_ipv4_range_update(db, range_name, ip_start, ip_end):
ctx = click.get_current_context()
if not ip_end:
ip_end = ip_start
if not validate_str_type("ipv4-address", ip_start) or not validate_str_type("ipv4-address", ip_end):
ctx.fail("ip_start or ip_end is not valid ipv4 address")
if count_ipv4(ip_start, ip_end) < 1:
ctx.fail("range value is illegal")
dbconn = db.db
key = "DHCP_SERVER_IPV4_RANGE|" + range_name
if dbconn.exists("CONFIG_DB", key):
dbconn.set("CONFIG_DB", key, "range", ip_start + "," + ip_end)
else:
ctx.fail("Range {} does not exist, cannot update".format(range_name))
@dhcp_server_ipv4_range.command(name="del")
@click.argument("range_name", required=True)
@click.option("--force", required=False, default=False, is_flag=True)
@clicommon.pass_db
def dhcp_sever_ipv4_range_del(db, range_name, force):
ctx = click.get_current_context()
dbconn = db.db
key = "DHCP_SERVER_IPV4_RANGE|" + range_name
if dbconn.exists("CONFIG_DB", key):
if not force:
for port in dbconn.keys("CONFIG_DB", "DHCP_SERVER_IPV4_PORT*"):
ranges = dbconn.get("CONFIG_DB", port, "ranges")
if ranges and range_name in ranges.split(","):
ctx.fail("Range {} is referenced in {}, cannot delete, add --force to bypass".format(range_name, port))
dbconn.delete("CONFIG_DB", key)
else:
ctx.fail("Range {} does not exist, cannot delete".format(range_name))
def register(cli): def register(cli):
# cli.add_command(dhcp_server) # cli.add_command(dhcp_server)
pass pass