code: def test_basic_workflows(workflow_start_regular_shared):
      @ray.remote
      def source1():
          return "[source1]"
  
      @ray.remote
      def append1(x):
          return x + "[append1]"
  
      @ray.remote
      def append2(x):
          return x + "[append2]"
  
      @ray.remote
      def simple_sequential():
          x = source1.bind()
          y = append1.bind(x)
          return workflow.continuation(append2.bind(y))
  
      @ray.remote
      def identity(x):
          return x
  
      @ray.remote
      def simple_sequential_with_input(x):
          y = append1.bind(x)
          return workflow.continuation(append2.bind(y))
  
      @ray.remote
      def loop_sequential(n):
          x = source1.bind()
          for _ in range(n):
              x = append1.bind(x)
          return workflow.continuation(append2.bind(x))
  
      @ray.remote
      def nested_task(x):
          return workflow.continuation(
              append2.bind(append1.bind(x + "~[nested]~"))
          )
  
      @ray.remote
      def nested(x):
          return workflow.continuation(nested_task.bind(x))
  
      @ray.remote
      def join(x, y):
          return f"join({x}, {y})"
  
      @ray.remote
      def fork_join():
          x = source1.bind()
          y = append1.bind(x)
          y = identity.bind(y)
          z = append2.bind(x)
          return workflow.continuation(join.bind(y, z))
  
      @ray.remote
      def mul(a, b):
          return a * b
  
      @ray.remote
      def factorial(n):
          if n == 1:
              return 1
          else:
              return workflow.continuation(
                  mul.bind(n, factorial.bind(n - 1))
              )
  
      # This test also shows different "style" of running workflows.
      assert (
          workflow.run(simple_sequential.bind())
          == "[source1][append1][append2]"
      )
  
      wf = simple_sequential_with_input.bind("start:")
      assert workflow.run(wf) == "start:[append1][append2]"
  
      wf = loop_sequential.bind(3)
      assert (
          workflow.run(wf)
          == "[source1]" + "[append1]" * 3 + "[append2]"
      )
  
      wf = nested.bind("nested:")
      assert workflow.run(wf) == "nested:~[nested]~[append1][append2]"
  
      wf = fork_join.bind()
      assert (
          workflow.run(wf)
          == "join([source1][append1], [source1][append2])"
      )
  
      assert workflow.run(factorial.bind(10)) == 3628800
  
file_path: ray/python/ray/workflow/tests/test_basic_workflows.py
imports: import time
  from ray.tests.conftest import *
  from filelock import FileLock
  import pytest
  import ray
  from ray import workflow
label_desc: Use image node_id 1 for calling a global function with example usage: test_basic_workflows(workflow_start_regular_shared) and returns: str, unknown, unknown, workflow, x, workflow, workflow, workflow, workflow, str, x, str, y, str, workflow, unknown, int, workflow