Fixing CLI plugin tests

Author: cshivashgit

dockers/docker-dhcp-relay/cli-plugin-tests/test_config_dhcp_relay.py

@@ -11,6 +11,9 @@
 Added DHCP relay address [{}] to Vlan1000
 Restarting DHCP relay service...
 """
+config_dhcp_relay_update_output = """\
+Updated DHCPv4 Servers as {} to Vlan1000
+"""
 config_dhcp_relay_del_output = """\
 Removed DHCP relay address [{}] from Vlan1000
 Restarting DHCP relay service...
@@ -104,8 +107,8 @@
 }
 
 
-@pytest.fixture(scope="module", params=["ipv4", "ipv6"])
-def ip_version(request):
+@pytest.fixture(scope="module", params=["ipv4", "ipv4_dhcp", "ipv6"])
+def version(request):
     """
     Parametrize Ip version
 
@@ -166,111 +169,131 @@ def test_config_dhcp_relay_del_with_nonexist_vlanid_ipv6(self):
             assert "Error: Vlan1001 doesn't exist" in result.output
             assert mock_run_command.call_count == 0
 
-    def test_config_add_del_dhcp_relay_with_invalid_ip(self, ip_version, op):
+    def test_config_add_del_dhcp_relay_with_invalid_ip(self, mock_cfgdb, version, op):
         runner = CliRunner()
-        invalid_ip = IP_VER_TEST_PARAM_MAP[ip_version]["invalid_ip"]
-
+        db = Db()
+        db.cfgdb = mock_cfgdb
+        invalid_ip = IP_VER_TEST_PARAM_MAP[version]["invalid_ip"]
+        ip_version = "ipv6" if version == "ipv6" else "ipv4"
+        if version == "ipv4_dhcp":
+            db.cfgdb.set_entry('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'})
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
                                    .commands[op], ["1000", invalid_ip])
             print(result.exit_code)
             print(result.output)
             assert result.exit_code != 0
             assert "Error: {} is invalid IP address".format(invalid_ip) in result.output
             assert mock_run_command.call_count == 0
 
-    def test_config_add_dhcp_with_exist_ip(self, mock_cfgdb, ip_version):
+    def test_config_add_dhcp_with_exist_ip(self, mock_cfgdb, version):
         runner = CliRunner()
         db = Db()
         db.cfgdb = mock_cfgdb
-        exist_ip = IP_VER_TEST_PARAM_MAP[ip_version]["exist_ip"]
-
+        exist_ip = IP_VER_TEST_PARAM_MAP[version]["exist_ip"]
+        ip_version = "ipv6" if version == "ipv6" else "ipv4"
+        if version == "ipv4_dhcp":
+            db.cfgdb.set_entry('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'})
+        table = db.cfgdb.get_entry("FEATURE", "dhcp_relay")
+        if('has_sonic_dhcpv4_relay' in table and table['has_sonic_dhcpv4_relay'] == 'True'):
+            print("has_sonic_dhcpv4_relay is set")
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
                                    .commands["add"], ["1000", exist_ip], obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code != 0
-            if ip_version == "ipv4":
-                assert "{} is already a DHCPv4 relay for Vlan1000".format(exist_ip) in result.output
+            if version == "ipv4_dhcp":
+                assert "DHCPv4 relay entry for Vlan1000 already exists. Use 'update' instead." in result.output
             else:
                 assert "{} is already a DHCP relay for Vlan1000".format(exist_ip) in result.output
             assert mock_run_command.call_count == 0
         db.cfgdb.set_entry.reset_mock()
 
-    def test_config_del_nonexist_dhcp_relay(self, mock_cfgdb, ip_version):
+    def test_config_del_nonexist_dhcp_relay(self, mock_cfgdb, version):
         runner = CliRunner()
         db = Db()
         db.cfgdb = mock_cfgdb
-        nonexist_ip = IP_VER_TEST_PARAM_MAP[ip_version]["nonexist_ip"]
+        nonexist_ip = IP_VER_TEST_PARAM_MAP[version]["nonexist_ip"]
+        ip_version = "ipv6" if version == "ipv6" else "ipv4"
+        if version == "ipv4_dhcp":
+            db.cfgdb.set_entry('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'})
 
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
                                    .commands["del"], ["1000", nonexist_ip], obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code != 0
-            if ip_version == "ipv4":
+            if version == "ipv4_dhcp":
                 assert "Error: {} is not a DHCPv4 relay for Vlan1000".format(nonexist_ip) in result.output
             else:
                 assert "Error: {} is not a DHCP relay for Vlan1000".format(nonexist_ip) in result.output
 
             assert mock_run_command.call_count == 0
         db.cfgdb.set_entry.reset_mock()
 
-    def test_config_add_del_dhcp_relay(self, mock_cfgdb, ip_version):
+    def test_config_add_del_dhcp_relay(self, mock_cfgdb, version):
         runner = CliRunner()
         db = Db()
         db.cfgdb = mock_cfgdb
-        test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0]
-        config_db_table = IP_VER_TEST_PARAM_MAP[ip_version]["table"]
+        test_ip = IP_VER_TEST_PARAM_MAP[version]["ips"][0]
+        ip_version = "ipv6" if version == "ipv6" else "ipv4"
+        config_db_table = IP_VER_TEST_PARAM_MAP[version]["table"]
 
-        if ip_version == "ipv4" :
+        if version == "ipv4_dhcp":
+            db.cfgdb.set_entry('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'})
             dhcp4_config_db_table = IP_VER_TEST_PARAM_MAP["ipv4_dhcp"]["table"]
+            op = "update"
+        else:
+            op = "add"
 
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             # add new dhcp relay
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
-                                   .commands["add"], ["1000", test_ip], obj=db)
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
+                                   .commands[op], ["1000", test_ip], obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code == 0
-            assert result.output == config_dhcp_relay_add_output.format(test_ip)
-            if ip_version != "ipv4":
+            if version != "ipv4_dhcp":
+                assert result.output == config_dhcp_relay_add_output.format(test_ip)
                 assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \
-                    == expected_dhcp_relay_add_config_db_output[ip_version]
-            assert mock_run_command.call_count == 3
+                    == expected_dhcp_relay_add_config_db_output[version]
+                assert mock_run_command.call_count == 3
 
-            if ip_version == "ipv4" :
+            if version == "ipv4_dhcp" :
+                assert result.output == config_dhcp_relay_update_output.format(test_ip)
                 expected_calls = [
+                    mock.call('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'}),
                     mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_add_config_db_output["ipv4_dhcp"])
                 ]
 
                 assert db.cfgdb.set_entry.call_args_list == expected_calls
             else:
                 db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000",
-                                                       expected_dhcp_relay_add_config_db_output[ip_version])
+                                                       expected_dhcp_relay_add_config_db_output[version])
 
         db.cfgdb.set_entry.reset_mock()
         # del dhcp relay
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
                                    .commands["del"], ["1000", test_ip], obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code == 0
-            assert result.output == config_dhcp_relay_del_output.format(test_ip)
-            assert mock_run_command.call_count == 3
-            if ip_version != "ipv4":
+            if version != "ipv4_dhcp":
+                assert result.output == config_dhcp_relay_del_output.format(test_ip)
                 assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \
-                    == expected_dhcp_relay_del_config_db_output[ip_version]
+                    == expected_dhcp_relay_del_config_db_output[version]
+                assert mock_run_command.call_count == 3
 
-            if ip_version == "ipv4" :
+            if version == "ipv4_dhcp" :
+                assert "Removed DHCP relay address [{}] from Vlan1000".format(test_ip) in result.output
                 expected_calls = [
                     mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_del_config_db_output["ipv4_dhcp"])
                 ]
@@ -279,7 +302,7 @@ def test_config_add_del_dhcp_relay(self, mock_cfgdb, ip_version):
 
             else:
                 db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000",
-                                                           expected_dhcp_relay_del_config_db_output[ip_version])
+                                                           expected_dhcp_relay_del_config_db_output[version])
 
     def test_config_add_del_dhcp_relay_with_enable_dhcp_server(self, mock_cfgdb):
         runner = CliRunner()
@@ -311,81 +334,96 @@ def test_config_add_del_dhcp_relay_with_enable_dhcp_server(self, mock_cfgdb):
             assert result.exit_code == 0
             assert "Cannot change ipv4 dhcp_relay configuration when dhcp_server feature is enabled" in result.output
 
-    def test_config_add_del_multiple_dhcp_relay(self, mock_cfgdb, ip_version):
+    def test_config_add_del_multiple_dhcp_relay(self, mock_cfgdb, version):
         runner = CliRunner()
         db = Db()
         db.cfgdb = mock_cfgdb
-        test_ips = IP_VER_TEST_PARAM_MAP[ip_version]["ips"]
-        config_db_table = IP_VER_TEST_PARAM_MAP[ip_version]["table"]
+        test_ips = IP_VER_TEST_PARAM_MAP[version]["ips"]
+        config_db_table = IP_VER_TEST_PARAM_MAP[version]["table"]
+        ip_version = "ipv6" if version == "ipv6" else "ipv4"
 
-        if ip_version == "ipv4" :
+        if version == "ipv4_dhcp" :
+            db.cfgdb.set_entry('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'})
             dhcp4_config_db_table = IP_VER_TEST_PARAM_MAP["ipv4_dhcp"]["table"]
+            op = "update"
+        else:
+            op = "add"
 
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             # add new dhcp relay
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
-                                   .commands["add"], ["1000"] + test_ips, obj=db)
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
+                                   .commands[op], ["1000"] + test_ips, obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code == 0
-            assert result.output == config_dhcp_relay_add_output.format(",".join(test_ips))
-            if ip_version != "ipv4":
+            if version != "ipv4_dhcp":
+                assert result.output == config_dhcp_relay_add_output.format(",".join(test_ips))
                 assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \
-                    == expected_dhcp_relay_add_multi_config_db_output[ip_version]
-            assert mock_run_command.call_count == 3
+                    == expected_dhcp_relay_add_multi_config_db_output[version]
+                assert mock_run_command.call_count == 3
 
-            if ip_version == "ipv4" :
+            if version == "ipv4_dhcp" :
+                assert result.output == config_dhcp_relay_update_output.format(",".join(test_ips))
                 expected_calls = [
+                    mock.call('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'}),
                     mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_add_multi_config_db_output["ipv4_dhcp"]),
                 ]
 
                 assert db.cfgdb.set_entry.call_args_list == expected_calls
             else:
                 db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000",
-                                                       expected_dhcp_relay_add_multi_config_db_output[ip_version])
+                                                       expected_dhcp_relay_add_multi_config_db_output[version])
 
         db.cfgdb.set_entry.reset_mock()
         # del dhcp relay
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
                                    .commands["del"], ["1000"] + test_ips, obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code == 0
-            assert result.output == config_dhcp_relay_del_output.format(",".join(test_ips))
-            assert mock_run_command.call_count == 3
-            if ip_version != "ipv4":
+            if version != "ipv4_dhcp":
+                assert result.output == config_dhcp_relay_del_output.format(",".join(test_ips))
+                assert mock_run_command.call_count == 3
                 assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \
-                    == expected_dhcp_relay_del_config_db_output[ip_version]
-            if ip_version == "ipv4":
+                    == expected_dhcp_relay_del_config_db_output[version]
+            if ip_version == "ipv4_dhcp":
+                assert "Removed DHCP relay address [{}] from Vlan1000".format(",".join(test_ip)) in result.output
                 expected_calls = [
                     mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_del_config_db_output["ipv4_dhcp"]),
                 ]
 
                 assert db.cfgdb.set_entry.call_args_list == expected_calls
             else:
                 db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000",
-                                                           expected_dhcp_relay_del_config_db_output[ip_version])
+                                                           expected_dhcp_relay_del_config_db_output[version])
 
         db.cfgdb.set_entry.reset_mock()
 
-    def test_config_add_del_duplicate_dhcp_relay(self, mock_cfgdb, ip_version, op):
+    def test_config_add_del_duplicate_dhcp_relay(self, mock_cfgdb, version, op):
         runner = CliRunner()
         db = Db()
         db.cfgdb = mock_cfgdb
-        test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] if op == "add" \
-            else IP_VER_TEST_PARAM_MAP[ip_version]["exist_ip"]
+        test_ip = IP_VER_TEST_PARAM_MAP[version]["ips"][0] if op == "add" \
+            else IP_VER_TEST_PARAM_MAP[version]["exist_ip"]
+        ip_version = "ipv6" if version == "ipv6" else "ipv4"
+
+        if version == "ipv4_dhcp" :
+            db.cfgdb.set_entry('FEATURE', 'dhcp_relay', {'has_sonic_dhcpv4_relay': 'True'})
 
         with mock.patch("utilities_common.cli.run_command") as mock_run_command:
             result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version]
-                                   .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]]
+                                   .commands[IP_VER_TEST_PARAM_MAP[version]["command"]]
                                    .commands[op], ["1000", test_ip, test_ip], obj=db)
             print(result.exit_code)
             print(result.output)
             assert result.exit_code != 0
-            assert "Error: Find duplicate DHCP relay ip {} in {} list".format(test_ip, op) in result.output
+            if version == "ipv4_dhcp" and op == "add":
+                assert "Error: DHCPv4 relay entry for Vlan1000 already exists. Use 'update' instead." in result.output
+            else:
+                assert "Error: Find duplicate DHCP relay ip {} in {} list".format(test_ip, op) in result.output
             assert mock_run_command.call_count == 0
 
     def test_config_add_dhcp_relay_ipv6_with_non_entry(self, mock_cfgdb):