WIP: Implement eth/66

tests/core/json-rpc/test_ipc.py

@@ -678,7 +678,7 @@ async def test_admin_peers(
             def to_remote_address(session):
                 return f"{session.remote.address.ip}:{session.remote.address.tcp_port}"
 
-            assert json_bob['caps'] == ['eth/63', 'eth/65']
+            assert json_bob['caps'] == ['eth/63', 'eth/66']
             assert json_bob['enode'] == alice.connection.session.remote.uri()
             assert json_bob['id'] == str(alice.connection.session.id)
             assert json_bob['name'] == 'bob'
@@ -687,7 +687,7 @@ def to_remote_address(session):
             assert bob_network['localAddress'] == '0.0.0.0:30303'
             assert bob_network['remoteAddress'] == to_remote_address(alice.connection.session)
 
-            assert json_alice['caps'] == ['eth/63', 'eth/65']
+            assert json_alice['caps'] == ['eth/63', 'eth/66']
             assert json_alice['enode'] == bob.connection.session.remote.uri()
             assert json_alice['id'] == str(bob.connection.session.id)
             assert json_alice['name'] == 'alice'

tests/core/p2p-proto/test_eth_api.py

@@ -15,9 +15,10 @@
 from trinity._utils.assertions import assert_type_equality
 from trinity.db.eth1.header import AsyncHeaderDB
 from trinity.exceptions import WrongForkIDFailure
-from trinity.protocol.eth.api import ETHV65API, ETHV63API, ETHV64API
+from trinity.protocol.eth.api import ETHV65API, ETHV63API, ETHV64API, ETHV66API
 from trinity.protocol.eth.commands import (
     GetBlockHeadersV65,
+    GetBlockHeadersV66,
     GetNodeDataV65,
     NewBlock,
     Status,
@@ -28,6 +29,7 @@
     ETHProtocolV63,
     ETHProtocolV64,
     ETHProtocolV65,
+    ETHProtocolV66,
 )
 
 from trinity.tools.factories.common import (
@@ -45,6 +47,13 @@
 )
 
 
+def get_highest_eth_protocol_version(client):
+    return max(
+        protocol.as_capability()[1] for protocol in client.connection.get_protocols()
+        if protocol.as_capability()[0] == 'eth'
+    )
+
+
 @pytest.fixture
 def bob_chain():
     chain = build(
@@ -118,6 +127,8 @@ def protocol_specific_classes(alice):
         return ETHV64API, ETHHandshakeReceipt, Status, StatusPayloadFactory
     elif alice.connection.has_protocol(ETHProtocolV65):
         return ETHV65API, ETHHandshakeReceipt, Status, StatusPayloadFactory
+    elif alice.connection.has_protocol(ETHProtocolV66):
+        return ETHV66API, ETHHandshakeReceipt, Status, StatusPayloadFactory
     else:
         raise Exception("No ETH protocol found")
 
@@ -225,6 +236,9 @@ async def _handle_cmd(connection, cmd):
 
 @pytest.mark.asyncio
 async def test_eth_api_send_get_node_data(alice, bob):
+    if get_highest_eth_protocol_version(alice) > ETHProtocolV65.version:
+        pytest.skip("Test not applicable above eth/65")
+
     payload = tuple(BlockHashFactory.create_batch(5))
 
     command_fut = asyncio.Future()
@@ -242,6 +256,9 @@ async def _handle_cmd(connection, cmd):
 
 @pytest.mark.asyncio
 async def test_eth_api_send_get_block_headers(alice, bob):
+    if get_highest_eth_protocol_version(alice) > ETHProtocolV65.version:
+        pytest.skip("Test not applicable above eth/65")
+
     payload = BlockHeadersQueryFactory()
 
     command_fut = asyncio.Future()
@@ -262,6 +279,32 @@ async def _handle_cmd(connection, cmd):
     assert_type_equality(payload, result.payload)
 
 
+@pytest.mark.asyncio
+async def test_eth_api_send_get_block_headers_with_request_id(alice, bob):
+    if get_highest_eth_protocol_version(alice) < ETHProtocolV66.version:
+        pytest.skip("Test not applicable below eth/66")
+
+    payload = BlockHeadersQueryFactory()
+
+    command_fut = asyncio.Future()
+
+    async def _handle_cmd(connection, cmd):
+        command_fut.set_result(cmd)
+
+    bob.connection.add_command_handler(GetBlockHeadersV66, _handle_cmd)
+    request_id = alice.eth_api.send_get_block_headers(
+        block_number_or_hash=payload.block_number_or_hash,
+        max_headers=payload.block_number_or_hash,
+        skip=payload.skip,
+        reverse=payload.reverse,
+    )
+
+    result = await asyncio.wait_for(command_fut, timeout=1)
+    assert isinstance(result, GetBlockHeadersV66)
+    assert_type_equality(payload, result.payload.query)
+    assert result.payload.request_id == request_id
+
+
 @pytest.mark.asyncio
 async def test_handshake_with_incompatible_fork_id(alice_chain, bob_chain):
 

tests/core/p2p-proto/test_peer.py

@@ -7,7 +7,7 @@
 from trinity.protocol.eth.peer import ETHPeer
 from trinity.protocol.eth.proto import (
     ETHProtocolV63,
-    ETHProtocolV65,
+    ETHProtocolV66,
 )
 from trinity.protocol.les.peer import LESPeer
 from trinity.protocol.les.proto import (
@@ -63,8 +63,8 @@ async def test_ETH_peers():
         assert isinstance(alice, ETHPeer)
         assert isinstance(bob, ETHPeer)
 
-        assert isinstance(alice.sub_proto, ETHProtocolV65)
-        assert isinstance(bob.sub_proto, ETHProtocolV65)
+        assert isinstance(alice.sub_proto, ETHProtocolV66)
+        assert isinstance(bob.sub_proto, ETHProtocolV66)
 
 
 @pytest.mark.asyncio

tests/core/p2p-proto/test_peer_block_body_validator_api.py

@@ -19,7 +19,7 @@
 
 from trinity.rlp.block_body import BlockBody
 
-from trinity.tools.factories import LatestETHPeerPairFactory
+from trinity.tools.factories import LatestETHPeerPairFactory, ETHV65PeerPairFactory
 
 
 def mk_uncle(block_number):
@@ -73,25 +73,36 @@ def mk_headers(*counts):
         yield mk_header_and_body(idx, num_transactions, num_uncles)
 
 
-@pytest.fixture
-async def eth_peer_and_remote():
-    async with LatestETHPeerPairFactory() as (peer, remote):
+@pytest.fixture(
+    params=(
+        ETHV65PeerPairFactory,
+        LatestETHPeerPairFactory,
+    )
+)
+async def eth_peer_and_remote(request):
+    async with request.param() as (peer, remote):
         yield peer, remote
 
 
+def get_request_id_or_none(request):
+    try:
+        return request.payload.request_id
+    except AttributeError:
+        return None
+
+
 @pytest.mark.asyncio
 async def test_eth_peer_get_block_bodies_round_trip_with_empty_response(eth_peer_and_remote):
     peer, remote = eth_peer_and_remote
 
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
 
-    async def send_block_bodies():
-        remote.eth_api.send_block_bodies([])
-        await asyncio.sleep(0)
+    async def send_block_bodies(_, cmd):
+        remote.eth_api.send_block_bodies([], get_request_id_or_none(cmd))
 
+    remote.connection.add_msg_handler(send_block_bodies)
     get_bodies_task = asyncio.ensure_future(peer.eth_api.get_block_bodies(headers))
-    asyncio.ensure_future(send_block_bodies())
 
     response = await get_bodies_task
 
@@ -105,15 +116,15 @@ async def test_eth_peer_get_block_bodies_round_trip_with_full_response(eth_peer_
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
 
-    async def send_block_bodies():
-        remote.eth_api.send_block_bodies(bodies)
-        await asyncio.sleep(0)
+    async def send_block_bodies(_, cmd):
+        remote.eth_api.send_block_bodies(bodies, get_request_id_or_none(cmd))
+
+    remote.connection.add_msg_handler(send_block_bodies)
 
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
 
     get_bodies_task = asyncio.ensure_future(peer.eth_api.get_block_bodies(headers))
-    asyncio.ensure_future(send_block_bodies())
 
     response = await get_bodies_task
 
@@ -128,15 +139,15 @@ async def test_eth_peer_get_block_bodies_round_trip_with_partial_response(eth_pe
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
 
-    async def send_block_bodies():
-        remote.eth_api.send_block_bodies(bodies[1:])
+    async def send_block_bodies(_, cmd):
+        remote.eth_api.send_block_bodies(bodies[1:], get_request_id_or_none(cmd))
         await asyncio.sleep(0)
 
+    remote.connection.add_msg_handler(send_block_bodies)
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
 
     get_bodies_task = asyncio.ensure_future(peer.eth_api.get_block_bodies(headers))
-    asyncio.ensure_future(send_block_bodies())
 
     response = await get_bodies_task
 
@@ -151,19 +162,19 @@ async def test_eth_peer_get_block_bodies_round_trip_with_noise(eth_peer_and_remo
     headers_bundle = mk_headers((2, 3), (8, 4), (0, 1), (0, 0))
     headers, bodies, transactions_roots, trie_data_dicts, uncles_hashes = zip(*headers_bundle)
 
-    async def send_block_bodies():
-        remote.eth_api.send_node_data((b'', b'arst'))
+    async def send_block_bodies(_, cmd):
+        remote.eth_api.send_node_data((b'', b'arst'), get_request_id_or_none(cmd))
         await asyncio.sleep(0)
-        remote.eth_api.send_block_bodies(bodies)
+        remote.eth_api.send_block_bodies(bodies, get_request_id_or_none(cmd))
         await asyncio.sleep(0)
-        remote.eth_api.send_node_data((b'', b'arst'))
+        remote.eth_api.send_node_data((b'', b'arst'), get_request_id_or_none(cmd))
         await asyncio.sleep(0)
 
+    remote.connection.add_msg_handler(send_block_bodies)
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
 
     get_bodies_task = asyncio.ensure_future(peer.eth_api.get_block_bodies(headers))
-    asyncio.ensure_future(send_block_bodies())
 
     response = await get_bodies_task
 
@@ -181,17 +192,17 @@ async def test_eth_peer_get_block_bodies_round_trip_no_match_invalid_response(et
     wrong_headers_bundle = mk_headers((4, 1), (3, 5), (2, 0), (7, 3))
     _, wrong_bodies, _, _, _ = zip(*wrong_headers_bundle)
 
-    async def send_block_bodies():
-        remote.eth_api.send_block_bodies(wrong_bodies)
+    async def send_block_bodies(_, cmd):
+        remote.eth_api.send_block_bodies(wrong_bodies, get_request_id_or_none(cmd))
         await asyncio.sleep(0)
-        remote.eth_api.send_block_bodies(bodies)
+        remote.eth_api.send_block_bodies(bodies, get_request_id_or_none(cmd))
         await asyncio.sleep(0)
 
+    remote.connection.add_msg_handler(send_block_bodies)
     transactions_bundles = tuple(zip(transactions_roots, trie_data_dicts))
     bodies_bundle = tuple(zip(bodies, transactions_bundles, uncles_hashes))
 
     get_bodies_task = asyncio.ensure_future(peer.eth_api.get_block_bodies(headers))
-    asyncio.ensure_future(send_block_bodies())
 
     response = await get_bodies_task
 

tests/core/p2p-proto/test_peer_block_header_validator_api.py

@@ -25,6 +25,13 @@ async def next_request_id(self):
         return msg.command.payload.request_id
 
 
+def get_request_id_or_none(request):
+    try:
+        return request.payload.request_id
+    except AttributeError:
+        return None
+
+
 @to_tuple
 def mk_header_chain(length):
     assert length >= 1
@@ -72,11 +79,11 @@ async def test_eth_peer_get_headers_round_trip(eth_peer_and_remote,
                                                headers):
     peer, remote = eth_peer_and_remote
 
-    async def send_headers():
-        remote.eth_api.send_block_headers(headers)
+    async def send_headers(_, cmd):
+        remote.eth_api.send_block_headers(headers, get_request_id_or_none(cmd))
 
+    remote.connection.add_msg_handler(send_headers)
     get_headers_task = asyncio.ensure_future(peer.chain_api.get_block_headers(*params))
-    asyncio.ensure_future(send_headers())
 
     response = await get_headers_task
 
@@ -90,13 +97,11 @@ async def test_eth_peer_get_headers_round_trip_concurrent_requests(eth_peer_and_
     peer, remote = eth_peer_and_remote
     headers = mk_header_chain(1)
 
-    async def send_headers():
-        await asyncio.sleep(0.01)
-        remote.eth_api.send_block_headers(headers)
+    async def send_headers(_, cmd):
         await asyncio.sleep(0.01)
-        remote.eth_api.send_block_headers(headers)
-        await asyncio.sleep(0.01)
-        remote.eth_api.send_block_headers(headers)
+        remote.eth_api.send_block_headers(headers, get_request_id_or_none(cmd))
+
+    remote.connection.add_msg_handler(send_headers)
 
     params = (0, 1, 0, False)
 
@@ -105,7 +110,7 @@ async def send_headers():
         asyncio.ensure_future(peer.chain_api.get_block_headers(*params)),
         asyncio.ensure_future(peer.chain_api.get_block_headers(*params)),
     ]
-    asyncio.ensure_future(send_headers())
+
     results = await asyncio.gather(*tasks)
 
     for response in results:
@@ -148,14 +153,14 @@ async def test_eth_peer_get_headers_round_trip_with_noise(eth_peer_and_remote):
 
     headers = mk_header_chain(10)
 
-    async def send_responses():
-        remote.eth_api.send_node_data([b'arst', b'tsra'])
+    async def send_responses(_, cmd):
+        remote.eth_api.send_node_data([b'arst', b'tsra'], get_request_id_or_none(cmd))
         await asyncio.sleep(0)
-        remote.eth_api.send_block_headers(headers)
+        remote.eth_api.send_block_headers(headers, get_request_id_or_none(cmd))
         await asyncio.sleep(0)
 
+    remote.connection.add_msg_handler(send_responses)
     get_headers_task = asyncio.ensure_future(peer.chain_api.get_block_headers(0, 10, 0, False))
-    asyncio.ensure_future(send_responses())
 
     response = await get_headers_task
 
@@ -172,16 +177,16 @@ async def test_eth_peer_get_headers_round_trip_does_not_match_invalid_response(e
 
     wrong_headers = mk_header_chain(10)[3:8]
 
-    async def send_responses():
-        remote.eth_api.send_node_data([b'arst', b'tsra'])
+    async def send_responses(_, cmd):
+        remote.eth_api.send_node_data([b'arst', b'tsra'], get_request_id_or_none(cmd))
         await asyncio.sleep(0)
-        remote.eth_api.send_block_headers(wrong_headers)
+        remote.eth_api.send_block_headers(wrong_headers, get_request_id_or_none(cmd))
         await asyncio.sleep(0)
-        remote.eth_api.send_block_headers(headers)
+        remote.eth_api.send_block_headers(headers, get_request_id_or_none(cmd))
         await asyncio.sleep(0)
 
+    remote.connection.add_msg_handler(send_responses)
     get_headers_task = asyncio.ensure_future(peer.chain_api.get_block_headers(0, 5, 0, False))
-    asyncio.ensure_future(send_responses())
 
     response = await get_headers_task