path: root/util/ec3po/interpreter_unittest.py

#!/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
from chromite.lib import cros_logging as logging
import mock
import multiprocessing
import tempfile
import unittest

import interpreter

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 = multiprocessing.Pipe()
    self.dbg_pipe_user, self.dbg_pipe_itpr = multiprocessing.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)

  @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.')



if __name__ == '__main__':
  unittest.main()