More sx manager additions:

- When subscribing, create typed subscriptions based on the
  subscription params.

- Add ``get_by_id()`` and ``get_by_label()`` methods to the
  ``SubscriptionManager`` class.

- Allow event to be passed to logs subscriptions for easier
  processing when the messages are received.

- Add ability to count handler calls on both the manager and on a per-subscription basis

- Add tests for subscription manager; tighten up typing.

- Fix some types:
  - ``process_log`` expects a ``LogReceipt``, not a hex.
  - ``FormattedEthSubscriptionResponse`` should also include
    ``HexBytes`` for pending tx hashes (full_transactions=False).

- Create live integration tests for ``eth_subscribe`` for "newHeads", "newPendingTransactions", "logs". Syncing still has to be stubbed out.

Author: fselmo

tests/core/subscriptions/test_subscription_manager.py

@@ -0,0 +1,104 @@
+import itertools
+import pytest
+from unittest.mock import (
+    AsyncMock,
+)
+
+import pytest_asyncio
+
+from web3 import (
+    AsyncWeb3,
+    PersistentConnectionProvider,
+)
+from web3.exceptions import (
+    Web3ValueError,
+)
+from web3.providers.persistent.subscription_manager import (
+    SubscriptionManager,
+)
+from web3.utils.subscriptions import (
+    LogsSubscription,
+    NewHeadsSubscription,
+    PendingTxSubscription,
+)
+
+
+class MockProvider(PersistentConnectionProvider):
+    socket_recv = AsyncMock()
+    socket_send = AsyncMock()
+
+
+@pytest_asyncio.fixture
+async def subscription_manager():
+    countr = itertools.count()
+    _w3 = AsyncWeb3(MockProvider())
+    _w3.eth._subscribe = AsyncMock()
+    _w3.eth._subscribe.side_effect = lambda *_: f"0x{str(next(countr))}"
+    _w3.eth._unsubscribe = AsyncMock()
+    _w3.eth._unsubscribe.return_value = True
+    yield SubscriptionManager(_w3)
+
+
+@pytest.mark.asyncio
+async def test_subscription_manager_raises_for_sx_with_the_same_label(
+    subscription_manager,
+):
+    sx1 = NewHeadsSubscription(label="foo")
+    await subscription_manager.subscribe(sx1)
+
+    with pytest.raises(
+        Web3ValueError,
+        match="Subscription label already exists. Subscriptions must have unique "
+        "labels.\n    label: foo",
+    ):
+        sx2 = LogsSubscription(label="foo")
+        await subscription_manager.subscribe(sx2)
+
+    # make sure the subscription was subscribed to and not added to the manager
+    assert subscription_manager.subscriptions == [sx1]
+    assert subscription_manager._subscriptions_by_label == {"foo": sx1}
+    assert subscription_manager._subscriptions_by_id == {"0x0": sx1}
+
+
+@pytest.mark.asyncio
+async def test_subscription_manager_get_by_id(subscription_manager):
+    sx = NewHeadsSubscription(label="foo")
+    await subscription_manager.subscribe(sx)
+    assert subscription_manager.get_by_id("0x0") == sx
+    assert subscription_manager.get_by_id("0x1") is None
+
+
+@pytest.mark.asyncio
+async def test_subscription_manager_get_by_label(subscription_manager):
+    sx = NewHeadsSubscription(label="foo")
+    await subscription_manager.subscribe(sx)
+    assert subscription_manager.get_by_label("foo") == sx
+    assert subscription_manager.get_by_label("bar") is None
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_one_by_one_clears_all_subscriptions(
+    subscription_manager,
+):
+    sx1 = NewHeadsSubscription(label="foo")
+    sx2 = PendingTxSubscription(label="bar")
+    await subscription_manager.subscribe(sx1)
+    await subscription_manager.subscribe(sx2)
+
+    await subscription_manager.unsubscribe(sx1)
+    assert subscription_manager.subscriptions == [sx2]
+
+    await subscription_manager.unsubscribe(sx2)
+    assert subscription_manager.subscriptions == []
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_all_clears_all_subscriptions(subscription_manager):
+    sx1 = NewHeadsSubscription(label="foo")
+    sx2 = PendingTxSubscription(label="bar")
+    await subscription_manager.subscribe([sx1, sx2])
+
+    await subscription_manager.unsubscribe_all()
+    assert subscription_manager.subscriptions == []
+    assert subscription_manager._subscriptions_by_id == {}
+    assert subscription_manager._subscriptions_by_label == {}

