lock.py (1723B)
1 # This file implements a shared lock that lets us ensure that the test cases in 2 # this directory run serially. Each test case obtains this lock as its first 3 # step, and releases it as its last. (The nel_test helper function in 4 # nel.sub.js automates this process.) Because the lock needs to be shared 5 # across all of the test cases, we use a hard-coded stash key. This hard-coded 6 # key is a random UUID, which should not conflict with any other auto-generated 7 # stash keys. 8 9 import time 10 11 _LOCK_KEY = b"67966d2e-a847-41d8-b7c3-5f6aee3375ba" 12 _TIMEOUT = 5 # seconds 13 14 def wait_for_lock(request): 15 t0 = time.time() 16 while time.time() - t0 < _TIMEOUT: 17 time.sleep(0.5) 18 value = request.server.stash.take(key=_LOCK_KEY) 19 if value is None: 20 return True 21 return False 22 23 def lock(request, report_id): 24 with request.server.stash.lock: 25 # Loop until the lock is free 26 if not wait_for_lock(request): 27 return (503, [], b"Cannot obtain lock") 28 request.server.stash.put(key=_LOCK_KEY, value=report_id) 29 return b"Obtained lock for %s" % report_id 30 31 def unlock(request, report_id): 32 with request.server.stash.lock: 33 lock_holder = request.server.stash.take(key=_LOCK_KEY) 34 if lock_holder != report_id: 35 # Return the lock holder to the stash 36 request.server.stash.put(key=_LOCK_KEY, value=lock_holder) 37 return (503, [], b"Cannot release lock held by %s" % lock_holder) 38 return b"Released lock for %s" % report_id 39 40 def main(request, response): 41 op = request.GET.first(b"op") 42 report_id = request.GET.first(b"reportID") 43 if op == b"lock": 44 return lock(request, report_id) 45 elif op == b"unlock": 46 return unlock(request, report_id) 47 else: 48 return (400, [], b"Invalid op")