diff --git a/config/console.py b/config/console.py index 1ecf80c38..d487b9ae2 100644 --- a/config/console.py +++ b/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='', 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,7 +97,9 @@ def disable_console_switch(db): @click.option('--baud', '-b', metavar='', required=True, type=click.INT) @click.option('--flowcontrol', '-f', metavar='', required=False, is_flag=True) @click.option('--devicename', '-d', metavar='', required=False) -def add_console_setting(db, linenum, baud, flowcontrol, devicename): +@click.option('--escape', '-e', metavar='', 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) @@ -67,6 +107,7 @@ def add_console_setting(db, linenum, baud, flowcontrol, devicename): 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,6 +153,7 @@ 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 ...') # @@ -116,7 +161,7 @@ def remove_console_setting(db, linenum): @clicommon.pass_db @click.argument('linenum', metavar='', required=True, type=click.IntRange(0, 65535)) @click.argument('devicename', metavar='', 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='', required=True, type=click.IntRange(0, 65535)) +@click.argument('escape', metavar='', 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) diff --git a/consutil/lib.py b/consutil/lib.py index e597e3b64..dbadcf8d1 100644 --- a/consutil/lib.py +++ b/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,7 +79,7 @@ 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 @@ -85,6 +87,21 @@ 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): diff --git a/consutil/main.py b/consutil/main.py index be6ac8af2..bec268203 100644 --- a/consutil/main.py +++ b/consutil/main.py @@ -5,6 +5,7 @@ # Command-line utility for interacting with switches over serial via console device # + try: import click import os @@ -16,6 +17,7 @@ except ImportError as e: raise ImportError("%s - required module not found" % str(e)) + @click.group() @clicommon.pass_db def consutil(db): @@ -28,13 +30,14 @@ def consutil(db): SysInfoProvider.init_device_prefix() + # 'show' subcommand @consutil.command() @clicommon.pass_db -@click.option('--brief', '-b', metavar='', required=False, is_flag=True) +@click.option('--brief', '-b', is_flag=True) def show(db, brief): """Show all ports and their info include available ttyUSB devices unless specified brief mode""" - port_provider = ConsolePortProvider(db, brief, refresh=True) + port_provider = ConsolePortProvider(db, brief, refresh=True) # noqa: F405 ports = list(port_provider.get_all()) # sort ports for table rendering @@ -50,14 +53,50 @@ def show(db, brief): date = port.session_start_date if port.session_start_date else "-" baud = port.baud flow_control = "Enabled" if port.flow_control else "Disabled" - body.append([busy+port.line_num, baud if baud else "-", flow_control, pid if pid else "-", date if date else "-", port.remote_device]) + body.append([ + busy+port.line_num, + baud if baud else "-", + flow_control, + pid if pid else "-", + date if date else "-", + port.remote_device, + ]) + click.echo(tabulate(body, header, stralign='right')) + + +# 'show-escape' subcommand +@consutil.command('show-escape') +@clicommon.pass_db +@click.option('--brief', '-b', is_flag=True) +def show_escape(db, brief): + """Show all default and line escape char info include available ttyUSB devices unless specified brief mode""" + port_provider = ConsolePortProvider(db, brief, refresh=True) # noqa: F405 + ports = list(port_provider.get_all()) + + # sort ports for table rendering + ports.sort(key=lambda p: int(p.line_num)) + + # set table header style + header = ["Line", "Default Escape Char", "Line Escape Char", "Final Escape Char"] + body = [] + for port in ports: + # runtime information + busy = "*" if port.busy else " " + body.append([ + busy+port.line_num, + port.default_escape_char if port.default_escape_char else "-", + port.line_escape_char if port.line_escape_char else "-", + port.escape_char if port.escape_char else "-", + ]) click.echo(tabulate(body, header, stralign='right')) + # 'clear' subcommand @consutil.command() @clicommon.pass_db @click.argument('target') -@click.option('--devicename', '-d', is_flag=True, help="clear by name - if flag is set, interpret target as device name instead") +@click.option('--devicename', '-d', is_flag=True, + help="clear by name - if flag is set, interpret target as device name instead") def clear(db, target, devicename): """Clear preexisting connection to line""" if os.geteuid() != 0: @@ -77,11 +116,13 @@ def clear(db, target, devicename): else: click.echo("Cleared line") + # 'connect' subcommand @consutil.command() @clicommon.pass_db @click.argument('target') -@click.option('--devicename', '-d', is_flag=True, help="connect by name - if flag is set, interpret target as device name instead") +@click.option('--devicename', '-d', is_flag=True, + 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 @@ -108,8 +149,10 @@ def connect(db, target, devicename): sys.exit(ERR_DEV) # interact - click.echo("Successful connection to line [{}]\nPress ^A ^X to disconnect".format(line_num)) + 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() + if __name__ == '__main__': consutil() diff --git a/show/main.py b/show/main.py index e73162e49..b4013014c 100755 --- a/show/main.py +++ b/show/main.py @@ -2476,11 +2476,13 @@ def information(): # 'line' command ("show line") # @cli.command('line') -@click.option('--brief', '-b', metavar='', required=False, is_flag=True) +@click.option('--brief', '-b', is_flag=True, + help="Show information for only configured console lines") +@click.option('--show-escape', "-e", is_flag=True, help="Show escape character for each line") @click.option('--verbose', is_flag=True, help="Enable verbose output") -def line(brief, verbose): +def line(brief, show_escape, verbose): """Show all console lines and their info include available ttyUSB devices unless specified brief mode""" - cmd = ['consutil', 'show'] + (["-b"] if brief else []) + cmd = ['consutil', 'show-escape' if show_escape else 'show'] + (["-b"] if brief else []) run_command(cmd, display_cmd=verbose) return diff --git a/tests/console_test.py b/tests/console_test.py index 3a0d9b28d..2d3a1b669 100644 --- a/tests/console_test.py +++ b/tests/console_test.py @@ -27,7 +27,7 @@ class TestConfigConsoleCommands(object): @classmethod def setup_class(cls): print("SETUP") - + def test_enable_console_switch(self): runner = CliRunner() db = Db() @@ -36,9 +36,10 @@ def test_enable_console_switch(self): print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code == 0 - + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", + mock.Mock(side_effect=ValueError)) def test_enable_console_switch_yang_validation(self): runner = CliRunner() db = Db() @@ -57,7 +58,8 @@ def test_disable_console_switch(self): assert result.exit_code == 0 @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", + mock.Mock(side_effect=ValueError)) def test_disable_console_switch_yang_validation(self): runner = CliRunner() db = Db() @@ -66,11 +68,11 @@ def test_disable_console_switch_yang_validation(self): print(result.exit_code) print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output - + def test_console_add_exists(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"baud_rate": "9600"}) # add a console setting which the port exists result = runner.invoke(config.config.commands["console"].commands["add"], ["1", '--baud', "9600"], obj=db) @@ -78,7 +80,7 @@ def test_console_add_exists(self): print(sys.stderr, result.output) assert result.exit_code != 0 assert "Trying to add console port setting, which is already exists." in result.output - + def test_console_add_no_baud(self): runner = CliRunner() db = Db() @@ -93,10 +95,11 @@ def test_console_add_no_baud(self): def test_console_add_name_conflict(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch1"}) # add a console setting which the device name has been used by other port - result = runner.invoke(config.config.commands["console"].commands["add"], ["1", '--baud', "9600", "--devicename", "switch1"], obj=db) + result = runner.invoke(config.config.commands["console"].commands["add"], + ["1", '--baud', "9600", "--devicename", "switch1"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code != 0 @@ -113,19 +116,42 @@ def test_console_add_success(self): assert result.exit_code == 0 # add a console setting with flow control option - result = runner.invoke(config.config.commands["console"].commands["add"], ["1", '--baud', "9600", "--flowcontrol"], obj=db) + result = runner.invoke(config.config.commands["console"].commands["add"], + ["1", '--baud', "9600", "--flowcontrol"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code == 0 # add a console setting with device name option - result = runner.invoke(config.config.commands["console"].commands["add"], ["2", '--baud', "9600", "--devicename", "switch1"], obj=db) + result = runner.invoke(config.config.commands["console"].commands["add"], + ["2", '--baud', "9600", "--devicename", "switch1"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # add a console setting with escape character option + result = runner.invoke( + config.config.commands["console"].commands["add"], + ["3", '--baud', "9600", "--escape", "A"], + obj=db, + ) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # add a console setting with all options (flow control, device name, and escape character) + result = runner.invoke( + config.config.commands["console"].commands["add"], + ["4", '--baud', "9600", "--flowcontrol", "--devicename", "switch2", "--escape", "b"], + obj=db, + ) print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code == 0 @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", + mock.Mock(side_effect=ValueError)) def test_console_add_yang_validation(self): runner = CliRunner() db = Db() @@ -135,7 +161,50 @@ def test_console_add_yang_validation(self): print(result.exit_code) print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output - + + def test_console_add_complete_database_state_verification(self): + """Test that all fields are correctly stored in the database with proper types and values""" + runner = CliRunner() + db = Db() + + # Add console port with all options + result = runner.invoke( + config.config.commands["console"].commands["add"], + ["5", '--baud', "115200", "--flowcontrol", "--devicename", "test-switch", "--escape", "X"], + obj=db, + ) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # Verify COMPLETE database entry + entry = db.cfgdb.get_entry("CONSOLE_PORT", "5") + assert entry is not None + assert entry["baud_rate"] == "115200" + assert entry["flow_control"] == "1" # flowcontrol flag should set this to "1" + assert entry["remote_device"] == "test-switch" + assert entry["escape_char"] == "x" # Should be converted to lowercase + + # Verify no unexpected fields + expected_fields = {"baud_rate", "flow_control", "remote_device", "escape_char"} + assert set(entry.keys()) == expected_fields + + def test_console_add_partial_database_state_verification(self): + """Test that only specified fields are stored, defaults are applied correctly""" + runner = CliRunner() + db = Db() + + # Add console port with minimal options (just baud rate) + result = runner.invoke(config.config.commands["console"].commands["add"], ["6", '--baud', "38400"], obj=db) + assert result.exit_code == 0 + + # Verify database entry has required fields and proper defaults + entry = db.cfgdb.get_entry("CONSOLE_PORT", "6") + assert entry["baud_rate"] == "38400" + assert entry["flow_control"] == "0" # Default when no --flowcontrol flag + assert "remote_device" not in entry # Optional field not provided + assert "escape_char" not in entry # Optional field not provided + def test_console_del_non_exists(self): runner = CliRunner() db = Db() @@ -150,7 +219,7 @@ def test_console_del_non_exists(self): def test_console_del_success(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # add a console setting which the port exists result = runner.invoke(config.config.commands["console"].commands["del"], ["1"], obj=db) @@ -159,18 +228,124 @@ def test_console_del_success(self): assert result.exit_code == 0 @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", mock.Mock(side_effect=JsonPatchConflict)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", + mock.Mock(side_effect=JsonPatchConflict)) def test_console_del_yang_validation(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # add a console setting which the port exists result = runner.invoke(config.config.commands["console"].commands["del"], ["1"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output - + + def test_console_default_escape_set_lowercase(self): + runner = CliRunner() + db = Db() + + # set console escape character to 'd' + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["d"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # verify the default_escape_char is stored in the config + console_mgmt = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert console_mgmt.get("default_escape_char") == "d" + + def test_console_default_escape_set_uppercase(self): + runner = CliRunner() + db = Db() + + # set console escape character to 'D' (uppercase) - should be converted to lowercase + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["D"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # verify the default_escape_char is stored as lowercase in the config + console_mgmt = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert console_mgmt.get("default_escape_char") == "d" + + def test_console_default_escape_clear(self): + runner = CliRunner() + db = Db() + + # first set an escape character + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes", "default_escape_char": "d"}) + + # clear the escape character + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["clear"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # verify the default_escape_char is removed but enabled is preserved + console_mgmt = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert "default_escape_char" not in console_mgmt + assert console_mgmt.get("enabled") == "yes" + + def test_console_default_escape_clear_when_enabled_exists(self): + """Test clearing default escape character preserves enabled field""" + runner = CliRunner() + db = Db() + + # Set up console switch with both enabled and default_escape_char + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", { + "enabled": "yes", + "default_escape_char": "d" + }) + + # Verify initial state + entry = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert len(entry) == 2 + assert entry["default_escape_char"] == "d" + assert entry["enabled"] == "yes" + + # Clear default escape character + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["clear"], obj=db) + assert result.exit_code == 0 + + # Verify default_escape_char is removed but enabled is preserved + entry = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert "default_escape_char" not in entry + assert entry["enabled"] == "yes" + assert len(entry) == 1 + + def test_console_default_escape_clear_no_escape(self): + """Test clearing default_escape_char when it was never set""" + runner = CliRunner() + db = Db() + + # Set up console switch with only enabled, no default_escape_char + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes"}) + + # Clear default escape character that doesn't exist + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["clear"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # Verify entry is unchanged + entry = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert "default_escape_char" not in entry + assert entry.get("enabled") == "yes" + + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", + mock.Mock(side_effect=ValueError)) + def test_console_default_escape_set_yang_validation(self): + runner = CliRunner() + db = Db() + + # set console escape character with yang validation error + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["d"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert "Invalid ConfigDB. Error" in result.output + def test_update_console_remote_device_name_non_exists(self): runner = CliRunner() db = Db() @@ -185,8 +360,8 @@ def test_update_console_remote_device_name_non_exists(self): def test_update_console_remote_device_name_conflict(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "baud": "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "baud": "9600", "remote_device" : "switch1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"baud": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"baud": "9600", "remote_device": "switch1"}) # trying to update a console line remote device configuration which is not exists result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["1", "switch1"], obj=db) @@ -194,22 +369,26 @@ def test_update_console_remote_device_name_conflict(self): print(sys.stderr, result.output) assert result.exit_code != 0 assert "Please enter a valid device name or remove the existing one" in result.output - + def test_update_console_remote_device_name_existing_and_same(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch1"}) - # trying to update a console line remote device configuration which is existing and same with user provided value + # trying to update a console line remote device configuration that exists and same with user provided value result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["2", "switch1"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code == 0 + # Verify that the device name remains unchanged in the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "2") + assert entry.get("remote_device") == "switch1" + def test_update_console_remote_device_name_reset(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch1"}) # trying to reset a console line remote device configuration which is not exists result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["2"], obj=db) @@ -217,23 +396,48 @@ def test_update_console_remote_device_name_reset(self): print(sys.stderr, result.output) assert result.exit_code == 0 - @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError)) + # Verify that the remote_device field has been removed from the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "2") + assert "remote_device" not in entry + + def test_update_console_remote_device_name_reset_no_device(self): + """Test resetting remote_device when the field doesn't exist on the port""" + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"baud_rate": "9600"}) + + # trying to reset remote_device that was never configured + result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["2"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # Verify the port entry is unchanged + entry = db.cfgdb.get_entry("CONSOLE_PORT", "2") + assert "remote_device" not in entry + assert entry.get("baud_rate") == "9600" + + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", + mock.Mock(return_value=True)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", + mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", + mock.Mock(side_effect=ValueError)) def test_update_console_remote_device_name_reset_yang_validation(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch1"}) # trying to reset a console line remote device configuration which is not exists result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["2"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output - + def test_update_console_remote_device_name_success(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # trying to set a console line remote device configuration result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["1", "switch1"], obj=db) @@ -241,30 +445,111 @@ def test_update_console_remote_device_name_success(self): print(sys.stderr, result.output) assert result.exit_code == 0 + # Verify that the remote_device field has been added to the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert entry.get("remote_device") == "switch1" + assert entry.get("baud_rate") == "9600" # Ensure other fields preserved + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", + mock.Mock(side_effect=ValueError)) def test_update_console_remote_device_name_yang_validation(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # trying to set a console line remote device configuration result = runner.invoke(config.config.commands["console"].commands["remote_device"], ["1", "switch1"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output - + + def test_update_console_escape_non_exists(self): + runner = CliRunner() + db = Db() + + # trying to set a console line escape character which is not exists + result = runner.invoke(config.config.commands["console"].commands["escape"], ["1", "d"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code != 0 + assert "Trying to update console port setting, which is not present." in result.output + + def test_update_console_escape_success(self): + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) + + # trying to set a console line escape character + result = runner.invoke(config.config.commands["console"].commands["escape"], ["1", "D"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + line_cfg = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert line_cfg.get("escape_char") == "d" + + def test_update_console_escape_clear(self): + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600", "escape_char": "d"}) + + # trying to clear a console line escape character + result = runner.invoke(config.config.commands["console"].commands["escape"], ["1", "clear"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + line_cfg = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert "escape_char" not in line_cfg + assert line_cfg.get("baud_rate") == "9600" + + def test_update_console_escape_clear_no_escape(self): + """Test clearing escape_char when the field doesn't exist on the port""" + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) + + # trying to clear escape_char that was never configured + result = runner.invoke(config.config.commands["console"].commands["escape"], ["1", "clear"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + + # Verify the port entry is unchanged + line_cfg = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert "escape_char" not in line_cfg + assert line_cfg.get("baud_rate") == "9600" + + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", + mock.Mock(side_effect=ValueError)) + def test_update_console_escape_yang_validation(self): + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) + + # trying to set a console line escape character with yang validation error + result = runner.invoke(config.config.commands["console"].commands["escape"], ["1", "d"], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert "Invalid ConfigDB. Error" in result.output + def test_update_console_baud_no_change(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # trying to set a console line baud which is same with existing one result = runner.invoke(config.config.commands["console"].commands["baud"], ["1", "9600"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code == 0 - + + # Verify that the baud_rate remains unchanged in the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert entry.get("baud_rate") == "9600" + def test_update_console_baud_non_exists(self): runner = CliRunner() db = Db() @@ -275,11 +560,11 @@ def test_update_console_baud_non_exists(self): print(sys.stderr, result.output) assert result.exit_code != 0 assert "Trying to update console port setting, which is not present." in result.output - + def test_update_console_baud_success(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # trying to set a console line baud result = runner.invoke(config.config.commands["console"].commands["baud"], ["1", "115200"], obj=db) @@ -287,23 +572,28 @@ def test_update_console_baud_success(self): print(sys.stderr, result.output) assert result.exit_code == 0 + # Verify that the baud_rate has been updated in the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert entry.get("baud_rate") == "115200" + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", + mock.Mock(side_effect=ValueError)) def test_update_console_baud_yang_validation(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) # trying to set a console line baud result = runner.invoke(config.config.commands["console"].commands["baud"], ["1", "115200"], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output - + def test_update_console_flow_control_no_change(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600", "flow_control" : "0" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600", "flow_control": "0"}) # trying to set a console line flow control option which is same with existing one result = runner.invoke(config.config.commands["console"].commands["flow_control"], ["disable", "1"], obj=db) @@ -311,6 +601,11 @@ def test_update_console_flow_control_no_change(self): print(sys.stderr, result.output) assert result.exit_code == 0 + # Verify that the flow_control remains unchanged in the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert entry.get("flow_control") == "0" + assert entry.get("baud_rate") == "9600" # Ensure other fields preserved + def test_update_console_flow_control_non_exists(self): runner = CliRunner() db = Db() @@ -325,7 +620,7 @@ def test_update_console_flow_control_non_exists(self): def test_update_console_flow_control_success(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600", "flow_control" : "0" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600", "flow_control": "0"}) # trying to set a console line flow control option result = runner.invoke(config.config.commands["console"].commands["flow_control"], ["enable", "1"], obj=db) @@ -333,12 +628,18 @@ def test_update_console_flow_control_success(self): print(sys.stderr, result.output) assert result.exit_code == 0 + # Verify that the flow_control has been updated in the database + entry = db.cfgdb.get_entry("CONSOLE_PORT", "1") + assert entry.get("flow_control") == "1" + assert entry.get("baud_rate") == "9600" # Ensure other fields preserved + @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True)) - @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", mock.Mock(side_effect=ValueError)) + @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_mod_entry", + mock.Mock(side_effect=ValueError)) def test_update_console_flow_control_yang_validation(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600", "flow_control" : "0" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600", "flow_control": "0"}) # trying to set a console line flow control option result = runner.invoke(config.config.commands["console"].commands["flow_control"], ["enable", "1"], obj=db) @@ -346,6 +647,149 @@ def test_update_console_flow_control_yang_validation(self): print(sys.stderr, result.output) assert "Invalid ConfigDB. Error" in result.output + def test_console_full_workflow_integration(self): + """Test complete workflow: add -> update various fields -> delete""" + runner = CliRunner() + db = Db() + + # Step 0: Set global default escape character + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["G"], obj=db) + assert result.exit_code == 0 + global_entry = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert global_entry["default_escape_char"] == "g" + + # Step 1: Add console port with basic settings + result = runner.invoke(config.config.commands["console"].commands["add"], ["7", '--baud', "9600"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry["baud_rate"] == "9600" + assert entry["flow_control"] == "0" + + # Step 2: Update baud rate + result = runner.invoke(config.config.commands["console"].commands["baud"], ["7", "115200"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry["baud_rate"] == "115200" + + # Step 3: Enable flow control + result = runner.invoke(config.config.commands["console"].commands["flow_control"], ["enable", "7"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry["flow_control"] == "1" + + # Step 4: Add device name + result = runner.invoke( + config.config.commands["console"].commands["remote_device"], + ["7", "workflow-device"], + obj=db, + ) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry["remote_device"] == "workflow-device" + + # Step 5: Add escape character + result = runner.invoke(config.config.commands["console"].commands["escape"], ["7", "Z"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry["escape_char"] == "z" + + # Step 6: Update escape character + result = runner.invoke(config.config.commands["console"].commands["escape"], ["7", "A"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry["escape_char"] == "a" + + # Step 7: Remove escape character + result = runner.invoke(config.config.commands["console"].commands["escape"], ["7", "clear"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert "escape_char" not in entry + + # Step 8: Remove device name + result = runner.invoke( + config.config.commands["console"].commands["remote_device"], + ["7"], + obj=db, + ) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert "remote_device" not in entry + + # Step 10: Clear global default escape character + result = runner.invoke(config.config.commands["console"].commands["default_escape"], ["clear"], obj=db) + assert result.exit_code == 0 + global_entry = db.cfgdb.get_entry("CONSOLE_SWITCH", "console_mgmt") + assert "default_escape_char" not in global_entry + + # Step 11: Delete entire console port + result = runner.invoke(config.config.commands["console"].commands["del"], ["7"], obj=db) + assert result.exit_code == 0 + entry = db.cfgdb.get_entry("CONSOLE_PORT", "7") + assert entry is None or len(entry) == 0 + + def test_console_multiple_operations_same_port(self): + """Test that multiple rapid updates to the same port work correctly""" + runner = CliRunner() + db = Db() + + # Add initial port + result = runner.invoke(config.config.commands["console"].commands["add"], ["8", '--baud', "9600"], obj=db) + assert result.exit_code == 0 + + # Perform multiple updates in sequence + operations = [ + ("baud", ["8", "19200"]), + ("flow_control", ["enable", "8"]), + ("remote_device", ["8", "multi-op-device"]), + ("escape", ["8", "M"]), + ("baud", ["8", "57600"]), + ("flow_control", ["disable", "8"]), + ("escape", ["8", "N"]), + ("remote_device", ["8", "updated-device"]) + ] + + for command, args in operations: + result = runner.invoke(config.config.commands["console"].commands[command], args, obj=db) + assert result.exit_code == 0 + + # Verify final state + entry = db.cfgdb.get_entry("CONSOLE_PORT", "8") + assert entry["baud_rate"] == "57600" + assert entry["flow_control"] == "0" + assert entry["remote_device"] == "updated-device" + assert entry["escape_char"] == "n" + + def test_update_console_escape_clear_when_multiple_fields_exist(self): + """Test clearing escape character preserves all other fields""" + runner = CliRunner() + db = Db() + + # Set up port with ALL possible fields + db.cfgdb.set_entry("CONSOLE_PORT", "9", { + "baud_rate": "115200", + "flow_control": "1", + "remote_device": "clear-test-device", + "escape_char": "t" + }) + + # Verify initial state + entry = db.cfgdb.get_entry("CONSOLE_PORT", "9") + assert len(entry) == 4 + assert entry["escape_char"] == "t" + + # Clear only the escape character + result = runner.invoke(config.config.commands["console"].commands["escape"], ["9", "clear"], obj=db) + assert result.exit_code == 0 + + # Verify escape_char is removed but other fields preserved + entry = db.cfgdb.get_entry("CONSOLE_PORT", "9") + assert "escape_char" not in entry + assert entry["baud_rate"] == "115200" + assert entry["flow_control"] == "1" + assert entry["remote_device"] == "clear-test-device" + assert len(entry) == 3 + + class TestConsutilLib(object): @classmethod def setup_class(cls): @@ -358,15 +802,16 @@ def test_console_port_provider_get_all_configured_only_empty(self): def test_console_port_provider_get_all_configured_only_nonempty(self): db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) provider = ConsolePortProvider(db, configured_only=True) assert len(list(provider.get_all())) == 1 - @mock.patch('consutil.lib.SysInfoProvider.list_console_ttys', mock.MagicMock(return_value=["/dev/ttyUSB0", "/dev/ttyUSB1"])) + @mock.patch('consutil.lib.SysInfoProvider.list_console_ttys', + mock.MagicMock(return_value=["/dev/ttyUSB0", "/dev/ttyUSB1"])) def test_console_port_provider_get_all_with_ttys(self): db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) provider = ConsolePortProvider(db, configured_only=False) ports = list(provider.get_all()) @@ -375,7 +820,7 @@ def test_console_port_provider_get_all_with_ttys(self): def test_console_port_provider_get_line_success(self): db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", "1", { "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", "1", {"baud_rate": "9600"}) provider = ConsolePortProvider(db, configured_only=True) port = provider.get("1") @@ -390,7 +835,7 @@ def test_console_port_provider_get_line_not_found(self): def test_console_port_provider_get_line_by_device_success(self): db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch2" }) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2"}) provider = ConsolePortProvider(db, configured_only=True) port = provider.get("switch2", use_device=True) @@ -400,34 +845,37 @@ def test_console_port_provider_get_line_by_device_success(self): def test_console_port_provider_get_line_by_device_not_found(self): with pytest.raises(LineNotFoundError): db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch2" }) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2"}) provider = ConsolePortProvider(db, configured_only=True) provider.get("switch1") - @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', mock.MagicMock(return_value={ "1" : ("223", "2020/11/2")})) + @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', + mock.MagicMock(return_value={"1": ("223", "2020/11/2")})) def test_console_port_info_refresh_without_session(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1" }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1"}) port.refresh() assert port.busy assert port.session_pid == "223" assert port.session_start_date == "2020/11/2" - @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', mock.MagicMock(return_value={ "2" : ("223", "2020/11/2")})) + @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', + mock.MagicMock(return_value={"2": ("223", "2020/11/2")})) def test_console_port_info_refresh_without_session_idle(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1" }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1"}) port.refresh() assert port.busy == False - @mock.patch('consutil.lib.SysInfoProvider.get_active_console_process_info', mock.MagicMock(return_value=("1", "223", "2020/11/2"))) + @mock.patch('consutil.lib.SysInfoProvider.get_active_console_process_info', + mock.MagicMock(return_value=("1", "223", "2020/11/2"))) def test_console_port_info_refresh_with_session(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1" }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1"}) port._session = ConsoleSession(port, mock.MagicMock(pid="223")) print(port) @@ -436,11 +884,12 @@ def test_console_port_info_refresh_with_session(self): assert port.session_pid == "223" assert port.session_start_date == "2020/11/2" - @mock.patch('consutil.lib.SysInfoProvider.get_active_console_process_info', mock.MagicMock(return_value=("2", "223", "2020/11/2"))) + @mock.patch('consutil.lib.SysInfoProvider.get_active_console_process_info', + mock.MagicMock(return_value=("2", "223", "2020/11/2"))) def test_console_port_info_refresh_with_session_line_mismatch(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1" }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1"}) port._session = ConsoleSession(port, mock.MagicMock(pid="223")) print(port) @@ -453,7 +902,7 @@ def test_console_port_info_refresh_with_session_line_mismatch(self): def test_console_port_info_refresh_with_session_process_ended(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1" }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1"}) port._session = ConsoleSession(port, mock.MagicMock(pid="223")) print(port) @@ -462,7 +911,7 @@ def test_console_port_info_refresh_with_session_process_ended(self): def test_console_port_info_connect_state_busy(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "CUR_STATE" : { "state" : "busy" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "CUR_STATE": {"state": "busy"}}) port.refresh = mock.MagicMock(return_value=None) with pytest.raises(LineBusyError): @@ -470,7 +919,7 @@ def test_console_port_info_connect_state_busy(self): def test_console_port_info_connect_invalid_config(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "CUR_STATE" : { "state" : "idle" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "CUR_STATE": {"state": "idle"}}) port.refresh = mock.MagicMock(return_value=None) with pytest.raises(InvalidConfigurationError): @@ -478,7 +927,7 @@ def test_console_port_info_connect_invalid_config(self): def test_console_port_info_connect_device_busy(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "baud_rate" : "9600", "CUR_STATE" : { "state" : "idle" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "baud_rate": "9600", "CUR_STATE": {"state": "idle"}}) port.refresh = mock.MagicMock(return_value=None) mock_proc = mock.MagicMock(spec=subprocess.Popen) @@ -490,7 +939,7 @@ def test_console_port_info_connect_device_busy(self): def test_console_port_info_connect_connection_fail(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "baud_rate" : "9600", "CUR_STATE" : { "state" : "idle" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "baud_rate": "9600", "CUR_STATE": {"state": "idle"}}) port.refresh = mock.MagicMock(return_value=None) mock_proc = mock.MagicMock(spec=subprocess.Popen) @@ -502,7 +951,7 @@ def test_console_port_info_connect_connection_fail(self): def test_console_port_info_connect_success(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "baud_rate" : "9600", "CUR_STATE" : { "state" : "idle" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "baud_rate": "9600", "CUR_STATE": {"state": "idle"}}) port.refresh = mock.MagicMock(return_value=None) mock_proc = mock.MagicMock(spec=subprocess.Popen, pid="223") @@ -515,7 +964,7 @@ def test_console_port_info_connect_success(self): def test_console_port_info_clear_session_line_not_busy(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "baud_rate" : "9600", "CUR_STATE" : { "state" : "idle" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "baud_rate": "9600", "CUR_STATE": {"state": "idle"}}) port.refresh = mock.MagicMock(return_value=None) assert not port.clear_session() @@ -523,26 +972,29 @@ def test_console_port_info_clear_session_line_not_busy(self): @mock.patch('consutil.lib.SysInfoProvider.run_command', mock.MagicMock(return_value=None)) def test_console_port_info_clear_session_with_state_db(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "baud_rate" : "9600", "CUR_STATE" : { "state" : "busy", "pid" : "223" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "baud_rate": "9600", + "CUR_STATE": {"state": "busy", "pid": "223"}}) port.refresh = mock.MagicMock(return_value=None) assert port.clear_session() def test_console_port_info_clear_session_with_existing_session(self): db = Db() - port = ConsolePortInfo(DbUtils(db), { "LINE" : "1", "baud_rate" : "9600", "CUR_STATE" : { "state" : "busy" } }) + port = ConsolePortInfo(DbUtils(db), {"LINE": "1", "baud_rate": "9600", "CUR_STATE": {"state": "busy"}}) port._session = ConsoleSession(port, None) port._session.close = mock.MagicMock(return_value=None) port.refresh = mock.MagicMock(return_value=None) assert port.clear_session() - @mock.patch('sonic_py_common.device_info.get_paths_to_platform_and_hwsku_dirs', mock.MagicMock(return_value=("dummy_path", None))) + @mock.patch('sonic_py_common.device_info.get_paths_to_platform_and_hwsku_dirs', + mock.MagicMock(return_value=("dummy_path", None))) @mock.patch('os.path.exists', mock.MagicMock(return_value=False)) def test_sys_info_provider_init_device_prefix_plugin_nonexists(self): SysInfoProvider.init_device_prefix() assert SysInfoProvider.DEVICE_PREFIX == "/dev/ttyUSB" - @mock.patch('sonic_py_common.device_info.get_paths_to_platform_and_hwsku_dirs', mock.MagicMock(return_value=("dummy_path", None))) + @mock.patch('sonic_py_common.device_info.get_paths_to_platform_and_hwsku_dirs', + mock.MagicMock(return_value=("dummy_path", None))) @mock.patch('os.path.exists', mock.MagicMock(return_value=True)) def test_sys_info_provider_init_device_prefix_plugin(self): with mock.patch("builtins.open", mock.mock_open(read_data="C0-")): @@ -562,7 +1014,7 @@ def test_sys_info_provider_list_console_ttys_device_not_exists(self): all_active_processes_output = ''+ \ """ PID STARTED CMD - 8 Mon Nov 2 04:29:41 2020 picocom /dev/ttyUSB0 + 8 Mon Nov 2 04:29:41 2020 picocom /dev/ttyUSB0 """ @mock.patch('consutil.lib.SysInfoProvider.run_command', mock.MagicMock(return_value=all_active_processes_output)) def test_sys_info_provider_list_active_console_processes(self): @@ -587,6 +1039,7 @@ def test_sys_info_provider_get_active_console_process_info_nonexists(self): proc = SysInfoProvider.get_active_console_process_info("2") assert proc is None + class TestConsutil(object): @classmethod def setup_class(cls): @@ -609,7 +1062,7 @@ def test_consutil_feature_disabled_null_config(self): def test_consutil_feature_disabled_config(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", { "enabled" : "no" }) + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "no"}) result = runner.invoke(consutil.consutil, ['show'], obj=db) print(result.exit_code) @@ -622,13 +1075,27 @@ def test_consutil_feature_disabled_config(self): def test_consutil_feature_enabled(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", { "enabled" : "yes" }) + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes"}) result = runner.invoke(consutil.consutil, ['show'], obj=db) print(result.exit_code) print(sys.stderr, result.output) assert result.exit_code == 0 + @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) + @mock.patch('consutil.main.show_escape', mock.MagicMock(return_value=None)) + def test_consutil_show_escape_feature_disabled(self): + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "no"}) + + result = runner.invoke(consutil.consutil, ['show-escape'], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 1 + assert result.output == "Console switch feature is disabled\n" + + class TestConsutilShow(object): @classmethod def setup_class(cls): @@ -642,13 +1109,14 @@ def setup_class(cls): 3 9600 Enabled - - """ @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) - @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', mock.MagicMock(return_value={ "2" : ("223", "Wed Mar 6 08:31:35 2019")})) + @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', + mock.MagicMock(return_value={"2": ("223", "Wed Mar 6 08:31:35 2019")})) def test_show(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch2", "baud_rate" : "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 3, { "baud_rate" : "9600", "flow_control" : "1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 3, {"baud_rate": "9600", "flow_control": "1"}) db.db.set(db.db.STATE_DB, "CONSOLE_PORT|2", "state", "busy") db.db.set(db.db.STATE_DB, "CONSOLE_PORT|2", "pid", "223") @@ -662,13 +1130,14 @@ def test_show(self): assert result.output == TestConsutilShow.expect_show_output @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) - @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', mock.MagicMock(return_value={ "2" : ("223", "Wed Mar 6 08:31:35 2019")})) + @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', + mock.MagicMock(return_value={"2": ("223", "Wed Mar 6 08:31:35 2019")})) def test_show_stale_idle_to_busy(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch2", "baud_rate" : "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 3, { "baud_rate" : "9600", "flow_control" : "1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 3, {"baud_rate": "9600", "flow_control": "1"}) # use '--brief' option to avoid access system result = runner.invoke(consutil.consutil.commands["show"], ['--brief'], obj=db) @@ -678,13 +1147,14 @@ def test_show_stale_idle_to_busy(self): assert result.output == TestConsutilShow.expect_show_output @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) - @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', mock.MagicMock(return_value={ "2" : ("223", "Wed Mar 6 08:31:35 2019")})) + @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', + mock.MagicMock(return_value={"2": ("223", "Wed Mar 6 08:31:35 2019")})) def test_show_stale_busy_to_idle(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 2, { "remote_device" : "switch2", "baud_rate" : "9600" }) - db.cfgdb.set_entry("CONSOLE_PORT", 3, { "baud_rate" : "9600", "flow_control" : "1" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 3, {"baud_rate": "9600", "flow_control": "1"}) db.db.set(db.db.STATE_DB, "CONSOLE_PORT|1", "state", "busy") db.db.set(db.db.STATE_DB, "CONSOLE_PORT|1", "pid", "222") @@ -701,6 +1171,84 @@ def test_show_stale_busy_to_idle(self): assert result.exit_code == 0 assert result.output == TestConsutilShow.expect_show_output + +class TestConsutilShowEscape(object): + @classmethod + def setup_class(cls): + print("SETUP") + + expect_show_escape_output_no_global = '' + \ + """ Line Default Escape Char Line Escape Char Final Escape Char +------ --------------------- ------------------ ------------------- + 1 - a a + 2 - - - + 3 - c c +""" + + expect_show_escape_output_with_global = '' + \ + """ Line Default Escape Char Line Escape Char Final Escape Char +------ --------------------- ------------------ ------------------- + 1 g a a + *2 g - g + 3 g c c +""" + + @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) + def test_show_escape_no_global_default(self): + """Test show_escape with no global default escape character set""" + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes"}) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600", "escape_char": "a"}) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 3, {"baud_rate": "9600", "flow_control": "1", "escape_char": "c"}) + + # use '--brief' option to avoid access system + result = runner.invoke(consutil.consutil.commands["show-escape"], ['--brief'], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + assert result.output == TestConsutilShowEscape.expect_show_escape_output_no_global + + @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) + @mock.patch('consutil.lib.SysInfoProvider.list_active_console_processes', + mock.MagicMock(return_value={"2": ("223", "Wed Mar 6 08:31:35 2019")})) + def test_show_escape_with_global_default(self): + """Test show_escape with global default escape character and per-line overrides""" + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes", "default_escape_char": "g"}) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600", "escape_char": "a"}) + db.cfgdb.set_entry("CONSOLE_PORT", 2, {"remote_device": "switch2", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_PORT", 3, {"baud_rate": "9600", "flow_control": "1", "escape_char": "c"}) + + # use '--brief' option to avoid access system + result = runner.invoke(consutil.consutil.commands["show-escape"], ['--brief'], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + assert result.output == TestConsutilShowEscape.expect_show_escape_output_with_global + + @mock.patch('consutil.lib.SysInfoProvider.init_device_prefix', mock.MagicMock(return_value=None)) + def test_show_escape_empty_config(self): + """Test show_escape with no console ports configured""" + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes"}) + + # use '--brief' option to avoid access system + result = runner.invoke(consutil.consutil.commands["show-escape"], ['--brief'], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + # Should just show header with no data rows + expected_output = '' + \ + """ Line Default Escape Char Line Escape Char Final Escape Char +------ --------------------- ------------------ ------------------- +""" + assert result.output == expected_output + + class TestConsutilConnect(object): @classmethod def setup_class(cls): @@ -711,7 +1259,7 @@ def setup_class(cls): def test_connect_target_nonexists(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) result = runner.invoke(consutil.consutil.commands["connect"], ['2'], obj=db) print(result.exit_code) @@ -731,7 +1279,7 @@ def test_connect_target_nonexists(self): def test_connect_line_busy(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) result = runner.invoke(consutil.consutil.commands["connect"], ['1'], obj=db) print(result.exit_code) @@ -772,11 +1320,12 @@ def test_connect_picocom_err(self): @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)))) + @mock.patch('consutil.lib.ConsolePortInfo.connect', + mock.MagicMock(return_value=mock.MagicMock(interact=mock.MagicMock(return_value=None)))) def test_connect_success(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) result = runner.invoke(consutil.consutil.commands["connect"], ['1'], obj=db) print(result.exit_code) @@ -784,6 +1333,42 @@ def test_connect_success(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_connect_with_custom_escape_char(self): + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes", "default_escape_char": "d"}) + + result = runner.invoke(consutil.consutil.commands["connect"], ['1'], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + assert result.output == "Successful connection to line [1]\nPress ^D ^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_connect_default_escape_after_clear(self): + runner = CliRunner() + db = Db() + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) + # Set CONSOLE_SWITCH with default_escape_char, and clear it later + db.cfgdb.set_entry("CONSOLE_SWITCH", "console_mgmt", {"enabled": "yes", "default_escape_char": "d"}) + + runner.invoke(config.config.commands["console"].commands["default_escape"], ["clear"], obj=db) + + result = runner.invoke(consutil.consutil.commands["connect"], ['1'], obj=db) + print(result.exit_code) + print(sys.stderr, result.output) + assert result.exit_code == 0 + assert result.output == "Successful connection to line [1]\nPress ^A ^X to disconnect\n" + + class TestConsutilClear(object): @classmethod def setup_class(cls): @@ -822,7 +1407,7 @@ def test_clear_line_not_found(self): def test_clear_idle(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) result = runner.invoke(consutil.consutil.commands["clear"], ['1'], obj=db) print(result.exit_code) @@ -837,7 +1422,7 @@ def test_clear_idle(self): def test_clear_success(self): runner = CliRunner() db = Db() - db.cfgdb.set_entry("CONSOLE_PORT", 1, { "remote_device" : "switch1", "baud_rate" : "9600" }) + db.cfgdb.set_entry("CONSOLE_PORT", 1, {"remote_device": "switch1", "baud_rate": "9600"}) result = runner.invoke(consutil.consutil.commands["clear"], ['1'], obj=db) print(result.exit_code)