code: def assert_amm_transfer_story(
      key: str, w_from: Worker, w_to: Worker
  ) -> None:
      """Test that an in-memory key was transferred from worker w_from to worker w_to by
      the Active Memory Manager and it was not recalculated on w_to
      """
      assert_story(
          w_to.state.story(key),
          [
              (key, "fetch", "flight", "flight", {}),
              ("request-dep", w_from.address, lambda set_: key in set_),
              ("receive-dep", w_from.address, lambda set_: key in set_),
              (key, "put-in-memory"),
              (key, "flight", "memory", "memory", {}),
          ],
          strict=False,
      )
      assert key in w_to.data
  
imports: from __future__ import annotations
  import asyncio
  import gc
  import importlib
  import itertools
  import logging
  import os
  import random
  import sys
  import tempfile
  import threading
  import traceback
  import warnings
  import weakref
  from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
  from concurrent.futures.process import BrokenProcessPool
  from numbers import Number
  from operator import add
  from time import sleep
  from unittest import mock
  import psutil
  import pytest
  from tlz import first, pluck, sliding_window
  from tornado.ioloop import IOLoop
  import dask
  from dask import delayed
  from dask.system import CPU_COUNT
  from dask.utils import tmpfile
  from distributed import Client, Event, Nanny, WorkerPlugin, default_client, get_client, get_worker, profile, wait
  from distributed.comm.registry import backends
  from distributed.comm.utils import OFFLOAD_THRESHOLD
  from distributed.compatibility import LINUX, WINDOWS
  from distributed.core import CommClosedError, Status, rpc
  from distributed.diagnostics.plugin import ForwardOutput
  from distributed.metrics import time
  from distributed.protocol import pickle
  from distributed.scheduler import KilledWorker, Scheduler
  from distributed.utils import wait_for
  from distributed.utils_test import NO_AMM, BlockedExecute, BlockedGatherDep, BlockedGetData, TaskStateMetadataPlugin, _LockedCommPool, assert_story, async_poll_for, captured_logger, dec, div, freeze_batched_send, freeze_data_fetching, gen_cluster, gen_test, inc, mul, nodebug, raises_with_cause, slowinc, slowsum, wait_for_state
  from distributed.worker import Worker, benchmark_disk, benchmark_memory, benchmark_network, error_message, logger
  from distributed.worker_state_machine import AcquireReplicasEvent, ComputeTaskEvent, ExecuteFailureEvent, ExecuteSuccessEvent, RemoveReplicasEvent
label_desc: Use image node_id 5 for calling a global function with example usage: assert_amm_transfer_story(key, w_from, w_to) without return types