code: def test_workflow_queuing_3(shutdown_only, tmp_path):
      """This test ensures the queuing workflow is indeed pending."""
      ray.init(storage=str(tmp_path))
      workflow.init(max_running_workflows=1, max_pending_workflows=1)
  
      import time
      import filelock
      from ray.exceptions import GetTimeoutError
  
      lock_path = str(tmp_path / ".lock")
  
      @ray.remote
      def long_running(x):
          (tmp_path / str(x)).write_text(str(x))
          with filelock.FileLock(lock_path):
              return x
  
      workflow_id = "test_workflow_queuing_3"
  
      with filelock.FileLock(lock_path):
          wf_1 = workflow.run_async(
              long_running.bind(1), workflow_id=f"{workflow_id}_1"
          )
          wf_2 = workflow.run_async(
              long_running.bind(2), workflow_id=f"{workflow_id}_2"
          )
          time.sleep(5)
          assert (tmp_path / str(1)).exists()
          assert not (tmp_path / str(2)).exists()
          assert (
              workflow.get_status(workflow_id=f"{workflow_id}_1")
              == workflow.RUNNING
          )
          assert (
              workflow.get_status(workflow_id=f"{workflow_id}_2")
              == workflow.PENDING
          )
          with pytest.raises(GetTimeoutError):
              ray.get(wf_2, timeout=5)
  
      assert ray.get([wf_1, wf_2]) == [1, 2]
  
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 5 for calling a global function with example usage: test_workflow_queuing_3(shutdown_only, tmp_path) and returns: x