[consutil][connect] Remove root need from connect line command (sonic…

…-net#1225)

* Remove root check for consutil connect
* Add unit tests

Signed-off-by: Jing Kan jika@microsoft.com

consutil/lib.py

@@ -158,7 +158,7 @@ def connect(self):
 
  # build and start picocom command
  flow_cmd = "h" if self.flow_control else "n"
- cmd = "sudo picocom -b {} -f {} {}{}".format(self.baud, flow_cmd, SysInfoProvider.DEVICE_PREFIX, self.line_num)
+ cmd = "picocom -b {} -f {} {}{}".format(self.baud, flow_cmd, SysInfoProvider.DEVICE_PREFIX, self.line_num)
 
  # start connection
  try:
@@ -228,10 +228,11 @@ def _update_state(self, state, pid, date, line_num=None):
  state_db.set(state_db.STATE_DB, line_key, STATE_KEY, state)
  state_db.set(state_db.STATE_DB, line_key, PID_KEY, pid)
  state_db.set(state_db.STATE_DB, line_key, START_TIME_KEY, date)
- self._info[CUR_STATE_KEY] = {} if CUR_STATE_KEY not in self._info else self._info[CUR_STATE_KEY]
- self._info[CUR_STATE_KEY][STATE_KEY] = state
- self._info[CUR_STATE_KEY][PID_KEY] = pid
- self._info[CUR_STATE_KEY][START_TIME_KEY] = date
+ self._info[CUR_STATE_KEY] = {
+ STATE_KEY: state,
+ PID_KEY: pid,
+ START_TIME_KEY: date
+ }
 
 class ConsoleSession(object):
  """

consutil/main.py

@@ -74,10 +74,6 @@ def clear(db, target, devicename):
 @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"""
- if os.geteuid() != 0:
- click.echo("Root privileges are required for this operation")
- sys.exit(ERR_CMD)
-
  # identify the target line
  port_provider = ConsolePortProvider(db, configured_only=False)
  try:
@@ -92,7 +88,7 @@ def connect(db, target, devicename):
  try:
  session = target_port.connect()
  except LineBusyError:
- click.echo("Cannot connect: line {} is busy".format(line_num))
+ click.echo("Cannot connect: line [{}] is busy".format(line_num))
  sys.exit(ERR_BUSY)
  except InvalidConfigurationError as cfg_err:
  click.echo("Cannot connect: {}".format(cfg_err.message))
@@ -102,7 +98,7 @@ 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 ^A ^X to disconnect".format(line_num))
  session.interact()
 
 if __name__ == '__main__':

tests/console_test.py

@@ -495,3 +495,86 @@ def test_show(self):
  print(sys.stderr, result.output)
  assert result.exit_code == 0
  assert result.output == TestConsutilShow.expect_show_output
+
+class TestConsutilConnect(object):
+ @classmethod
+ def setup_class(cls):
+ print("SETUP")
+
+ @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))
+ def test_connect_target_nonexists(self):
+ runner = CliRunner()
+ db = Db()
+ 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)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 3
+ assert result.output == "Cannot connect: target [2] does not exist\n"
+
+ result = runner.invoke(consutil.consutil.commands["connect"], ['--devicename', 'switch2'], obj=db)
+ print(result.exit_code)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 3
+ assert result.output == "Cannot connect: target [switch2] does not exist\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(side_effect=LineBusyError()))
+ def test_connect_line_busy(self):
+ runner = CliRunner()
+ db = Db()
+ 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)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 5
+ assert result.output == "Cannot connect: line [1] is busy\n"
+
+ result = runner.invoke(consutil.consutil.commands["connect"], ['--devicename', 'switch1'], obj=db)
+ print(result.exit_code)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 5
+ assert result.output == "Cannot connect: line [1] is busy\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))
+ def test_connect_no_baud(self):
+ runner = CliRunner()
+ db = Db()
+
+ result = runner.invoke(consutil.consutil.commands["connect"], ['1'], obj=db)
+ print(result.exit_code)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 4
+ assert result.output == "Cannot connect: line [1] has no baud rate\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(side_effect=ConnectionFailedError()))
+ def test_connect_picocom_err(self):
+ runner = CliRunner()
+ db = Db()
+
+ result = runner.invoke(consutil.consutil.commands["connect"], ['1'], obj=db)
+ print(result.exit_code)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 3
+ assert result.output == "Cannot connect: unable to open picocom process\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_success(self):
+ runner = CliRunner()
+ db = Db()
+ 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)
+ print(sys.stderr, result.output)
+ assert result.exit_code == 0
+ assert result.output == "Successful connection to line [1]\nPress ^A ^X to disconnect\n"