[console] improve picocom launch efficiency (#376)

replace pexpect calls with direct function call and exec while preserving the original logic

<!--
 Please make sure you've read and understood our contributing guidelines:
 https://github.com/Azure/SONiC/blob/gh-pages/CONTRIBUTING.md

 CODE_OF_CONDUCT.md LICENSE README.md SECURITY.md SUPPORT.md azure-pipelines failure_prs.log scripts skip_prs.log Make sure all your commits include a signature generated with `git commit -s` **

 If this is a bug fix, make sure your description includes "closes #xxxx",
 "fixes #xxxx" or "resolves #xxxx" so that GitHub automatically closes the related
 issue when the PR is merged.

 If you are adding/modifying/removing any command or utility script, please also
 make sure to add/modify/remove any unit tests from the tests
 directory as appropriate.

 If you are modifying or removing an existing 'show', 'config' or 'sonic-clear'
 subcommand, or you are adding a new subcommand, please make sure you also
 update the Command Line Reference Guide (doc/Command-Reference.md) to reflect
 your changes.

 Please provide the following information:
-->

#### What I did
streamlined the picocom launch process

#### How I did it
replaced the first pexpect call with a direct function call
wrapped the picocom launch in a bash script
replaced the second pexpect call with execvp

#### How to verify it
first, launch console app just like before
then, use ps -ef to make sure "connect line" is no longer a standalone process

#### Previous command output (if the output of a command-line utility has changed)
connect line X or just reverse ssh

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

Author: mssonicbld

connect/main.py

@@ -1,6 +1,5 @@
 import configparser
 import os
-import pexpect
 import sys
 
 import click
@@ -66,15 +65,6 @@ def get_command(self, ctx, cmd_name):
             return click.Group.get_command(self, ctx, matches[0])
         ctx.fail('Too many matches: %s' % ', '.join(sorted(matches)))
 
-def run_command(command, display_cmd=False):
-    if display_cmd:
-        click.echo(click.style("Command: ", fg='cyan') + click.style(command, fg='green'))
-
-    proc = pexpect.spawn(command)
-    proc.interact()
-    proc.close()
-    return proc.exitstatus
-
 CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help', '-?'])
 
 #
@@ -95,8 +85,8 @@ def connect():
 @click.option('--devicename', '-d', is_flag=True, help="connect by name - if flag is set, interpret target as device name instead")
 def line(target, devicename):
     """Connect to line LINENUM via serial connection"""
-    cmd = "consutil connect {}".format("--devicename " if devicename else "") + str(target)
-    sys.exit(run_command(cmd))
+    from consutil.lib import console_connect
+    console_connect(target, use_device=devicename)
 
 #
 # 'device' command ("connect device")
@@ -105,8 +95,8 @@ def line(target, devicename):
 @click.argument('devicename')
 def device(devicename):
     """Connect to device DEVICENAME via serial connection"""
-    cmd = "consutil connect -d " + devicename
-    sys.exit(run_command(cmd))
+    from consutil.lib import console_connect
+    console_connect(devicename, use_device=True)
 
 if __name__ == '__main__':
     connect()

consutil/lib.py

@@ -384,6 +384,50 @@ def update_state(self, line_num, state, pid="", date=""):
             START_TIME_KEY: date
         }
 
+
+def initialize_console_runtime(db):
+    """Validate console feature state and initialize device prefix."""
+    config_db = db.cfgdb
+    data = config_db.get_entry(CONSOLE_SWITCH_TABLE, FEATURE_KEY)
+    if FEATURE_ENABLED_KEY not in data or data[FEATURE_ENABLED_KEY] == "no":
+        click.echo("Console switch feature is disabled")
+        sys.exit(ERR_DISABLE)
+
+    SysInfoProvider.init_device_prefix()
+
+
+def console_connect(target, use_device=False, db=None):
+    """Connect to a console port. Can be called directly without Click context."""
+    if db is None:
+        from utilities_common.db import Db
+        db = Db()
+        initialize_console_runtime(db)
+
+    port_provider = ConsolePortProvider(db, configured_only=False)
+    try:
+        target_port = port_provider.get(target, use_device=use_device)
+    except LineNotFoundError:
+        click.echo("Cannot connect: target [{}] does not exist".format(target))
+        sys.exit(ERR_DEV)
+
+    line_num = target_port.line_num
+
+    try:
+        session = target_port.connect()
+    except LineBusyError:
+        click.echo("Cannot connect: line [{}] is busy".format(line_num))
+        sys.exit(ERR_BUSY)
+    except InvalidConfigurationError as cfg_err:
+        click.echo("Cannot connect: {}".format(cfg_err.message))
+        sys.exit(ERR_CFG)
+    except ConnectionFailedError:
+        click.echo("Cannot connect: unable to open picocom process")
+        sys.exit(ERR_DEV)
+
+    click.echo("Successful connection to line [{}]\nPress ^{} ^X to disconnect"
+               .format(line_num, target_port.escape_char.upper() if target_port.escape_char is not None else "A"))
+    session.interact()
+
 class InvalidConfigurationError(Exception):
     def __init__(self, config_key, message):
         self.config_key = config_key

