| 16 | |
| 17 | |
| 18 | def test_filelock(): |
| 19 | temp_dir = tempfile.mkdtemp() |
| 20 | lock_path = os.path.join(temp_dir, "file.lock") |
| 21 | out_file_path = os.path.join(temp_dir, "out.txt") |
| 22 | |
| 23 | process_count = 5 |
| 24 | processes = [] |
| 25 | for i in range(process_count): |
| 26 | process = multiprocessing.Process(target=_acquire_lock, args=(lock_path, out_file_path)) |
| 27 | process.start() |
| 28 | processes.append(process) |
| 29 | |
| 30 | for process in processes: |
| 31 | process.join() |
| 32 | |
| 33 | pids = [] |
| 34 | times = [] |
| 35 | with open(out_file_path) as out: |
| 36 | content = out.read() |
| 37 | logging.info("Times:\n%s", content) |
| 38 | for line in content.strip().split("\n"): |
| 39 | pid, time_val = line.split(":") |
| 40 | pids.append(pid) |
| 41 | times.append(float(time_val)) |
| 42 | |
| 43 | assert len(set(pids)) == process_count |
| 44 | time1 = times.pop() |
| 45 | while times: |
| 46 | time2 = times.pop() |
| 47 | assert int(time1) - int(time2) >= 2 |
| 48 | time1 = time2 |
| 49 | |
| 50 | |
| 51 | def test_filelock_init_acquired(): |