(target, args=(), timeout=60)
| 276 | |
| 277 | @contextmanager |
| 278 | def background(target, args=(), timeout=60): |
| 279 | thread = threading.Thread(target=target, args=args) |
| 280 | thread.daemon = True |
| 281 | thread.start() |
| 282 | try: |
| 283 | yield target |
| 284 | finally: |
| 285 | thread.join(timeout=timeout) |
| 286 | if thread.is_alive(): |
| 287 | msg = 'Background task did not exit in a timely manner.' |
| 288 | raise BackgroundTaskFailed(msg) |
| 289 | |
| 290 | |
| 291 | def run_server(handler, port): |
no test coverage detected