count UT case in the end of log

test/xpu/extended/run_test_with_skip.py

@@ -1,22 +1,27 @@
 import os
 import sys
+sys.path.append("..")
 from skip_list_common import skip_dict
 from skip_list_win import skip_dict as skip_dict_win
+from xpu_test_utils import launch_test
 
 IS_WINDOWS = sys.platform == "win32"
 
 skip_list = skip_dict["test_ops_xpu.py"]
 if IS_WINDOWS:
     skip_list += skip_dict_win["test_ops_xpu.py"]
 
-skip_options = " -k \"not " + skip_list[0]
-for skip_case in skip_list[1:]:
-    skip_option = " and not " + skip_case
-    skip_options += skip_option
-skip_options += "\""
+return_code, count_buf, fails = launch_test("test_ops_xpu.py", skip_list)
 
-os.environ["PYTORCH_TEST_WITH_SLOW"]="1"
-test_command = "pytest -v test_ops_xpu.py"
-test_command += skip_options
-res = os.system(test_command)
-sys.exit(res)
+if fails:
+    print("="*10," failures list ","="*10)
+    for fail in fails:
+        print(fail)
+print("="*10," case count ","="*10)
+print(count_buf)
+
+if os.name == "nt":
+    sys.exit(return_code)
+else:    
+    exit_code = os.WEXITSTATUS(return_code)
+    sys.exit(exit_code)

test/xpu/extended/run_test_with_skip_arc.py

@@ -1,24 +1,29 @@
 import os
 import sys
+sys.path.append("..")
 from skip_list_common import skip_dict
 from skip_list_arc import skip_dict as skip_dict_specifical
 from skip_list_win import skip_dict as skip_dict_win
 from skip_list_win_arc import skip_dict as skip_dict_win_arc
+from xpu_test_utils import launch_test
 
 IS_WINDOWS = sys.platform == "win32"
 
 skip_list = skip_dict["test_ops_xpu.py"] + skip_dict_specifical["test_ops_xpu.py"]
 if IS_WINDOWS:
     skip_list += skip_dict_win["test_ops_xpu.py"] + skip_dict_win_arc["test_ops_xpu.py"]
 
-skip_options = " -k \"not " + skip_list[0]
-for skip_case in skip_list[1:]:
-    skip_option = " and not " + skip_case
-    skip_options += skip_option
-skip_options += "\""
+return_code, count_buf, fails = launch_test("test_ops_xpu.py", skip_list)
 
-os.environ["PYTORCH_TEST_WITH_SLOW"]="1"
-test_command = "pytest -v test_ops_xpu.py"
-test_command += skip_options
-res = os.system(test_command)
-sys.exit(res)
+if fails:
+    print("="*10," failures list ","="*10)
+    for fail in fails:
+        print(fail)
+print("="*10," case count ","="*10)
+print(count_buf)
+
+if os.name == "nt":
+    sys.exit(return_code)
+else:    
+    exit_code = os.WEXITSTATUS(return_code)
+    sys.exit(exit_code)

test/xpu/run_test_with_only.py

@@ -1,35 +1,10 @@
 import os
 import sys
+from xpu_test_utils import launch_test
 
 
 # Cases in the file is too slow to run all suites on CPU. So add white list.
 
-
-def launch_test(test_case, skip_list=None, exe_list=None):
-    os.environ["PYTORCH_ENABLE_XPU_FALLBACK"]="1"
-    os.environ["PYTORCH_TEST_WITH_SLOW"]="1"
-    if skip_list != None:
-        skip_options = " -k \"not " + skip_list[0]
-        for skip_case in skip_list[1:]:
-            skip_option = " and not " + skip_case
-            skip_options += skip_option
-        skip_options += "\""
-        test_command   = "pytest -v "  + test_case  + skip_options
-        return os.system(test_command)
-    elif exe_list != None:
-        exe_options = " -k \"" + exe_list[0]
-        for exe_case in exe_list[1:]:
-            exe_option = " or " + exe_case
-            exe_options += exe_option
-        exe_options += "\""
-        test_command   = "pytest -v "  + test_case  + exe_options
-        return os.system(test_command)
-    else:
-        test_command   = "pytest -v " + test_case  
-        return os.system(test_command)     
-
-res = 0
-
 # test_decomp
 # full skip_list is in Issue #470
 execute_list = (
@@ -39,10 +14,18 @@ def launch_test(test_case, skip_list=None, exe_list=None):
     "test_comprehensive_nn_functional_nll_loss_xpu_float64",
     "bincount",
 )
