1 files changed, 0 insertions, 380 deletions
diff --git a/util/ec3po/interpreter_unittest.py b/util/ec3po/interpreter_unittest.py
deleted file mode 100755
index fe4d43c351..0000000000
--- a/util/ec3po/interpreter_unittest.py
+++ /dev/null
@@ -1,380 +0,0 @@
-#!/usr/bin/env python
-# 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."""
-
-# Note: This is a py2/3 compatible file.
-
-from __future__ import print_function
-
-import logging
-import mock
-import tempfile
-import unittest
-
-import six
-
-from ec3po import interpreter
-from ec3po import threadproc_shim
-
-
-def GetBuiltins(func):
-  if six.PY2:
-    return '__builtin__.' + func
-  return 'builtins.' + func
-
-
-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(GetBuiltins('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('ec3po.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, 'r+b', buffering=0)]
-    # 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 = b'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 = b'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('ec3po.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, 'r+b', buffering=0)]
-    # 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 = b'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 = [b'&&EE', b'&&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 = b'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 = b'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('ec3po.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 = [b'[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(GetBuiltins('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(b'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(b'taskinfo')
-    self.itpr.HandleUserData()
-    self.cmd_pipe_user.send(b'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.encode('utf-8'))
-      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(b'reconnect')
-
-    with mock.patch(GetBuiltins('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(b'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(b'reconnect')
-
-    with mock.patch(GetBuiltins('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()