Add tests for MplexConn

libp2p/host/basic_host.py

@@ -1,4 +1,3 @@
-import asyncio
 from typing import List, Sequence
 
 import multiaddr

libp2p/stream_muxer/mplex/mplex.py

@@ -38,7 +38,6 @@ class Mplex(IMuxedConn):
 
     _tasks: List["asyncio.Future[Any]"]
 
-    # TODO: `generic_protocol_handler` should be refactored out of mplex conn.
     def __init__(self, secured_conn: ISecureConn, peer_id: ID) -> None:
         """
         create a new muxed connection

tests/network/test_net_stream.py

@@ -7,9 +7,6 @@ from tests.constants import MAX_READ_LEN
 
 DATA = b"data_123"
 
-# TODO: Move `muxed_stream` specific(currently we are using `MplexStream`) tests to its
-#   own file, after `generic_protocol_handler` is refactored out of `Mplex`.
-
 
 @pytest.mark.asyncio
 async def test_net_stream_read_write(net_stream_pair):

tests/stream_muxer/conftest.py

@@ -10,6 +10,8 @@ async def mplex_conn_pair(is_host_secure):
     mplex_conn_0, swarm_0, mplex_conn_1, swarm_1 = await mplex_conn_pair_factory(
         is_host_secure
     )
+    assert mplex_conn_0.initiator
+    assert not mplex_conn_1.initiator
     try:
         yield mplex_conn_0, mplex_conn_1
     finally:

tests/stream_muxer/test_mplex_conn.py

@@ -1,6 +1,50 @@
+import asyncio
+
 import pytest
 
 
 @pytest.mark.asyncio
 async def test_mplex_conn(mplex_conn_pair):
     conn_0, conn_1 = mplex_conn_pair
+
+    assert len(conn_0.streams) == 0
+    assert len(conn_1.streams) == 0
+    assert not conn_0.event_shutting_down.is_set()
+    assert not conn_1.event_shutting_down.is_set()
+    assert not conn_0.event_closed.is_set()
+    assert not conn_1.event_closed.is_set()
+
+    # Test: Open a stream, and both side get 1 more stream.
+    stream_0 = await conn_0.open_stream()
+    await asyncio.sleep(0.01)
+    assert len(conn_0.streams) == 1
+    assert len(conn_1.streams) == 1
+    # Test: From another side.
+    stream_1 = await conn_1.open_stream()
+    await asyncio.sleep(0.01)
+    assert len(conn_0.streams) == 2
+    assert len(conn_1.streams) == 2
+
+    # Close from one side.
+    await conn_0.close()
+    # Sleep for a while for both side to handle `close`.
+    await asyncio.sleep(0.01)
+    # Test: Both side is closed.
+    assert conn_0.event_shutting_down.is_set()
+    assert conn_0.event_closed.is_set()
+    assert conn_1.event_shutting_down.is_set()
+    assert conn_1.event_closed.is_set()
+    # Test: All streams should have been closed.
+    assert stream_0.event_remote_closed.is_set()
+    assert stream_0.event_reset.is_set()
+    assert stream_0.event_local_closed.is_set()
+    assert conn_0.streams is None
+    # Test: All streams on the other side are also closed.
+    assert stream_1.event_remote_closed.is_set()
+    assert stream_1.event_reset.is_set()
+    assert stream_1.event_local_closed.is_set()
+    assert conn_1.streams is None
+
+    # Test: No effect to close more than once between two side.
+    await conn_0.close()
+    await conn_1.close()