-res += launch_test("test_decomp_xpu.py", exe_list=execute_list)
+return_code, count_buf, fails = launch_test("test_decomp_xpu.py", exe_list=execute_list)
+
+if fails:
+    print("="*10," failures list ","="*10)
+    for fail in fails:
+        print(fail)
+print("="*10," case count ","="*10)
+print(count_buf)
+
 
 if os.name == "nt":
-    sys.exit(res)
+    sys.exit(return_code)
 else:    
-    exit_code = os.WEXITSTATUS(res)
+    exit_code = os.WEXITSTATUS(return_code)
     sys.exit(exit_code)

test/xpu/run_test_with_skip.py

@@ -6,16 +6,22 @@
 
 
 res = 0
-fail_test = []
+fails = []
+count = ""
 
 for key in skip_dict:
     skip_list = skip_dict[key]
-    fail = launch_test(key, skip_list)
-    res += fail
-    if fail:
-        fail_test.append(key)
-if fail_test:
-    print(",".join(fail_test) + " have failures")
+    return_code, count_buf, fail = launch_test(key, skip_list)
+    res += return_code
+    count=count+count_buf+"\n"
+    if return_code:
+        fails.extend(fail)
+if fails:
+    print("="*10," failures list ","="*10)
+    for fail in fails:
+        print(fail)
+print("="*10," case count ","="*10)
+print(count)
 
 
 if os.name == "nt":

test/xpu/run_test_with_skip_arc.py

@@ -1,13 +1,16 @@
 import os
 import sys
-from skip_list_common import skip_dict
+
 from skip_list_arc import skip_dict as skip_dict_specifical
+from skip_list_common import skip_dict
 from skip_list_win import skip_dict as skip_dict_win
 from skip_list_win_arc import skip_dict as skip_dict_win_arc
 from xpu_test_utils import launch_test
 
 
 res = 0
+fails = []
+count = ""
 IS_WINDOWS = sys.platform == "win32"
 
 for key in skip_dict:
@@ -18,10 +21,20 @@
         skip_list += skip_dict_win[key]
     if IS_WINDOWS and key in skip_dict_win_arc:
         skip_list += skip_dict_win_arc[key]
-    res += launch_test(key, skip_list)
+    return_code, count_buf, fail = launch_test(key, skip_list)
+    res += return_code
+    count = count + count_buf + "\n"
+    if return_code:
+        fails.extend(fail)
+if fails:
+    print("=" * 10, " failures list ", "=" * 10)
+    for fail in fails:
+        print(fail)
+print("=" * 10, " case count ", "=" * 10)
+print(count)
 
 if os.name == "nt":
     sys.exit(res)
-else:    
+else:
     exit_code = os.WEXITSTATUS(res)
-    sys.exit(exit_code)
+    sys.exit(exit_code)

test/xpu/run_test_with_skip_mtl.py

@@ -6,6 +6,8 @@
 from xpu_test_utils import launch_test
 
 res = 0
+fails = []
+count = ""
 IS_WINDOWS = sys.platform == "win32"
 
 for key in skip_dict:
@@ -14,10 +16,20 @@
         skip_list += skip_dict_specifical[key]
     if IS_WINDOWS and key in skip_dict_win:
         skip_list += skip_dict_win[key]
-    res += launch_test(key, skip_list)
+    return_code, count_buf, fail = launch_test(key, skip_list)
+    res += return_code
+    count = count + count_buf + "\n"
+    if return_code:
+        fails.extend(fail)
+if fails:
+    print("=" * 10, " failures list ", "=" * 10)
+    for fail in fails:
+        print(fail)
+print("=" * 10, " case count ", "=" * 10)
+print(count)
 
 if os.name == "nt":
     sys.exit(res)
 else:    
     exit_code = os.WEXITSTATUS(res)
