diff --git a/rcli/__init__.py b/rcli/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000 diff --git a/rcli/linecard.py b/rcli/linecard.py deleted file mode 100644 index fdc6882ed16..00000000000 --- a/rcli/linecard.py +++ /dev/null @@ -1,151 +0,0 @@ -import click -import os -import paramiko -import sys -import select -import socket -import sys -import termios -import tty - -from .utils import get_linecard_ip -from paramiko.py3compat import u -from paramiko import Channel - -EMPTY_OUTPUTS = ['', '\x1b[?2004l\r'] - -class Linecard: - - def __init__(self, linecard_name, username, password): - """ - Initialize Linecard object and store credentials, connection, and channel - - :param linecard_name: The name of the linecard you want to connect to - :param username: The username to use to connect to the linecard - :param password: The linecard password. If password not provided, it - will prompt the user for it - :param use_ssh_keys: Whether or not to use SSH keys to authenticate. - """ - self.ip = get_linecard_ip(linecard_name) - - if not self.ip: - sys.exit(1) - - self.linecard_name = linecard_name - self.username = username - self.password = password - - self.connection = self._connect() - - - def _connect(self): - connection = paramiko.SSHClient() - # if ip address not in known_hosts, ignore known_hosts error - connection.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - try: - connection.connect(self.ip, username=self.username, password=self.password) - except paramiko.ssh_exception.NoValidConnectionsError as e: - connection = None - click.echo(e) - return connection - - def _get_password(self): - """ - Prompts the user for a password, and returns the password - - :param username: The username that we want to get the password for - :type username: str - :return: The password for the username. - """ - - return getpass( - "Password for username '{}': ".format(self.username), - # Pass in click stdout stream - this is similar to using click.echo - stream=click.get_text_stream('stdout') - ) - - def _set_tty_params(self): - tty.setraw(sys.stdin.fileno()) - tty.setcbreak(sys.stdin.fileno()) - - def _is_data_to_read(self, read): - if self.channel in read: - return True - return False - - def _is_data_to_write(self, read): - if sys.stdin in read: - return True - return False - - def _write_to_terminal(self, data): - # Write channel output to terminal - sys.stdout.write(data) - sys.stdout.flush() - - def _start_interactive_shell(self): - oldtty = termios.tcgetattr(sys.stdin) - try: - self._set_tty_params() - self.channel.settimeout(0.0) - - while True: - #Continuously wait for commands and execute them - read, write, ex = select.select([self.channel, sys.stdin], [], []) - if self._is_data_to_read(read): - try: - # Get output from channel - x = u(self.channel.recv(1024)) - if len(x) == 0: - # logout message will be displayed - break - self._write_to_terminal(x) - except socket.timeout as e: - click.echo("Connection timed out") - break - if self._is_data_to_write(read): - # If we are able to send input, get the input from stdin - x = sys.stdin.read(1) - if len(x) == 0: - break - # Send the input to the channel - self.channel.send(x) - finally: - # Now that the channel has been exited, return to the previously-saved old tty - termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty) - pass - - - def start_shell(self) -> None: - """ - Opens a session, gets a pseudo-terminal, invokes a shell, and then - attaches the host shell to the remote shell. - """ - # Create shell session - self.channel = self.connection.get_transport().open_session() - self.channel.get_pty() - self.channel.invoke_shell() - # Use Paramiko Interactive script to connect to the shell - self._start_interactive_shell() - # After user exits interactive shell, close the connection - self.connection.close() - - - def execute_cmd(self, command) -> str: - """ - Takes a command as an argument, executes it on the remote shell, and returns the output - - :param command: The command to execute on the remote shell - :return: The output of the command. - """ - # Execute the command and gather errors and output - _, stdout, stderr = self.connection.exec_command(command + "\n") - output = stdout.read().decode('utf-8') - - if stderr: - # Error was present, add message to output - output += stderr.read().decode('utf-8') - - # Close connection and return output - self.connection.close() - return output diff --git a/rcli/rexec.py b/rcli/rexec.py deleted file mode 100644 index fb56df83511..00000000000 --- a/rcli/rexec.py +++ /dev/null @@ -1,44 +0,0 @@ -import os -import click -import paramiko -import sys - -from .linecard import Linecard -from rcli import utils as rcli_utils -from sonic_py_common import device_info - -@click.command() -@click.argument('linecard_names', nargs=-1, type=str, required=True) -@click.option('-c', '--command', type=str, required=True) -def cli(linecard_names, command): - """ - Executes a command on one or many linecards - - :param linecard_names: A list of linecard names to execute the command on, - use `all` to execute on all linecards. - :param command: The command to execute on the linecard(s) - """ - if not device_info.is_chassis(): - click.echo("This commmand is only supported Chassis") - sys.exit(1) - - username = os.getlogin() - password = rcli_utils.get_password(username) - - if list(linecard_names) == ["all"]: - # Get all linecard names using autocompletion helper - linecard_names = rcli_utils.get_all_linecards(None, None, "") - - # Iterate through each linecard, execute command, and gather output - for linecard_name in linecard_names: - try: - lc = Linecard(linecard_name, username, password) - if lc.connection: - # If connection was created, connection exists. Otherwise, user will see an error message. - click.echo("======== {} output: ========".format(lc.linecard_name)) - click.echo(lc.execute_cmd(command)) - except paramiko.ssh_exception.AuthenticationException: - click.echo("Login failed on '{}' with username '{}'".format(linecard_name, lc.username)) - -if __name__=="__main__": - cli(prog_name='rexec') diff --git a/rcli/rshell.py b/rcli/rshell.py deleted file mode 100644 index decda6cd59b..00000000000 --- a/rcli/rshell.py +++ /dev/null @@ -1,38 +0,0 @@ -import os -import click -import paramiko -import sys - -from .linecard import Linecard -from sonic_py_common import device_info -from rcli import utils as rcli_utils - - -@click.command() -@click.argument('linecard_name', type=str, autocompletion=rcli_utils.get_all_linecards) -def cli(linecard_name): - """ - Open interactive shell for one linecard - - :param linecard_name: The name of the linecard to connect to - """ - if not device_info.is_chassis(): - click.echo("This commmand is only supported Chassis") - sys.exit(1) - - username = os.getlogin() - password = rcli_utils.get_password(username) - - try: - lc =Linecard(linecard_name, username, password) - if lc.connection: - click.echo("Connecting to {}".format(lc.linecard_name)) - # If connection was created, connection exists. Otherwise, user will see an error message. - lc.start_shell() - click.echo("Connection Closed") - except paramiko.ssh_exception.AuthenticationException: - click.echo("Login failed on '{}' with username '{}'".format(linecard_name, lc.username)) - - -if __name__=="__main__": - cli(prog_name='rshell') diff --git a/rcli/utils.py b/rcli/utils.py deleted file mode 100644 index 933043d0698..00000000000 --- a/rcli/utils.py +++ /dev/null @@ -1,149 +0,0 @@ -import click -from getpass import getpass -import os -import sys - -from swsscommon.swsscommon import SonicV2Connector - -CHASSIS_MODULE_INFO_TABLE = 'CHASSIS_MODULE_TABLE' -CHASSIS_MODULE_INFO_KEY_TEMPLATE = 'CHASSIS_MODULE {}' -CHASSIS_MODULE_INFO_DESC_FIELD = 'desc' -CHASSIS_MODULE_INFO_SLOT_FIELD = 'slot' -CHASSIS_MODULE_INFO_OPERSTATUS_FIELD = 'oper_status' -CHASSIS_MODULE_INFO_ADMINSTATUS_FIELD = 'admin_status' - -CHASSIS_MIDPLANE_INFO_TABLE = 'CHASSIS_MIDPLANE_TABLE' -CHASSIS_MIDPLANE_INFO_IP_FIELD = 'ip_address' -CHASSIS_MIDPLANE_INFO_ACCESS_FIELD = 'access' - -CHASSIS_MODULE_HOSTNAME_TABLE = 'CHASSIS_MODULE_HOSTNAME_TABLE' -CHASSIS_MODULE_HOSTNAME = 'module_hostname' - -def connect_to_chassis_state_db(): - chassis_state_db = SonicV2Connector(host="127.0.0.1") - chassis_state_db.connect(chassis_state_db.CHASSIS_STATE_DB) - return chassis_state_db - - -def connect_state_db(): - state_db = SonicV2Connector(host="127.0.0.1") - state_db.connect(state_db.STATE_DB) - return state_db - - - -def get_linecard_module_name_from_hostname(linecard_name: str): - - chassis_state_db = connect_to_chassis_state_db() - - keys = chassis_state_db.keys(chassis_state_db.CHASSIS_STATE_DB , '{}|{}'.format(CHASSIS_MODULE_HOSTNAME_TABLE, '*')) - for key in keys: - module_name = key.split('|')[1] - hostname = chassis_state_db.get(chassis_state_db.CHASSIS_STATE_DB, key, CHASSIS_MODULE_HOSTNAME) - if hostname.replace('-', '').lower() == linecard_name.replace('-', '').lower(): - return module_name - - return None - -def get_linecard_ip(linecard_name: str): - """ - Given a linecard name, lookup its IP address in the midplane table - - :param linecard_name: The name of the linecard you want to connect to - :type linecard_name: str - :return: IP address of the linecard - """ - # Adapted from `show chassis modules midplane-status` command logic: - # https://github.com/sonic-net/sonic-utilities/blob/master/show/chassis_modules.py - - # if the user passes linecard hostname, then try to get the module name for that linecard - module_name = get_linecard_module_name_from_hostname(linecard_name) - # if the module name cannot be found from host, assume the user has passed module name - if module_name is None: - module_name = linecard_name - module_ip, module_access = get_module_ip_and_access_from_state_db(module_name) - - if not module_ip: - click.echo('Linecard {} not found'.format(linecard_name)) - return None - - if module_access != 'True': - click.echo('Linecard {} not accessible'.format(linecard_name)) - return None - - - return module_ip - -def get_module_ip_and_access_from_state_db(module_name): - state_db = connect_state_db() - data_dict = state_db.get_all( - state_db.STATE_DB, '{}|{}'.format(CHASSIS_MIDPLANE_INFO_TABLE,module_name )) - if data_dict is None: - return None, None - - linecard_ip = data_dict.get(CHASSIS_MIDPLANE_INFO_IP_FIELD, None) - access = data_dict.get(CHASSIS_MIDPLANE_INFO_ACCESS_FIELD, None) - - return linecard_ip, access - - -def get_all_linecards(ctx, args, incomplete) -> list: - """ - Return a list of all accessible linecard names. This function is used to - autocomplete linecard names in the CLI. - - :param ctx: The Click context object that is passed to the command function - :param args: The arguments passed to the Click command - :param incomplete: The string that the user has typed so far - :return: A list of all accessible linecard names. - """ - # Adapted from `show chassis modules midplane-status` command logic: - # https://github.com/sonic-net/sonic-utilities/blob/master/show/chassis_modules.py - - - chassis_state_db = connect_to_chassis_state_db() - state_db = connect_state_db() - - linecards = [] - keys = state_db.keys(state_db.STATE_DB,'{}|*'.format(CHASSIS_MIDPLANE_INFO_TABLE)) - for key in keys: - key_list = key.split('|') - if len(key_list) != 2: # error data in DB, log it and ignore - click.echo('Warn: Invalid Key {} in {} table'.format(key, CHASSIS_MIDPLANE_INFO_TABLE )) - continue - module_name = key_list[1] - linecard_ip, access = get_module_ip_and_access_from_state_db(module_name) - if linecard_ip is None: - continue - - if access != "True" : - continue - - # get the hostname for this module - hostname = chassis_state_db.get(chassis_state_db.CHASSIS_STATE_DB, '{}|{}'.format(CHASSIS_MODULE_HOSTNAME_TABLE, module_name), CHASSIS_MODULE_HOSTNAME) - if hostname: - linecards.append(hostname) - else: - linecards.append(module_name) - - # Return a list of all matched linecards - return [lc for lc in linecards if incomplete in lc] - - -def get_password(username=None): - """ - Prompts the user for a password, and returns the password - - :param username: The username that we want to get the password for - :type username: str - :return: The password for the username. - """ - - if username is None: - username =os.getlogin() - - return getpass( - "Password for username '{}': ".format(username), - # Pass in click stdout stream - this is similar to using click.echo - stream=click.get_text_stream('stdout') - ) \ No newline at end of file diff --git a/setup.py b/setup.py index cdeddb8f42e..ea0e949ab99 100644 --- a/setup.py +++ b/setup.py @@ -74,7 +74,6 @@ 'pddf_thermalutil', 'pddf_ledutil', 'syslog_util', - 'rcli', 'show', 'show.interfaces', 'show.plugins', @@ -207,8 +206,6 @@ 'pddf_psuutil = pddf_psuutil.main:cli', 'pddf_thermalutil = pddf_thermalutil.main:cli', 'pddf_ledutil = pddf_ledutil.main:cli', - 'rexec = rcli.rexec:cli', - 'rshell = rcli.rshell:cli', 'show = show.main:cli', 'sonic-clear = clear.main:cli', 'sonic-installer = sonic_installer.main:sonic_installer', @@ -237,7 +234,6 @@ 'natsort>=6.2.1', # 6.2.1 is the last version which supports Python 2. Can update once we no longer support Python 2 'netaddr>=0.8.0', 'netifaces>=0.10.7', - 'paramiko==2.11.0', 'pexpect>=4.8.0', 'semantic-version>=2.8.5', 'prettyprinter>=0.18.0', diff --git a/sonic-utilities-data/bash_completion.d/rexec b/sonic-utilities-data/bash_completion.d/rexec deleted file mode 100644 index 1199fd06769..00000000000 --- a/sonic-utilities-data/bash_completion.d/rexec +++ /dev/null @@ -1,21 +0,0 @@ -_rexec_completion() { - local IFS=$' -' - COMPREPLY=( $( env COMP_WORDS="${COMP_WORDS[*]}" \ - COMP_CWORD=$COMP_CWORD \ - _REXEC_COMPLETE=complete $1 ) ) - return 0 -} - -_rexec_completionetup() { - local COMPLETION_OPTIONS="" - local BASH_VERSION_ARR=(${BASH_VERSION//./ }) - # Only BASH version 4.4 and later have the nosort option. - if [ ${BASH_VERSION_ARR[0]} -gt 4 ] || ([ ${BASH_VERSION_ARR[0]} -eq 4 ] && [ ${BASH_VERSION_ARR[1]} -ge 4 ]); then - COMPLETION_OPTIONS="-o nosort" - fi - - complete $COMPLETION_OPTIONS -F _rexec_completion rexec -} - -_rexec_completionetup; \ No newline at end of file diff --git a/sonic-utilities-data/bash_completion.d/rshell b/sonic-utilities-data/bash_completion.d/rshell deleted file mode 100644 index 012f754dd7e..00000000000 --- a/sonic-utilities-data/bash_completion.d/rshell +++ /dev/null @@ -1,21 +0,0 @@ -_rshell_completion() { - local IFS=$' -' - COMPREPLY=( $( env COMP_WORDS="${COMP_WORDS[*]}" \ - COMP_CWORD=$COMP_CWORD \ - _RSHELL_COMPLETE=complete $1 ) ) - return 0 -} - -_rshell_completionetup() { - local COMPLETION_OPTIONS="" - local BASH_VERSION_ARR=(${BASH_VERSION//./ }) - # Only BASH version 4.4 and later have the nosort option. - if [ ${BASH_VERSION_ARR[0]} -gt 4 ] || ([ ${BASH_VERSION_ARR[0]} -eq 4 ] && [ ${BASH_VERSION_ARR[1]} -ge 4 ]); then - COMPLETION_OPTIONS="-o nosort" - fi - - complete $COMPLETION_OPTIONS -F _rshell_completion rshell -} - -_rshell_completionetup; \ No newline at end of file diff --git a/tests/chassis_modules_test.py b/tests/chassis_modules_test.py index fa8cd608dd3..e6dbe569d2c 100644 --- a/tests/chassis_modules_test.py +++ b/tests/chassis_modules_test.py @@ -33,11 +33,11 @@ """ show_chassis_midplane_output="""\ - Name IP-Address Reachability ----------- ------------- -------------- -LINE-CARD0 192.168.1.100 True -LINE-CARD1 192.168.1.2 False -LINE-CARD2 192.168.1.1 True + Name IP-Address Reachability +----------- ------------- -------------- + LINE-CARD0 192.168.1.1 True + LINE-CARD1 192.168.1.2 False +SUPERVISOR0 192.168.1.100 True """ show_chassis_system_ports_output_asic0="""\ @@ -225,7 +225,7 @@ def test_midplane_show_all_count_lines(self): result = runner.invoke(show.cli.commands["chassis"].commands["modules"].commands["midplane-status"], []) print(result.output) result_lines = result.output.strip('\n').split('\n') - modules = ["LINE-CARD0", "LINE-CARD1", "LINE-CARD2"] + modules = ["LINE-CARD0", "LINE-CARD1", "SUPERVISOR0"] for i, module in enumerate(modules): assert module in result_lines[i + warning_lines + header_lines] assert len(result_lines) == warning_lines + header_lines + len(modules) diff --git a/tests/mock_tables/asic0/state_db.json b/tests/mock_tables/asic0/state_db.json index 6ae0258be05..559af048260 100644 --- a/tests/mock_tables/asic0/state_db.json +++ b/tests/mock_tables/asic0/state_db.json @@ -287,18 +287,6 @@ "REMOTE_MOD": "0", "REMOTE_PORT": "93" }, - "CHASSIS_MIDPLANE_TABLE|LINE-CARD0": { - "ip_address": "127.0.0.1", - "access": "True" - }, - "CHASSIS_MIDPLANE_TABLE|LINE-CARD1": { - "ip_address": "127.0.0.1", - "access": "True" - }, - "CHASSIS_MIDPLANE_TABLE|LINE-CARD2": { - "ip_address": "127.0.0.1", - "access": "False" - }, "ACL_TABLE_TABLE|DATAACL_5" : { "status": "Active" }, diff --git a/tests/mock_tables/chassis_state_db.json b/tests/mock_tables/chassis_state_db.json deleted file mode 100644 index 5178c49ca06..00000000000 --- a/tests/mock_tables/chassis_state_db.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "CHASSIS_MODULE_HOSTNAME_TABLE|LINE-CARD0": { - "module_hostname": "sonic-lc1" - }, - "CHASSIS_MODULE_HOSTNAME_TABLE|LINE-CARD1": { - "module_hostname": "sonic-lc2" - } - -} \ No newline at end of file diff --git a/tests/mock_tables/database_config.json b/tests/mock_tables/database_config.json index f55c0734c22..d12ba054146 100644 --- a/tests/mock_tables/database_config.json +++ b/tests/mock_tables/database_config.json @@ -56,11 +56,6 @@ "id" : 12, "separator": "|", "instance" : "redis" - }, - "CHASSIS_STATE_DB" : { - "id" : 13, - "separator": "|", - "instance" : "redis" } }, "VERSION" : "1.1" diff --git a/tests/mock_tables/state_db.json b/tests/mock_tables/state_db.json index 1d8f46297d5..cd1a194ba8e 100644 --- a/tests/mock_tables/state_db.json +++ b/tests/mock_tables/state_db.json @@ -935,11 +935,11 @@ "max_queues": "20", "max_priority_groups": "8" }, - "CHASSIS_MIDPLANE_TABLE|LINE-CARD0": { + "CHASSIS_MIDPLANE_TABLE|SUPERVISOR0": { "ip_address": "192.168.1.100", "access": "True" }, - "CHASSIS_MIDPLANE_TABLE|LINE-CARD2": { + "CHASSIS_MIDPLANE_TABLE|LINE-CARD0": { "ip_address": "192.168.1.1", "access": "True" }, diff --git a/tests/remote_cli_test.py b/tests/remote_cli_test.py deleted file mode 100644 index 67545dd1b31..00000000000 --- a/tests/remote_cli_test.py +++ /dev/null @@ -1,260 +0,0 @@ -import os -from click.testing import CliRunner -import paramiko -from rcli import rexec -from rcli import rshell -from rcli import linecard -from rcli import utils as rcli_utils -import sys -from io import BytesIO, StringIO -from unittest import mock -import select -import socket -import termios - -MULTI_LC_REXEC_OUTPUT = '''======== sonic-lc1 output: ======== -hello world -======== LINE-CARD2 output: ======== -hello world -''' -REXEC_HELP = '''Usage: cli [OPTIONS] LINECARD_NAMES... - - Executes a command on one or many linecards - - :param linecard_names: A list of linecard names to execute the command on, - use `all` to execute on all linecards. :param command: The command to - execute on the linecard(s) - -Options: - -c, --command TEXT [required] - --help Show this message and exit. -''' - -def mock_exec_command(): - - mock_stdout = BytesIO(b"""hello world""") - mock_stderr = BytesIO() - return '', mock_stdout, None - -def mock_exec_error_cmd(): - mock_stdout = BytesIO() - mock_stderr = BytesIO(b"""Command not found""") - return '', mock_stdout, mock_stderr - -def mock_connection_channel(): - c = mock.MagicMock(return_value="channel") - c.get_pty = mock.MagicMock(return_value='') - c.invoke_shell = mock.MagicMock() - c.recv = mock.MagicMock(side_effect=['abcd', '']) - return c - -def mock_connection_channel_with_timeout(): - c = mock.MagicMock(return_value="channel") - c.get_pty = mock.MagicMock(return_value='') - c.invoke_shell = mock.MagicMock() - c.recv = mock.MagicMock(side_effect=['abcd', socket.timeout(10, 'timeout')]) - return c - -def mock_paramiko_connection(channel): - # Create a mock to return for connection. - conn = mock.MagicMock() - #create a mock return for transport - t = mock.MagicMock() - t.open_session = mock.MagicMock(return_value=channel) - conn.get_transport = mock.MagicMock(return_value=t) - conn.connect = mock.MagicMock() - conn.close = mock.MagicMock() - return conn - -class TestRemoteExec(object): - @classmethod - def setup_class(cls): - print("SETUP") - from .mock_tables import dbconnector - dbconnector.load_database_config() - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - #@mock.patch.object(linecard.Linecard, '_get_password', mock.MagicMock(return_value='dummmy')) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(paramiko.SSHClient, 'exec_command', mock.MagicMock(return_value = mock_exec_command())) - def test_rexec_with_module_name(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "pwd"]) - print(result.output) - assert result.exit_code == 0, result.output - assert "hello world" in result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(paramiko.SSHClient, 'exec_command', mock.MagicMock(return_value = mock_exec_command())) - def test_rexec_with_hostname(self): - runner = CliRunner() - LINECARD_NAME = "sonic-lc1" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "pwd"]) - print(result.output) - assert result.exit_code == 0, result.output - assert "hello world" in result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(paramiko.SSHClient, 'exec_command', mock.MagicMock(return_value = mock_exec_error_cmd())) - def test_rexec_error_with_module_name(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "pwd"]) - print(result.output) - assert result.exit_code == 0, result.output - assert "Command not found" in result.output - - def test_rexec_error(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "show version"]) - print(result.output) - assert result.exit_code == 1, result.output - assert "This commmand is only supported Chassis" in result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(linecard.Linecard, 'execute_cmd', mock.MagicMock(return_value = "hello world")) - def test_rexec_all(self): - runner = CliRunner() - LINECARD_NAME = "all" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "show version"]) - print(result.output) - assert result.exit_code == 0, result.output - assert MULTI_LC_REXEC_OUTPUT == result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(linecard.Linecard, 'execute_cmd', mock.MagicMock(return_value = "hello world")) - def test_rexec_invalid_lc(self): - runner = CliRunner() - LINECARD_NAME = "sonic-lc-3" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "show version"]) - print(result.output) - assert result.exit_code == 1, result.output - assert "Linecard sonic-lc-3 not found\n" == result.output - - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(linecard.Linecard, 'execute_cmd', mock.MagicMock(return_value = "hello world")) - def test_rexec_unreachable_lc(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD1" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "show version"]) - print(result.output) - assert result.exit_code == 1, result.output - assert "Linecard LINE-CARD1 not accessible\n" == result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock()) - @mock.patch.object(linecard.Linecard, 'execute_cmd', mock.MagicMock(return_value = "hello world")) - def test_rexec_help(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD1" - result = runner.invoke(rexec.cli, ["--help"]) - print(result.output) - assert result.exit_code == 0, result.output - assert REXEC_HELP == result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(paramiko.SSHClient, 'connect', mock.MagicMock(side_effect=paramiko.ssh_exception.NoValidConnectionsError({('192.168.0.1', - 22): "None" }))) - @mock.patch.object(linecard.Linecard, 'execute_cmd', mock.MagicMock(return_value = "hello world")) - def test_rexec_exception(self): - runner = CliRunner() - LINECARD_NAME = "sonic-lc1" - result = runner.invoke(rexec.cli, [LINECARD_NAME, "-c", "show version"]) - print(result.output) - assert result.exit_code == 0, result.output - assert "[Errno None] Unable to connect to port 22 on 192.168.0.1\n" == result.output - - -class TestRemoteCLI(object): - @classmethod - def setup_class(cls): - print("SETUP") - from .mock_tables import dbconnector - dbconnector.load_database_config() - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(linecard.Linecard, '_set_tty_params', mock.MagicMock()) - @mock.patch.object(termios, 'tcsetattr', mock.MagicMock()) - @mock.patch.object(termios, 'tcgetattr', mock.MagicMock(return_value=[])) - def test_rcli_with_module_name(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - channel = mock_connection_channel() - - with mock.patch('paramiko.SSHClient', mock.MagicMock(return_value=mock_paramiko_connection(channel))), \ - mock.patch('select.select', mock.MagicMock(return_value=([channel], [], []))): - result = runner.invoke(rshell.cli, [LINECARD_NAME]) - print(result.output) - assert result.exit_code == 0, result.output - assert "abcd" in result.output - - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(linecard.Linecard, '_set_tty_params', mock.MagicMock()) - @mock.patch.object(termios, 'tcsetattr', mock.MagicMock()) - @mock.patch.object(termios, 'tcgetattr', mock.MagicMock(return_value=[])) - def test_rcli_with_module_name_2(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - channel = mock_connection_channel() - - with mock.patch('paramiko.SSHClient', mock.MagicMock(return_value=mock_paramiko_connection(channel))), \ - mock.patch('select.select', mock.MagicMock(side_effect=[([], [], []), ([channel], [], []),([channel], [], [])])): - result = runner.invoke(rshell.cli, [LINECARD_NAME]) - print(result.output) - assert result.exit_code == 0, result.output - assert "Connecting to LINE-CARD0" in result.output - - @mock.patch("sonic_py_common.device_info.is_chassis", mock.MagicMock(return_value=True)) - @mock.patch("os.getlogin", mock.MagicMock(return_value="admin")) - @mock.patch("rcli.utils.get_password", mock.MagicMock(return_value="dummy")) - @mock.patch.object(linecard.Linecard, '_set_tty_params', mock.MagicMock()) - @mock.patch.object(termios, 'tcsetattr', mock.MagicMock()) - @mock.patch.object(termios, 'tcgetattr', mock.MagicMock(return_value=[])) - def test_rcli_with_module_name_3(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - channel = mock_connection_channel_with_timeout() - - with mock.patch('paramiko.SSHClient', mock.MagicMock(return_value=mock_paramiko_connection(channel))), \ - mock.patch('select.select', mock.MagicMock(return_value=([channel], [], []))): - result = runner.invoke(rshell.cli, [LINECARD_NAME]) - print(result.output) - assert result.exit_code == 0, result.output - assert "Connecting to LINE-CARD0" in result.output - - def test_rcli_error(self): - runner = CliRunner() - LINECARD_NAME = "LINE-CARD0" - result = runner.invoke(rshell.cli, [LINECARD_NAME]) - print(result.output) - assert result.exit_code == 1, result.output - assert "This commmand is only supported Chassis" in result.output \ No newline at end of file