From 35efa23cde6cfdce2d56f0eeeff0933cb7bbb078 Mon Sep 17 00:00:00 2001
From: sascha <sascha@sascha-lutz.de>
Date: Mon, 7 Sep 2026 11:38:34 +0200
Subject: [PATCH] tests: FD-lock serialization and NFS regressions

---
 usenet-scripts/test_movetdarr_parallel.py | 110 ++++++++++++++++++++++
 1 file changed, 110 insertions(+)
 create mode 100644 usenet-scripts/test_movetdarr_parallel.py

diff --git a/usenet-scripts/test_movetdarr_parallel.py b/usenet-scripts/test_movetdarr_parallel.py
new file mode 100644
index 0000000..f3cad42
--- /dev/null
+++ b/usenet-scripts/test_movetdarr_parallel.py
@@ -0,0 +1,110 @@
+import json
+import os
+import subprocess
+import tempfile
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+
+SCRIPT = Path(__file__).with_name("movetdarr.sh")
+state_lock = threading.Lock()
+next_job = 0
+open_leases = 0
+max_open_leases = 0
+job_ids = {}
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_POST(self):
+        global next_job, open_leases, max_open_leases
+        data = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))))
+        action = data.get("action")
+        if action == "start":
+            with state_lock:
+                next_job += 1
+                job_id = f"job-{next_job}"
+                job_ids[data["directory"]] = job_id
+                open_leases += 1
+                max_open_leases = max(max_open_leases, open_leases)
+            time.sleep(0.3)
+            result = {"ok": True, "jobId": job_id, "state": "registered"}
+        elif action == "moved":
+            with state_lock:
+                open_leases -= 1
+            result = {"ok": True, "jobId": data["jobId"], "state": "processing"}
+        elif action == "status":
+            result = {"ok": True, "jobId": data["jobId"], "state": "success"}
+        else:
+            result = {"ok": False, "state": "failed"}
+        body = json.dumps(result).encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *_args):
+        pass
+
+
+def start_script(root, name, server_port):
+    source_root = root / "source"
+    source = source_root / "serien" / name
+    source.mkdir(parents=True)
+    (source / f"{name}.mkv").write_bytes(b"video")
+    env = os.environ | {
+        "SOURCE_ROOT": str(source_root),
+        "DEST_ROOT": str(root / "dest"),
+        "LOGFILE": str(root / "postprocess.log"),
+        "LEASE_LOCK": str(root / "lease.lock"),
+        "HANDOFF_URL": f"http://127.0.0.1:{server_port}/media/handoff",
+        "POLL_SECONDS": "0",
+        "MAX_POLLS": "2",
+    }
+    return subprocess.Popen(
+        ["bash", str(SCRIPT), str(source), name + ".nzb", name, "", "serien", "", "0"],
+        env=env,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+
+def test_parallel_lease_blocks_are_serialized():
+    global next_job, open_leases, max_open_leases, job_ids
+    next_job = 0
+    open_leases = 0
+    max_open_leases = 0
+    job_ids = {}
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        with tempfile.TemporaryDirectory() as td:
+            root = Path(td)
+            first = start_script(root, "Show.S01E01", server.server_port)
+            second = start_script(root, "Show.S01E02", server.server_port)
+            first_result = first.communicate(timeout=20)
+            second_result = second.communicate(timeout=20)
+            assert first.returncode == 0, first_result
+            assert second.returncode == 0, second_result
+            assert len(set(job_ids.values())) == 2
+            assert max_open_leases == 1, "start→move→moved blocks overlapped"
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+def test_lock_is_held_on_a_file_descriptor_until_register_returns():
+    source = SCRIPT.read_text()
+    assert 'exec 9>"$LEASE_LOCK"' in source
+    assert 'flock -n 9' in source
+    assert 'flock -u 9' in source
+
+
+if __name__ == "__main__":
+    test_lock_is_held_on_a_file_descriptor_until_register_returns()
+    test_parallel_lease_blocks_are_serialized()
+    print("PASS: parallel start→move→moved blocks are serialized")