Start a server in a daemon thread that waits `delay` seconds before responding.
(port_holder: list[int], delay: float = 2.0)
| 646 | |
| 647 | |
| 648 | def make_slow_server(port_holder: list[int], delay: float = 2.0) -> None: |
| 649 | """Start a server in a daemon thread that waits `delay` seconds before responding.""" |
| 650 | server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 651 | server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 652 | server.bind(('127.0.0.1', 0)) |
| 653 | port_holder[0] = server.getsockname()[1] |
| 654 | server.listen(1) |
| 655 | |
| 656 | def _serve() -> None: |
| 657 | try: |
| 658 | conn, _ = server.accept() |
| 659 | conn.recv(1024) |
| 660 | time.sleep(delay) |
| 661 | conn.send(b'HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK') |
| 662 | conn.close() |
| 663 | finally: |
| 664 | server.close() |
| 665 | |
| 666 | threading.Thread(target=_serve, daemon=True).start() |
| 667 | |
| 668 | |
| 669 | class TestTimeoutBehaviour: |
no outgoing calls
no test coverage detected
searching dependent graphs…