tests/core/subscriptions/test_subscriptions.py

@@ -0,0 +1,86 @@
+import pytest
+from unittest.mock import (
+    Mock,
+)
+
+from web3.utils.subscriptions import (
+    LogsSubscription,
+    NewHeadsSubscription,
+    PendingTxSubscription,
+    SyncingSubscription,
+)
+
+
+@pytest.fixture
+def handler():
+    pass
+
+
+def test_new_heads_subscription_properties(handler):
+    new_heads_subscription = NewHeadsSubscription(
+        handler=handler, label="new heads label"
+    )
+    assert new_heads_subscription._handler is handler
+    assert new_heads_subscription.label == "new heads label"
+    assert new_heads_subscription.subscription_params == ("newHeads",)
+
+
+def test_pending_tx_subscription_properties_default(handler):
+    pending_tx_subscription = PendingTxSubscription(
+        handler=handler, label="pending tx label"
+    )
+    assert pending_tx_subscription._handler is handler
+    assert pending_tx_subscription.label == "pending tx label"
+    assert pending_tx_subscription.subscription_params == (
+        "newPendingTransactions",
+        False,
+    )
+
+
+def test_pending_tx_subscription_properties_full_transactions(handler):
+    pending_tx_subscription = PendingTxSubscription(
+        full_transactions=True, handler=handler, label="pending tx label"
+    )
+    assert pending_tx_subscription._handler is handler
+    assert pending_tx_subscription.label == "pending tx label"
+    assert pending_tx_subscription.subscription_params == (
+        "newPendingTransactions",
+        True,
+    )
+
+
+def test_logs_subscription_properties_default(handler):
+    logs_subscription = LogsSubscription(handler=handler, label="logs label")
+    assert logs_subscription._handler is handler
+    assert logs_subscription.label == "logs label"
+    assert logs_subscription.subscription_params == ("logs", {})
+    assert logs_subscription.address is None
+    assert logs_subscription.topics is None
+
+
+def test_logs_subscription_properties(handler):
+    address = "0x1234567890123456789012345678901234567890"
+    topics = [
+        "0x0000000000000000000000000000000000000000000000000000000000000001",
+        "0x0000000000000000000000000000000000000000000000000000000000000002",
+    ]
+    event = Mock()
+    logs_subscription = LogsSubscription(
+        address=address, topics=topics, handler=handler, event=event, label="logs label"
+    )
+    assert logs_subscription._handler is handler
+    assert logs_subscription.label == "logs label"
+    assert logs_subscription.subscription_params == (
+        "logs",
+        {"address": address, "topics": topics},
+    )
+    assert logs_subscription.address == address
+    assert logs_subscription.topics == topics
+    assert logs_subscription.event is event
+
+
+def test_syncing_subscription_properties(handler):
+    syncing_subscription = SyncingSubscription(handler=handler, label="syncing label")
+    assert syncing_subscription._handler is handler
+    assert syncing_subscription.label == "syncing label"
+    assert syncing_subscription.subscription_params == ("syncing",)

web3/_utils/module_testing/eth_module.py

@@ -1991,9 +1991,8 @@ async def test_async_eth_get_logs_with_logs_none_topic_args(
         # Test with None overflowing
         filter_params: FilterParams = {
             "fromBlock": BlockNumber(0),
-            "topics": [None, None, None],
+            "topics": [None, None, None, None],
         }
-
         result = await async_w3.eth.get_logs(filter_params)
         assert len(result) == 0
 

web3/_utils/module_testing/module_testing_utils.py

@@ -106,7 +106,6 @@ def assert_contains_log(
     emitter_contract_address: ChecksumAddress,
     txn_hash_with_log: HexStr,
 ) -> None:
-    assert len(result) == 1
     log_entry = result[0]
     assert log_entry["blockNumber"] == block_with_txn_with_log["number"]
     assert log_entry["blockHash"] == block_with_txn_with_log["hash"]