code: def test_secede_cancelled_or_resumed_workerstate(
      ws, resume_to_fetch, resume_to_executing, done_ev_cls
  ):
      """Test what happens when a cancelled or resumed(fetch) task calls secede().
      See also test_secede_cancelled_or_resumed_scheduler
      """
      ws2 = "127.0.0.1:2"
      ws.handle_stimulus(
          ComputeTaskEvent.dummy("x", stimulus_id="s1"),
          FreeKeysEvent(keys=["x"], stimulus_id="s2"),
      )
      if resume_to_fetch:
          ws.handle_stimulus(
              ComputeTaskEvent.dummy(
                  "y", who_has={"x": [ws2]}, stimulus_id="s3"
              ),
          )
      ts = ws.tasks["x"]
      assert ts.previous == "executing"
      assert ts in ws.executing
      assert ts not in ws.long_running
  
      instructions = ws.handle_stimulus(
          SecedeEvent(key="x", compute_duration=1, stimulus_id="s4")
      )
      assert not instructions  # Do not send RescheduleMsg
      assert ts.previous == "long-running"
      assert ts not in ws.executing
      assert ts in ws.long_running
  
      if resume_to_executing:
          instructions = ws.handle_stimulus(
              FreeKeysEvent(keys=["y"], stimulus_id="s5"),
              ComputeTaskEvent.dummy("x", stimulus_id="s6"),
          )
          # Inform the scheduler of the SecedeEvent that happened in the past
          assert instructions == [
              LongRunningMsg(
                  key="x", compute_duration=None, stimulus_id="s6"
              )
          ]
          assert ts.state == "long-running"
          assert ts not in ws.executing
          assert ts in ws.long_running
  
      ws.handle_stimulus(done_ev_cls.dummy(key="x", stimulus_id="s7"))
      assert ts not in ws.executing
      assert ts not in ws.long_running
  
imports: from __future__ import annotations
  import asyncio
  import pytest
  import distributed
  from distributed import Event, Lock, Worker
  from distributed.client import wait
  from distributed.utils_test import BlockedExecute, BlockedGatherDep, BlockedGetData, _LockedCommPool, assert_story, async_poll_for, freeze_batched_send, gen_cluster, inc, lock_inc, wait_for_state, wait_for_stimulus
  from distributed.worker_state_machine import AddKeysMsg, ComputeTaskEvent, Execute, ExecuteFailureEvent, ExecuteSuccessEvent, FreeKeysEvent, GatherDep, GatherDepFailureEvent, GatherDepNetworkFailureEvent, GatherDepSuccessEvent, LongRunningMsg, ReleaseWorkerDataMsg, RemoveReplicasEvent, RescheduleEvent, SecedeEvent, TaskFinishedMsg, UpdateDataEvent
label_desc: Use image node_id 13 for calling a global function with example usage: test_secede_cancelled_or_resumed_workerstate(ws, resume_to_fetch, resume_to_executing, done_ev_cls) without return types