Test that TCP connections are closed after the configured timeout period.
(caplog_async)
| 99 | |
| 100 | |
| 101 | async def test_tcp_timeout(caplog_async): |
| 102 | """Test that TCP connections are closed after the configured timeout period.""" |
| 103 | caplog_async.set_level("INFO") |
| 104 | manager = MagicMock() |
| 105 | |
| 106 | with taddons.context() as tctx: |
| 107 | # Set timeout to 0 for immediate timeout (fastest test) |
| 108 | tctx.options.tcp_timeout = 0 |
| 109 | |
| 110 | inst = ServerInstance.make("regular@127.0.0.1:0", manager) |
| 111 | await inst.start() |
| 112 | assert await caplog_async.await_log("proxy listening") |
| 113 | |
| 114 | host, port, *_ = inst.listen_addrs[0] |
| 115 | reader, writer = await asyncio.open_connection(host, port) |
| 116 | assert await caplog_async.await_log("client connect") |
| 117 | |
| 118 | # Keep connection open but inactive - it should timeout after 1s |
| 119 | # The await_log below will wait for the timeout to trigger |
| 120 | |
| 121 | async with asyncio.timeout(30): |
| 122 | # Verify the connection was closed due to inactivity in <60s |
| 123 | assert await caplog_async.await_log("Closing connection due to inactivity") |
| 124 | assert await caplog_async.await_log("client disconnect") |
| 125 | |
| 126 | # Try to read from the closed connection to confirm it's really closed |
| 127 | data = await reader.read(1) |
| 128 | assert data == b"" # EOF indicates connection is closed |
| 129 | |
| 130 | writer.close() |
| 131 | await writer.wait_closed() |
| 132 | |
| 133 | await inst.stop() |
| 134 | assert await caplog_async.await_log("stopped") |
| 135 | |
| 136 | |
| 137 | @pytest.mark.parametrize("failure", [True, False]) |
nothing calls this directly
no test coverage detected
searching dependent graphs…