code: def test_constrained_vs_ready_priority_1(
      ws, p1, p2, expect_key, swap
  ):
      """If there are both ready and constrained tasks, those with the highest priority
      win (note: on the Worker, priorities have their sign inverted)
      """
      ws.available_resources = {"R": 1}
      ws.total_resources = {"R": 1}
      RR = {"resource_restrictions": {"R": 1}}
  
      ws.handle_stimulus(
          ComputeTaskEvent.dummy(key="clog", stimulus_id="clog")
      )
  
      stimuli = [
          ComputeTaskEvent.dummy("x", priority=(p1,), stimulus_id="s1"),
          ComputeTaskEvent.dummy(
              "y", priority=(p2,), **RR, stimulus_id="s2"
          ),
      ]
      if swap:
          stimuli = stimuli[::-1]  # This must be inconsequential
  
      instructions = ws.handle_stimulus(
          *stimuli,
          ExecuteSuccessEvent.dummy("clog", stimulus_id="s3"),
      )
      assert instructions == [
          TaskFinishedMsg.match(key="clog", stimulus_id="s3"),
          Execute(key=expect_key, stimulus_id="s3"),
      ]
  
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 1 for calling a global function with example usage: test_constrained_vs_ready_priority_1(ws, p1, p2, expect_key, swap) without return types