code: def test_workflow_queuing_1(shutdown_only, tmp_path):
      ray.init(storage=str(tmp_path))
      workflow.init(max_running_workflows=2, max_pending_workflows=2)
  
      import queue
      import filelock
  
      lock_path = str(tmp_path / ".lock")
  
      @ray.remote
      def long_running(x):
          with filelock.FileLock(lock_path):
              return x
  
      wfs = [long_running.bind(i) for i in range(5)]
  
      with filelock.FileLock(lock_path):
          refs = [
              workflow.run_async(wfs[i], workflow_id=f"workflow_{i}")
              for i in range(4)
          ]
  
          assert sorted(
              x[0] for x in workflow.list_all({workflow.RUNNING})
          ) == [
              "workflow_0",
              "workflow_1",
          ]
          assert sorted(
              x[0] for x in workflow.list_all({workflow.PENDING})
          ) == [
              "workflow_2",
              "workflow_3",
          ]
  
          with pytest.raises(
              queue.Full, match="Workflow queue has been full"
          ):
              workflow.run(wfs[4], workflow_id="workflow_4")
  
      assert ray.get(refs) == [0, 1, 2, 3]
      assert workflow.run(wfs[4], workflow_id="workflow_4") == 4
      assert sorted(
          x[0] for x in workflow.list_all({workflow.SUCCESSFUL})
      ) == [
          "workflow_0",
          "workflow_1",
          "workflow_2",
          "workflow_3",
          "workflow_4",
      ]
      for i in range(5):
          assert workflow.get_output(f"workflow_{i}") == i
  
file_path: ray/python/ray/workflow/tests/test_workflow_queuing.py
imports: import pytest
  import ray
  from ray import workflow
  from ray.tests.conftest import *
label_desc: Use image node_id 3 for calling a global function with example usage: test_workflow_queuing_1(shutdown_only, tmp_path) and returns: x