From d01f6ed93073d2247f2307d6cbd8ab79eb927da6 Mon Sep 17 00:00:00 2001 From: Karstein Phobic Nyvold Kvistad Date: Wed, 29 Apr 2026 08:34:06 +0200 Subject: [PATCH] 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' --- tests/mock_watcher.py | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/tests/mock_watcher.py b/tests/mock_watcher.py index 08544c2..b1b1867 100644 --- a/tests/mock_watcher.py +++ b/tests/mock_watcher.py @@ -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