ec3po: Add compatibility for older EC images.

The EC-3PO console and interpreter could be used to talk to EC images
which do not have the necessary changes to support the new enhancements.
If this was the case, the interpreter would be very confused and the
user wouldn't be able to use the console.  This commit adds
compatibility support for talking to both non-enhanced and enhanced EC
images.

When the console and interpreter are instantiated, they assume by
default that the EC image they are talking to is non-enhanced.  When the
user presses the carriage return key, the console initiates an
interrogation with the EC image.  The interrogation is a simple
EC_SYN(0xEC) and waits EC_INTERROGATION_TIMEOUT for the correct
EC_ACK(0xC0).  Enhanced EC images will try to reply immediately to a
EC_SYN. Non-enhanced EC images will just ignore the EC_SYN as it's not a
printable character.  Once the interrogation is complete, the console
will either simply pass everything forwards to the EC or provide the
console interface itself.

BUG=chrome-os-partner:46063
BRANCH=None
TEST=Enabled CONFIG_EXPERIMENTAL_CONSOLE on GLaDOS.  Entered some
commands and verified console was working.  Disabled
CONFIG_EXPERIMENTAL_CONSOLE on GLaDOS, reflashed, and verified console
was still working without restarting the EC-3PO console.
TEST=./util/ec3po/console_unittest.py -b
TEST=./util/ec3po/interpreter_unittest.py -b
TEST=cros lint --debug util/ec3po/console.py
TEST=cros lint --debug util/ec3po/console_unittest.py
TEST=cros lint --debug util/ec3po/interpreter.py
TEST=cros lint --debug util/ec3po/interpreter_unittest.py

Change-Id: I4f472afbdd7e898bee308c239b68ace0f4049842
Signed-off-by: Aseda Aboagye <aaboagye@google.com>
Reviewed-on: https://chromium-review.googlesource.com/313002
Commit-Ready: Aseda Aboagye <aaboagye@chromium.org>
Tested-by: Aseda Aboagye <aaboagye@chromium.org>
Reviewed-by: Randall Spangler <rspangler@chromium.org>

1 files changed, 252 insertions, 0 deletions

diff --git a/util/ec3po/interpreter_unittest.py b/util/ec3po/interpreter_unittest.py
new file mode 100755
index 0000000000..46bbcf8e93
--- /dev/null
+++ b/util/ec3po/interpreter_unittest.py
@@ -0,0 +1,252 @@
+#!/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()