consutil/main.py

@@ -14,6 +14,7 @@
 
     from tabulate import tabulate
     from .lib import *
+    from .lib import initialize_console_runtime, console_connect
 except ImportError as e:
     raise ImportError("%s - required module not found" % str(e))
 
@@ -22,13 +23,7 @@
 @clicommon.pass_db
 def consutil(db):
     """consutil - Command-line utility for interacting with switches via console device"""
-    config_db = db.cfgdb
-    data = config_db.get_entry(CONSOLE_SWITCH_TABLE, FEATURE_KEY)
-    if FEATURE_ENABLED_KEY not in data or data[FEATURE_ENABLED_KEY] == "no":
-        click.echo("Console switch feature is disabled")
-        sys.exit(ERR_DISABLE)
-
-    SysInfoProvider.init_device_prefix()
+    initialize_console_runtime(db)
 
 
 # 'show' subcommand
@@ -116,7 +111,6 @@ def clear(db, target, devicename):
     else:
         click.echo("Cleared line")
 
-
 # 'connect' subcommand
 @consutil.command()
 @clicommon.pass_db
@@ -125,33 +119,7 @@ def clear(db, target, devicename):
               help="connect by name - if flag is set, interpret target as device name instead")
 def connect(db, target, devicename):
     """Connect to switch via console device - TARGET is line number or device name of switch"""
-    # identify the target line
-    port_provider = ConsolePortProvider(db, configured_only=False)
-    try:
-        target_port = port_provider.get(target, use_device=devicename)
-    except LineNotFoundError:
-        click.echo("Cannot connect: target [{}] does not exist".format(target))
-        sys.exit(ERR_DEV)
-
-    line_num = target_port.line_num
-
-    # connect
-    try:
-        session = target_port.connect()
-    except LineBusyError:
-        click.echo("Cannot connect: line [{}] is busy".format(line_num))
-        sys.exit(ERR_BUSY)
-    except InvalidConfigurationError as cfg_err:
-        click.echo("Cannot connect: {}".format(cfg_err.message))
-        sys.exit(ERR_CFG)
-    except ConnectionFailedError:
-        click.echo("Cannot connect: unable to open picocom process")
-        sys.exit(ERR_DEV)
-
-    # interact
-    click.echo("Successful connection to line [{}]\nPress ^{} ^X to disconnect"
-               .format(line_num, target_port.escape_char.upper() if target_port.escape_char is not None else "A"))
-    session.interact()
+    console_connect(target, use_device=devicename, db=db)
 
 
 if __name__ == '__main__':

tests/console_test.py

@@ -15,7 +15,7 @@
 from click.testing import CliRunner
 from utilities_common.db import Db
 from consutil.lib import ConsolePortProvider, ConsolePortInfo, ConsoleSession, SysInfoProvider, DbUtils, \
-    InvalidConfigurationError, LineBusyError, LineNotFoundError, ConnectionFailedError
+    InvalidConfigurationError, LineBusyError, LineNotFoundError, ConnectionFailedError, console_connect
 from sonic_py_common import device_info
 from jsonpatch import JsonPatchConflict
 
@@ -1368,6 +1368,19 @@ def test_connect_default_escape_after_clear(self):
         assert result.exit_code == 0
         assert result.output == "Successful connection to line [1]\nPress ^A ^X to disconnect\n"
 
+    @mock.patch('consutil.lib.SysInfoProvider.list_console_ttys', mock.MagicMock(return_value=["/dev/ttyUSB1"]))
+    @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None))
+    @mock.patch('consutil.lib.ConsolePortInfo.connect',
+                mock.MagicMock(return_value=mock.MagicMock(interact=mock.MagicMock(return_value=None))))
+    def test_console_connect_without_db(self):
+        prepared_db = Db()
+        prepared_db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes"})
+        prepared_db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"})
+
+        with mock.patch('utilities_common.db.Db', return_value=prepared_db) as mock_db_cls:
+            console_connect('1')
+            mock_db_cls.assert_called_once()
+
 
 class TestConsutilClear(object):
     @classmethod