content: import pytest
  import asyncio
  from dataclasses import (
      dataclass,
  )
  from typing import (
      TYPE_CHECKING,
      Any,
      Generator,
      Sequence,
      cast,
  )
  
  from eth_typing import (
      ChecksumAddress,
      HexStr,
  )
  from eth_utils import (
      is_hexstr,
  )
  from hexbytes import (
      HexBytes,
  )
  
  from web3 import (
      AsyncWeb3,
      PersistentConnectionProvider,
  )
  from web3.beacon import (
      AsyncBeacon,
  )
  from web3.datastructures import (
      AttributeDict,
  )
  from web3.middleware import (
      ExtraDataToPOAMiddleware,
  )
  from web3.types import (
      BlockData,
      FormattedEthSubscriptionResponse,
      LogReceipt,
      Nonce,
      RPCEndpoint,
      TopicFilter,
      TxData,
      Wei,
  )
  from web3.utils import (
      EthSubscription,
  )
  from web3.utils.subscriptions import (
      LogsSubscription,
      LogsSubscriptionContext,
      NewHeadsSubscription,
      NewHeadsSubscriptionContext,
      PendingTxSubscription,
      PendingTxSubscriptionContext,
  )
  
  if TYPE_CHECKING:
      from web3.contract.async_contract import (
          AsyncContract,
          AsyncContractFunction,
      )
      from web3.providers.persistent.subscription_manager import (
          SubscriptionContainer,
      )
  
  
  # LogIndexedAndNotIndexed event args
  INDEXED_ADDR = "0xdEad000000000000000000000000000000000000"
  INDEXED_UINT256 = 1337
  NON_INDEXED_ADDR = "0xbeeF000000000000000000000000000000000000"
  NON_INDEXED_UINT256 = 1999
  NON_INDEXED_STRING = "test logs subscriptions"
  
  SOME_BLOCK_KEYS = [
      "number",
      "hash",
      "parentHash",
      "transactionsRoot",
      "stateRoot",
      "receiptsRoot",
      "gasLimit",
      "gasUsed",
      "timestamp",
      "baseFeePerGas",
      "withdrawalsRoot",
  ]
  
  
  @dataclass
  class SubscriptionHandlerTest:
      passed: bool = False
  
  
  async def new_heads_handler(
      handler_context: NewHeadsSubscriptionContext,
  ) -> None:
      w3 = handler_context.async_w3
      sub = handler_context.subscription
      assert isinstance(w3, AsyncWeb3)
      provider = cast(PersistentConnectionProvider, w3.provider)
      assert isinstance(provider.get_endpoint_uri_or_ipc_path(), str)
  
      assert isinstance(sub, EthSubscription)
  
      block = handler_context.result
      assert block is not None
      assert all(k in block.keys() for k in SOME_BLOCK_KEYS)
  
      assert handler_context.new_heads_handler_test.passed is False
      handler_context.new_heads_handler_test.passed = True
      assert await sub.unsubscribe()
  
  
  async def pending_tx_handler(
      handler_context: PendingTxSubscriptionContext,
  ) -> None:
      w3 = handler_context.async_w3
      sub = handler_context.subscription
      tx = handler_context.result
  
      assert w3 is not None
      provider = cast(PersistentConnectionProvider, w3.provider)
      assert isinstance(provider.get_endpoint_uri_or_ipc_path(), str)
  
      assert isinstance(sub, PendingTxSubscription)
  
      assert tx is not None
      tx = cast(TxData, tx)
      accts = await w3.eth.accounts
      assert tx["from"] == accts[0]
      await w3.eth.wait_for_transaction_receipt(tx["hash"])
  
      assert handler_context.pending_tx_handler_test.passed is False
      handler_context.pending_tx_handler_test.passed = True
      assert await sub.unsubscribe()
  
  
  async def logs_handler(
      handler_context: LogsSubscriptionContext,
  ) -> None:
      w3 = handler_context.async_w3
      sub = handler_context.subscription
      log_receipt = handler_context.result
  
      assert w3 is not None
      provider = cast(PersistentConnectionProvider, w3.provider)
      assert isinstance(provider.get_endpoint_uri_or_ipc_path(), str)
  
      assert isinstance(sub, LogsSubscription)
      event_data = handler_context.event.process_log(log_receipt)
      assert event_data.args.indexedAddress == INDEXED_ADDR
      assert event_data.args.indexedUint256 == INDEXED_UINT256
      assert event_data.args.nonIndexedAddress == NON_INDEXED_ADDR
      assert event_data.args.nonIndexedUint256 == NON_INDEXED_UINT256
      assert event_data.args.nonIndexedString == NON_INDEXED_STRING
  
      assert handler_context.logs_handler_test.passed is False
      handler_context.logs_handler_test.passed = True
      assert await sub.unsubscribe()
  
  
  async def idle_handler(
      _handler_context: Any,
  ) -> None:
      pass
  
  
  async def emit_contract_event(
      async_w3: "AsyncWeb3[Any]",
      acct: ChecksumAddress,
      contract_function: "AsyncContractFunction",
      args: Any = (),
      delay: float = 0.25,
  ) -> None:
      await asyncio.sleep(delay)
      tx_hash = await contract_function(*args).transact({"from": acct})
      receipt = await async_w3.eth.wait_for_transaction_receipt(tx_hash)
      assert receipt["status"] == 1
  
  
  async def log_indexed_and_non_indexed_args_task(
      async_w3: "AsyncWeb3[Any]",
      async_emitter_contract: "AsyncContract",
      acct: ChecksumAddress,
      delay: float = 0.1,
  ) -> "asyncio.Task[None]":
      return asyncio.create_task(
          emit_contract_event(
              async_w3,
              acct,
              async_emitter_contract.functions.logIndexedAndNotIndexedArgs,
              args=(
                  INDEXED_ADDR,
                  INDEXED_UINT256,
                  NON_INDEXED_ADDR,
                  NON_INDEXED_UINT256,
                  NON_INDEXED_STRING,
              ),
              delay=delay,
          )
      )
  
  
  def assert_no_subscriptions_left(sub_container: "SubscriptionContainer") -> None:
      assert len(sub_container) == 0
      assert len(sub_container.subscriptions) == 0
      assert len(sub_container.subscriptions_by_id) == 0
      assert len(sub_container.subscriptions_by_label) == 0
      assert len(sub_container.handler_subscriptions) == 0
  
  
  async def clean_up_task(task: "asyncio.Task[Any]") -> None:
      task.cancel()
      try:
          await task
      except asyncio.CancelledError:
          pass
  
  
  class PersistentConnectionProviderTest:
      @pytest.fixture(autouse=True)
      def clear_caches(self, async_w3: "AsyncWeb3[Any]") -> Generator[None, None, None]:
          yield
          async_w3.provider._request_processor.clear_caches()
          async_w3.subscription_manager.total_handler_calls = 0
  
      @staticmethod
      async def seed_transactions_to_geth(
          async_w3: "AsyncWeb3[Any]",
          acct: ChecksumAddress,
          num_txs: int = 1,
          delay: float = 0.1,
      ) -> None:
          nonce = int(await async_w3.eth.get_transaction_count(acct))
  
          async def send_tx() -> None:
              nonlocal nonce
              await async_w3.eth.send_transaction(
                  {
                      "from": acct,
                      "to": acct,
                      "value": Wei(nonce),
                      "gas": 21000,
                      "nonce": Nonce(nonce),
                  }
              )
              nonce += 1
  
          for _ in range(num_txs):
              await asyncio.sleep(delay)
              await send_tx()
  
      @pytest.mark.asyncio
      @pytest.mark.parametrize(
          "subscription_params,ws_subscription_response,expected_formatted_result",
          (
              (
                  ("syncing",),
                  {
                      "jsonrpc": "2.0",
                      "method": "eth_subscription",
                      "params": {
                          "subscription": "THIS_WILL_BE_REPLACED_IN_THE_TEST",
                          "result": False,
                      },
                  },
                  False,
              ),
              (
                  ("syncing",),
                  {
                      "jsonrpc": "2.0",
                      "method": "eth_subscription",
                      "params": {
                          "subscription": "THIS_WILL_BE_REPLACED_IN_THE_TEST",
                          "result": {
                              "isSyncing": True,
                              "startingBlock": "0x0",
                              "currentBlock": "0x4346fe",
                              "highestBlock": "0x434806",
                          },
                      },
                  },
                  AttributeDict(
                      {
                          "isSyncing": True,
                          "startingBlock": 0,
                          "currentBlock": 4409086,
                          "highestBlock": 4409350,
                      }
                  ),
              ),
          ),
          ids=[
              "syncing-False",
              "syncing-True",
          ],
      )
      async def test_async_eth_subscribe_syncing_mocked(
          self,
          async_w3: "AsyncWeb3[Any]",
          subscription_params: tuple[Any, ...],
          ws_subscription_response: dict[str, Any],
          expected_formatted_result: Any,
      ) -> None:
          sub_id = await async_w3.eth.subscribe(*subscription_params)
          assert is_hexstr(sub_id)
  
          # stub out the subscription id so we know how to process the response
          ws_subscription_response["params"]["subscription"] = sub_id
  
          # add the response to the subscription response cache as if it came from the
          # websocket connection
          await async_w3.provider._request_processor.cache_raw_response(
              ws_subscription_response, subscription=True
          )
  
          async for msg in async_w3.socket.process_subscriptions():
              response = cast(FormattedEthSubscriptionResponse, msg)
              assert response["subscription"] == sub_id
              assert response["result"] == expected_formatted_result
  
              # only testing one message, so break here
              await async_w3.eth.unsubscribe(sub_id)
              break
  
          assert_no_subscriptions_left(
              async_w3.subscription_manager._subscription_container
          )
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_new_heads(
          self, async_w3: "AsyncWeb3[Any]"
      ) -> None:
          sub_id = await async_w3.eth.subscribe("newHeads")
          assert is_hexstr(sub_id)
  
          async for msg in async_w3.socket.process_subscriptions():
              response = cast(FormattedEthSubscriptionResponse, msg)
              assert response["subscription"] == sub_id
              result = cast(BlockData, response["result"])
              assert all(k in result.keys() for k in SOME_BLOCK_KEYS)
              break
  
          assert await async_w3.eth.unsubscribe(sub_id)
          assert_no_subscriptions_left(
              async_w3.subscription_manager._subscription_container
          )
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_creates_and_handles_new_heads_subscription_type(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          sub_manager = async_w3.subscription_manager
          new_heads_handler_test = SubscriptionHandlerTest()
  
          sub_id = await async_w3.eth.subscribe(
              "newHeads",
              handler=new_heads_handler,
              handler_context={"new_heads_handler_test": new_heads_handler_test},
          )
          assert is_hexstr(sub_id)
  
          assert len(sub_manager.subscriptions) == 1
          sub = sub_manager.subscriptions[0]
          assert isinstance(sub, NewHeadsSubscription)
  
          await sub_manager.handle_subscriptions()
  
          assert new_heads_handler_test.passed
          assert len(sub_manager.subscriptions) == 0
  
          assert sub_manager.total_handler_calls == 1
          assert sub.handler_call_count == 1
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_process_pending_tx_true(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          sub_id = await async_w3.eth.subscribe("newPendingTransactions", True)
          assert is_hexstr(sub_id)
  
          accts = await async_w3.eth.accounts
          acct = accts[0]
  
          num_txs = 2
          original_nonce = await async_w3.eth.get_transaction_count(acct)
          tx_seeder_task = asyncio.create_task(
              self.seed_transactions_to_geth(async_w3, acct, num_txs=num_txs)
          )
  
          nonce = int(original_nonce)
          tx_hash = None
          async for msg in async_w3.socket.process_subscriptions():
              response = cast(FormattedEthSubscriptionResponse, msg)
              assert response["subscription"] == sub_id
              result = cast(TxData, response["result"])
              assert result["gas"] == 21000
              assert result["from"] == acct
              assert result["to"] == acct
              assert int(result["value"]) == int(nonce)
              tx_hash = result["hash"]
              assert tx_hash is not None
  
              nonce += 1
              if nonce == int(original_nonce) + num_txs:
                  break
  
          # cleanup
          assert await async_w3.eth.unsubscribe(sub_id)
          assert_no_subscriptions_left(
              async_w3.subscription_manager._subscription_container
          )
          async_w3.provider._request_processor.clear_caches()
          await async_w3.eth.wait_for_transaction_receipt(tx_hash)
          await clean_up_task(tx_seeder_task)
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_and_process_pending_tx_false(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          sub_id = await async_w3.eth.subscribe("newPendingTransactions")
          assert is_hexstr(sub_id)
  
          accts = await async_w3.eth.accounts
          acct = accts[0]
          await async_w3.eth.get_transaction_count(acct)
  
          num_txs = 2
          tx_seeder_task = asyncio.create_task(
              self.seed_transactions_to_geth(async_w3, acct, num_txs=num_txs)
          )
  
          tx_hash = None
          i = 0
          async for msg in async_w3.socket.process_subscriptions():
              response = cast(FormattedEthSubscriptionResponse, msg)
              assert response["subscription"] == sub_id
              assert isinstance(response["result"], HexBytes)
              tx_hash = response["result"]
  
              i += 1
              if i == num_txs:
                  break
  
          # cleanup
          await async_w3.eth.unsubscribe(sub_id)
          assert_no_subscriptions_left(
              async_w3.subscription_manager._subscription_container
          )
          await async_w3.eth.wait_for_transaction_receipt(tx_hash)
          await clean_up_task(tx_seeder_task)
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_creates_and_handles_pending_tx_subscription_type(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          sub_manager = async_w3.subscription_manager
          pending_tx_handler_test = SubscriptionHandlerTest()
  
          sub_id = await async_w3.eth.subscribe(
              "newPendingTransactions",
              True,
              handler=pending_tx_handler,
              handler_context={"pending_tx_handler_test": pending_tx_handler_test},
          )
          assert is_hexstr(sub_id)
  
          assert len(sub_manager.subscriptions) == 1
          sub = sub_manager.subscriptions[0]
          assert isinstance(sub, PendingTxSubscription)
  
          # seed transactions to geth
          accts = await async_w3.eth.accounts
          acct = accts[0]
          tx_seeder_task = asyncio.create_task(
              self.seed_transactions_to_geth(async_w3, acct)
          )
          await sub_manager.handle_subscriptions()
  
          assert pending_tx_handler_test.passed
          assert len(sub_manager.subscriptions) == 0
  
          assert sub_manager.total_handler_calls == 1
          assert sub.handler_call_count == 1
  
          # cleanup
          sub_manager.total_handler_calls = 0
          await clean_up_task(tx_seeder_task)
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_and_process_logs(
          self, async_w3: "AsyncWeb3[Any]", async_emitter_contract: "AsyncContract"
      ) -> None:
          event = async_emitter_contract.events.LogIndexedAndNotIndexed
          event_topic = async_w3.keccak(text=event.abi_element_identifier).to_0x_hex()
  
          sub_id = await async_w3.eth.subscribe(
              "logs",
              {
                  "address": async_emitter_contract.address,
                  "topics": [HexStr(event_topic)],
              },
          )
          assert is_hexstr(sub_id)
  
          accts = await async_w3.eth.accounts
          acct = accts[0]
          emit_event_task = await log_indexed_and_non_indexed_args_task(
              async_w3, async_emitter_contract, acct
          )
  
          async for msg in async_w3.socket.process_subscriptions():
              response = cast(FormattedEthSubscriptionResponse, msg)
              assert response["subscription"] == sub_id
              log_receipt = cast(LogReceipt, response["result"])
              event_data = event.process_log(log_receipt)
              assert event_data.args.indexedAddress == INDEXED_ADDR
              assert event_data.args.indexedUint256 == INDEXED_UINT256
              assert event_data.args.nonIndexedAddress == NON_INDEXED_ADDR
              assert event_data.args.nonIndexedUint256 == NON_INDEXED_UINT256
              assert event_data.args.nonIndexedString == NON_INDEXED_STRING
              break
  
          assert await async_w3.eth.unsubscribe(sub_id)
          assert len(async_w3.subscription_manager.subscriptions) == 0
          await clean_up_task(emit_event_task)
  
      @pytest.mark.asyncio
      async def test_async_eth_subscribe_creates_and_handles_logs_subscription_type(
          self,
          async_w3: "AsyncWeb3[Any]",
          async_emitter_contract: "AsyncContract",
      ) -> None:
          sub_manager = async_w3.subscription_manager
  
          event = async_emitter_contract.events.LogIndexedAndNotIndexed
  
          logs_handler_test = SubscriptionHandlerTest()
          sub_id = await async_w3.eth.subscribe(
              "logs",
              {
                  "address": async_emitter_contract.address,
                  "topics": [event.topic],
              },
              handler=logs_handler,
              handler_context={
                  "logs_handler_test": logs_handler_test,
                  "event": event,
              },
          )
          assert is_hexstr(sub_id)
  
          assert len(sub_manager.subscriptions) == 1
          sub = sub_manager.subscriptions[0]
          assert isinstance(sub, LogsSubscription)
  
          accts = await async_w3.eth.accounts
          acct = accts[0]
          emit_event_task = await log_indexed_and_non_indexed_args_task(
              async_w3, async_emitter_contract, acct
          )
  
          await sub_manager.handle_subscriptions()
  
          assert logs_handler_test.passed
          assert len(sub_manager.subscriptions) == 0
  
          assert sub_manager.total_handler_calls == 1
          assert sub.handler_call_count == 1
  
          # cleanup
          sub_manager.total_handler_calls = 0
          await clean_up_task(emit_event_task)
  
      @pytest.mark.asyncio
      @pytest.mark.parametrize(
          "topics",
          [
              pytest.param(
                  [
                      HexStr(
                          "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"  # noqa: E501
                      )
                  ],
                  id="Single specific topic at position 0",
              ),
              pytest.param(
                  [
                      None,
                      HexStr(
                          "0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890"  # noqa: E501
                      ),
                  ],
                  id="Wildcard at position 0, specific topic at position 1",
              ),
              pytest.param(
                  [
                      [
                          HexStr(
                              "0x1111111111111111111111111111111111111111111111111111111111111111"  # noqa: E501
                          ),
                          HexStr(
                              "0x2222222222222222222222222222222222222222222222222222222222222222"  # noqa: E501
                          ),
                      ]
                  ],
                  id="OR pattern: topic A or B at position 0",
              ),
              pytest.param(
                  [
                      [
                          HexStr(
                              "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"  # noqa: E501
                          ),
                          HexStr(
                              "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"  # noqa: E501
                          ),
                      ],
                      HexStr(
                          "0xcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"  # noqa: E501
                      ),
                  ],
                  id="Complex: (A or B) at position 0 AND C at position 1",
              ),
              pytest.param(
                  [
                      HexBytes(
                          "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"  # noqa: E501
                      )
                  ],
                  id="Single specific topic at position 0 with HexBytes",
              ),
              pytest.param(
                  [
                      [
                          HexBytes(
                              "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"  # noqa: E501
                          ),
                          HexStr(
                              "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"  # noqa: E501
                          ),
                          b"\xcc" * 32,
                      ]
                  ],
                  id="OR pattern with mixed HexBytes, HexStr, and bytes at position 0",
              ),
          ],
      )
      async def test_async_logs_subscription_with_and_or_topic_patterns(
          self,
          async_w3: "AsyncWeb3[Any]",
          async_emitter_contract: "AsyncContract",
          topics: Sequence[TopicFilter],
      ) -> None:
          """Test that LogsSubscription properly handles AND/OR topic patterns."""
          sub_manager = async_w3.subscription_manager
  
          subscription = LogsSubscription(
              address=async_emitter_contract.address,
              topics=topics,
              handler=idle_handler,
          )
  
          await sub_manager.subscribe(subscription)
          assert len(sub_manager.subscriptions) == 1
          assert isinstance(sub_manager.subscriptions[0], LogsSubscription)
          assert sub_manager.subscriptions[0].topics == topics
  
          assert subscription.subscription_params == (
              "logs",
              {
                  "address": async_emitter_contract.address,
                  "topics": topics,
              },
          )
  
          # clean up
          await sub_manager.unsubscribe(subscription)
          assert len(sub_manager.subscriptions) == 0
  
      @pytest.mark.asyncio
      async def test_async_extradata_poa_middleware_on_eth_subscription(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          async_w3.middleware_onion.inject(
              ExtraDataToPOAMiddleware, "poa_middleware", layer=0
          )
  
          sub_id = await async_w3.eth.subscribe("newHeads")
          assert is_hexstr(sub_id)
  
          # add the response to the subscription response cache as if it came from the
          # websocket connection
          await async_w3.provider._request_processor.cache_raw_response(
              {
                  "jsonrpc": "2.0",
                  "method": "eth_subscription",
                  "params": {
                      "subscription": sub_id,
                      "result": {
                          "extraData": f"0x{'00' * 100}",
                      },
                  },
              },
              subscription=True,
          )
  
          async for msg in async_w3.socket.process_subscriptions():
              response = cast(FormattedEthSubscriptionResponse, msg)
              assert response.keys() == {"subscription", "result"}
              assert response["subscription"] == sub_id
              assert response["result"]["proofOfAuthorityData"] == HexBytes(  # type: ignore  # noqa: E501
                  f"0x{'00' * 100}"
              )
  
              # only testing one message, so break here
              break
  
          # clean up
          assert await async_w3.eth.unsubscribe(sub_id)
          async_w3.middleware_onion.remove("poa_middleware")
  
      @pytest.mark.asyncio
      async def test_asyncio_gather_for_multiple_requests_matches_the_responses(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          (
              latest,
              chain_id,
              block_num,
              chain_id2,
              pending,
              chain_id3,
          ) = await asyncio.gather(
              async_w3.eth.get_block("latest"),
              async_w3.eth.chain_id,
              async_w3.eth.block_number,
              async_w3.eth.chain_id,
              async_w3.eth.get_block("pending"),
              async_w3.eth.chain_id,
          )
  
          # assert attrdict middleware was applied appropriately
          assert isinstance(latest, AttributeDict)
          assert isinstance(pending, AttributeDict)
  
          # assert block values
          assert latest is not None
          assert all(k in latest.keys() for k in SOME_BLOCK_KEYS)
          assert pending is not None
          assert all(k in pending.keys() for k in SOME_BLOCK_KEYS)
  
          assert isinstance(block_num, int)
          assert isinstance(chain_id, int)
          assert isinstance(chain_id2, int)
          assert isinstance(chain_id3, int)
          # chain id is set in fixture file
          assert chain_id == chain_id2 == chain_id3 == 131277322940537
  
      @pytest.mark.asyncio
      async def test_async_public_socket_api(self, async_w3: "AsyncWeb3[Any]") -> None:
          # clear all caches and queues
          async_w3.provider._request_processor.clear_caches()
  
          # send a request over the socket
          await async_w3.socket.send(
              RPCEndpoint("eth_getBlockByNumber"), ["latest", True]
          )
  
          # recv and validate the unprocessed response
          response = await async_w3.socket.recv()
          assert "id" in response, "Expected 'id' key in response."
          assert "jsonrpc" in response, "Expected 'jsonrpc' key in response."
          assert "result" in response, "Expected 'result' key in response."
          assert all(k in response["result"].keys() for k in SOME_BLOCK_KEYS)
          assert not isinstance(response["result"]["number"], int)  # assert not processed
  
          # make a request over the socket
          response = await async_w3.socket.make_request(
              RPCEndpoint("eth_getBlockByNumber"), ["latest", True]
          )
          assert "id" in response, "Expected 'id' key in response."
          assert "jsonrpc" in response, "Expected 'jsonrpc' key in response."
          assert "result" in response, "Expected 'result' key in response."
          assert all(k in response["result"].keys() for k in SOME_BLOCK_KEYS)
          assert not isinstance(response["result"]["number"], int)  # assert not processed
  
      @pytest.mark.asyncio
      async def test_async_subscription_manager_subscribes_to_many_subscriptions(
          self, async_w3: "AsyncWeb3[Any]", async_emitter_contract: "AsyncContract"
      ) -> None:
          sub_manager = async_w3.subscription_manager
  
          event = async_emitter_contract.events.LogIndexedAndNotIndexed
          event_topic = async_w3.keccak(text=event.abi_element_identifier).to_0x_hex()
  
          new_heads_handler_test = SubscriptionHandlerTest()
          pending_tx_handler_test = SubscriptionHandlerTest()
          logs_handler_test = SubscriptionHandlerTest()
  
          await sub_manager.subscribe(
              [
                  NewHeadsSubscription(
                      handler=new_heads_handler,
                      handler_context={"new_heads_handler_test": new_heads_handler_test},
                  ),
                  PendingTxSubscription(
                      full_transactions=True,
                      handler=pending_tx_handler,
                      handler_context={
                          "pending_tx_handler_test": pending_tx_handler_test
                      },
                  ),
                  LogsSubscription(
                      address=async_emitter_contract.address,
                      topics=[HexStr(event_topic)],
                      handler=logs_handler,
                      handler_context={
                          "logs_handler_test": logs_handler_test,
                          "event": event,
                      },
                  ),
              ]
          )
  
          # emit contract event for `logs` subscription
          accts = await async_w3.eth.accounts
          acct = accts[0]
          emit_event_task = await log_indexed_and_non_indexed_args_task(
              async_w3, async_emitter_contract, acct
          )
  
          # get subscriptions before they are unsubscribed and removed
          subs = sub_manager.subscriptions
  
          await sub_manager.handle_subscriptions()
  
          # assert unsubscribed and removed subscriptions
          assert len(sub_manager.subscriptions) == 0
  
          assert sub_manager.total_handler_calls == 3
          assert all(sub.handler_call_count == 1 for sub in subs)
  
          assert new_heads_handler_test.passed
          assert pending_tx_handler_test.passed
          assert logs_handler_test.passed
  
          # cleanup
          sub_manager.total_handler_calls = 0
          await clean_up_task(emit_event_task)
  
      @pytest.mark.asyncio
      async def test_subscription_handler_context(
          self, async_w3: "AsyncWeb3[Any]"
      ) -> None:
          base_url = "http://localhost:1337"
          async_beacon = AsyncBeacon(base_url)
          handler_test = SubscriptionHandlerTest()
  
          async def test_sub_handler(
              handler_context: NewHeadsSubscriptionContext,
          ) -> None:
              beacon = handler_context.beacon
              assert isinstance(beacon, AsyncBeacon)
              assert beacon.base_url == base_url
              assert handler_context.int1 == 1337
              assert handler_context.str1 == "foo"
              assert handler_context.int2 == 1999
              assert handler_context.str2 == "bar"
  
              handler_context.handler_test.passed = True
              unsubscribed = await handler_context.subscription.unsubscribe()
              assert unsubscribed
  
          subscribed = await async_w3.eth.subscribe(
              "newHeads",
              label="foo",
              handler=test_sub_handler,
              handler_context={
                  "beacon": async_beacon,
                  "int1": 1337,
                  "str1": "foo",
                  "int2": 1999,
                  "str2": "bar",
                  "handler_test": handler_test,
              },
          )
          assert is_hexstr(subscribed)
  
          sub_manager = async_w3.subscription_manager
  
          await sub_manager.handle_subscriptions()
  
          assert len(sub_manager.subscriptions) == 0
          assert sub_manager.total_handler_calls == 1
          assert handler_test.passed
  
          # cleanup
          sub_manager.total_handler_calls = 0
  
      @pytest.mark.asyncio
      async def test_subscriptions_with_handler_and_without(
          self, async_w3: "AsyncWeb3[Any]"
      ) -> None:
          handler_test = SubscriptionHandlerTest()
          stream_passed = False
  
          async def test_sub_handler(
              handler_context: NewHeadsSubscriptionContext,
          ) -> None:
              handler_context.handler_test.passed = True
              await handler_context.subscription.unsubscribe()
  
          async def handle_subscription_stream() -> None:
              nonlocal stream_passed
              async for msg in async_w3.socket.process_subscriptions():
                  response = cast(FormattedEthSubscriptionResponse, msg)
                  assert sub_manager.get_by_id(response["subscription"]) is not None
                  assert response["result"] is not None
                  # wait for the handler to unsubscribe:
                  stream_passed = True
                  await async_w3.eth.unsubscribe(response["subscription"])
                  break
  
          await async_w3.eth.subscribe(
              "newHeads",
              handler=test_sub_handler,
              label="managed",
              handler_context={"handler_test": handler_test},
          )
          await async_w3.eth.subscribe("newHeads", label="streamed")
  
          sub_manager = async_w3.subscription_manager
          assert len(sub_manager.subscriptions) == 2
  
          await asyncio.gather(
              sub_manager.handle_subscriptions(),
              handle_subscription_stream(),
          )
  
          assert len(sub_manager.subscriptions) == 0
          assert sub_manager.total_handler_calls == 1
          assert handler_test.passed
          assert stream_passed
  
          # cleanup
          sub_manager.total_handler_calls = 0
  
      @pytest.mark.asyncio
      async def test_handle_subscriptions_breaks_on_unsubscribe(
          self,
          async_w3: "AsyncWeb3[Any]",
      ) -> None:
          async def unsubscribe_subs(
              subs: list[NewHeadsSubscription | LogsSubscription],
          ) -> None:
              for sub in subs:
                  await sub.unsubscribe()
  
          sub_manager = async_w3.subscription_manager
          sub1 = NewHeadsSubscription(label="foo", handler=idle_handler)
          sub2 = LogsSubscription(label="bar", handler=idle_handler)
  
          await sub_manager.subscribe([sub1, sub2])
  
  