code: def test_serve_complex_model(protocols, replicas, ctxt_manager):
      class InputComplexDoc(BaseDoc):
          img: ImageDoc
  
      class OutputComplexDoc(BaseDoc):
          tensor: Optional[AnyTensor]
          url: ImageUrl
          lll: List[List[List[int]]] = [[[5]]]
          fff: List[List[List[float]]] = [[[5.2]]]
          single_text: TextDoc
          texts: DocList[TextDoc]
          d: Dict[str, str] = {"a": "b"}
          u: Union[str, int]
          lu: List[Union[str, int]] = [0, 1, 2]
  
      class MyComplexServeExec(Executor):
          @requests(on="/bar")
          def bar(
              self, docs: DocList[InputComplexDoc], **kwargs
          ) -> DocList[OutputComplexDoc]:
              docs_return = DocList[OutputComplexDoc](
                  [
                      OutputComplexDoc(
                          url="photo.jpg",
                          lll=[[[40]]],
                          fff=[[[40.2]]],
                          d={"b": "a"},
                          texts=DocList[TextDoc](
                              [
                                  TextDoc(
                                      text="hey ha",
                                      embedding=np.zeros(3),
                                  )
                              ]
                          ),
                          single_text=TextDoc(
                              text="single hey ha",
                              embedding=np.zeros(2),
                          ),
                          u="a",
                          lu=[3, 4],
                      )
                      for _ in range(len(docs))
                  ]
              )
              return docs_return
  
      ports = [random_port() for _ in protocols]
      if ctxt_manager == "flow":
          ctxt = Flow(port=ports, protocol=protocols).add(
              replicas=replicas, uses=MyComplexServeExec
          )
      else:
          ctxt = Deployment(
              port=ports,
              protocol=protocols,
              replicas=replicas,
              uses=MyComplexServeExec,
          )
      with ctxt:
          for port, protocol in zip(ports, protocols):
              c = Client(port=port, protocol=protocol)
              docs = c.post(
                  on="/bar",
                  inputs=InputComplexDoc(
                      img=ImageDoc(tensor=np.zeros((3, 224, 224)))
                  ),
                  return_type=DocList[OutputComplexDoc],
              )
              assert docs[0].url == "photo.jpg"
              assert docs[0].lll == [[[40]]]
              assert docs[0].fff == [[[40.2]]]
              assert docs[0].d == {"b": "a"}
              assert docs[0].u == "a"
              assert docs[0].lu == ["3", "4"]
              assert len(docs[0].texts) == 1
              assert docs[0].single_text.text == "single hey ha"
              assert docs[0].single_text.embedding.shape == (2,)
              assert len(docs[0].texts) == 1
              assert docs[0].texts[0].text == "hey ha"
              assert docs[0].texts[0].embedding.shape == (3,)
  
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 33 for calling a global function with example usage: test_serve_complex_model(protocols, replicas, ctxt_manager) and returns: docs_return