Skip to content

Commit eca5488

Browse files
committed
added new fragment and tests
1 parent 39375fb commit eca5488

File tree

2 files changed

+38
-0
lines changed

2 files changed

+38
-0
lines changed

newsfragments/752.internal.rst

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
[mplex] Add timeout and error handling during stream close

tests/core/stream_muxer/test_mplex_stream.py

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
MplexStreamClosed,
99
MplexStreamEOF,
1010
MplexStreamReset,
11+
MuxedConnUnavailable,
1112
)
1213
from libp2p.stream_muxer.mplex.mplex import (
1314
MPLEX_MESSAGE_CHANNEL_SIZE,
@@ -213,3 +214,39 @@ async def test_mplex_stream_reset(mplex_stream_pair):
213214
# `reset` should do nothing as well.
214215
await stream_0.reset()
215216
await stream_1.reset()
217+
218+
219+
@pytest.mark.trio
220+
async def test_mplex_stream_close_timeout(monkeypatch, mplex_stream_pair):
221+
stream_0, stream_1 = mplex_stream_pair
222+
223+
# (simulate hanging)
224+
async def fake_send_message(*args, **kwargs):
225+
await trio.sleep_forever()
226+
227+
monkeypatch.setattr(stream_0.muxed_conn, "send_message", fake_send_message)
228+
229+
with pytest.raises(TimeoutError):
230+
await stream_0.close()
231+
232+
233+
@pytest.mark.trio
234+
async def test_mplex_stream_close_mux_unavailable(monkeypatch, mplex_stream_pair):
235+
stream_0, _ = mplex_stream_pair
236+
237+
# Patch send_message to raise MuxedConnUnavailable
238+
def raise_unavailable(*args, **kwargs):
239+
raise MuxedConnUnavailable("Simulated conn drop")
240+
241+
monkeypatch.setattr(stream_0.muxed_conn, "send_message", raise_unavailable)
242+
243+
# Case 1: Mplex is shutting down — should not raise
244+
stream_0.muxed_conn.event_shutting_down.set()
245+
await stream_0.close() # Should NOT raise
246+
247+
# Case 2: Mplex is NOT shutting down — should raise RuntimeError
248+
stream_0.event_local_closed = trio.Event() # Reset since it was set in first call
249+
stream_0.muxed_conn.event_shutting_down = trio.Event() # Unset the shutdown flag
250+
251+
with pytest.raises(RuntimeError, match="Failed to send close message"):
252+
await stream_0.close()

0 commit comments

Comments
 (0)