code: def test_floating_needs_more_complex(needs, tmpdir):
      NUM_REQ = 20
      print(f" tmpdir {tmpdir}")
      file_name1 = os.path.join(str(tmpdir), "file1.txt")
      file_name2 = os.path.join(str(tmpdir), "file2.txt")
  
      class FloatingTestExecutorWriteDocs(Executor):
          def __init__(self, file_name, *args, **kwargs):
              super().__init__(*args, **kwargs)
              self.file_name = file_name
  
          @requests
          def foo(self, docs, **kwargs):
              time.sleep(TIME_SLEEP_FLOATING)
              with open(self.file_name, "a+", encoding="utf-8") as f:
                  for d in docs:
                      f.write(d.text)
  
              for d in docs:
                  d.text = "change it"
  
      expected_str1 = ""
      if needs == "executor0":
          expected_str1 = "Hello World" * NUM_REQ
      elif needs == "executor1":
          expected_str1 = "Hello World from FastAddExecutor" * NUM_REQ
      expected_str2 = "change it" * NUM_REQ
  
      class FastChangingExecutor(Executor):
          @requests()
          def foo(self, docs, **kwargs):
              for doc in docs:
                  doc.text = "Hello World"
  
      class FastAddExecutor(Executor):
          @requests()
          def foo(self, docs, **kwargs):
              for doc in docs:
                  doc.text += " from FastAddExecutor"
  
      f = (
          Flow()
          .add(name="executor0", uses=FastChangingExecutor)
          .add(
              name="executor1",
              uses=FastAddExecutor,
              needs=["executor0"],
          )
          .add(
              name="floating_executor",
              uses=FloatingTestExecutorWriteDocs,
              uses_with={"file_name": file_name1},
              needs=[needs],
              floating=True,
          )
          .add(
              name="floating_executor_2",
              uses=FloatingTestExecutorWriteDocs,
              uses_with={"file_name": file_name2},
              needs=["floating_executor"],
              floating=True,
          )
      )
      with f:
          for j in range(NUM_REQ):
              start_time = time.time()
              response = f.post(
                  on="/endpoint", inputs=DocumentArray.empty(1)
              )
              end_time = time.time()
              assert (end_time - start_time) < TIME_SLEEP_FLOATING
              assert response.texts == [
                  "Hello World from FastAddExecutor",
              ]
  
      with open(file_name1, "r", encoding="utf-8") as f:
          resulted_str = f.read()
  
      assert resulted_str == expected_str1
  
      with open(file_name2, "r", encoding="utf-8") as f:
          resulted_str = f.read()
  
      assert resulted_str == expected_str2
  
file_path: jina/tests/integration/floating_deployments/test_floating_deployments.py
imports: import os
  import time
  import pytest
  from jina import DocumentArray, Executor, Flow, requests
  from jina.constants import __default_endpoint__
label_desc: Use image node_id 6 for calling a global function with example usage: test_floating_needs_more_complex(needs, tmpdir) without return types