code: def test_flow_all_floating(protocol, tmpdir):
      file_name = os.path.join(str(tmpdir), "file.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):
              length_of_docs = len(docs)
              with open(self.file_name, "a+", encoding="utf-8") as f:
                  f.write(str(len(docs)))
  
      flow = Flow(protocol=protocol).add(
          name="A",
          floating=True,
          uses=FloatingTestExecutorWriteDocs,
          uses_with={"file_name": file_name},
      )
      with flow:
          flow.post(on="/", inputs=DocumentArray.empty(1))
  
      with open(file_name, "r", encoding="utf-8") as f:
          resulted_str = f.read()
  
      assert resulted_str == "1"
  
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 7 for calling a global function with example usage: test_flow_all_floating(protocol, tmpdir) without return types