+ def persist_output(self, test: TestToRun, message: str, output: str) -> None:
+ """Called to save the output of a test run."""
+
+ dest = f'{test.name}-output.txt'
+ with open(f'./test_output/{dest}', 'w') as wf:
+ print(message, file=wf)
+ print('-' * len(message), file=wf)
+ wf.write(output)
+
+ def execute_commandline(
+ self,
+ test: TestToRun,
+ *,
+ timeout: float = 120.0,
+ ) -> TestResults:
+ """Execute a particular commandline to run a test."""
+
+ try:
+ output = exec_utils.cmd(
+ test.cmdline,
+ timeout_seconds=timeout,
+ )
+ self.persist_output(test, f'{test.name} ({test.cmdline}) succeeded.', output)
+ logger.debug('%s: %s (%s) succeeded', self.get_name(), test.name, test.cmdline)
+ return TestResults(test.name, [test.name], [test.name], [], [])
+ except subprocess.TimeoutExpired as e:
+ msg = f'{self.get_name()}: {test.name} ({test.cmdline}) timed out after {e.timeout:.1f} seconds.'
+ logger.error(msg)
+ logger.debug(
+ '%s: %s output when it timed out: %s', self.get_name(), test.name, e.output
+ )
+ self.persist_output(test, msg, e.output.decode('utf-8'))
+ return TestResults(
+ test.name,
+ [test.name],
+ [],
+ [],
+ [test.name],
+ )
+ except subprocess.CalledProcessError as e:
+ msg = (
+ f'{self.get_name()}: {test.name} ({test.cmdline}) failed; exit code {e.returncode}'
+ )
+ logger.error(msg)
+ logger.debug('%s: %s output when it failed: %s', self.get_name(), test.name, e.output)
+ self.persist_output(test, msg, e.output.decode('utf-8'))
+ return TestResults(
+ test.name,
+ [test.name],
+ [],
+ [test.name],
+ [],
+ )