code: def test_task_cancelled_and_readded_with_resources(ws):
      """See https://github.com/dask/distributed/issues/6710
  
      A task is enqueued without resources, then cancelled by the client, then re-added
      with the same key, this time with resources.
      Test that resources are respected.
      """
      ws.available_resources = {"R": 1}
      ws.total_resources = {"R": 1}
      RR = {"resource_restrictions": {"R": 1}}
  
      ws.handle_stimulus(
          ComputeTaskEvent.dummy(key="clog", **RR, stimulus_id="s1"),
          ComputeTaskEvent.dummy(key="x", stimulus_id="s2"),
      )
      ts = ws.tasks["x"]
      assert ts.state == "ready"
      assert ts in ws.ready
      assert ts not in ws.constrained
      assert ts.resource_restrictions == {}
  
      ws.handle_stimulus(
          FreeKeysEvent(keys=["x"], stimulus_id="clog"),
          ComputeTaskEvent.dummy(key="x", **RR, stimulus_id="s2"),
      )
      ts = ws.tasks["x"]
      assert ts.state == "constrained"
      assert ts not in ws.ready
      assert ts in ws.constrained
      assert ts.resource_restrictions == {"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 4 for calling a global function with example usage: test_task_cancelled_and_readded_with_resources(ws) without return types