code: def test_resumed_with_resources(ws_with_running_task, done_ev_cls):
      """Test transition loop of a task with resources:
  
      executing -> cancelled -> resumed(fetch) -> (complete execution)
      """
      ws = ws_with_running_task
      ws2 = "127.0.0.1:2"
      assert ws.available_resources == {"R": 0}
  
      ws.handle_stimulus(FreeKeysEvent(keys=["x"], stimulus_id="s1"))
      assert ws.available_resources == {"R": 0}
  
      ws.handle_stimulus(
          ComputeTaskEvent.dummy(
              "y", who_has={"x": [ws2]}, stimulus_id="s2"
          )
      )
      assert ws.available_resources == {"R": 0}
  
      ws.handle_stimulus(done_ev_cls.dummy("x", stimulus_id="s3"))
      assert ws.available_resources == {"R": 1}
  
file_path: distributed/distributed/tests/test_resources.py
imports: from __future__ import annotations
  import asyncio
  import pytest
  import dask
  from dask import delayed
  from distributed import Lock, Worker
  from distributed.client import wait
  from distributed.utils_test import NO_AMM, gen_cluster, inc, lock_inc, slowadd, slowinc
  from distributed.worker_state_machine import ComputeTaskEvent, Execute, ExecuteFailureEvent, ExecuteSuccessEvent, FreeKeysEvent, LongRunningMsg, RescheduleEvent, TaskFinishedMsg
label_desc: Use image node_id 7 for calling a global function with example usage: test_resumed_with_resources(ws_with_running_task, done_ev_cls) without return types