110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
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")
|