test(cron): cover bounded TERMINAL_CWD lock acquisition

Lock-primitive timeout tests from #63959, applied onto the timeout API
that landed via #80912.
This commit is contained in:
dsad 2026-08-07 17:56:14 +05:30 committed by kshitij
parent 11ce6419c3
commit 5fcca432f5
1 changed files with 25 additions and 0 deletions

View File

@ -11,6 +11,7 @@ These tests assert that contract.
"""
import threading
import time
def _lock():
@ -84,6 +85,30 @@ def test_writer_waits_for_active_reader():
assert order == ["reader-release", "writer-acquire"]
def test_writer_acquire_times_out_behind_active_reader():
"""Bounded acquire prevents a stuck reader from blocking a writer forever."""
lock = _lock()
lock.acquire_read()
started = time.monotonic()
try:
assert lock.acquire_write(timeout=0.01) is False
finally:
lock.release_read()
assert time.monotonic() - started < 1.0
def test_reader_acquire_times_out_behind_active_writer():
"""Bounded acquire prevents a stuck writer from blocking readers forever."""
lock = _lock()
lock.acquire_write()
started = time.monotonic()
try:
assert lock.acquire_read(timeout=0.01) is False
finally:
lock.release_write()
assert time.monotonic() - started < 1.0
def test_reader_never_observes_writer_override():
"""Regression: the cross-pool TERMINAL_CWD corruption.