0
0
Fork 0

fix(test): retry on PermissionError in mock_watcher (Win race)

On Windows, the renamed .command.json / script .py file is briefly
locked by Defender or NTFS rename finalization, which surfaces as a
transient PermissionError when the watcher's open() runs immediately
after the producer's atomicWrite + rename.

Wrap the two reads in process_command() with read_with_retry(), which
retries up to 10 times with 20 ms sleeps before giving up. POSIX
behaviour unchanged (first attempt always succeeds).

Repro on Windows:
  npm test
  > FAIL tests/unit/ipc.test.ts > sendCommand handles script error
  > PermissionError: [Errno 13] Permission denied: '...command.json'
This commit is contained in:
Karstein Phobic Nyvold Kvistad 2026-04-29 08:34:06 +02:00
parent aaf76d2eec
commit d01f6ed930

View file

@ -40,6 +40,25 @@ def atomic_write(file_path, content):
os.rename(tmp_path, file_path)
def read_with_retry(path, attempts=10, delay=0.02):
"""Read a file with retries on PermissionError.
On Windows the renamed file can be briefly locked by Defender / NTFS
finalization, producing a transient PermissionError. Retry a few times
with a small sleep before giving up.
"""
last = None
for _ in range(attempts):
try:
with open(path, "r") as f:
return f.read()
except (IOError, OSError) as e:
# PermissionError is OSError on py3, IOError on py2
last = e
time.sleep(delay)
raise last
def process_command(commands_dir, results_dir, command_file):
"""Process a single .command.json file."""
command_path = os.path.join(commands_dir, command_file)
@ -51,16 +70,14 @@ def process_command(commands_dir, results_dir, command_file):
error = ""
try:
with open(command_path, "r") as f:
command_data = json.loads(f.read())
command_data = json.loads(read_with_retry(command_path))
script_path = command_data.get("scriptPath", "")
if not os.path.exists(script_path):
raise IOError("Script file not found: %s" % script_path)
with open(script_path, "r") as f:
script_code = f.read()
script_code = read_with_retry(script_path)
# Capture stdout/stderr
old_stdout = sys.stdout