text: #!/usr/bin/python2
  # Copyright 2015 The Chromium OS Authors. All rights reserved.
  # Use of this source code is governed by a BSD-style license that can be
  # found in the LICENSE file.
  
  """Unit tests for the EC-3PO interpreter."""
  
  from __future__ import print_function
  
  # pylint: disable=cros-logging-import
  import logging
  import mock
  import tempfile
  import unittest
  
  import interpreter
  import threadproc_shim
  
  class TestEnhancedECBehaviour(unittest.TestCase):
    """Test case to verify all enhanced EC interpretation tasks."""
    def setUp(self):
      """Setup the test harness."""
      # Setup logging with a timestamp, the module, and the log level.
      logging.basicConfig(level=logging.DEBUG,
                          format=('%(asctime)s - %(module)s -'
                                  ' %(levelname)s - %(message)s'))
  
      # Create a tempfile that would represent the EC UART PTY.
      self.tempfile = tempfile.NamedTemporaryFile()
  
      # Create the pipes that the interpreter will use.
      self.cmd_pipe_user, self.cmd_pipe_itpr = threadproc_shim.Pipe()
      self.dbg_pipe_user, self.dbg_pipe_itpr = threadproc_shim.Pipe(duplex=False)
  
      # Mock the open() function so we can inspect reads/writes to the EC.
      self.ec_uart_pty = mock.mock_open()
      with mock.patch('__builtin__.open', self.ec_uart_pty):
        # Create an interpreter.
        self.itpr = interpreter.Interpreter(self.tempfile.name,
                                            self.cmd_pipe_itpr,
                                            self.dbg_pipe_itpr,
                                            log_level=logging.DEBUG,
                                            name="EC")
  
    @mock.patch('interpreter.os')
    def test_HandlingCommandsThatProduceNoOutput(self, mock_os):
      """Verify that the Interpreter correctly handles non-output commands.
  
      Args:
        mock_os: MagicMock object replacing the 'os' module for this test
          case.
      """
      # The interpreter init should open the EC UART PTY.
      expected_ec_calls = [mock.call(self.tempfile.name, 'a+')]
      # Have a command come in the command pipe.  The first command will be an
      # interrogation to determine if the EC is enhanced or not.
      self.cmd_pipe_user.send(interpreter.EC_SYN)
      self.itpr.HandleUserData()
      # At this point, the command should be queued up waiting to be sent, so
      # let's actually send it to the EC.
      self.itpr.SendCmdToEC()
      expected_ec_calls.extend([mock.call().write(interpreter.EC_SYN),
                                mock.call().flush()])
      # Now, assume that the EC sends only 1 response back of EC_ACK.
      mock_os.read.side_effect = [interpreter.EC_ACK]
      # When reading the EC, the interpreter will call file.fileno() to pass to
      # os.read().
      expected_ec_calls.append(mock.call().fileno())
      # Simulate the response.
      self.itpr.HandleECData()
  
      # Now that the interrogation was complete, it's time to send down the real
      # command.
      test_cmd = 'chan save'
      # Send the test command down the pipe.
      self.cmd_pipe_user.send(test_cmd)
      self.itpr.HandleUserData()
      self.itpr.SendCmdToEC()
      # Since the EC image is enhanced, we should have sent a packed command.
      expected_ec_calls.append(mock.call().write(self.itpr.PackCommand(test_cmd)))
      expected_ec_calls.append(mock.call().flush())
  
      # Now that the first command was sent, we should send another command which
      # produces no output.  The console would send another interrogation.
      self.cmd_pipe_user.send(interpreter.EC_SYN)
      self.itpr.HandleUserData()
      self.itpr.SendCmdToEC()
      expected_ec_calls.extend([mock.call().write(interpreter.EC_SYN),
                                mock.call().flush()])
      # Again, assume that the EC sends only 1 response back of EC_ACK.
      mock_os.read.side_effect = [interpreter.EC_ACK]
      # When reading the EC, the interpreter will call file.fileno() to pass to
      # os.read().
      expected_ec_calls.append(mock.call().fileno())
      # Simulate the response.
      self.itpr.HandleECData()
  
      # Now send the second test command.
      test_cmd = 'chan 0'
      self.cmd_pipe_user.send(test_cmd)
      self.itpr.HandleUserData()
      self.itpr.SendCmdToEC()
      # Since the EC image is enhanced, we should have sent a packed command.
      expected_ec_calls.append(mock.call().write(self.itpr.PackCommand(test_cmd)))
      expected_ec_calls.append(mock.call().flush())
  
      # Finally, verify that the appropriate writes were actually sent to the EC.
      self.ec_uart_pty.assert_has_calls(expected_ec_calls)
  
    @mock.patch('interpreter.os')
    def test_CommandRetryingOnError(self, mock_os):
      """Verify that commands are retried if an error is encountered.
  
      Args:
        mock_os: MagicMock object replacing the 'os' module for this test
          case.
      """
      # The interpreter init should open the EC UART PTY.
      expected_ec_calls = [mock.call(self.tempfile.name, 'a+')]
      # Have a command come in the command pipe.  The first command will be an
      # interrogation to determine if the EC is enhanced or not.
      self.cmd_pipe_user.send(interpreter.EC_SYN)
      self.itpr.HandleUserData()
      # At this point, the command should be queued up waiting to be sent, so
      # let's actually send it to the EC.
      self.itpr.SendCmdToEC()
      expected_ec_calls.extend([mock.call().write(interpreter.EC_SYN),
                                mock.call().flush()])
      # Now, assume that the EC sends only 1 response back of EC_ACK.
      mock_os.read.side_effect = [interpreter.EC_ACK]
      # When reading the EC, the interpreter will call file.fileno() to pass to
      # os.read().
      expected_ec_calls.append(mock.call().fileno())
      # Simulate the response.
      self.itpr.HandleECData()
  
      # Let's send a command that is received on the EC-side with an error.
      test_cmd = 'accelinfo'
      self.cmd_pipe_user.send(test_cmd)
      self.itpr.HandleUserData()
      self.itpr.SendCmdToEC()
      packed_cmd = self.itpr.PackCommand(test_cmd)
      expected_ec_calls.extend([mock.call().write(packed_cmd),
                                mock.call().flush()])
      # Have the EC return the error string twice.
      mock_os.read.side_effect = ['&&EE', '&&EE']
      for i in range(2):
        # When reading the EC, the interpreter will call file.fileno() to pass to
        # os.read().
        expected_ec_calls.append(mock.call().fileno())
        # Simulate the response.
        self.itpr.HandleECData()
  
        # Since an error was received, the EC should attempt to retry the command.
        expected_ec_calls.extend([mock.call().write(packed_cmd),
                                  mock.call().flush()])
        # Verify that the retry count was decremented.
        self.assertEqual(interpreter.COMMAND_RETRIES-i-1, self.itpr.cmd_retries,
                         'Unexpected cmd_remaining count.')
        # Actually retry the command.
        self.itpr.SendCmdToEC()
  
      # Now assume that the last one goes through with no trouble.
      expected_ec_calls.extend([mock.call().write(packed_cmd),
                                mock.call().flush()])
      self.itpr.SendCmdToEC()
  
      # Verify all the calls.
      self.ec_uart_pty.assert_has_calls(expected_ec_calls)
  
    def test_PackCommandsForEnhancedEC(self):
      """Verify that the interpreter packs commands for enhanced EC images."""
      # Assume current EC image is enhanced.
      self.itpr.enhanced_ec = True
      # Receive a command from the user.
      test_cmd = 'gettime'
      self.cmd_pipe_user.send(test_cmd)
      # Mock out PackCommand to see if it was called.
      self.itpr.PackCommand = mock.MagicMock()
      # Have the interpreter handle the command.
      self.itpr.HandleUserData()
      # Verify that PackCommand() was called.
      self.itpr.PackCommand.assert_called_once_with(test_cmd)
  
    def test_DontPackCommandsForNonEnhancedEC(self):
      """Verify the interpreter doesn't pack commands for non-enhanced images."""
      # Assume current EC image is not enhanced.
      self.itpr.enhanced_ec = False
      # Receive a command from the user.
      test_cmd = 'gettime'
      self.cmd_pipe_user.send(test_cmd)
      # Mock out PackCommand to see if it was called.
      self.itpr.PackCommand = mock.MagicMock()
      # Have the interpreter handle the command.
      self.itpr.HandleUserData()
      # Verify that PackCommand() was called.
      self.itpr.PackCommand.assert_not_called()
  
    @mock.patch('interpreter.os')
    def test_KeepingTrackOfInterrogation(self, mock_os):
      """Verify that the interpreter can track the state of the interrogation.
  
      Args:
        mock_os: MagicMock object replacing the 'os' module. for this test
          case.
      """
      # Upon init, the interpreter should assume that the current EC image is not
      # enhanced.
      self.assertFalse(self.itpr.enhanced_ec, msg=('State of enhanced_ec upon'
                                                   ' init is not False.'))
  
      # Assume an interrogation request comes in from the user.
      self.cmd_pipe_user.send(interpreter.EC_SYN)
      self.itpr.HandleUserData()
  
      # Verify the state is now within an interrogation.
      self.assertTrue(self.itpr.interrogating, 'interrogating should be True')
      # The state of enhanced_ec should not be changed yet because we haven't
      # received a valid response yet.
      self.assertFalse(self.itpr.enhanced_ec, msg=('State of enhanced_ec is '
                                                   'not False.'))
  
      # Assume that the EC responds with an EC_ACK.
      mock_os.read.side_effect = [interpreter.EC_ACK]
      self.itpr.HandleECData()
  
      # Now, the interrogation should be complete and we should know that the
      # current EC image is enhanced.
      self.assertFalse(self.itpr.interrogating, msg=('interrogating should be '
                                                     'False'))
      self.assertTrue(self.itpr.enhanced_ec, msg='enhanced_ec sholud be True')
  
      # Now let's perform another interrogation, but pretend that the EC ignores
      # it.
      self.cmd_pipe_user.send(interpreter.EC_SYN)
      self.itpr.HandleUserData()
  
      # Verify interrogating state.
      self.assertTrue(self.itpr.interrogating, 'interrogating sholud be True')
      # We should assume that the image is not enhanced until we get the valid
      # response.
      self.assertFalse(self.itpr.enhanced_ec, 'enhanced_ec should be False now.')
  
      # Let's pretend that we get a random debug print.  This should clear the
      # interrogating flag.
      mock_os.read.side_effect = '[1660.593076 HC 0x103]'
      self.itpr.HandleECData()
  
      # Verify that interrogating flag is cleared and enhanced_ec is still False.
      self.assertFalse(self.itpr.interrogating, 'interrogating should be False.')
      self.assertFalse(self.itpr.enhanced_ec,
                       'enhanced_ec should still be False.')
  
  
  class TestUARTDisconnection(unittest.TestCase):
    """Test case to verify interpreter disconnection/reconnection."""
    def setUp(self):
      """Setup the test harness."""
      # Setup logging with a timestamp, the module, and the log level.
      logging.basicConfig(level=logging.DEBUG,
                          format=('%(asctime)s - %(module)s -'
                                  ' %(levelname)s - %(message)s'))
  
      # Create a tempfile that would represent the EC UART PTY.
      self.tempfile = tempfile.NamedTemporaryFile()
  
      # Create the pipes that the interpreter will use.
      self.cmd_pipe_user, self.cmd_pipe_itpr = threadproc_shim.Pipe()
      self.dbg_pipe_user, self.dbg_pipe_itpr = threadproc_shim.Pipe(duplex=False)
  
      # Mock the open() function so we can inspect reads/writes to the EC.
      self.ec_uart_pty = mock.mock_open()
      with mock.patch('__builtin__.open', self.ec_uart_pty):
        # Create an interpreter.
        self.itpr = interpreter.Interpreter(self.tempfile.name,
                                            self.cmd_pipe_itpr,
                                            self.dbg_pipe_itpr,
                                            log_level=logging.DEBUG,
                                            name="EC")
  
      # First, check that interpreter is initialized to connected.
      self.assertTrue(self.itpr.connected, ('The interpreter should be'
                                            ' initialized in a connected state'))
  
    def test_DisconnectStopsECTraffic(self):
      """Verify that when in disconnected state, no debug prints are sent."""
      # Let's send a disconnect command through the command pipe.
      self.cmd_pipe_user.send('disconnect')
      self.itpr.HandleUserData()
  
      # Verify interpreter is disconnected from EC.
      self.assertFalse(self.itpr.connected, ('The interpreter should be'
                                             'disconnected.'))
      # Verify that the EC UART is no longer a member of the inputs.  The
      # interpreter will never pull data from the EC if it's not a member of the
      # inputs list.
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.inputs)
  
    def test_CommandsDroppedWhenDisconnected(self):
      """Verify that when in disconnected state, commands are dropped."""
      # Send a command, followed by 'disconnect'.
      self.cmd_pipe_user.send('taskinfo')
      self.itpr.HandleUserData()
      self.cmd_pipe_user.send('disconnect')
      self.itpr.HandleUserData()
  
      # Verify interpreter is disconnected from EC.
      self.assertFalse(self.itpr.connected, ('The interpreter should be'
                                             'disconnected.'))
      # Verify that the EC UART is no longer a member of the inputs nor outputs.
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.inputs)
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.outputs)
  
      # Have the user send a few more commands in the disconnected state.
      command = 'help\n'
      for char in command:
        self.cmd_pipe_user.send(char)
        self.itpr.HandleUserData()
  
      # The command queue should be empty.
      self.assertEqual(0, self.itpr.ec_cmd_queue.qsize())
  
      # Now send the reconnect command.
      self.cmd_pipe_user.send('reconnect')
      with mock.patch('__builtin__.open', mock.mock_open()):
        self.itpr.HandleUserData()
  
      # Verify interpreter is connected.
      self.assertTrue(self.itpr.connected)
      # Verify that EC UART is a member of the inputs.
      self.assertTrue(self.itpr.ec_uart_pty in self.itpr.inputs)
      # Since no command was sent after reconnection, verify that the EC UART is
      # not a member of the outputs.
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.outputs)
  
    def test_ReconnectAllowsECTraffic(self):
      """Verify that when connected, EC UART traffic is allowed."""
      # Let's send a disconnect command through the command pipe.
      self.cmd_pipe_user.send('disconnect')
      self.itpr.HandleUserData()
  
      # Verify interpreter is disconnected.
      self.assertFalse(self.itpr.connected, ('The interpreter should be'
                                             'disconnected.'))
      # Verify that the EC UART is no longer a member of the inputs nor outputs.
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.inputs)
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.outputs)
  
      # Issue reconnect command through the command pipe.
      self.cmd_pipe_user.send('reconnect')
      with mock.patch('__builtin__.open', mock.mock_open()):
        self.itpr.HandleUserData()
  
      # Verify interpreter is connected.
      self.assertTrue(self.itpr.connected, ('The interpreter should be'
                                            'connected.'))
      # Verify that the EC UART is now a member of the inputs.
      self.assertTrue(self.itpr.ec_uart_pty in self.itpr.inputs)
      # Since we have issued no commands during the disconnected state, no
      # commands are pending and therefore the PTY should not be added to the
      # outputs.
      self.assertFalse(self.itpr.ec_uart_pty in self.itpr.outputs)
  
  
  if __name__ == '__main__':
    unittest.main()
  