diff --git a/libp2p/transport/quic/stream.py b/libp2p/transport/quic/stream.py index ba250c98c..2ff35ab23 100644 --- a/libp2p/transport/quic/stream.py +++ b/libp2p/transport/quic/stream.py @@ -251,7 +251,7 @@ async def read(self, n: int | None = None) -> bytes: return data # Check if stream was reset (unblocking read loop) - if self._state == StreamState.RESET: + if self.is_reset(): raise QUICStreamResetError( f"Stream {self.stream_id} was reset" ) diff --git a/newsfragments/1373.internal.rst b/newsfragments/1373.internal.rst new file mode 100644 index 000000000..76db57030 --- /dev/null +++ b/newsfragments/1373.internal.rst @@ -0,0 +1 @@ +Fixed CI lint failure on ``main`` caused by mypy 2.2 ``comparison-overlap`` in the QUIC stream read loop. Reads blocked on ``_receive_event`` still raise ``QUICStreamResetError`` when the stream is reset concurrently. diff --git a/tests/core/transport/quic/test_stream.py b/tests/core/transport/quic/test_stream.py new file mode 100644 index 000000000..3b2459fc2 --- /dev/null +++ b/tests/core/transport/quic/test_stream.py @@ -0,0 +1,75 @@ +"""Unit tests for QUICStream behavior.""" + +from unittest.mock import Mock + +import pytest +from multiaddr.multiaddr import Multiaddr +import trio +from trio.testing import wait_all_tasks_blocked + +from libp2p.crypto.ed25519 import create_new_key_pair +from libp2p.peer.id import ID +from libp2p.transport.quic.config import QUICTransportConfig +from libp2p.transport.quic.connection import QUICConnection +from libp2p.transport.quic.exceptions import QUICStreamResetError +from libp2p.transport.quic.stream import QUICStream, StreamDirection + + +@pytest.fixture +def quic_connection() -> QUICConnection: + mock_quic = Mock() + mock_quic.next_event.return_value = None + mock_quic.datagrams_to_send.return_value = [] + mock_quic.get_timer.return_value = None + mock_quic.connect = Mock() + mock_quic.close = Mock() + mock_quic.send_stream_data = Mock() + mock_quic.reset_stream = Mock() + + mock_transport = Mock() + mock_transport._config = QUICTransportConfig() + + private_key = create_new_key_pair().private_key + peer_id = ID.from_pubkey(private_key.get_public_key()) + + return QUICConnection( + quic_connection=mock_quic, + remote_addr=("127.0.0.1", 4001), + remote_peer_id=None, + local_peer_id=peer_id, + is_initiator=True, + maddr=Multiaddr("/ip4/127.0.0.1/udp/4001/quic"), + transport=mock_transport, + resource_scope=None, + security_manager=Mock(), + ) + + +@pytest.mark.trio +async def test_read_raises_reset_error_when_reset_during_wait( + quic_connection: QUICConnection, +) -> None: + """read() must detect peer reset while blocked on _receive_event.""" + stream = QUICStream( + connection=quic_connection, + stream_id=1, + direction=StreamDirection.INBOUND, + remote_addr=("127.0.0.1", 4001), + ) + + reset_error: QUICStreamResetError | None = None + + async def read_and_capture() -> None: + nonlocal reset_error + try: + await stream.read() + except QUICStreamResetError as exc: + reset_error = exc + + async with trio.open_nursery() as nursery: + nursery.start_soon(read_and_capture) + await wait_all_tasks_blocked() + await stream.handle_reset(error_code=1) + + assert reset_error is not None + assert "reset" in str(reset_error).lower()