\n" % stream.name)
+ self.f.write("\n")
+ self.f.write("
")
+ if implicit:
+ self.f.write("")
+ self.f.write(self._escape(data))
+ if implicit:
+ self.f.write("")
+ self.last_stream = stream
+
+ def flush(self):
+ '''Flush the log stream, to ensure correct log interleaving.
+
+ Args:
+ None.
+
+ Returns:
+ Nothing.
+ '''
+
+ self.f.flush()
diff --git a/test/py/pytest.ini b/test/py/pytest.ini
new file mode 100644
index 00000000000..67e514f4205
--- /dev/null
+++ b/test/py/pytest.ini
@@ -0,0 +1,11 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# Static configuration data for pytest. pytest reads this at startup time.
+
+[pytest]
+markers =
+ boardspec: U-Boot: Describes the set of boards a test can/can't run on.
+ buildconfigspec: U-Boot: Describes Kconfig/config-header constraints.
diff --git a/test/py/test.py b/test/py/test.py
new file mode 100755
index 00000000000..9c23898774e
--- /dev/null
+++ b/test/py/test.py
@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# Wrapper script to invoke pytest with the directory name that contains the
+# U-Boot tests.
+
+import os
+import os.path
+import sys
+
+# Get rid of argv[0]
+sys.argv.pop(0)
+
+# argv; py.test test_directory_name user-supplied-arguments
+args = ["py.test", os.path.dirname(__file__) + "/tests"]
+args.extend(sys.argv)
+
+try:
+ os.execvp("py.test", args)
+except:
+ # Log full details of any exception for detailed analysis
+ import traceback
+ traceback.print_exc()
+ # Hint to the user that they likely simply haven't installed the required
+ # dependencies.
+ print >>sys.stderr, """
+exec(py.test) failed; perhaps you are missing some dependencies?
+See test/py/README.md for the list."""
diff --git a/test/py/tests/test_000_version.py b/test/py/tests/test_000_version.py
new file mode 100644
index 00000000000..d262f0534ef
--- /dev/null
+++ b/test/py/tests/test_000_version.py
@@ -0,0 +1,20 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# pytest runs tests the order of their module path, which is related to the
+# filename containing the test. This file is named such that it is sorted
+# first, simply as a very basic sanity check of the functionality of the U-Boot
+# command prompt.
+
+def test_version(u_boot_console):
+ '''Test that the "version" command prints the U-Boot version.'''
+
+ # "version" prints the U-Boot sign-on message. This is usually considered
+ # an error, so that any unexpected reboot causes an error. Here, this
+ # error detection is disabled since the sign-on message is expected.
+ with u_boot_console.disable_check('main_signon'):
+ response = u_boot_console.run_command('version')
+ # Ensure "version" printed what we expected.
+ u_boot_console.validate_version_string_in_text(response)
diff --git a/test/py/tests/test_help.py b/test/py/tests/test_help.py
new file mode 100644
index 00000000000..894f3b5f170
--- /dev/null
+++ b/test/py/tests/test_help.py
@@ -0,0 +1,9 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+def test_help(u_boot_console):
+ '''Test that the "help" command can be executed.'''
+
+ u_boot_console.run_command('help')
diff --git a/test/py/tests/test_unknown_cmd.py b/test/py/tests/test_unknown_cmd.py
new file mode 100644
index 00000000000..2de93e0026f
--- /dev/null
+++ b/test/py/tests/test_unknown_cmd.py
@@ -0,0 +1,14 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+def test_unknown_command(u_boot_console):
+ '''Test that executing an unknown command causes U-Boot to print an
+ error.'''
+
+ # The "unknown command" error is actively expected here,
+ # so error detection for it is disabled.
+ with u_boot_console.disable_check('unknown_command'):
+ response = u_boot_console.run_command('non_existent_cmd')
+ assert('Unknown command \'non_existent_cmd\' - try \'help\'' in response)
diff --git a/test/py/u_boot_console_base.py b/test/py/u_boot_console_base.py
new file mode 100644
index 00000000000..520f9a9e9f3
--- /dev/null
+++ b/test/py/u_boot_console_base.py
@@ -0,0 +1,360 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# Common logic to interact with U-Boot via the console. This class provides
+# the interface that tests use to execute U-Boot shell commands and wait for
+# their results. Sub-classes exist to perform board-type-specific setup
+# operations, such as spawning a sub-process for Sandbox, or attaching to the
+# serial console of real hardware.
+
+import multiplexed_log
+import os
+import pytest
+import re
+import sys
+
+# Regexes for text we expect U-Boot to send to the console.
+pattern_u_boot_spl_signon = re.compile('(U-Boot SPL \\d{4}\\.\\d{2}-[^\r\n]*)')
+pattern_u_boot_main_signon = re.compile('(U-Boot \\d{4}\\.\\d{2}-[^\r\n]*)')
+pattern_stop_autoboot_prompt = re.compile('Hit any key to stop autoboot: ')
+pattern_unknown_command = re.compile('Unknown command \'.*\' - try \'help\'')
+pattern_error_notification = re.compile('## Error: ')
+
+class ConsoleDisableCheck(object):
+ '''Context manager (for Python's with statement) that temporarily disables
+ the specified console output error check. This is useful when deliberately
+ executing a command that is known to trigger one of the error checks, in
+ order to test that the error condition is actually raised. This class is
+ used internally by ConsoleBase::disable_check(); it is not intended for
+ direct usage.'''
+
+ def __init__(self, console, check_type):
+ self.console = console
+ self.check_type = check_type
+
+ def __enter__(self):
+ self.console.disable_check_count[self.check_type] += 1
+
+ def __exit__(self, extype, value, traceback):
+ self.console.disable_check_count[self.check_type] -= 1
+
+class ConsoleBase(object):
+ '''The interface through which test functions interact with the U-Boot
+ console. This primarily involves executing shell commands, capturing their
+ results, and checking for common error conditions. Some common utilities
+ are also provided too.'''
+
+ def __init__(self, log, config, max_fifo_fill):
+ '''Initialize a U-Boot console connection.
+
+ Can only usefully be called by sub-classes.
+
+ Args:
+ log: A mulptiplex_log.Logfile object, to which the U-Boot output
+ will be logged.
+ config: A configuration data structure, as built by conftest.py.
+ max_fifo_fill: The maximum number of characters to send to U-Boot
+ command-line before waiting for U-Boot to echo the characters
+ back. For UART-based HW without HW flow control, this value
+ should be set less than the UART RX FIFO size to avoid
+ overflow, assuming that U-Boot can't keep up with full-rate
+ traffic at the baud rate.
+
+ Returns:
+ Nothing.
+ '''
+
+ self.log = log
+ self.config = config
+ self.max_fifo_fill = max_fifo_fill
+
+ self.logstream = self.log.get_stream('console', sys.stdout)
+
+ # Array slice removes leading/trailing quotes
+ self.prompt = self.config.buildconfig['config_sys_prompt'][1:-1]
+ self.prompt_escaped = re.escape(self.prompt)
+ self.p = None
+ self.disable_check_count = {
+ 'spl_signon': 0,
+ 'main_signon': 0,
+ 'unknown_command': 0,
+ 'error_notification': 0,
+ }
+
+ self.at_prompt = False
+ self.at_prompt_logevt = None
+ self.ram_base = None
+
+ def close(self):
+ '''Terminate the connection to the U-Boot console.
+
+ This function is only useful once all interaction with U-Boot is
+ complete. Once this function is called, data cannot be sent to or
+ received from U-Boot.
+
+ Args:
+ None.
+
+ Returns:
+ Nothing.
+ '''
+
+ if self.p:
+ self.p.close()
+ self.logstream.close()
+
+ def run_command(self, cmd, wait_for_echo=True, send_nl=True,
+ wait_for_prompt=True):
+ '''Execute a command via the U-Boot console.
+
+ The command is always sent to U-Boot.
+
+ U-Boot echoes any command back to its output, and this function
+ typically waits for that to occur. The wait can be disabled by setting
+ wait_for_echo=False, which is useful e.g. when sending CTRL-C to
+ interrupt a long-running command such as "ums".
+
+ Command execution is typically triggered by sending a newline
+ character. This can be disabled by setting send_nl=False, which is
+ also useful when sending CTRL-C.
+
+ This function typically waits for the command to finish executing, and
+ returns the console output that it generated. This can be disabled by
+ setting wait_for_prompt=False, which is useful when invoking a long-
+ running command such as "ums".
+
+ Args:
+ cmd: The command to send.
+ wait_for_each: Boolean indicating whether to wait for U-Boot to
+ echo the command text back to its output.
+ send_nl: Boolean indicating whether to send a newline character
+ after the command string.
+ wait_for_prompt: Boolean indicating whether to wait for the
+ command prompt to be sent by U-Boot. This typically occurs
+ immediately after the command has been executed.
+
+ Returns:
+ If wait_for_prompt == False:
+ Nothing.
+ Else:
+ The output from U-Boot during command execution. In other
+ words, the text U-Boot emitted between the point it echod the
+ command string and emitted the subsequent command prompts.
+ '''
+
+ self.ensure_spawned()
+
+ if self.at_prompt and \
+ self.at_prompt_logevt != self.logstream.logfile.cur_evt:
+ self.logstream.write(self.prompt, implicit=True)
+
+ bad_patterns = []
+ bad_pattern_ids = []
+ if (self.disable_check_count['spl_signon'] == 0 and
+ self.u_boot_spl_signon):
+ bad_patterns.append(self.u_boot_spl_signon_escaped)
+ bad_pattern_ids.append('SPL signon')
+ if self.disable_check_count['main_signon'] == 0:
+ bad_patterns.append(self.u_boot_main_signon_escaped)
+ bad_pattern_ids.append('U-Boot main signon')
+ if self.disable_check_count['unknown_command'] == 0:
+ bad_patterns.append(pattern_unknown_command)
+ bad_pattern_ids.append('Unknown command')
+ if self.disable_check_count['error_notification'] == 0:
+ bad_patterns.append(pattern_error_notification)
+ bad_pattern_ids.append('Error notification')
+ try:
+ self.at_prompt = False
+ if send_nl:
+ cmd += '\n'
+ while cmd:
+ # Limit max outstanding data, so UART FIFOs don't overflow
+ chunk = cmd[:self.max_fifo_fill]
+ cmd = cmd[self.max_fifo_fill:]
+ self.p.send(chunk)
+ if not wait_for_echo:
+ continue
+ chunk = re.escape(chunk)
+ chunk = chunk.replace('\\\n', '[\r\n]')
+ m = self.p.expect([chunk] + bad_patterns)
+ if m != 0:
+ self.at_prompt = False
+ raise Exception('Bad pattern found on console: ' +
+ bad_pattern_ids[m - 1])
+ if not wait_for_prompt:
+ return
+ m = self.p.expect([self.prompt_escaped] + bad_patterns)
+ if m != 0:
+ self.at_prompt = False
+ raise Exception('Bad pattern found on console: ' +
+ bad_pattern_ids[m - 1])
+ self.at_prompt = True
+ self.at_prompt_logevt = self.logstream.logfile.cur_evt
+ # Only strip \r\n; space/TAB might be significant if testing
+ # indentation.
+ return self.p.before.strip('\r\n')
+ except Exception as ex:
+ self.log.error(str(ex))
+ self.cleanup_spawn()
+ raise
+
+ def ctrlc(self):
+ '''Send a CTRL-C character to U-Boot.
+
+ This is useful in order to stop execution of long-running synchronous
+ commands such as "ums".
+
+ Args:
+ None.
+
+ Returns:
+ Nothing.
+ '''
+
+ self.run_command(chr(3), wait_for_echo=False, send_nl=False)
+
+ def ensure_spawned(self):
+ '''Ensure a connection to a correctly running U-Boot instance.
+
+ This may require spawning a new Sandbox process or resetting target
+ hardware, as defined by the implementation sub-class.
+
+ This is an internal function and should not be called directly.
+
+ Args:
+ None.
+
+ Returns:
+ Nothing.
+ '''
+
+ if self.p:
+ return
+ try:
+ self.at_prompt = False
+ self.log.action('Starting U-Boot')
+ self.p = self.get_spawn()
+ # Real targets can take a long time to scroll large amounts of
+ # text if LCD is enabled. This value may need tweaking in the
+ # future, possibly per-test to be optimal. This works for 'help'
+ # on board 'seaboard'.
+ self.p.timeout = 30000
+ self.p.logfile_read = self.logstream
+ if self.config.buildconfig.get('CONFIG_SPL', False) == 'y':
+ self.p.expect([pattern_u_boot_spl_signon])
+ self.u_boot_spl_signon = self.p.after
+ self.u_boot_spl_signon_escaped = re.escape(self.p.after)
+ else:
+ self.u_boot_spl_signon = None
+ self.p.expect([pattern_u_boot_main_signon])
+ self.u_boot_main_signon = self.p.after
+ self.u_boot_main_signon_escaped = re.escape(self.p.after)
+ build_idx = self.u_boot_main_signon.find(', Build:')
+ if build_idx == -1:
+ self.u_boot_version_string = self.u_boot_main_signon
+ else:
+ self.u_boot_version_string = self.u_boot_main_signon[:build_idx]
+ while True:
+ match = self.p.expect([self.prompt_escaped,
+ pattern_stop_autoboot_prompt])
+ if match == 1:
+ self.p.send(chr(3)) # CTRL-C
+ continue
+ break
+ self.at_prompt = True
+ self.at_prompt_logevt = self.logstream.logfile.cur_evt
+ except Exception as ex:
+ self.log.error(str(ex))
+ self.cleanup_spawn()
+ raise
+
+ def cleanup_spawn(self):
+ '''Shut down all interaction with the U-Boot instance.
+
+ This is used when an error is detected prior to re-establishing a
+ connection with a fresh U-Boot instance.
+
+ This is an internal function and should not be called directly.
+
+ Args:
+ None.
+
+ Returns:
+ Nothing.
+ '''
+
+ try:
+ if self.p:
+ self.p.close()
+ except:
+ pass
+ self.p = None
+
+ def validate_version_string_in_text(self, text):
+ '''Assert that a command's output includes the U-Boot signon message.
+
+ This is primarily useful for validating the "version" command without
+ duplicating the signon text regex in a test function.
+
+ Args:
+ text: The command output text to check.
+
+ Returns:
+ Nothing. An exception is raised if the validation fails.
+ '''
+
+ assert(self.u_boot_version_string in text)
+
+ def disable_check(self, check_type):
+ '''Temporarily disable an error check of U-Boot's output.
+
+ Create a new context manager (for use with the "with" statement) which
+ temporarily disables a particular console output error check.
+
+ Args:
+ check_type: The type of error-check to disable. Valid values may
+ be found in self.disable_check_count above.
+
+ Returns:
+ A context manager object.
+ '''
+
+ return ConsoleDisableCheck(self, check_type)
+
+ def find_ram_base(self):
+ '''Find the running U-Boot's RAM location.
+
+ Probe the running U-Boot to determine the address of the first bank
+ of RAM. This is useful for tests that test reading/writing RAM, or
+ load/save files that aren't associated with some standard address
+ typically represented in an environment variable such as
+ ${kernel_addr_r}. The value is cached so that it only needs to be
+ actively read once.
+
+ Args:
+ None.
+
+ Returns:
+ The address of U-Boot's first RAM bank, as an integer.
+ '''
+
+ if self.config.buildconfig.get('config_cmd_bdi', 'n') != 'y':
+ pytest.skip('bdinfo command not supported')
+ if self.ram_base == -1:
+ pytest.skip('Previously failed to find RAM bank start')
+ if self.ram_base is not None:
+ return self.ram_base
+
+ with self.log.section('find_ram_base'):
+ response = self.run_command('bdinfo')
+ for l in response.split('\n'):
+ if '-> start' in l:
+ self.ram_base = int(l.split('=')[1].strip(), 16)
+ break
+ if self.ram_base is None:
+ self.ram_base = -1
+ raise Exception('Failed to find RAM bank start in `bdinfo`')
+
+ return self.ram_base
diff --git a/test/py/u_boot_console_exec_attach.py b/test/py/u_boot_console_exec_attach.py
new file mode 100644
index 00000000000..0ca9e7c178e
--- /dev/null
+++ b/test/py/u_boot_console_exec_attach.py
@@ -0,0 +1,65 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# Logic to interact with U-Boot running on real hardware, typically via a
+# physical serial port.
+
+import sys
+from u_boot_spawn import Spawn
+from u_boot_console_base import ConsoleBase
+
+class ConsoleExecAttach(ConsoleBase):
+ '''Represents a physical connection to a U-Boot console, typically via a
+ serial port. This implementation executes a sub-process to attach to the
+ console, expecting that the stdin/out of the sub-process will be forwarded
+ to/from the physical hardware. This approach isolates the test infra-
+ structure from the user-/installation-specific details of how to
+ communicate with, and the identity of, serial ports etc.'''
+
+ def __init__(self, log, config):
+ '''Initialize a U-Boot console connection.
+
+ Args:
+ log: A multiplexed_log.Logfile instance.
+ config: A "configuration" object as defined in conftest.py.
+
+ Returns:
+ Nothing.
+ '''
+
+ # The max_fifo_fill value might need tweaking per-board/-SoC?
+ # 1 would be safe anywhere, but is very slow (a pexpect issue?).
+ # 16 is a common FIFO size.
+ # HW flow control would mean this could be infinite.
+ super(ConsoleExecAttach, self).__init__(log, config, max_fifo_fill=16)
+
+ self.log.action('Flashing U-Boot')
+ cmd = ['u-boot-test-flash', config.board_type, config.board_identity]
+ runner = self.log.get_runner(cmd[0], sys.stdout)
+ runner.run(cmd)
+ runner.close()
+
+ def get_spawn(self):
+ '''Connect to a fresh U-Boot instance.
+
+ The target board is reset, so that U-Boot begins running from scratch.
+
+ Args:
+ None.
+
+ Returns:
+ A u_boot_spawn.Spawn object that is attached to U-Boot.
+ '''
+
+ args = [self.config.board_type, self.config.board_identity]
+ s = Spawn(['u-boot-test-console'] + args)
+
+ self.log.action('Resetting board')
+ cmd = ['u-boot-test-reset'] + args
+ runner = self.log.get_runner(cmd[0], sys.stdout)
+ runner.run(cmd)
+ runner.close()
+
+ return s
diff --git a/test/py/u_boot_console_sandbox.py b/test/py/u_boot_console_sandbox.py
new file mode 100644
index 00000000000..88b137e8c3a
--- /dev/null
+++ b/test/py/u_boot_console_sandbox.py
@@ -0,0 +1,79 @@
+# Copyright (c) 2015 Stephen Warren
+# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# Logic to interact with the sandbox port of U-Boot, running as a sub-process.
+
+import time
+from u_boot_spawn import Spawn
+from u_boot_console_base import ConsoleBase
+
+class ConsoleSandbox(ConsoleBase):
+ '''Represents a connection to a sandbox U-Boot console, executed as a sub-
+ process.'''
+
+ def __init__(self, log, config):
+ '''Initialize a U-Boot console connection.
+
+ Args:
+ log: A multiplexed_log.Logfile instance.
+ config: A "configuration" object as defined in conftest.py.
+
+ Returns:
+ Nothing.
+ '''
+
+ super(ConsoleSandbox, self).__init__(log, config, max_fifo_fill=1024)
+
+ def get_spawn(self):
+ '''Connect to a fresh U-Boot instance.
+
+ A new sandbox process is created, so that U-Boot begins running from
+ scratch.
+
+ Args:
+ None.
+
+ Returns:
+ A u_boot_spawn.Spawn object that is attached to U-Boot.
+ '''
+
+ return Spawn([self.config.build_dir + '/u-boot'])
+
+ def kill(self, sig):
+ '''Send a specific Unix signal to the sandbox process.
+
+ Args:
+ sig: The Unix signal to send to the process.
+
+ Returns:
+ Nothing.
+ '''
+
+ self.ensure_spawned()
+ self.log.action('kill %d' % sig)
+ self.p.kill(sig)
+
+ def validate_exited(self):
+ '''Determine whether the sandbox process has exited.
+
+ If required, this function waits a reasonable time for the process to
+ exit.
+
+ Args:
+ None.
+
+ Returns:
+ Boolean indicating whether the process has exited.
+ '''
+
+ p = self.p
+ self.p = None
+ for i in xrange(100):
+ ret = not p.isalive()
+ if ret:
+ break
+ time.sleep(0.1)
+ p.close()
+ return ret
diff --git a/test/py/u_boot_spawn.py b/test/py/u_boot_spawn.py
new file mode 100644
index 00000000000..1baee63df25
--- /dev/null
+++ b/test/py/u_boot_spawn.py
@@ -0,0 +1,174 @@
+# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
+#
+# SPDX-License-Identifier: GPL-2.0
+
+# Logic to spawn a sub-process and interact with its stdio.
+
+import os
+import re
+import pty
+import signal
+import select
+import time
+
+class Timeout(Exception):
+ '''An exception sub-class that indicates that a timeout occurred.'''
+ pass
+
+class Spawn(object):
+ '''Represents the stdio of a freshly created sub-process. Commands may be
+ sent to the process, and responses waited for.
+ '''
+
+ def __init__(self, args):
+ '''Spawn (fork/exec) the sub-process.
+
+ Args:
+ args: array of processs arguments. argv[0] is the command to execute.
+
+ Returns:
+ Nothing.
+ '''
+
+ self.waited = False
+ self.buf = ''
+ self.logfile_read = None
+ self.before = ''
+ self.after = ''
+ self.timeout = None
+
+ (self.pid, self.fd) = pty.fork()
+ if self.pid == 0:
+ try:
+ # For some reason, SIGHUP is set to SIG_IGN at this point when
+ # run under "go" (www.go.cd). Perhaps this happens under any
+ # background (non-interactive) system?
+ signal.signal(signal.SIGHUP, signal.SIG_DFL)
+ os.execvp(args[0], args)
+ except:
+ print 'CHILD EXECEPTION:'
+ import traceback
+ traceback.print_exc()
+ finally:
+ os._exit(255)
+
+ self.poll = select.poll()
+ self.poll.register(self.fd, select.POLLIN | select.POLLPRI | select.POLLERR | select.POLLHUP | select.POLLNVAL)
+
+ def kill(self, sig):
+ '''Send unix signal "sig" to the child process.
+
+ Args:
+ sig: The signal number to send.
+
+ Returns:
+ Nothing.
+ '''
+
+ os.kill(self.pid, sig)
+
+ def isalive(self):
+ '''Determine whether the child process is still running.
+
+ Args:
+ None.
+
+ Returns:
+ Boolean indicating whether process is alive.
+ '''
+
+ if self.waited:
+ return False
+
+ w = os.waitpid(self.pid, os.WNOHANG)
+ if w[0] == 0:
+ return True
+
+ self.waited = True
+ return False
+
+ def send(self, data):
+ '''Send data to the sub-process's stdin.
+
+ Args:
+ data: The data to send to the process.
+
+ Returns:
+ Nothing.
+ '''
+
+ os.write(self.fd, data)
+
+ def expect(self, patterns):
+ '''Wait for the sub-process to emit specific data.
+
+ This function waits for the process to emit one pattern from the
+ supplied list of patterns, or for a timeout to occur.
+
+ Args:
+ patterns: A list of strings or regex objects that we expect to
+ see in the sub-process' stdout.
+
+ Returns:
+ The index within the patterns array of the pattern the process
+ emitted.
+
+ Notable exceptions:
+ Timeout, if the process did not emit any of the patterns within
+ the expected time.
+ '''
+
+ for pi in xrange(len(patterns)):
+ if type(patterns[pi]) == type(''):
+ patterns[pi] = re.compile(patterns[pi])
+
+ try:
+ while True:
+ earliest_m = None
+ earliest_pi = None
+ for pi in xrange(len(patterns)):
+ pattern = patterns[pi]
+ m = pattern.search(self.buf)
+ if not m:
+ continue
+ if earliest_m and m.start() > earliest_m.start():
+ continue
+ earliest_m = m
+ earliest_pi = pi
+ if earliest_m:
+ pos = earliest_m.start()
+ posafter = earliest_m.end() + 1
+ self.before = self.buf[:pos]
+ self.after = self.buf[pos:posafter]
+ self.buf = self.buf[posafter:]
+ return earliest_pi
+ events = self.poll.poll(self.timeout)
+ if not events:
+ raise Timeout()
+ c = os.read(self.fd, 1024)
+ if not c:
+ raise EOFError()
+ if self.logfile_read:
+ self.logfile_read.write(c)
+ self.buf += c
+ finally:
+ if self.logfile_read:
+ self.logfile_read.flush()
+
+ def close(self):
+ '''Close the stdio connection to the sub-process.
+
+ This also waits a reasonable time for the sub-process to stop running.
+
+ Args:
+ None.
+
+ Returns:
+ Nothing.
+ '''
+
+ os.close(self.fd)
+ for i in xrange(100):
+ if not self.isalive():
+ break
+ time.sleep(0.1)