PR Feedback 2

Author: VaggelisD

sqlmesh/core/test/__init__.py

@@ -21,57 +21,16 @@
     load_model_test_file as load_model_test_file,
 )
 from sqlmesh.core.test.result import ModelTextTestResult as ModelTextTestResult
-from sqlmesh.core.test.runner import ModelTextTestRunner as ModelTextTestRunner
+from sqlmesh.core.test.runner import (
+    ModelTextTestRunner,
+    log_test_report,
+)
 from sqlmesh.utils import UniqueKeyDict, Verbosity
 
 if t.TYPE_CHECKING:
     from sqlmesh.core.config.loader import C
 
 
-def log_test_report(results: ModelTextTestResult, test_duration: float) -> None:
-    # Aggregate parallel test run results
-    tests_run = results.testsRun
-    errors = results.errors
-    failures = results.failures
-    skipped = results.skipped
-
-    is_success = not (errors or failures)
-
-    # Compute test info
-    infos = []
-    if failures:
-        infos.append(f"failures={len(failures)}")
-    if errors:
-        infos.append(f"errors={len(errors)}")
-    if skipped:
-        infos.append(f"skipped={skipped}")
-
-    # Report test errors
-    stream = results.stream
-
-    stream.write("\n")
-
-    for test_case, err in failures:
-        stream.writeln(unittest.TextTestResult.separator1)
-        stream.writeln(f"FAIL: {test_case}")
-        stream.writeln(unittest.TextTestResult.separator2)
-        stream.writeln(err)
-
-    for error in errors:
-        stream.writeln(unittest.TextTestResult.separator1)
-        stream.writeln(f"ERROR: {error[1]}")
-        stream.writeln(unittest.TextTestResult.separator2)
-
-    # Test report
-    stream.writeln(unittest.TextTestResult.separator2)
-    stream.writeln(
-        f'Ran {tests_run} {"tests" if tests_run > 1 else "test"} in {test_duration:.3f}s \n'
-    )
-    stream.writeln(
-        f'{"OK" if is_success else "FAILED"}{" (" + ", ".join(infos) + ")" if infos else ""}'
-    )
-
-
 def run_tests(
     model_test_metadata: list[ModelTestMetadata],
     models: UniqueKeyDict[str, Model],

sqlmesh/core/test/runner.py

@@ -20,3 +20,47 @@ def __init__(
             resultclass=ModelTextTestResult,
             **kwargs,
         )
+
+
+def log_test_report(results: ModelTextTestResult, test_duration: float) -> None:
+    # Aggregate parallel test run results
+    tests_run = results.testsRun
+    errors = results.errors
+    failures = results.failures
+    skipped = results.skipped
+
+    is_success = not (errors or failures)
+
+    # Compute test info
+    infos = []
+    if failures:
+        infos.append(f"failures={len(failures)}")
+    if errors:
+        infos.append(f"errors={len(errors)}")
+    if skipped:
+        infos.append(f"skipped={skipped}")
+
+    # Report test errors
+    stream = results.stream
+
+    stream.write("\n")
+
+    for test_case, err in failures:
+        stream.writeln(unittest.TextTestResult.separator1)
+        stream.writeln(f"FAIL: {test_case}")
+        stream.writeln(unittest.TextTestResult.separator2)
+        stream.writeln(err)
+
+    for error in errors:
+        stream.writeln(unittest.TextTestResult.separator1)
+        stream.writeln(f"ERROR: {error[1]}")
+        stream.writeln(unittest.TextTestResult.separator2)
+
+    # Test report
+    stream.writeln(unittest.TextTestResult.separator2)
+    stream.writeln(
+        f'Ran {tests_run} {"tests" if tests_run > 1 else "test"} in {test_duration:.3f}s \n'
+    )
+    stream.writeln(
+        f'{"OK" if is_success else "FAILED"}{" (" + ", ".join(infos) + ")" if infos else ""}'
+    )