code: def test_complex_spec_multiple_endpoints(
      tmp_path, lightning_squeezenet1_1_obj
  ):
      from tests.core.serve.models import (
          ClassificationInferenceComposable,
      )
  
      comp1 = ClassificationInferenceComposable(
          lightning_squeezenet1_1_obj
      )
      comp2 = ClassificationInferenceComposable(
          lightning_squeezenet1_1_obj
      )
      comp3 = ClassificationInferenceComposable(
          lightning_squeezenet1_1_obj
      )
  
      (
          comp1.outputs.predicted_tag >> comp3.inputs.tag
      )  # skipcq: PYL-W0104
      comp2.outputs.cropped_img >> comp3.inputs.img  # skipcq: PYL-W0104
      (
          comp1.outputs.predicted_tag >> comp2.inputs.tag
      )  # skipcq: PYL-W0104
  
      ep1 = Endpoint(
          route="/predict",
          inputs={
              "img_1": comp1.inputs.img,
              "img_2": comp2.inputs.img,
              "tag_1": comp1.inputs.tag,
          },
          outputs={"prediction": comp3.outputs.predicted_tag},
      )
  
      ep2 = Endpoint(
          route="/other_predict",
          inputs={
              "img_1": comp1.inputs.img,
              "img_2": comp2.inputs.img,
              "tag_1": comp1.inputs.tag,
          },
          outputs={
              "prediction_3": comp3.outputs.predicted_tag,
              "prediction_2": comp2.outputs.cropped_img,
          },
      )
  
      composit = Composition(
          comp1=comp1,
          comp2=comp2,
          comp3=comp3,
          predict_compositon_ep=ep1,
          other_predict_ep=ep2,
      )
      connections = [str(c) for c in composit.connections]
      assert connections == [
          "callnum_1.outputs.predicted_tag >> callnum_3.inputs.tag",
          "callnum_1.outputs.predicted_tag >> callnum_2.inputs.tag",
          "callnum_2.outputs.cropped_img >> callnum_3.inputs.img",
      ]
      assert composit.component_uid_names == {
          "callnum_1": "comp1",
          "callnum_2": "comp2",
          "callnum_3": "comp3",
      }
  
      actual_endpoints = {
          k: asdict(v) for k, v in composit.endpoints.items()
      }
      assert actual_endpoints == {
          "predict_compositon_ep": {
              "inputs": {
                  "img_1": "callnum_1.inputs.img",
                  "img_2": "callnum_2.inputs.img",
                  "tag_1": "callnum_1.inputs.tag",
              },
              "outputs": {
                  "prediction": "callnum_3.outputs.predicted_tag",
              },
              "route": "/predict",
          },
          "other_predict_ep": {
              "inputs": {
                  "img_1": "callnum_1.inputs.img",
                  "img_2": "callnum_2.inputs.img",
                  "tag_1": "callnum_1.inputs.tag",
              },
              "outputs": {
                  "prediction_3": "callnum_3.outputs.predicted_tag",
                  "prediction_2": "callnum_2.outputs.cropped_img",
              },
              "route": "/other_predict",
          },
      }
  
file_path: lightning-flash/tests/core/serve/test_composition.py
imports: import base64
  from dataclasses import asdict
  import pytest
  from flash.core.serve import Composition, Endpoint
  from flash.core.utilities.imports import _FASTAPI_AVAILABLE, _TOPIC_SERVE_AVAILABLE
label_desc: Use image node_id 8 for calling a global function with example usage: test_complex_spec_multiple_endpoints(tmp_path, lightning_squeezenet1_1_obj) without return types