code: def test_retry_does_retry_and_sleep(cleanup):
      # test the retry and sleep pattern of `retry`
      n_calls = 0
  
      class MyEx(Exception):
          pass
  
      async def coro():
          nonlocal n_calls
          n_calls += 1
          raise MyEx(f"RT_ERROR {n_calls}")
  
      sleep_calls = []
  
      async def my_sleep(amount):
          sleep_calls.append(amount)
          return
  
      async def f():
          return await retry(
              coro,
              retry_on_exceptions=(MyEx,),
              count=5,
              delay_min=1.0,
              delay_max=6.0,
              jitter_fraction=0.0,
          )
  
      with mock.patch("asyncio.sleep", my_sleep):
          with pytest.raises(MyEx, match="RT_ERROR 6"):
              asyncio_run(f(), loop_factory=get_loop_factory())
  
      assert n_calls == 6
      assert sleep_calls == [0.0, 1.0, 3.0, 6.0, 6.0]
  
imports: from __future__ import annotations
  import asyncio
  import random
  from unittest import mock
  import pytest
  from dask.optimization import SubgraphCallable
  from distributed import wait
  from distributed.compatibility import asyncio_run
  from distributed.config import get_loop_factory
  from distributed.core import ConnectionPool, Status
  from distributed.utils_comm import WrappedKey, gather_from_workers, pack_data, retry, subs_multiple, unpack_remotedata
  from distributed.utils_test import BarrierGetData, BrokenComm, gen_cluster, inc
label_desc: Use image node_id 5 for calling a global function with example usage: test_retry_does_retry_and_sleep(cleanup) and returns: unknown