From b8b5ac5e06026fed267ac6972cd572098a04b833 Mon Sep 17 00:00:00 2001 From: mhchia Date: Tue, 17 Sep 2019 21:54:20 +0800 Subject: [PATCH] Add test for notifee disconnected --- libp2p/host/basic_host.py | 2 +- libp2p/network/network_interface.py | 8 +- libp2p/network/swarm.py | 15 +- libp2p/pubsub/pubsub.py | 2 +- tests/network/test_notify.py | 393 ++++++---------------------- 5 files changed, 87 insertions(+), 333 deletions(-) diff --git a/libp2p/host/basic_host.py b/libp2p/host/basic_host.py index 862fd5c9..5f0ccfd1 100644 --- a/libp2p/host/basic_host.py +++ b/libp2p/host/basic_host.py @@ -100,7 +100,7 @@ class BasicHost(IHost): :return: stream: new stream created """ - net_stream = await self._network.new_stream(peer_id, protocol_ids) + net_stream = await self._network.new_stream(peer_id) # Perform protocol muxing to determine protocol to use selected_protocol = await self.multiselect_client.select_one_of( diff --git a/libp2p/network/network_interface.py b/libp2p/network/network_interface.py index 470da1a9..94ddba2c 100644 --- a/libp2p/network/network_interface.py +++ b/libp2p/network/network_interface.py @@ -7,7 +7,7 @@ from libp2p.network.connection.net_connection_interface import INetConn from libp2p.peer.id import ID from libp2p.peer.peerstore_interface import IPeerStore from libp2p.transport.listener_interface import IListener -from libp2p.typing import StreamHandlerFn, TProtocol +from libp2p.typing import StreamHandlerFn from .stream.net_stream_interface import INetStream @@ -38,9 +38,7 @@ class INetwork(ABC): """ @abstractmethod - async def new_stream( - self, peer_id: ID, protocol_ids: Sequence[TProtocol] - ) -> INetStream: + async def new_stream(self, peer_id: ID) -> INetStream: """ :param peer_id: peer_id of destination :param protocol_ids: available protocol ids to use for stream @@ -61,7 +59,7 @@ class INetwork(ABC): """ @abstractmethod - def notify(self, notifee: "INotifee") -> bool: + def register_notifee(self, notifee: "INotifee") -> None: """ :param notifee: object implementing Notifee interface :return: true if notifee registered successfully, false otherwise diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py index a0d9d77a..18218d67 100644 --- a/libp2p/network/swarm.py +++ b/libp2p/network/swarm.py @@ -1,6 +1,6 @@ import asyncio import logging -from typing import Dict, List, Optional, Sequence +from typing import Dict, List, Optional from multiaddr import Multiaddr @@ -14,7 +14,7 @@ from libp2p.transport.exceptions import MuxerUpgradeFailure, SecurityUpgradeFail from libp2p.transport.listener_interface import IListener from libp2p.transport.transport_interface import ITransport from libp2p.transport.upgrader import TransportUpgrader -from libp2p.typing import StreamHandlerFn, TProtocol +from libp2p.typing import StreamHandlerFn from .connection.raw_connection import RawConnection from .connection.swarm_connection import SwarmConn @@ -131,9 +131,7 @@ class Swarm(INetwork): return swarm_conn - async def new_stream( - self, peer_id: ID, protocol_ids: Sequence[TProtocol] - ) -> INetStream: + async def new_stream(self, peer_id: ID) -> INetStream: """ :param peer_id: peer_id of destination :param protocol_id: protocol id @@ -229,15 +227,12 @@ class Swarm(INetwork): # No maddr succeeded return False - def notify(self, notifee: INotifee) -> bool: + def register_notifee(self, notifee: INotifee) -> None: """ :param notifee: object implementing Notifee interface :return: true if notifee registered successfully, false otherwise """ - if isinstance(notifee, INotifee): - self.notifees.append(notifee) - return True - return False + self.notifees.append(notifee) def add_router(self, router: IPeerRouting) -> None: self.router = router diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index 17cf0344..a295b697 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -95,7 +95,7 @@ class Pubsub: # Register a notifee self.peer_queue = asyncio.Queue() - self.host.get_network().notify(PubsubNotifee(self.peer_queue)) + self.host.get_network().register_notifee(PubsubNotifee(self.peer_queue)) # Register stream handlers for each pubsub router protocol to handle # the pubsub streams opened on those protocols diff --git a/tests/network/test_notify.py b/tests/network/test_notify.py index e47a044d..aaf0ed55 100644 --- a/tests/network/test_notify.py +++ b/tests/network/test_notify.py @@ -5,347 +5,108 @@ called, and that the stream passed into opened_stream is correct Note: Listen event does not get hit because MyNotifee is passed into network after network has already started listening -TODO: Add tests for closed_stream disconnected, listen_close when those +TODO: Add tests for closed_stream, listen_close when those features are implemented in swarm """ -import multiaddr +import asyncio +import enum + import pytest -from libp2p import initialize_default_swarm, new_node -from libp2p.crypto.rsa import create_new_key_pair -from libp2p.host.basic_host import BasicHost from libp2p.network.notifee_interface import INotifee -from tests.constants import MAX_READ_LEN -from tests.utils import perform_two_host_set_up +from tests.configs import LISTEN_MADDR +from tests.factories import SwarmFactory +from tests.utils import connect_swarm -ACK = "ack:" + +class Event(enum.Enum): + OpenedStream = 0 + ClosedStream = 1 # Not implemented + Connected = 2 + Disconnected = 3 + Listen = 4 + ListenClose = 5 # Not implemented class MyNotifee(INotifee): - def __init__(self, events, val_to_append_to_event): + def __init__(self, events): self.events = events - self.val_to_append_to_event = val_to_append_to_event async def opened_stream(self, network, stream): - self.events.append(["opened_stream" + self.val_to_append_to_event, stream]) + self.events.append(Event.OpenedStream) async def closed_stream(self, network, stream): + # TODO: It is not implemented yet. pass async def connected(self, network, conn): - self.events.append(["connected" + self.val_to_append_to_event, conn.conn]) + self.events.append(Event.Connected) async def disconnected(self, network, conn): - pass + self.events.append(Event.Disconnected) async def listen(self, network, _multiaddr): - self.events.append(["listened" + self.val_to_append_to_event, _multiaddr]) + self.events.append(Event.Listen) async def listen_close(self, network, _multiaddr): + # TODO: It is not implemented yet. pass -class InvalidNotifee: - def __init__(self): - pass - - async def opened_stream(self): - assert False - - async def closed_stream(self): - assert False - - async def connected(self): - assert False - - async def disconnected(self): - assert False - - async def listen(self): - assert False - - @pytest.mark.asyncio -async def test_one_notifier(): - node_a, node_b = await perform_two_host_set_up() - - # Add notifee for node_a - events = [] - assert node_a.get_network().notify(MyNotifee(events, "0")) - - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # Ensure the connected and opened_stream events were hit in MyNotifee obj - # and that stream passed into opened_stream matches the stream created on - # node_a - assert events == [["connected0", stream.muxed_conn], ["opened_stream0", stream]] - - messages = ["hello", "hello"] - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. - - -@pytest.mark.asyncio -async def test_one_notifier_on_two_nodes(): - events_b = [] - messages = ["hello", "hello"] - - async def my_stream_handler(stream): - # Ensure the connected and opened_stream events were hit in Notifee obj - # and that the stream passed into opened_stream matches the stream created on - # node_b - assert events_b == [ - ["connectedb", stream.muxed_conn], - ["opened_streamb", stream], - ] - for message in messages: - read_string = (await stream.read(len(message))).decode() - - resp = ACK + read_string - await stream.write(resp.encode()) - - node_a, node_b = await perform_two_host_set_up(my_stream_handler) - - # Add notifee for node_a - events_a = [] - assert node_a.get_network().notify(MyNotifee(events_a, "a")) - - # Add notifee for node_b - assert node_b.get_network().notify(MyNotifee(events_b, "b")) - - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # Ensure the connected and opened_stream events were hit in MyNotifee obj - # and that stream passed into opened_stream matches the stream created on - # node_a - assert events_a == [["connecteda", stream.muxed_conn], ["opened_streama", stream]] - - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. - - -@pytest.mark.asyncio -async def test_one_notifier_on_two_nodes_with_listen(): - events_b = [] - messages = ["hello", "hello"] - - node_a_key_pair = create_new_key_pair() - node_a_transport_opt = ["/ip4/127.0.0.1/tcp/0"] - node_a = await new_node(node_a_key_pair, transport_opt=node_a_transport_opt) - await node_a.get_network().listen(multiaddr.Multiaddr(node_a_transport_opt[0])) - - # Set up node_b swarm to pass into host - node_b_key_pair = create_new_key_pair() - node_b_transport_opt = ["/ip4/127.0.0.1/tcp/0"] - node_b_multiaddr = multiaddr.Multiaddr(node_b_transport_opt[0]) - node_b_swarm = initialize_default_swarm( - node_b_key_pair, transport_opt=node_b_transport_opt - ) - node_b = BasicHost(node_b_swarm) - - async def my_stream_handler(stream): - # Ensure the listened, connected and opened_stream events were hit in Notifee obj - # and that the stream passed into opened_stream matches the stream created on - # node_b - assert events_b == [ - ["listenedb", node_b_multiaddr], - ["connectedb", stream.muxed_conn], - ["opened_streamb", stream], - ] - for message in messages: - read_string = (await stream.read(len(message))).decode() - resp = ACK + read_string - await stream.write(resp.encode()) - - # Add notifee for node_a - events_a = [] - assert node_a.get_network().notify(MyNotifee(events_a, "a")) - - # Add notifee for node_b - assert node_b.get_network().notify(MyNotifee(events_b, "b")) - - # start listen on node_b_swarm - await node_b.get_network().listen(node_b_multiaddr) - - node_b.set_stream_handler("/echo/1.0.0", my_stream_handler) - # Associate the peer with local ip address (see default parameters of Libp2p()) - node_a.get_peerstore().add_addrs(node_b.get_id(), node_b.get_addrs(), 10) - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # Ensure the connected and opened_stream events were hit in MyNotifee obj - # and that stream passed into opened_stream matches the stream created on - # node_a - assert events_a == [["connecteda", stream.muxed_conn], ["opened_streama", stream]] - - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. - - -@pytest.mark.asyncio -async def test_two_notifiers(): - node_a, node_b = await perform_two_host_set_up() - - # Add notifee for node_a - events0 = [] - assert node_a.get_network().notify(MyNotifee(events0, "0")) - - events1 = [] - assert node_a.get_network().notify(MyNotifee(events1, "1")) - - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # Ensure the connected and opened_stream events were hit in both Notifee objs - # and that the stream passed into opened_stream matches the stream created on - # node_a - assert events0 == [["connected0", stream.muxed_conn], ["opened_stream0", stream]] - assert events1 == [["connected1", stream.muxed_conn], ["opened_stream1", stream]] - - messages = ["hello", "hello"] - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. - - -@pytest.mark.asyncio -async def test_ten_notifiers(): - num_notifiers = 10 - - node_a, node_b = await perform_two_host_set_up() - - # Add notifee for node_a - events_lst = [] - for i in range(num_notifiers): - events_lst.append([]) - assert node_a.get_network().notify(MyNotifee(events_lst[i], str(i))) - - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # Ensure the connected and opened_stream events were hit in both Notifee objs - # and that the stream passed into opened_stream matches the stream created on - # node_a - for i in range(num_notifiers): - assert events_lst[i] == [ - ["connected" + str(i), stream.muxed_conn], - ["opened_stream" + str(i), stream], - ] - - messages = ["hello", "hello"] - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. - - -@pytest.mark.asyncio -async def test_ten_notifiers_on_two_nodes(): - num_notifiers = 10 - events_lst_b = [] - - async def my_stream_handler(stream): - # Ensure the connected and opened_stream events were hit in all Notifee objs - # and that the stream passed into opened_stream matches the stream created on - # node_b - for i in range(num_notifiers): - assert events_lst_b[i] == [ - ["connectedb" + str(i), stream.muxed_conn], - ["opened_streamb" + str(i), stream], - ] - while True: - read_string = (await stream.read(MAX_READ_LEN)).decode() - - resp = ACK + read_string - await stream.write(resp.encode()) - - node_a, node_b = await perform_two_host_set_up(my_stream_handler) - - # Add notifee for node_a and node_b - events_lst_a = [] - for i in range(num_notifiers): - events_lst_a.append([]) - events_lst_b.append([]) - assert node_a.get_network().notify(MyNotifee(events_lst_a[i], "a" + str(i))) - assert node_b.get_network().notify(MyNotifee(events_lst_b[i], "b" + str(i))) - - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # Ensure the connected and opened_stream events were hit in all Notifee objs - # and that the stream passed into opened_stream matches the stream created on - # node_a - for i in range(num_notifiers): - assert events_lst_a[i] == [ - ["connecteda" + str(i), stream.muxed_conn], - ["opened_streama" + str(i), stream], - ] - - messages = ["hello", "hello"] - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. - - -@pytest.mark.asyncio -async def test_invalid_notifee(): - num_notifiers = 10 - - node_a, node_b = await perform_two_host_set_up() - - # Add notifee for node_a - events_lst = [] - for _ in range(num_notifiers): - events_lst.append([]) - assert not node_a.get_network().notify(InvalidNotifee()) - - stream = await node_a.new_stream(node_b.get_id(), ["/echo/1.0.0"]) - - # If this point is reached, this implies that the InvalidNotifee instance - # did not assert false, i.e. no functions of InvalidNotifee were called (which is correct - # given that InvalidNotifee should not have been added as a notifee) - messages = ["hello", "hello"] - for message in messages: - expected_resp = ACK + message - await stream.write(message.encode()) - - response = (await stream.read(len(expected_resp))).decode() - - assert response == expected_resp - - # Success, terminate pending tasks. +async def test_notify(is_host_secure): + swarms = [SwarmFactory(is_host_secure) for _ in range(2)] + + events_0_0 = [] + events_1_0 = [] + events_0_without_listen = [] + swarms[0].register_notifee(MyNotifee(events_0_0)) + swarms[1].register_notifee(MyNotifee(events_1_0)) + # Listen + await asyncio.gather(*[swarm.listen(LISTEN_MADDR) for swarm in swarms]) + + swarms[0].register_notifee(MyNotifee(events_0_without_listen)) + + # Connected + await connect_swarm(swarms[0], swarms[1]) + # OpenedStream: first + await swarms[0].new_stream(swarms[1].get_peer_id()) + # OpenedStream: second + await swarms[0].new_stream(swarms[1].get_peer_id()) + # OpenedStream: third, but different direction. + await swarms[1].new_stream(swarms[0].get_peer_id()) + + await asyncio.sleep(0.01) + + # TODO: Check `ClosedStream` and `ListenClose` events after they are ready. + + # Disconnected + await swarms[0].close_peer(swarms[1].get_peer_id()) + await asyncio.sleep(0.01) + + # Connected again, but different direction. + await connect_swarm(swarms[1], swarms[0]) + await asyncio.sleep(0.01) + + # Disconnected again, but different direction. + await swarms[1].close_peer(swarms[0].get_peer_id()) + await asyncio.sleep(0.01) + + expected_events_without_listen = [ + Event.Connected, + Event.OpenedStream, + Event.OpenedStream, + Event.OpenedStream, + Event.Disconnected, + Event.Connected, + Event.Disconnected, + ] + expected_events = [Event.Listen] + expected_events_without_listen + + assert events_0_0 == expected_events + assert events_1_0 == expected_events + assert events_0_without_listen == expected_events_without_listen + + # Clean up + await asyncio.gather(*[swarm.close() for swarm in swarms])