diff --git a/.travis.yml b/.travis.yml index ce9b061e..6b6b3ccc 100644 --- a/.travis.yml +++ b/.travis.yml @@ -2,6 +2,9 @@ language: python matrix: include: + - python: 3.6-dev + dist: xenial + env: TOXENV=py36-test - python: 3.7-dev dist: xenial env: TOXENV=py37-test diff --git a/libp2p/network/connection/raw_connection.py b/libp2p/network/connection/raw_connection.py index 08d22055..4f01b124 100644 --- a/libp2p/network/connection/raw_connection.py +++ b/libp2p/network/connection/raw_connection.py @@ -1,4 +1,5 @@ import asyncio +import sys from .exceptions import RawConnError from .raw_connection_interface import IRawConnection @@ -52,4 +53,6 @@ class RawConnection(IRawConnection): async def close(self) -> None: self.writer.close() + if sys.version_info < (3, 7): + return await self.writer.wait_closed() diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index 25ab81e6..a44aa052 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -149,7 +149,7 @@ class Pubsub: # Map of topic to topic validator self.topic_validators = {} - self.counter = time.time_ns() + self.counter = int(time.time()) self._tasks = [] # Call handle peer to keep waiting for updates to peer queue diff --git a/libp2p/tools/factories.py b/libp2p/tools/factories.py index 1b98eaa6..2d1f99ad 100644 --- a/libp2p/tools/factories.py +++ b/libp2p/tools/factories.py @@ -1,7 +1,8 @@ import asyncio -from contextlib import asynccontextmanager from typing import Any, AsyncIterator, Dict, Tuple, cast +# NOTE: import ``asynccontextmanager`` from ``contextlib`` when support for python 3.6 is dropped. +from async_generator import asynccontextmanager import factory from libp2p import generate_new_rsa_identity, generate_peer_id_from @@ -173,7 +174,7 @@ async def host_pair_factory(is_secure: bool) -> Tuple[BasicHost, BasicHost]: return hosts[0], hosts[1] -@asynccontextmanager +@asynccontextmanager # type: ignore async def pair_of_connected_hosts( is_secure: bool = True ) -> AsyncIterator[Tuple[BasicHost, BasicHost]]: diff --git a/libp2p/tools/pubsub/floodsub_integration_test_settings.py b/libp2p/tools/pubsub/floodsub_integration_test_settings.py index b7b66244..d6b5b678 100644 --- a/libp2p/tools/pubsub/floodsub_integration_test_settings.py +++ b/libp2p/tools/pubsub/floodsub_integration_test_settings.py @@ -143,6 +143,14 @@ floodsub_protocol_pytest_params = [ ] +def _collect_node_ids(adj_list): + node_ids = set() + for node, neighbors in adj_list.items(): + node_ids.add(node) + node_ids.update(set(neighbors)) + return node_ids + + async def perform_test_from_obj(obj, router_factory) -> None: """ Perform pubsub tests from a test object, which is composed as follows: @@ -180,59 +188,43 @@ async def perform_test_from_obj(obj, router_factory) -> None: node_map = {} pubsub_map = {} - async def add_node(node_id_str: str) -> None: + async def add_node(node_id_str: str): pubsub_router = router_factory(protocols=obj["supported_protocols"]) pubsub = PubsubFactory(router=pubsub_router) await pubsub.host.get_network().listen(LISTEN_MADDR) node_map[node_id_str] = pubsub.host pubsub_map[node_id_str] = pubsub - tasks_connect = [] - for start_node_id in adj_list: - # Create node if node does not yet exist - if start_node_id not in node_map: - await add_node(start_node_id) + all_node_ids = _collect_node_ids(adj_list) - # For each neighbor of start_node, create if does not yet exist, - # then connect start_node to neighbor - for neighbor_id in adj_list[start_node_id]: - # Create neighbor if neighbor does not yet exist - if neighbor_id not in node_map: - await add_node(neighbor_id) - tasks_connect.append( - connect(node_map[start_node_id], node_map[neighbor_id]) - ) - # Connect nodes and wait at least for 2 seconds - await asyncio.gather(*tasks_connect, asyncio.sleep(2)) + for node in all_node_ids: + await add_node(node) + + for node, neighbors in adj_list.items(): + for neighbor_id in neighbors: + await connect(node_map[node], node_map[neighbor_id]) + + # NOTE: the test using this routine will fail w/o these sleeps... + await asyncio.sleep(1) # Step 2) Subscribe to topics queues_map = {} topic_map = obj["topic_map"] - tasks_topic = [] - tasks_topic_data = [] for topic, node_ids in topic_map.items(): for node_id in node_ids: - tasks_topic.append(pubsub_map[node_id].subscribe(topic)) - tasks_topic_data.append((node_id, topic)) - tasks_topic.append(asyncio.sleep(2)) + queue = await pubsub_map[node_id].subscribe(topic) + if node_id not in queues_map: + queues_map[node_id] = {} + # Store queue in topic-queue map for node + queues_map[node_id][topic] = queue - # Gather is like Promise.all - responses = await asyncio.gather(*tasks_topic) - for i in range(len(responses) - 1): - node_id, topic = tasks_topic_data[i] - if node_id not in queues_map: - queues_map[node_id] = {} - # Store queue in topic-queue map for node - queues_map[node_id][topic] = responses[i] - - # Allow time for subscribing before continuing - await asyncio.sleep(0.01) + # NOTE: the test using this routine will fail w/o these sleeps... + await asyncio.sleep(1) # Step 3) Publish messages topics_in_msgs_ordered = [] messages = obj["messages"] - tasks_publish = [] for msg in messages: topics = msg["topics"] @@ -242,21 +234,17 @@ async def perform_test_from_obj(obj, router_factory) -> None: # Publish message # TODO: Should be single RPC package with several topics for topic in topics: - tasks_publish.append(pubsub_map[node_id].publish(topic, data)) - - # For each topic in topics, add (topic, node_id, data) tuple to ordered test list - for topic in topics: + await pubsub_map[node_id].publish(topic, data) + # For each topic in topics, add (topic, node_id, data) tuple to ordered test list topics_in_msgs_ordered.append((topic, node_id, data)) - # Allow time for publishing before continuing - await asyncio.gather(*tasks_publish, asyncio.sleep(2)) - # Step 4) Check that all messages were received correctly. for topic, origin_node_id, data in topics_in_msgs_ordered: # Look at each node in each topic for node_id in topic_map[topic]: # Get message from subscription queue - msg = await queues_map[node_id][topic].get() + queue = queues_map[node_id][topic] + msg = await queue.get() assert data == msg.data # Check the message origin assert node_map[origin_node_id].get_id().to_bytes() == msg.from_id diff --git a/libp2p/transport/tcp/tcp.py b/libp2p/transport/tcp/tcp.py index 7470510d..f5c2aa40 100644 --- a/libp2p/transport/tcp/tcp.py +++ b/libp2p/transport/tcp/tcp.py @@ -1,5 +1,6 @@ import asyncio from socket import socket +import sys from typing import List from multiaddr import Multiaddr @@ -53,8 +54,11 @@ class TCPListener(IListener): if self.server is None: return self.server.close() - await self.server.wait_closed() + server = self.server self.server = None + if sys.version_info < (3, 7): + return + await server.wait_closed() class TCP(ITransport): diff --git a/setup.py b/setup.py index a7b1c367..56a6bf20 100644 --- a/setup.py +++ b/setup.py @@ -58,6 +58,8 @@ install_requires = [ "protobuf>=3.10.0,<4.0.0", "coincurve>=10.0.0,<11.0.0", "pynacl==1.3.0", + "dataclasses>=0.7, <1;python_version<'3.7'", + "async_generator==1.10", ] @@ -80,7 +82,7 @@ setup( url="https://github.com/libp2p/py-libp2p", include_package_data=True, install_requires=install_requires, - python_requires=">=3.7,<4", + python_requires=">=3.6,<4", extras_require=extras_require, py_modules=["libp2p"], license="MIT/APACHE2.0", @@ -94,6 +96,7 @@ setup( "License :: OSI Approved :: Apache Software License", "Natural Language :: English", "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.6", "Programming Language :: Python :: 3.7", ], platforms=["unix", "linux", "osx"], diff --git a/tests/pubsub/test_pubsub.py b/tests/pubsub/test_pubsub.py index 01d8ba74..6f3c6725 100644 --- a/tests/pubsub/test_pubsub.py +++ b/tests/pubsub/test_pubsub.py @@ -58,11 +58,11 @@ async def test_peers_subscribe(pubsubs_fsub): await connect(pubsubs_fsub[0].host, pubsubs_fsub[1].host) await pubsubs_fsub[0].subscribe(TESTING_TOPIC) # Yield to let 0 notify 1 - await asyncio.sleep(0.1) + await asyncio.sleep(1) assert pubsubs_fsub[0].my_id in pubsubs_fsub[1].peer_topics[TESTING_TOPIC] await pubsubs_fsub[0].unsubscribe(TESTING_TOPIC) # Yield to let 0 notify 1 - await asyncio.sleep(0.1) + await asyncio.sleep(1) assert pubsubs_fsub[0].my_id not in pubsubs_fsub[1].peer_topics[TESTING_TOPIC] diff --git a/tests/security/test_secio.py b/tests/security/test_secio.py index c7808b46..50374809 100644 --- a/tests/security/test_secio.py +++ b/tests/security/test_secio.py @@ -76,8 +76,8 @@ async def test_create_secure_session(): local_conn = InMemoryConnection(local_peer, is_initiator=True) remote_conn = InMemoryConnection(remote_peer) - local_pipe_task = asyncio.create_task(create_pipe(local_conn, remote_conn)) - remote_pipe_task = asyncio.create_task(create_pipe(remote_conn, local_conn)) + local_pipe_task = asyncio.ensure_future(create_pipe(local_conn, remote_conn)) + remote_pipe_task = asyncio.ensure_future(create_pipe(remote_conn, local_conn)) local_session_builder = create_secure_session( local_nonce, local_peer, local_key_pair.private_key, local_conn, remote_peer diff --git a/tests_interop/conftest.py b/tests_interop/conftest.py index 8ae9769b..08df614c 100644 --- a/tests_interop/conftest.py +++ b/tests_interop/conftest.py @@ -151,6 +151,8 @@ class DaemonStream(ReadWriteCloser): async def close(self) -> None: self.writer.close() + if sys.version_info < (3, 7): + return await self.writer.wait_closed() async def read(self, n: int = -1) -> bytes: @@ -196,7 +198,8 @@ async def py_to_daemon_stream_pair(hosts, p2pds, is_to_fail_daemon_stream): # some day. listener = p2pds[0].control.control.listener listener.close() - await listener.wait_closed() + if sys.version_info[0:2] > (3, 6): + await listener.wait_closed() stream_py = await host.new_stream(p2pd.peer_id, [protocol_id]) if not is_to_fail_daemon_stream: await event_stream_handled.wait() diff --git a/tox.ini b/tox.ini index de4f8f1b..21f46435 100644 --- a/tox.ini +++ b/tox.ini @@ -1,9 +1,9 @@ # Reference: https://github.com/ethereum/ethereum-python-project-template/blob/master/tox.ini -# TODO: consider py36 and pypy3 support +# TODO: consider pypy3 support [tox] envlist = - py37-test + py{36,37}-test py37-interop lint docs @@ -37,6 +37,7 @@ commands = basepython = docs: python py37: python3.7 + py36: python3.6 extras = test docs: doc