fix(test): create ModelTests before concurrent pool workers

sqlmesh/core/test/runner.py

@@ -125,25 +125,7 @@ def run_tests(
     # Ensure workers are not greater than the number of tests
     num_workers = min(len(model_test_metadata) or 1, default_test_connection.concurrent_tasks)
 
-    def _run_single_test(
-        metadata: ModelTestMetadata, engine_adapter: EngineAdapter
-    ) -> t.Optional[ModelTextTestResult]:
-        test = ModelTest.create_test(
-            body=metadata.body,
-            test_name=metadata.test_name,
-            models=models,
-            engine_adapter=engine_adapter,
-            dialect=dialect,
-            path=metadata.path,
-            default_catalog=default_catalog,
-            preserve_fixtures=preserve_fixtures,
-            concurrency=num_workers > 1,
-            verbosity=verbosity,
-        )
-
-        if not test:
-            return None
-
+    def _run_single_test(test: ModelTest) -> ModelTextTestResult:
         result = t.cast(
             ModelTextTestResult,
             ModelTextTestRunner().run(t.cast(unittest.TestCase, test)),
@@ -158,11 +140,30 @@ def _run_single_test(
 
     start_time = time.perf_counter()
     try:
+        # Build ModelTest instances on the calling thread before workers start. create_test()
+        # can call to_datetime() / ttl_cache (time.time()), which races with another worker's
+        # time_machine freeze when execution_time is set under concurrent_tasks > 1.
+        # NOTE: We can run create_tests in a separate parallel stage for a future optimization.
+        # We just can't overlap runs/creations.
+        tests: list[ModelTest] = []
+        for metadata, engine_adapter in metadata_to_adapter.items():
+            test = ModelTest.create_test(
+                body=metadata.body,
+                test_name=metadata.test_name,
+                models=models,
+                engine_adapter=engine_adapter,
+                dialect=dialect,
+                path=metadata.path,
+                default_catalog=default_catalog,
+                preserve_fixtures=preserve_fixtures,
+                concurrency=num_workers > 1,
+                verbosity=verbosity,
+            )
+            if test:
+                tests.append(test)
+
         with ThreadPoolExecutor(max_workers=num_workers) as pool:
-            futures = [
-                pool.submit(_run_single_test, metadata=metadata, engine_adapter=engine_adapter)
-                for metadata, engine_adapter in metadata_to_adapter.items()
-            ]
+            futures = [pool.submit(_run_single_test, test) for test in tests]
 
             for future in concurrent.futures.as_completed(futures):
                 test_results.append(future.result())