code: def test_running_and_canceled_workflow(
      workflow_start_regular, tmp_path
  ):
      workflow_id = "simple"
      flag = tmp_path / "flag"
  
      @ray.remote
      def simple():
          flag.touch()
          time.sleep(1000)
          return 0
  
      workflow.run_async(simple.bind(), workflow_id=workflow_id)
  
      # Wait until task runs to make sure pre-run metadata is written
      while not flag.exists():
          time.sleep(1)
  
      workflow_metadata = workflow.get_metadata(workflow_id)
      assert workflow_metadata["status"] == "RUNNING"
      assert "start_time" in workflow_metadata["stats"]
      assert "end_time" not in workflow_metadata["stats"]
  
      workflow.cancel(workflow_id)
  
      workflow_metadata = workflow.get_metadata(workflow_id)
      assert workflow_metadata["status"] == "CANCELED"
      assert "start_time" in workflow_metadata["stats"]
      assert "end_time" not in workflow_metadata["stats"]
  
file_path: ray/python/ray/workflow/tests/test_metadata.py
imports: import time
  from ray.tests.conftest import *
  import pytest
  import ray
  from ray import workflow
label_desc: Use image node_id 7 for calling a global function with example usage: test_running_and_canceled_workflow(workflow_start_regular, tmp_path) and returns: int