code: def test_condition_feature(protocol, temp_workspace, tmpdir):
      class ProcessingTestDocConditions(BaseDoc):
          text: str
          tags: Dict[str, int]
  
      class ConditionDumpExecutor(Executor):
          @requests
          def foo(
              self, docs: DocList[ProcessingTestDocConditions], **kwargs
          ) -> DocList[ProcessingTestDocConditions]:
              with open(
                  os.path.join(
                      str(self.workspace), f"{self.metas.name}.txt"
                  ),
                  "w",
                  encoding="utf-8",
              ) as fp:
                  for doc in docs:
                      fp.write(doc.text)
                      doc.text += f" processed by {self.metas.name}"
  
      class FirstExec(Executor):
          @requests
          def foo(
              self, docs: DocList[LegacyDocument], **kwargs
          ) -> DocList[ProcessingTestDocConditions]:
              output_da = DocList[ProcessingTestDocConditions](
                  [
                      ProcessingTestDocConditions(
                          text="type1", tags={"type": 1}
                      ),
                      ProcessingTestDocConditions(
                          text="type2", tags={"type": 2}
                      ),
                  ]
              )
              return output_da
  
      class JoinerExec(Executor):
          @requests
          def foo(
              self, docs: DocList[ProcessingTestDocConditions], **kwargs
          ) -> DocList[ProcessingTestDocConditions]:
              pass
  
      f = (
          Flow(protocol=protocol)
          .add(uses=FirstExec, name="first")
          .add(
              uses=ConditionDumpExecutor,
              uses_metas={"name": "exec1"},
              workspace=os.environ["TEMP_WORKSPACE"],
              name="exec1",
              needs=["first"],
              when={"tags__type": {"$eq": 1}},
          )
          .add(
              uses=ConditionDumpExecutor,
              workspace=os.environ["TEMP_WORKSPACE"],
              uses_metas={"name": "exec2"},
              name="exec2",
              needs="first",
              when={"tags__type": {"$gt": 1}},
          )
          .needs_all("joiner", uses=JoinerExec)
      )
  
      with f:
          input_da = DocList[LegacyDocument]([])
  
          ret = f.post(
              on="/bar",
              inputs=input_da,
              return_type=DocList[ProcessingTestDocConditions],
          )
          assert len(ret) == 2
          types_set = set()
          for doc in ret:
              if doc.tags["type"] == 1:
                  assert doc.text == "type1 processed by exec1"
              else:
                  assert doc.tags["type"] == 2
                  assert doc.text == "type2 processed by exec2"
              types_set.add(doc.tags["type"])
  
          assert types_set == {1, 2}
  
          with open(
              os.path.join(str(tmpdir), "exec1", "0", f"exec1.txt"),
              "r",
              encoding="utf-8",
          ) as fp:
              assert fp.read() == "type1"
  
          with open(
              os.path.join(str(tmpdir), "exec2", "0", f"exec2.txt"),
              "r",
              encoding="utf-8",
          ) as fp:
              assert fp.read() == "type2"
  
imports: import os
  import time
  from typing import Any, AsyncGenerator, AsyncIterator, Dict, Generator, Iterator, List, Optional, Union
  import numpy
  import pytest
  from docarray import BaseDoc, DocList
  from docarray.documents import ImageDoc, TextDoc
  from docarray.documents.legacy import LegacyDocument
  from docarray.typing import AnyTensor, ImageUrl, NdArray
  from jina import Client, Deployment, Executor, Flow, dynamic_batching, requests
  from jina.excepts import RuntimeFailToStart
  from jina.helper import random_port
  import uuid
label_desc: Use image node_id 11 for calling a global function with example usage: test_condition_feature(protocol, temp_workspace, tmpdir) and returns: output_da