code: def test_secede_without_stealing_issue_1262():
      """
      Tests that seceding works with the Stealing extension disabled
      https://github.com/dask/distributed/issues/1262
      """
  
      # run the loop as an inner function so all workers are closed
      # and exceptions can be examined
      @gen_cluster(
          client=True,
          scheduler_kwargs={"extensions": {}},
          worker_kwargs={"extensions": {}},
      )
      async def secede_test(c, s, a, b):
          def func(x):
              with worker_client() as wc:
                  y = wc.submit(lambda: 1 + x)
                  return wc.gather(y)
  
          f = await c.gather(c.submit(func, 1))
  
          return c, s, a, b, f
  
      c, s, a, b, f = secede_test()
  
      assert f == 2
  
file_path: distributed/distributed/tests/test_worker_client.py
imports: from __future__ import annotations
  import asyncio
  import random
  import threading
  from collections import defaultdict
  from time import sleep
  import pytest
  import dask
  from dask import delayed
  from distributed import Client, as_completed, get_client, get_worker, wait, worker_client
  from distributed.metrics import time
  from distributed.utils_test import double, gen_cluster, inc, slowinc
label_desc: Use image node_id 4 for calling a global function with example usage: test_secede_without_stealing_issue_1262() and returns: c,  s,  a,  b,  f, wc