diff options
Diffstat (limited to 'util/ec3po/interpreter_unittest.py')
-rwxr-xr-x | util/ec3po/interpreter_unittest.py | 380 |
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() |