[dhcp_server] add config dhcp server option

dockers/docker-dhcp-server/cli-plugin-tests/mock_config_db.json

@@ -33,14 +33,18 @@
         "lease_time": "3600",
         "mode": "PORT",
         "netmask": "255.255.255.0",
-        "customized_options": "option60",
         "state": "disabled"
     },
     "DHCP_SERVER_IPV4_CUSTOMIZED_OPTIONS|option60": {
         "id": "60",
         "type": "string",
         "value": "dummy_value"
     },
+    "DHCP_SERVER_IPV4_CUSTOMIZED_OPTIONS|option61": {
+        "id": "61",
+        "type": "string",
+        "value": "dummy_value"
+    },
     "DHCP_SERVER_IPV4_RANGE|range1": {
         "range": "100.1.1.3,100.1.1.5"
     },

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

@@ -606,3 +606,137 @@ def test_config_dhcp_server_ipv4_unbind_unbind_ip(self, mock_db):
                 ["Vlan100", "Ethernet4", "100.1.1.13,100.1.1.14"], 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_option_add(self, mock_db):
+        expected_value = {
+            "option_id": "62",
+            "type": "string",
+            "value": "dummy_value"
+        }
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["add"], \
+                ["option62", "62", "string", "dummy_value"], 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_CUSTOMIZED_OPTIONS|option62") == expected_value
+
+    def test_config_dhcp_server_ipv4_option_add_existing(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["add"], \
+                ["option60", "60", "string", "dummy_value"], 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_option_add_illegal_argument(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["add"], \
+                ["option62", "-5", "string", "dummy_value"], 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_option_del(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["del"], \
+                ["option61"], 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_CUSTOMIZED_OPTIONS|option61") == False
+
+    def test_config_dhcp_server_ipv4_option_del_nonexisting(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["del"], \
+                ["option62"], 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_option_del_referenced(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["del"], \
+                ["option60"], 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_option_bind(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["bind"], \
+                ["Vlan300", "option60"], 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|Vlan300", "customized_options") == "option60"
+
+    def test_config_dhcp_server_ipv4_option_bind_multiple_options(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["bind"], \
+                ["Vlan300", "option60,option61"], obj=db)
+        assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
+        result = mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan300", "customized_options")
+        assert result and set(result.split(",")) == set("option60,option61".split(","))
+
+    def test_config_dhcp_server_ipv4_option_bind_to_existing(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["bind"], \
+                ["Vlan100", "option61"], obj=db)
+        assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
+        result = mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "customized_options")
+        assert result and set(result.split(",")) == set("option60,option61".split(","))
+
+    def test_config_dhcp_server_ipv4_option_bind_same_option_to_existing(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["bind"], \
+                ["Vlan100", "option60"], 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", "customized_options") == "option60"
+
+    def test_config_dhcp_server_ipv4_option_bind_to_nonexisting_intf(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["bind"], \
+                ["Vlan200", "option60"], 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_option_bind_nonexisting_option(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["bind"], \
+                ["Vlan300", "option62"], 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_option_unbind(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["unbind"], \
+                ["Vlan100", "option60"], obj=db)
+        assert result.exit_code == 0, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)
+        result = mock_db.get("CONFIG_DB", "DHCP_SERVER_IPV4|Vlan100", "customized_options")
+        assert result == None or result == ""
+
+    def test_config_dhcp_server_ipv4_option_unbind_nonexisting_intf(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["unbind"], \
+                ["Vlan200", "option60"], 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_option_unbind_nonexisting_option(self, mock_db):
+        runner = CliRunner()
+        db = clicommon.Db()
+        db.db = mock_db
+        result = runner.invoke(dhcp_server.dhcp_server.commands["ipv4"].commands["option"].commands["unbind"], \
+                ["Vlan100", "option61"], obj=db)
+        assert result.exit_code == 2, "exit code: {}, Exception: {}, Traceback: {}".format(result.exit_code, result.exception, result.exc_info)

dockers/docker-dhcp-server/cli-plugin-tests/test_show_dhcp_server.py

@@ -187,6 +187,8 @@ def test_show_dhcp_server_ipv4_option_without_name(self, mock_db):
 +===============+=============+=============+========+
 | option60      |          60 | dummy_value | string |
 +---------------+-------------+-------------+--------+
+| option61      |          61 | dummy_value | string |
++---------------+-------------+-------------+--------+
 """
         runner = CliRunner()
         db = clicommon.Db()

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

@@ -46,6 +46,7 @@ def validate_str_type(type_, value):
 
 
 @click.group(cls=clicommon.AbbreviationGroup, name="dhcp_server")
+@clicommon.pass_db
 def dhcp_server():
     """config DHCP Server information"""
     ctx = click.get_current_context()
@@ -341,10 +342,98 @@ def dhcp_server_ipv4_ip_unbind(db, dhcp_interface, member_interface, range_, ip_
                 ctx.fail("Attempting to unbind range or ip that is not binded")
 
 
-def register(cli):
-    # cli.add_command(dhcp_server)
+@dhcp_server_ipv4.group(cls=clicommon.AliasedGroup, name="option")
+def dhcp_server_ipv4_option():
     pass
 
 
+@dhcp_server_ipv4_option.command(name="add")
+@click.argument("option_name", required=True)
+@click.argument("option_id", required=True)
+@click.argument("type_", required=True)
+@click.argument("value", required=True)
+@clicommon.pass_db
+def dhcp_server_ipv4_option_add(db, option_name, option_id, type_, value):
+    ctx = click.get_current_context()
+    if not validate_str_type("uint32", option_id):
+        ctx.fail("option_id must be uint32")
+    if type_ != "string":
+        ctx.fail("Currently only string type is supported")
+    dbconn = db.db
+    key = "DHCP_SERVER_IPV4_CUSTOMIZED_OPTIONS|" + option_name
+    if dbconn.exists("CONFIG_DB", key):
+        ctx.fail("Option {} already exist".format(option_name))
+    dbconn.hmset("CONFIG_DB", key, {
+        "option_id": option_id,
+        "type": type_,
+        "value": value,
+        })
+
+
+@dhcp_server_ipv4_option.command(name="del")
+@click.argument("option_name", required=True)
+@clicommon.pass_db
+def dhcp_server_ipv4_option_del(db, option_name):
+    ctx = click.get_current_context()
+    dbconn = db.db
+    option_key = "DHCP_SERVER_IPV4_CUSTOMIZED_OPTIONS|" + option_name
+    if not dbconn.exists("CONFIG_DB", option_key):
+        ctx.fail("Option {} does not exist, cannot delete".format(option_name))
+    for key in dbconn.keys("CONFIG_DB", "DHCP_SERVER_IPV4|*"):
+        existing_options = dbconn.get("CONFIG_DB", key, "customized_options")
+        if existing_options and option_name in existing_options.split(","):
+            ctx.fail("Option {} is referenced in {}, cannot delete".format(option_name, key[len("DHCP_SERVER_IPV4|"):]))
+    dbconn.delete("CONFIG_DB", option_key)
+
+
+@dhcp_server_ipv4_option.command(name="bind")
+@click.argument("dhcp_interface", required=True)
+@click.argument("option_list", required=True)
+@clicommon.pass_db
+def dhcp_server_ipv4_option_bind(db, dhcp_interface, option_list):
+    ctx = click.get_current_context()
+    dbconn = db.db
+    key = "DHCP_SERVER_IPV4|" + dhcp_interface
+    if not dbconn.exists("CONFIG_DB", key):
+        ctx.fail("Interface {} is not valid dhcp interface".format(dhcp_interface))
+    option_list = option_list.split(",")
+    for option_name in option_list:
+        option_key = "DHCP_SERVER_IPV4_CUSTOMIZED_OPTIONS|" + option_name
+        if not dbconn.exists("CONFIG_DB", option_key):
+            ctx.fail("Option {} does not exist, cannot bind".format(option_name))
+    existing_value = dbconn.get("CONFIG_DB", key, "customized_options")
+    value_set = set(existing_value.split(",")) if existing_value else set()
+    new_value_set = value_set.union(option_list)
+    dbconn.set("CONFIG_DB", key, "customized_options", ",".join(new_value_set))
+
+
+@dhcp_server_ipv4_option.command(name="unbind")
+@click.argument("dhcp_interface", required=True)
+@click.argument("option_list", required=False)
+@click.option("--all", "all_", required=False, default=False, is_flag=True)
+@clicommon.pass_db
+def dhcp_server_ipv4_option_unbind(db, dhcp_interface, option_list, all_):
+    ctx = click.get_current_context()
+    dbconn = db.db
+    key = "DHCP_SERVER_IPV4|" + dhcp_interface
+    if not dbconn.exists("CONFIG_DB", key):
+        ctx.fail("Interface {} is not valid dhcp interface".format(dhcp_interface))
+    if all_:
+        dbconn.set("CONFIG_DB", key, "customized_options", "")
+    else:
+        unbind_value = set(option_list.split(","))
+        existing_value = dbconn.get("CONFIG_DB", key, "customized_options")
+        value_set = set(existing_value.split(",")) if existing_value else set()
+        if value_set.issuperset(unbind_value):
+            new_value_set = value_set.difference(unbind_value)
+            dbconn.set("CONFIG_DB", key, "customized_options", ",".join(new_value_set))
+        else:
+            ctx.fail("Attempting to unbind option that is not binded")
+
+
+def register(cli):
+    cli.add_command(dhcp_server)
+
+
 if __name__ == '__main__':
     dhcp_server()

dockers/docker-dhcp-server/cli/show/plugins/show_dhcp_server.py

@@ -13,7 +13,7 @@ def ts_to_str(ts):
     return datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M:%S")
 
 
-@click.group(cls=clicommon.AliasedGroup)
+@click.group(cls=clicommon.AbbreviationGroup, name="dhcp_server")
 @clicommon.pass_db
 def dhcp_server(db):
     """Show dhcp_server related info"""