+ def persist_output(self, test_name: str, message: str, output: str) -> None:
+ basename = file_utils.without_path(test_name)
+ dest = f'{basename}-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_name: str,
+ cmdline: str,
+ *,
+ timeout: float = 120.0,
+ ) -> TestResults:
+
+ try:
+ logger.debug('%s: Running %s (%s)', self.get_name(), test_name, cmdline)
+ output = exec_utils.cmd(
+ cmdline,
+ timeout_seconds=timeout,
+ )
+ self.persist_output(test_name, f'{test_name} ({cmdline}) succeeded.', output)
+ logger.debug('%s (%s) succeeded', test_name, cmdline)
+ return TestResults(test_name, [test_name], [test_name], [], [])
+ except subprocess.TimeoutExpired as e:
+ msg = f'{self.get_name()}: {test_name} ({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_name, msg, e.output)
+ return TestResults(
+ test_name,
+ [test_name],
+ [],
+ [],
+ [test_name],
+ )
+ except subprocess.CalledProcessError as e:
+ msg = f'{self.get_name()}: {test_name} ({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_name, msg, e.output)
+ return TestResults(
+ test_name,
+ [test_name],
+ [],
+ [test_name],
+ [],
+ )
+