Repository navigation
Concurrent model tests crash when execution_time is set #6039
Description
sqlmesh test with concurrent_tasks > 1 can fail if tests set vars.execution_time:
TestError: Failed to create test ...
IndexError: list index out of range
If a unit test sets execution_time, we freeze the clock so datetime.now() and similar calls match that timestamp. The freeze (time_machine) changes time.time() for every thread in the process.
We take a lock around that freeze when rendering SQL (_concurrent_render_context in sqlmesh/core/test/definition.py), but we still build each test on a worker thread with no lock (ModelTest.create_test() in sqlmesh/core/test/runner.py).
Building a test calls to_datetime() (ModelTest.__init__ in definition.py), which goes through ttl_cache (sqlmesh/utils/__init__.py) and hits time.time(). If another worker is starting or stopping the freeze at the same time, that call crashes.
Caught by tests/core/test_test.py::test_freeze_time_concurrent. Do not skip or weaken this test.
Fix
Create the tests before we submit to the thread pool.
File: sqlmesh/core/test/runner.py
Function: run_tests()
Example
tests = [] # <----NEW
for metadata, engine_adapter in metadata_to_adapter.items(): # <----NEW
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: # <----NEW
tests.append(test) # <----NEW
def _run_single_test(test: ModelTest) -> ModelTextTestResult: # <----NEW (signature)
result = t.cast(
ModelTextTestResult,
ModelTextTestRunner().run(t.cast(unittest.TestCase, test)),
)
with lock:
combined_results.merge(result)
return result
test_results = []
start_time = time.perf_counter()
try:
with ThreadPoolExecutor(max_workers=num_workers) as pool:
futures = [pool.submit(_run_single_test, test) for test in tests] # <----NEW
for future in concurrent.futures.as_completed(futures):
test_results.append(future.result())Validation
pytest tests/core/test_test.py::test_freeze_time_concurrent -v a few times. This is a race, so a single pass is not enough.
Disclaimer: AI Notes
Activity
devtechedge commentedon Sep 9, 2026
Taking a look at this. Plan is to create the ModelTest instances on the main thread before submitting work to the pool, so create_test / to_datetime no longer races with another worker's time_machine freeze. Will keep test_freeze_time_concurrent and open a PR.