Make CTS resistant to segfaults/aborts

test/conformance/cts_exe.py

@@ -28,33 +28,15 @@ def _print_cmdline(cmd_args, env, cwd, file=sys.stderr):
     cmd_str = " ".join(map(shlex.quote, cmd_args))
     print(f"### env -C {cwd} -i {env_args} {cmd_str}", file=file)
 
-
-if __name__ == "__main__":
-
-    parser = ArgumentParser()
-    parser.add_argument("--test_command", help="Ctest test case")
-    parser.add_argument("--devices_count", type=str, help="Number of devices on which tests will be run")
-    parser.add_argument("--platforms_count", type=str, help="Number of platforms on which tests will be run")
-    parser.add_argument("--backend", type=str, help="Number of platforms on which tests will be run")
-    args = parser.parse_args()
-    invocation = [
-        args.test_command,
-        "--gtest_brief=1",
-        f"--devices_count={args.devices_count}",
-        f"--platforms_count={args.platforms_count}",
-        f"--backend={args.backend}",
-    ]
-    _print_cmdline(invocation, os.environ, os.getcwd())
-
+def _collect_output(full_output, match_output):
     result = subprocess.Popen(  # nosec B603
         invocation, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True
     )
 
     pat = re.compile(r'\[( )*FAILED( )*\]')
-    output_list = []
     test_cases = []
     for line in result.stdout:
-        output_list.append(line)
+        full_output.append(line)
         if pat.search(line):
             test_case = line.split(" ")[5]
             test_case = test_case.rstrip(',')
@@ -65,14 +47,36 @@ def _print_cmdline(cmd_args, env, cwd, file=sys.stderr):
     # message. To avoid matching mismatch, remove lines that differ only by device ID.
     test_cases = [re.sub(r'ID[0-9]ID', 'X', tc) for tc in test_cases]
     test_cases = list(OrderedDict.fromkeys(test_cases))
-
     for tc in test_cases:
-        print(tc)
+        match_output.append(tc)
+
+    return result.wait()
+
+
+if __name__ == "__main__":
+    parser = ArgumentParser()
+    parser.add_argument("--test_command", help="Ctest test case")
+    parser.add_argument("--devices_count", type=str, help="Number of devices on which tests will be run")
+    parser.add_argument("--platforms_count", type=str, help="Number of platforms on which tests will be run")
+    parser.add_argument("--backend", type=str, help="Number of platforms on which tests will be run")
+    args = parser.parse_args()
+    invocation = [
+        args.test_command,
+        "--gtest_brief=1",
+        f"--devices_count={args.devices_count}",
+        f"--platforms_count={args.platforms_count}",
+        f"--backend={args.backend}",
+    ]
+    _print_cmdline(invocation, os.environ, os.getcwd())
+
+    full_output = []
+    match_output = []
+    rc = _collect_output(full_output, match_output)
 
-    rc = result.wait()
     if rc < 0:
-        print(signal.strsignal(abs(rc)))
+        match_output.append(signal.strsignal(abs(rc)))
 
+    print("".join(match_output))
     print("#### GTEST_OUTPUT ####", file=sys.stderr)
-    print(''.join(output_list), file=sys.stderr)
+    print(''.join(full_output), file=sys.stderr)
     print("#### GTEST_OUTPUT_END ####", file=sys.stderr)