Function
run_test_timeout_thread
(name, test_env, input_queue, param)
Source from the content-addressed store, hash-verified
| 319 | |
| 320 | |
| 321 | def run_test_timeout_thread(name, test_env, input_queue, param): |
| 322 | def thread(): |
| 323 | outputs = [] |
| 324 | while True: |
| 325 | timeout = param.remaining_duration(test_env) |
| 326 | try: |
| 327 | new_output = input_queue.get(timeout=timeout) |
| 328 | except queue.Empty: |
| 329 | for output in outputs: |
| 330 | output.put(TestTimeout()) |
| 331 | break |
| 332 | else: |
| 333 | outputs.append(new_output) |
| 334 | |
| 335 | Thread(name=name, target=thread, daemon=True).start() |
| 336 | |
| 337 | |
| 338 | class Runnable: |
Tested by
no test coverage detected