[config] support for configuring muxcable to manual mode of operation

config/muxcable.py

@@ -73,43 +73,48 @@ def lookup_statedb_and_update_configdb(per_npu_statedb, config_db, port, state_c
     ipv6_value = get_value_for_key_in_config_tbl(config_db, port, "server_ipv6", "MUX_CABLE")
 
     state = get_value_for_key_in_dict(muxcable_statedb_dict, port, "state", "MUX_CABLE_TABLE")
-    if (state == "active" and configdb_state == "active") or (state == "standby" and configdb_state == "active") or (state == "unknown" and configdb_state == "active"):
-        if state_cfg_val == "active":
-            # status is already active, so right back error
+    if state in ["active", "standby", "unknown"] and configdb_state in ["active", "manual"]:
+        if state_cfg_val == configdb_state:
+            # status is already active or manual, so nothing to do
             port_status_dict[port] = 'OK'
-        if state_cfg_val == "auto":
+        elif state_cfg_val in ["auto", "manual"]:
             # display ok and write to cfgdb auto
             port_status_dict[port] = 'OK'
-            config_db.set_entry("MUX_CABLE", port, {"state": "auto",
+            config_db.set_entry("MUX_CABLE", port, {"state": state_cfg_val,
                                                     "server_ipv4": ipv4_value, "server_ipv6": ipv6_value})
-    elif state == "active" and configdb_state == "auto":
-        if state_cfg_val == "active":
+    elif state == "active" and configdb_state in ["auto", "manual"]:
+        if state_cfg_val == configdb_state:
+            # dont write anything to db, write OK to user
+            port_status_dict[port] = 'OK'
+        elif state_cfg_val in ["active", "manual"]:
             # make the state active and write back OK
-            config_db.set_entry("MUX_CABLE", port, {"state": "active",
+            config_db.set_entry("MUX_CABLE", port, {"state": state_cfg_val,
                                                     "server_ipv4": ipv4_value, "server_ipv6": ipv6_value})
             port_status_dict[port] = 'OK'
-        if state_cfg_val == "auto":
-            # dont write anything to db, write OK to user
-            port_status_dict[port] = 'OK'
 
-    elif (state == "standby" and configdb_state == "auto") or (state == "unknown" and configdb_state == "auto"):
+    elif state in ["standby", "unknown"] and configdb_state in ["auto", "manual"]:
         if state_cfg_val == "active":
             # make the state active
             config_db.set_entry("MUX_CABLE", port, {"state": "active",
                                                     "server_ipv4": ipv4_value, "server_ipv6": ipv6_value})
             port_status_dict[port] = 'INPROGRESS'
-        if state_cfg_val == "auto":
+        elif state_cfg_val == configdb_state:
             # dont write anything to db
             port_status_dict[port] = 'OK'
+        elif state_cfg_val in ["auto", "manual"]:
+            # make the state what is configured and write back OK
+            config_db.set_entry("MUX_CABLE", port, {"state": state_cfg_val,
+                                                    "server_ipv4": ipv4_value, "server_ipv6": ipv6_value})
+            port_status_dict[port] = 'OK'
 
 
 # 'muxcable' command ("config muxcable mode <port|all> active|auto")
 @muxcable.command()
-@click.argument('state', metavar='<operation_status>', required=True, type=click.Choice(["active", "auto"]))
+@click.argument('state', metavar='<operation_status>', required=True, type=click.Choice(["active", "auto", "manual"]))
 @click.argument('port', metavar='<port_name>', required=True, default=None)
 @click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL)
 def mode(state, port, json_output):
-    """Show muxcable summary information"""
+    """Config muxcable mode"""
 
     port_table_keys = {}
     y_cable_asic_table_keys = {}

tests/mock_tables/config_db.json

@@ -1540,6 +1540,11 @@
 	"server_ipv4": "10.1.1.1",
 	"server_ipv6": "fc00::75"
     },
+    "MUX_CABLE|Ethernet28": {
+        "state": "manual",
+	"server_ipv4": "10.1.1.1",
+	"server_ipv6": "fc00::75"
+    },
     "MUX_CABLE|Ethernet0": {
         "state": "active",
 	"server_ipv4": "10.2.1.1",

tests/muxcable_test.py

@@ -72,6 +72,7 @@
 Ethernet4   auto     10.3.1.1  e801::46
 Ethernet8   active   10.4.1.1  e802::46
 Ethernet12  active   10.4.1.1  e802::46
+Ethernet28  manual   10.1.1.1  fc00::75
 Ethernet32  auto     10.1.1.1  fc00::75
 """
 
@@ -108,6 +109,13 @@
                     "IPv6": "e802::46"
                 }
             },
+            "Ethernet28": {
+                "STATE": "manual",
+                "SERVER": {
+                    "IPv4": "10.1.1.1",
+                    "IPv6": "fc00::75"
+                }
+            },
             "Ethernet32": {
                 "STATE": "auto",
                 "SERVER": {
@@ -363,6 +371,16 @@ def test_config_muxcable_tabular_port_Ethernet8_auto(self):
 
         assert result.exit_code == 0
 
+    def test_config_muxcable_tabular_port_Ethernet8_manual(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('sonic_platform_base.sonic_sfp.sfputilhelper') as patched_util:
+            patched_util.SfpUtilHelper.return_value.get_asic_id_for_logical_port.return_value = 0
+            result = runner.invoke(config.config.commands["muxcable"].commands["mode"], ["manual", "Ethernet8"], obj=db)
+
+        assert result.exit_code == 0
+
     def test_config_muxcable_mode_auto_json(self):
         runner = CliRunner()
         db = Db()