[action] [PR:4266] [console] add config console escape (#320)

#### What I did
Add escape char config to console feature.
In CONFIG_DB, an optional new entry under CONSOLE_SWITCH, "console_mgmt", is added, "default_escape_char" as the key and we expect a lower case letter as value. Similarly, an optional new entry under CONSOLE_PORT, and then each line, "escape_char" as the key and we also expect a lower case letter as value.
Commands are:
1. config console default_escape (upper and lower case letters, or clear), e.g. config console default_escape k, config console default_escape B, config console default_escape clear.
2. config console escape [LINE_NUMBER] (upper and lower case letters, or clear), e.g. config console escape 6 C, config console escape 5 clear
3. show line --show-escape, e.g. show line --show-escape, show line -s -b
Connect command work as usual, except that it will add --escape <escape_char> in the underlying picocom command if it is configured.

Additionally,
1. some style change to old code to pass pre-commit check
2. unit tests for new functions
3. enhancement to old unit tests and its bug fix

#### How I did it
Implement them

#### How to verify it
Run the commands in dut.

Signed-off-by: Sonic Build Admin <sonicbld@microsoft.com>

Author: mssonicbld

config/console.py

@@ -1,7 +1,10 @@
 import click
+import string
 import utilities_common.cli as clicommon
 from .validated_config_db_connector import ValidatedConfigDBConnector
 from jsonpatch import JsonPatchConflict
+
+
 #
 # 'console' group ('config console ...')
 #
@@ -10,6 +13,7 @@ def console():
     """Console-related configuration tasks"""
     pass
 
+
 #
 # 'console enable' group ('config console enable')
 #
@@ -30,6 +34,7 @@ def enable_console_switch(db):
         ctx = click.get_current_context()
         ctx.fail("Invalid ConfigDB. Error: {}".format(e))
 
+
 #
 # 'console disable' group ('config console disable')
 #
@@ -50,6 +55,39 @@ def disable_console_switch(db):
         ctx = click.get_current_context()
         ctx.fail("Invalid ConfigDB. Error: {}".format(e))
 
+
+#
+# 'console default_escape' group ('config console default_escape A|B|...')
+#
+@console.command('default_escape')
+@clicommon.pass_db
+@click.argument('escape', metavar='<escape_char|clear>', required=True,
+                type=click.Choice(list(string.ascii_letters) + ["clear"], case_sensitive=True))
+def set_console_default_escape_char(db, escape):
+    """Set console escape character or clear the existing one"""
+    config_db = ValidatedConfigDBConnector(db.cfgdb)
+
+    table = "CONSOLE_SWITCH"
+    dataKey1 = 'console_mgmt'
+    dataKey2 = 'default_escape_char'
+
+    existing_entry = config_db.get_entry(table, dataKey1) or {}
+    if escape == "clear":
+        # Remove the default_escape_char field while preserving other keys (e.g., 'enabled')
+        if dataKey2 in existing_entry:
+            del existing_entry[dataKey2]
+        data = existing_entry
+    else:
+        existing_entry[dataKey2] = escape.lower()
+        data = existing_entry
+
+    try:
+        config_db.set_entry(table, dataKey1, data)
+    except ValueError as e:
+        ctx = click.get_current_context()
+        ctx.fail("Invalid ConfigDB. Error: {}".format(e))
+
+
 #
 # 'console add' group ('config console add ...')
 #
@@ -59,14 +97,17 @@ def disable_console_switch(db):
 @click.option('--baud', '-b', metavar='<baud>', required=True, type=click.INT)
 @click.option('--flowcontrol', '-f', metavar='<flow_control>', required=False, is_flag=True)
 @click.option('--devicename', '-d', metavar='<device_name>', required=False)
-def add_console_setting(db, linenum, baud, flowcontrol, devicename):
+@click.option('--escape', '-e', metavar='<escape_char>', required=False,
+              type=click.Choice(list(string.ascii_letters), case_sensitive=True))
+def add_console_setting(db, linenum, baud, flowcontrol, devicename, escape):
     """Add Console-realted configuration tasks"""
     config_db = ValidatedConfigDBConnector(db.cfgdb)
 
     table = "CONSOLE_PORT"
     dataKey1 = 'baud_rate'
     dataKey2 = 'flow_control'
     dataKey3 = 'remote_device'
+    dataKey4 = 'escape_char'
 
     ctx = click.get_current_context()
     data = config_db.get_entry(table, linenum)
@@ -81,6 +122,9 @@ def add_console_setting(db, linenum, baud, flowcontrol, devicename):
                 ctx.fail("Given device name {} has been used. Please enter a valid device name or remove the existing one !!".format(devicename))
             console_entry[dataKey3] = devicename
 
+        if escape:
+            console_entry[dataKey4] = escape.lower()
+
         try:
             config_db.set_entry(table, linenum, console_entry)
         except ValueError as e:
@@ -109,14 +153,15 @@ def remove_console_setting(db, linenum):
     else:
         ctx.fail("Trying to delete console port setting, which is not present.")
 
+
 #
 # 'console remote_device' group ('config console remote_device ...')
 #
 @console.command('remote_device')
 @clicommon.pass_db
 @click.argument('linenum', metavar='<line_number>', required=True, type=click.IntRange(0, 65535))
 @click.argument('devicename', metavar='<device_name>', required=False)
-def upate_console_remote_device_name(db, linenum, devicename):
+def update_console_remote_device_name(db, linenum, devicename):
     """Update remote device name for a console line"""
     config_db = ValidatedConfigDBConnector(db.cfgdb)
     ctx = click.get_current_context()
@@ -131,11 +176,12 @@ def upate_console_remote_device_name(db, linenum, devicename):
             return
         elif not devicename:
             # remove configuration key from console setting if user not give a remote device name
-            data.pop(dataKey, None)
-            try:
-                config_db.mod_entry(table, linenum, data)
-            except ValueError as e:
-                ctx.fail("Invalid ConfigDB. Error: {}".format(e))
+            if dataKey in data:
+                del data[dataKey]
+                try:
+                    config_db.set_entry(table, linenum, data)
+                except ValueError as e:
+                    ctx.fail("Invalid ConfigDB. Error: {}".format(e))
         elif isExistingSameDevice(config_db, devicename, table):
             ctx.fail("Given device name {} has been used. Please enter a valid device name or remove the existing one !!".format(devicename))
         else:
@@ -147,6 +193,7 @@ def upate_console_remote_device_name(db, linenum, devicename):
     else:
         ctx.fail("Trying to update console port setting, which is not present.")
 
+
 #
 # 'console baud' group ('config console baud ...')
 #
@@ -177,6 +224,7 @@ def update_console_baud(db, linenum, baud):
     else:
         ctx.fail("Trying to update console port setting, which is not present.")
 
+
 #
 # 'console flow_control' group ('config console flow_control ...')
 #
@@ -208,6 +256,39 @@ def update_console_flow_control(db, mode, linenum):
     else:
         ctx.fail("Trying to update console port setting, which is not present.")
 
+
+#
+# 'console escape' group ('config console escape ...')
+#
+@console.command('escape')
+@clicommon.pass_db
+@click.argument('linenum', metavar='<line_number>', required=True, type=click.IntRange(0, 65535))
+@click.argument('escape', metavar='<escape_char|clear>', required=True,
+                type=click.Choice(list(string.ascii_letters) + ["clear"], case_sensitive=True))
+def update_console_escape_char(db, linenum, escape):
+    """Update escape character for a console line"""
+    config_db = ValidatedConfigDBConnector(db.cfgdb)
+    ctx = click.get_current_context()
+
+    table = "CONSOLE_PORT"
+    dataKey = 'escape_char'
+
+    data = config_db.get_entry(table, linenum)
+    if data:
+        if escape == "clear":
+            if dataKey in data:
+                del data[dataKey]
+        else:
+            data[dataKey] = escape.lower()
+
+        try:
+            config_db.set_entry(table, linenum, data)
+        except ValueError as e:
+            ctx.fail("Invalid ConfigDB. Error: {}".format(e))
+    else:
+        ctx.fail("Trying to update console port setting, which is not present.")
+
+
 def isExistingSameDevice(config_db, deviceName, table):
     """Check if the given device name is conflict with existing device"""
     settings = config_db.get_table(table)

consutil/lib.py

@@ -32,6 +32,8 @@
 FLOW_KEY = "flow_control"
 FEATURE_KEY = "console_mgmt"
 FEATURE_ENABLED_KEY = "enabled"
+DEFAULT_FEATURE_ESCAPE_KEY = "default_escape_char"
+FEATURE_ESCAPE_KEY = "escape_char"
 
 # STATE_DB Keys
 STATE_KEY = "state"
@@ -65,7 +67,7 @@ def __init__(self, db, configured_only, refresh=False):
     def get_all(self):
         """Gets all console ports information"""
         for port in self._ports:
-            yield ConsolePortInfo(self._db_utils, port)
+            yield ConsolePortInfo(self._db_utils, port, self._default_escape_char)
 
     def get(self, target, use_device=False):
         """Gets information of a ports, the target is the line number by default"""
@@ -77,14 +79,29 @@ def get(self, target, use_device=False):
         # identify the line number by searching configuration
         for port in self._ports:
             if search_key in port and port[search_key] == target:
-                return ConsolePortInfo(self._db_utils, port)
+                return ConsolePortInfo(self._db_utils, port, self._default_escape_char)
 
         raise LineNotFoundError
 
     def _init_all(self, refresh):
         config_db = self._db.cfgdb
         state_db = self._db.db
 
+        # Querying CONFIG_DB to get console management feature state
+        feature_state = config_db.get_entry(CONSOLE_SWITCH_TABLE, FEATURE_KEY)
+        # Default to no escape character when console management feature is disabled or missing.
+        self._default_escape_char = None
+        if feature_state and feature_state.get(FEATURE_ENABLED_KEY, "no") == "yes":
+            self._default_escape_char = feature_state.get(
+                DEFAULT_FEATURE_ESCAPE_KEY,
+                feature_state.get(DEFAULT_FEATURE_ESCAPE_KEY, None),
+            )
+            if self._default_escape_char is not None and not self._default_escape_char.islower():
+                raise InvalidConfigurationError(
+                    DEFAULT_FEATURE_ESCAPE_KEY,
+                    "default console escape character is not valid",
+                )
+
         # Querying CONFIG_DB to get configured console ports
         keys = config_db.get_keys(CONSOLE_PORT_TABLE)
         ports = []
@@ -114,11 +131,12 @@ def _init_all(self, refresh):
         self._ports = ports
 
 class ConsolePortInfo(object):
-    def __init__(self, db_utils, info):
+    def __init__(self, db_utils, info, default_escape_char=None):
         self._db_utils = db_utils
         self._info = info
         self._session = None
-    
+        self._default_escape_char = default_escape_char
+
     def __str__(self):
         return "({}, {}, {})".format(self.line_num, self.baud, self.remote_device)
 
@@ -137,7 +155,19 @@ def flow_control(self):
     @property
     def remote_device(self):
         return self._info[DEVICE_KEY] if DEVICE_KEY in self._info else None
-    
+
+    @property
+    def default_escape_char(self):
+        return self._default_escape_char
+
+    @property
+    def line_escape_char(self):
+        return self._info.get(FEATURE_ESCAPE_KEY, None)
+
+    @property
+    def escape_char(self):
+        return self._info.get(FEATURE_ESCAPE_KEY, self._default_escape_char)
+
     @property
     def busy(self):
         return STATE_KEY in self.cur_state and self.cur_state[STATE_KEY] == BUSY_FLAG
@@ -170,7 +200,9 @@ def connect(self):
 
         # build and start picocom command
         flow_cmd = "h" if self.flow_control else "n"
-        cmd = "picocom -b {} -f {} {}{}".format(self.baud, flow_cmd, SysInfoProvider.DEVICE_PREFIX, self.line_num)
+        escape_cmd = "-e {}".format(self.escape_char) if self.escape_char else ""
+        cmd = "picocom {} -b {} -f {} {}{}".format(escape_cmd, self.baud, flow_cmd,
+                                                   SysInfoProvider.DEVICE_PREFIX, self.line_num)
 
         # start connection
         try:
@@ -207,7 +239,7 @@ def clear_session(self):
         finally:
             self.refresh()
             self._session = None
-        
+
         return True
 
     def refresh(self):