-    sys.exit(exit_code)
+    sys.exit(exit_code)

test/xpu/xpu_test_utils.py

@@ -5,6 +5,8 @@
 import os
 import sys
 import unittest
+import subprocess
+import re
 
 import torch
 from torch import bfloat16, cuda
@@ -339,10 +341,10 @@
         }
     },
     "histogram": {
-        ("TestCommonXPU", "test_out_histogram_xpu_float32"):{
-            torch.float32: tol(atol=3e-5, rtol=5e-5),
+        ("TestCommon", "test_out"): {
+            torch.float32: tol(atol=3e-5, rtol=5e-6),
         }
-    }
+    },
 }
 
 
@@ -971,33 +973,68 @@ def copy_tests(
 
 
 def launch_test(test_case, skip_list=None, exe_list=None):
-    os.environ["PYTORCH_ENABLE_XPU_FALLBACK"]="1"
-    os.environ["PYTORCH_TEST_WITH_SLOW"]="1"
     if skip_list != None:
-        skip_options = " -k \"not " + skip_list[0]
+        skip_options = ' -k "not ' + skip_list[0]
         for skip_case in skip_list[1:]:
             skip_option = " and not " + skip_case
             skip_options += skip_option
-        skip_options += "\""
-        test_command = (
-            "pytest -v "
-            + test_case
-        )
+        skip_options += '"'
+        test_command = "pytest -v " + test_case
         test_command += skip_options
     elif exe_list != None:
-        exe_options = " -k \"" + exe_list[0]
+        exe_options = ' -k "' + exe_list[0]
         for exe_case in exe_list[1:]:
             exe_option = " or " + exe_case
             exe_options += exe_option
-        exe_options += "\""
-        test_command = (
-            "pytest -v "
-            + test_case
-        )
+        exe_options += '"'
+        test_command = "pytest -v " + test_case
         test_command += exe_options
     else:
-        test_command = (
-            "pytest -v "
-            + test_case
-        )
-    return os.system(test_command)
+        test_command = "pytest -v " + test_case
+    env = os.environ.copy()
+    env["PYTORCH_ENABLE_XPU_FALLBACK"] = "1"
+    env["PYTORCH_TEST_WITH_SLOW"] = "1"
+    test_process = subprocess.run(
+        test_command, capture_output=True, shell=True, env=env
+    )
+    output = test_process.stdout.decode()
+    print(output)
+    outputs = output.split("\n")
+
+    result_count = {}
+    for line in outputs:
+        if line.find("collected") >= 0:
+            pattern = r"(\d+) items"
+            res = re.search(pattern, line)
+            assert res != None
+            result_count["items"] = int(res.group(1))
+            for key in ["selected", "deselected", "errors"]:
+                pattern = r"(\d+) " + key
+                res = re.search(pattern, line)
+                result_count[key] = int(res.group(1)) if res else 0
+    current = len(outputs) - 1
+    while outputs[current]=="" or outputs[current][0] != "=":
+        current -= 1
+    if test_process.returncode == 0 or test_process.returncode == 1:
+        for key in ["passed", "failed", "skipped", "xfailed"]:
+            pattern = r"(\d+) " + key
+            res = re.search(pattern, line)
+            result_count[key] = int(res.group(1)) if res else 0
+        count_buf = test_case + "\t"
+        for key, value in result_count.items():
+            count_buf += f"{key}:{value} "
+        if test_process.returncode == 0:
+            return (0, count_buf, [])
+        else:
+            current -= 1
+            fails = []
+            while outputs[current][0] != "=":
+                if outputs[current].startswith("FAILED "):
+                    tmp = outputs[current][7:]
+                    fails.append(tmp[: tmp.find(" ")])
+                    current -= 1
+                else:
+                    break
+            return (1, count_buf, fails)
+    return (2, count_buf, [test_case + "has error"])
+