Repository: jina-ai/serve
Pull request: fix: list-like args passed as string
<!---Decomposing the complex issue into subtasks can help you build it step-by-step. Thanks for your pull request! :rocket: --->
<!---We know that dev life is hectic, but **please provide a (brief) description** of what your PR does, and how it does it. **Otherwise, your PR cannot be reviewed!** --->
<!---This policy was agreed upon in a past company retro, and makes everyone's life a little easier. Thanks for your collaboration!--->

**Goals:**
<!---https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword--->

- resolves #5352 

- [x] parser can receive single value, str with "," and list of str
- [x] Flow can reveice list of values
- [x] Deployment can accept list of values + remove parsing logic
- [x] add tests
- [x] check and update documentation. See [guide](https://github.com/jina-ai/jina/blob/master/CONTRIBUTING.md#-contributing-documentation) and ask the team.


Problem statement:
bug: list-like args passed as string
**Describe the bug**
Right now in order to have `port-monitoring` (AND OTHER PARAMETERS) support `str` and `List[str]` at the same time, we have to do a hack to pass a string with "," comma separated values. 

This feels unnatural in the CLI and Python API.

We should be able to pass:

```python
from jina import Flow

f = Flow().add(port_monitoring=['8080', '8081'])
```

and in CLI:

```shell
jina executor --port-monitoring "8080" "8081"
```

Hints from issue discussion:

Commit message: test: opentelemetry-sdk is updated to 1.15.0

Patch to review:
diff --git a/docs/concepts/flow/executor-args.md b/docs/concepts/flow/executor-args.md
index 693eb971b7cc6..641203ff678cd 100644
--- a/docs/concepts/flow/executor-args.md
+++ b/docs/concepts/flow/executor-args.md
@@ -24,16 +24,15 @@
 | `volumes` | The path on the host to be mounted inside the container. <br><br>Note, <br>- If separated by `:`, then the first part will be considered as the local host path and the second part is the path in the container system. <br>- If no split provided, then the basename of that directory will be mounted into container's root path, e.g. `--volumes="/user/test/my-workspace"` will be mounted into `/my-workspace` inside the container. <br>- All volumes are mounted with read-write mode. | `array` | `None` |
 | `gpus` | This argument allows dockerized Jina Executors to discover local gpu devices.<br>    <br>    Note, <br>    - To access all gpus, use `--gpus all`.<br>    - To access multiple gpus, e.g. make use of 2 gpus, use `--gpus 2`.<br>    - To access specified gpus based on device id, use `--gpus device=[YOUR-GPU-DEVICE-ID]`<br>    - To access specified gpus based on multiple device id, use `--gpus device=[YOUR-GPU-DEVICE-ID1],device=[YOUR-GPU-DEVICE-ID2]`<br>    - To specify more parameters, use `--gpus device=[YOUR-GPU-DEVICE-ID],runtime=nvidia,capabilities=display | `string` | `None` |
 | `disable_auto_volume` | Do not automatically mount a volume for dockerized Executors. | `boolean` | `False` |
-| `host` | The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor. | `string` | `0.0.0.0` |
 | `runtime_cls` | The runtime class to run inside the Pod | `string` | `WorkerRuntime` |
 | `timeout_ready` | The timeout in milliseconds of a Pod waits for the runtime to be ready, -1 for waiting forever | `number` | `600000` |
 | `env` | The map of environment variables that are available inside runtime | `object` | `None` |
 | `floating` | If set, the current Pod/Deployment can not be further chained, and the next `.add()` will chain after the last Pod/Deployment not this current one. | `boolean` | `False` |
 | `restart` | If set, the Executor will restart while serving if the YAML configuration source is changed. This differs from `reload` argument in that this will restart the server and more configuration can be changed, like number of replicas. | `boolean` | `False` |
-| `port` | The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports, separated by commas. Then, every resulting address will be considered as one replica of the Executor. | `None` | `random in [49152, 65535]` |
+| `port` | The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports. Then, every resulting address will be considered as one replica of the Executor. | `number` | `random in [49152, 65535]` |
 | `reload` | If set, the Executor reloads the modules as they change | `boolean` | `False` |
 | `monitoring` | If set, spawn an http server with a prometheus endpoint to expose metrics | `boolean` | `False` |
-| `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `string` | `random in [49152, 65535]` |
+| `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
 | `retries` | Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas) | `number` | `-1` |
 | `tracing` | If set, the sdk implementation of the OpenTelemetry tracer will be available and will be enabled for automatic tracing of requests and customer span creation. Otherwise a no-op implementation will be provided. | `boolean` | `False` |
 | `traces_exporter_host` | If tracing is enabled, this hostname will be used to configure the trace exporter agent. | `string` | `None` |
diff --git a/docs/concepts/flow/gateway-args.md b/docs/concepts/flow/gateway-args.md
index 9363d4cc1557d..f6e99e193afaf 100644
--- a/docs/concepts/flow/gateway-args.md
+++ b/docs/concepts/flow/gateway-args.md
@@ -20,7 +20,7 @@
 | `ssl_keyfile` | the path to the key file | `string` | `None` |
 | `expose_graphql_endpoint` | If set, /graphql endpoint is added to HTTP interface. | `boolean` | `False` |
 | `protocol` | Communication protocol of the server exposed by the Gateway. This can be a single value or a list of protocols, depending on your chosen Gateway. Choose the convenient protocols from: ['GRPC', 'HTTP', 'WEBSOCKET']. | `array` | `[<GatewayProtocolType.GRPC: 0>]` |
-| `host` | The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor. | `string` | `0.0.0.0` |
+| `host` | The host address of the runtime, by default it is 0.0.0.0. | `string` | `0.0.0.0` |
 | `proxy` | If set, respect the http_proxy and https_proxy environment variables. otherwise, it will unset these proxy variables before start. gRPC seems to prefer no proxy | `boolean` | `False` |
 | `uses` | The config of the gateway, it could be one of the followings:<br>        * the string literal of an Gateway class name<br>        * a Gateway YAML file (.yml, .yaml, .jaml)<br>        * a docker image (must start with `docker://`)<br>        * the string literal of a YAML config (must start with `!` or `jtype: `)<br>        * the string literal of a JSON config<br><br>        When use it under Python, one can use the following values additionally:<br>        - a Python dict that represents the config<br>        - a text file stream has `.read()` interface | `string` | `None` |
 | `uses_with` | Dictionary of keyword arguments that will override the `with` configuration in `uses` | `object` | `None` |
@@ -38,9 +38,9 @@
 | `env` | The map of environment variables that are available inside runtime | `object` | `None` |
 | `floating` | If set, the current Pod/Deployment can not be further chained, and the next `.add()` will chain after the last Pod/Deployment not this current one. | `boolean` | `False` |
 | `restart` | If set, the Gateway will restart while serving if the YAML configuration source is changed. | `boolean` | `False` |
-| `port` | The port for input data to bind the gateway server to, by default, random ports between range [49152, 65535] will be assigned. The port argument can be either 1 single value in case only 1 protocol is used or multiple values when many protocols are used. | `None` | `random in [49152, 65535]` |
+| `port` | The port for input data to bind the gateway server to, by default, random ports between range [49152, 65535] will be assigned. The port argument can be either 1 single value in case only 1 protocol is used or multiple values when many protocols are used. | `number` | `random in [49152, 65535]` |
 | `monitoring` | If set, spawn an http server with a prometheus endpoint to expose metrics | `boolean` | `False` |
-| `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `string` | `random in [49152, 65535]` |
+| `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
 | `retries` | Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas) | `number` | `-1` |
 | `tracing` | If set, the sdk implementation of the OpenTelemetry tracer will be available and will be enabled for automatic tracing of requests and customer span creation. Otherwise a no-op implementation will be provided. | `boolean` | `False` |
 | `traces_exporter_host` | If tracing is enabled, this hostname will be used to configure the trace exporter agent. | `string` | `None` |
diff --git a/docs/concepts/flow/scale-out.md b/docs/concepts/flow/scale-out.md
index d73314b81ec06..1e6bbb2bf3c93 100644
--- a/docs/concepts/flow/scale-out.md
+++ b/docs/concepts/flow/scale-out.md
@@ -132,11 +132,11 @@ If you have external Executors with multiple replicas running elsewhere, you can
 ```python
 from jina import Flow
 
-replica_hosts, replica_ports = 'localhost,91.198.174.192', '12345,12346'
+replica_hosts, replica_ports = ['localhost','91.198.174.192'], ['12345','12346']
 Flow().add(host=replica_hosts, port=replica_ports, external=True)
 
 # alternative syntax
-Flow().add(host='localhost:12345,91.198.174.192:12346', external=True)
+Flow().add(host=['localhost:12345','91.198.174.192:12346'], external=True)
 ```
 
 This connects to `grpc://localhost:12345` and `grpc://91.198.174.192:12346` as two replicas of the external Executor.
diff --git a/jina/clients/__init__.py b/jina/clients/__init__.py
index 93d5e0a226f5b..fcd77427a0e1c 100644
--- a/jina/clients/__init__.py
+++ b/jina/clients/__init__.py
@@ -33,7 +33,7 @@ def Client(*,
     """Create a Client. Client is how user interact with Flow
 
     :param asyncio: If set, then the input and output of this Client work in an asynchronous manner.
-    :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+    :param host: The host of the Gateway, which the client should connect to, by default it is 0.0.0.0.
     :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
     :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
     :param metrics_exporter_port: If tracing is enabled, this port will be used to configure the metrics exporter agent.
@@ -83,7 +83,7 @@ def Client(
         c.post(on='/index', inputs=Document(text='hello!'))
 
     :param asyncio: If set, then the input and output of this Client work in an asynchronous manner.
-    :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+    :param host: The host of the Gateway, which the client should connect to, by default it is 0.0.0.0.
     :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
     :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
     :param metrics_exporter_port: If tracing is enabled, this port will be used to configure the metrics exporter agent.
diff --git a/jina/helper.py b/jina/helper.py
index a58a80a3ee677..b517b5a230bef 100644
--- a/jina/helper.py
+++ b/jina/helper.py
@@ -1643,60 +1643,18 @@ def _single_port_free(host: str, port: int) -> bool:
             return True
 
 
-def is_port_free(host: str, port: Union[int, List[int]]) -> bool:
+def is_port_free(host: Union[str, List[str]], port: Union[int, List[int]]) -> bool:
     if isinstance(port, list):
-        return all([_single_port_free(host, _p) for _p in port])
+        if isinstance(host, str):
+            return all([_single_port_free(host, _p) for _p in port])
+        else:
+            return all([all([_single_port_free(_h, _p) for _p in port]) for _h in host])
     else:
-        return _single_port_free(host, port)
-
-
-def _parse_ports(port: str) -> Union[int, List]:
-    """Parse port
-
-    EXAMPLE USAGE
-
-    .. code-block:: python
-
-
-        _parse_port('8000')
-        8000
-
-        _parse_port('8001,8002,8005')
-        [80001, 8002, 8005]
-
-    :param port: the string to parse
-    :return: the port or the iterable ports
-    """
-    try:
-        port = int(port)
-    except ValueError as e:
-        if ',' in port:
-            port = [int(port_) for port_ in port.split(',')]
-        elif not isinstance(port, list):
-            raise e
-    return port
-
-
-def _parse_hosts(host: str) -> Union[str, List[str]]:
-    """Parse port
-
-    EXAMPLE USAGE
-
-    .. code-block:: python
-
-
-        _parse_hosts('localhost')
-        'localhost'
-
-        _parse_port('localhost,91.198.174.192')
-        ['localhost', '91.198.174.192']
-
-    :param host: the string to parse
-    :return: the host or the iterable of hosts
-    """
-    hosts = host.split(',')
-    return hosts[0] if len(hosts) == 1 else hosts
-
+        if isinstance(host, str):
+            return _single_port_free(host, port)
+        else:
+            return all([_single_port_free(_h, port) for _h in host])
+    
 
 def send_telemetry_event(event: str, obj: Any, **kwargs) -> None:
     """Sends in a thread a request with telemetry for a given event
@@ -1730,35 +1688,3 @@ def _telemetry():
 
     threading.Thread(target=_telemetry, daemon=True).start()
 
-
-def make_iterable(o: object) -> Iterable:
-    """
-    Make an object an iterable by wrapping it as a singleton list.
-    If the input is already an iterable (except str and bytes), it will be returned as is.
-    Str and bytes are treated as non-iterable, and thus wrapped in a list.
-
-    EXAMPLE USAGE
-
-    .. code-block:: python
-
-
-        make_iter(1)
-        [1]
-
-        make_iter('a')
-        ['a']
-
-        make_iter([1, 2, 3])
-        [1, 2, 3]
-
-        make_iter((1, 2, 3))
-        (1, 2, 3)
-
-
-    :param o: the object to be converted to an iterable
-    :return: the iterable
-    """
-    if isinstance(o, Iterable) and not isinstance(o, (str, bytes)):
-        return o
-    else:
-        return [o]
diff --git a/jina/orchestrate/deployments/__init__.py b/jina/orchestrate/deployments/__init__.py
index 500755ca35528..0330e430b7ba8 100644
--- a/jina/orchestrate/deployments/__init__.py
+++ b/jina/orchestrate/deployments/__init__.py
@@ -17,12 +17,8 @@
 from jina.enums import DeploymentRoleType, PodRoleType, PollingType
 from jina.helper import (
     CatchAllCleanupContextManager,
-    _parse_hosts,
-    _parse_ports,
-    make_iterable,
     parse_host_scheme,
 )
-from jina.jaml.helper import complete_path
 from jina.orchestrate.pods.factory import PodFactory
 from jina.parsers.helper import _update_gateway_args
 from jina.serve.networking import host_is_local, in_docker
@@ -99,10 +95,8 @@ def _copy_to_head_args(args: Namespace) -> Namespace:
 
         _head_args = copy.deepcopy(args)
         _head_args.polling = args.polling
-        if not hasattr(args, 'port') or not args.port:
-            _head_args.port = helper.random_port()
-        else:
-            _head_args.port = args.port
+        _head_args.port = args.port[0]
+        _head_args.host = args.host[0]
         _head_args.uses = args.uses
         _head_args.pod_role = PodRoleType.HEAD
         _head_args.runtime_cls = 'HeadRuntime'
@@ -243,7 +237,7 @@ def __init__(
         self.uses_after_pod = None
         self.head_pod = None
         self.shards = {}
-        self._update_port_args()
+        self._update_port_monitoring_args()
         self.update_pod_args()
         self._sandbox_deployed = False
 
@@ -253,12 +247,14 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
 
     def _parse_addresses_into_host_and_port(self):
         # splits addresses passed to `host` into separate `host` and `port`
-        _hostname, port, scheme, tls = parse_host_scheme(self.args.host)
-        if _hostname != self.args.host:  # more than just hostname was passed to `host`
-            self.args.host = _hostname
-            self.args.port = port
-            self.args.scheme = scheme
-            self.args.tls = tls
+
+        for i, _host in enumerate(self.args.host):
+            _hostname, port, scheme, tls = parse_host_scheme(_host)
+            if _hostname != _host:  # more than just hostname was passed to `host`
+                self.args.host[i] = _hostname
+                self.args.port[i] = port
+                self.args.scheme = scheme
+                self.args.tls = tls
         for i, repl_host in enumerate(self.ext_repl_hosts):
             _hostname, port, scheme, tls = parse_host_scheme(repl_host)
             if (
@@ -271,23 +267,25 @@ def _parse_addresses_into_host_and_port(self):
 
     def _parse_external_replica_hosts_and_ports(self):
         # splits user provided lists of hosts and ports into a host and port for every distributed replica
-        ext_repl_ports = make_iterable(_parse_ports(str(self.args.port)))
-        ext_repl_hosts = make_iterable(_parse_hosts(str(self.args.host)))
+        ext_repl_ports: List = self.args.port.copy()
+        ext_repl_hosts: List = self.args.host.copy()
         if len(ext_repl_hosts) < len(ext_repl_ports):
             if (
                 len(ext_repl_hosts) == 1
             ):  # only one host given, assume replicas are on the same host
                 ext_repl_hosts = ext_repl_hosts * len(ext_repl_ports)
+                self.args.host = self.args.host * len(ext_repl_ports)
         elif len(ext_repl_hosts) > len(ext_repl_ports):
             if (
                 len(ext_repl_ports) == 1
             ):  # only one port given, assume replicas are on the same port
                 ext_repl_ports = ext_repl_ports * len(ext_repl_hosts)
+                self.args.port = self.args.port * len(ext_repl_hosts)
         if len(ext_repl_hosts) != len(ext_repl_ports):
             raise ValueError(
                 f'Number of hosts ({len(ext_repl_hosts)}) does not match the number of ports ({len(ext_repl_ports)})'
             )
-        self.args.port, self.args.host = int(ext_repl_ports[0]), ext_repl_hosts[0]
+
         self.ext_repl_hosts, self.ext_repl_ports = ext_repl_hosts, ext_repl_ports
         # varying tls and schemes other than 'grpc' only implemented if the entire address is passed to `host`
         self.ext_repl_schemes = [
@@ -297,8 +295,8 @@ def _parse_external_replica_hosts_and_ports(self):
             getattr(self.args, 'tls', None) for _ in self.ext_repl_ports
         ]
 
-    def _update_port_args(self):
-        _all_port_monitoring = _parse_ports(self.args.port_monitoring)
+    def _update_port_monitoring_args(self):
+        _all_port_monitoring = self.args.port_monitoring
         self.args.all_port_monitoring = (
             [_all_port_monitoring]
             if not type(_all_port_monitoring) == list
@@ -318,19 +316,6 @@ def update_pod_args(self):
         else:
             self.pod_args = self._parse_args(self.args)
 
-        if self.external:
-            for pod, port, host, scheme, tls in zip(
-                self.pod_args['pods'][0],
-                self.ext_repl_ports,
-                self.ext_repl_hosts,
-                self.ext_repl_schemes,
-                self.ext_repl_tls,
-            ):
-                pod.port = port
-                pod.host = host
-                pod.scheme = scheme
-                pod.tls = tls
-
     def update_sandbox_args(self):
         """Update args of all its pods based on the host and port returned by Hubble"""
         if self.is_sandbox:
@@ -344,7 +329,7 @@ def update_sandbox_args(self):
 
     def update_worker_pod_args(self):
         """Update args of all its worker pods based on Deployment args. Does not touch head and tail"""
-        self.pod_args['pods'] = self._set_pod_args(self.args)
+        self.pod_args['pods'] = self._set_pod_args()
 
     @property
     def is_sandbox(self) -> bool:
@@ -788,32 +773,40 @@ def _roundrobin_cuda_device(device_str: str, replicas: int):
             _c = cycle(selected_devices)
             return {j: next(_c) for j in range(replicas)}
 
-    @staticmethod
-    def _set_pod_args(args: Namespace) -> Dict[int, List[Namespace]]:
+    def _set_pod_args(self) -> Dict[int, List[Namespace]]:
         result = {}
-        shards = getattr(args, 'shards', 1)
-        replicas = getattr(args, 'replicas', 1)
+        shards = getattr(self.args, 'shards', 1)
+        replicas = getattr(self.args, 'replicas', 1)
         sharding_enabled = shards and shards > 1
 
         cuda_device_map = None
-        if args.env:
+        if self.args.env:
             cuda_device_map = Deployment._roundrobin_cuda_device(
-                args.env.get('CUDA_VISIBLE_DEVICES'), replicas
+                self.args.env.get('CUDA_VISIBLE_DEVICES'), replicas
             )
 
         for shard_id in range(shards):
             replica_args = []
             for replica_id in range(replicas):
-                _args = copy.deepcopy(args)
+                _args = copy.deepcopy(self.args)
                 _args.shard_id = shard_id
                 # for gateway pods, the pod role shouldn't be changed
                 if _args.pod_role != PodRoleType.GATEWAY:
                     _args.pod_role = PodRoleType.WORKER
+                    if len(self.args.host) == 1:
+                        _args.host = self.args.host[0]
+                    elif len(self.args.host) == replicas:
+                        _args.host = self.args.host[replica_id]
+                    else:
+                        raise ValueError(
+                            f'Number of hosts ({len(self.args.host)}) does not match the number of replicas ({replicas})'
+                        )
+                else:
+                    _args.host = self.args.host
 
                 if cuda_device_map:
                     _args.env['CUDA_VISIBLE_DEVICES'] = str(cuda_device_map[replica_id])
-
-                _args.host = args.host
+                
                 if _args.name:
                     _args.name += (
                         f'/shard-{shard_id}/rep-{replica_id}'
@@ -824,40 +817,52 @@ def _set_pod_args(args: Namespace) -> Dict[int, List[Namespace]]:
                     _args.name = f'{replica_id}'
 
                 # the gateway needs to respect the assigned port
-                if args.deployment_role == DeploymentRoleType.GATEWAY or args.external:
-                    _args.port = args.port
-                elif shards == 1 and replicas == 1:
-                    _args.port = args.port
-                    _args.port_monitoring = args.port_monitoring
-
-                elif shards == 1:
-                    _args.port_monitoring = (
-                        helper.random_port()
-                        if replica_id >= len(args.all_port_monitoring)
-                        else args.all_port_monitoring[replica_id]
-                    )
-                    # if there are no shards/replicas, we dont need to distribute ports randomly
-                    # we should rather use the pre assigned one
-                    _args.port = helper.random_port()
-                elif shards > 1:
-                    port_monitoring_index = (
-                        replica_id + replicas * shard_id + 1
-                    )  # the first index is for the head
-                    _args.port_monitoring = (
-                        helper.random_port()
-                        if port_monitoring_index >= len(args.all_port_monitoring)
-                        else args.all_port_monitoring[
-                            port_monitoring_index
-                        ]  # we skip the head port here
-                    )
-                    _args.port = helper.random_port()
+                if self.args.deployment_role == DeploymentRoleType.GATEWAY:
+                    _args.port = self.args.port
+                
+                elif not self.external:
+                    if shards == 1 and replicas == 1:
+                        _args.port = self.args.port[0]
+                        _args.port_monitoring = self.args.port_monitoring
+
+                    elif shards == 1:
+                        _args.port_monitoring = (
+                            helper.random_port()
+                            if replica_id >= len(self.args.all_port_monitoring)
+                            else self.args.all_port_monitoring[replica_id]
+                        )
+                        # if there are no shards/replicas, we dont need to distribute ports randomly
+                        # we should rather use the pre assigned one
+                        _args.port = helper.random_port()
+                    elif shards > 1:
+                        port_monitoring_index = (
+                            replica_id + replicas * shard_id + 1
+                        )  # the first index is for the head
+                        _args.port_monitoring = (
+                            helper.random_port()
+                            if port_monitoring_index >= len(self.args.all_port_monitoring)
+                            else self.args.all_port_monitoring[
+                                port_monitoring_index
+                            ]  # we skip the head port here
+                        )
+                        _args.port = helper.random_port()
+                    else:
+                        _args.port = helper.random_port()
+                        _args.port_monitoring = helper.random_port()
+                
                 else:
-                    _args.port = helper.random_port()
-                    _args.port_monitoring = helper.random_port()
+                    # if shards > 1:
+                    #     raise ValueError(
+                    #         f'external deployment with multiple shards is not supported'
+                    #     )
+                    _args.port = self.ext_repl_ports[replica_id]
+                    _args.host = self.ext_repl_hosts[replica_id]
+                    _args.scheme = self.ext_repl_schemes[replica_id]
+                    _args.tls = self.ext_repl_tls[replica_id]
 
                 # pod workspace if not set then derive from workspace
                 if not _args.workspace:
-                    _args.workspace = args.workspace
+                    _args.workspace = self.args.workspace
                 replica_args.append(_args)
             result[shard_id] = replica_args
         return result
@@ -867,7 +872,7 @@ def _set_uses_before_after_args(args: Namespace, entity_type: str) -> Namespace:
 
         _args = copy.deepcopy(args)
         _args.pod_role = PodRoleType.WORKER
-        _args.host = _args.host or __default_host__
+        _args.host = _args.host[0] or __default_host__
         _args.port = helper.random_port()
 
         if _args.name:
@@ -927,7 +932,7 @@ def _parse_base_deployment_args(self, args):
 
             parsed_args['head'] = BaseDeployment._copy_to_head_args(args)
 
-        parsed_args['pods'] = self._set_pod_args(args)
+        parsed_args['pods'] = self._set_pod_args()
 
         if parsed_args['head'] is not None:
             connection_list = defaultdict(list)
diff --git a/jina/orchestrate/deployments/config/docker_compose.py b/jina/orchestrate/deployments/config/docker_compose.py
index 72772d0995010..7d70d67c95d4e 100644
--- a/jina/orchestrate/deployments/config/docker_compose.py
+++ b/jina/orchestrate/deployments/config/docker_compose.py
@@ -93,7 +93,7 @@ def get_gateway_config(
             protocol = str(non_defaults.get('protocol', ['grpc'])[0]).lower()
 
             ports = cargs.port + (
-                [f'{cargs.port_monitoring}'] if cargs.monitoring else []
+                [cargs.port_monitoring] if cargs.monitoring else []
             )
 
             envs = [f'JINA_LOG_LEVEL={os.getenv("JINA_LOG_LEVEL", "INFO")}']
@@ -351,6 +351,7 @@ def _get_services_args(self, args):
             uses_before_cargs.uses_with = None
             uses_before_cargs.uses_metas = None
             uses_before_cargs.env = None
+            uses_before_cargs.host = args.host[0]
             uses_before_cargs.port = port
             uses_before_cargs.uses_before_address = None
             uses_before_cargs.uses_after_address = None
@@ -374,6 +375,7 @@ def _get_services_args(self, args):
             uses_after_cargs.uses_with = None
             uses_after_cargs.uses_metas = None
             uses_after_cargs.env = None
+            uses_after_cargs.host = args.host[0]
             uses_after_cargs.port = port
             uses_after_cargs.uses_before_address = None
             uses_after_cargs.uses_after_address = None
@@ -400,6 +402,7 @@ def _get_services_args(self, args):
                 cargs.pod_role = PodRoleType.GATEWAY
             else:
                 cargs.port = port
+                cargs.host = args.host[0]
             parsed_args['services'].append(cargs)
 
         return parsed_args
diff --git a/jina/orchestrate/deployments/config/k8s.py b/jina/orchestrate/deployments/config/k8s.py
index 1007de2b6d1aa..da5764689f05d 100644
--- a/jina/orchestrate/deployments/config/k8s.py
+++ b/jina/orchestrate/deployments/config/k8s.py
@@ -330,6 +330,7 @@ def _get_deployment_args(self, args):
 
         for i in range(shards):
             cargs = copy.deepcopy(args)
+            cargs.host = args.host[0]
             cargs.shard_id = i
             cargs.uses_before = None
             cargs.uses_after = None
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 797490c21e69d..f0e20619fc098 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -131,7 +131,7 @@ def __init__(
         """Create a Flow. Flow is how Jina streamlines and scales Executors. This overloaded method provides arguments from `jina client` CLI.
 
         :param asyncio: If set, then the input and output of this Client work in an asynchronous manner.
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param host: The host of the Gateway, which the client should connect to, by default it is 0.0.0.0.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
         :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
         :param metrics_exporter_port: If tracing is enabled, this port will be used to configure the metrics exporter agent.
@@ -177,8 +177,8 @@ def __init__(
         name: Optional[str] = 'gateway', 
         no_crud_endpoints: Optional[bool] = False, 
         no_debug_endpoints: Optional[bool] = False, 
-        port: Optional[Union[int, List[int]]] = None, 
-        port_monitoring: Optional[str] = None, 
+        port: Optional[int] = None, 
+        port_monitoring: Optional[int] = None, 
         prefetch: Optional[int] = 1000, 
         protocol: Optional[Union[str, List[str]]] = ['GRPC'], 
         proxy: Optional[bool] = False, 
@@ -222,7 +222,7 @@ def __init__(
         :param graph_conditions: Dictionary stating which filtering conditions each Executor in the graph requires to receive Documents.
         :param graph_description: Routing graph for the gateway
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param host: The host address of the runtime, by default it is 0.0.0.0.
         :param log_config: The YAML config of the logger used in this object.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
         :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
@@ -363,7 +363,7 @@ def __init__(
                     f.bock()  # serve Flow
 
         :param asyncio: If set, then the input and output of this Client work in an asynchronous manner.
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param host: The host of the Gateway, which the client should connect to, by default it is 0.0.0.0.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
         :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
         :param metrics_exporter_port: If tracing is enabled, this port will be used to configure the metrics exporter agent.
@@ -392,7 +392,7 @@ def __init__(
         :param graph_conditions: Dictionary stating which filtering conditions each Executor in the graph requires to receive Documents.
         :param graph_description: Routing graph for the gateway
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param host: The host address of the runtime, by default it is 0.0.0.0.
         :param log_config: The YAML config of the logger used in this object.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
         :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
@@ -840,7 +840,6 @@ def add(
         gpus: Optional[str] = None, 
         grpc_metadata: Optional[dict] = None, 
         grpc_server_options: Optional[dict] = None, 
-        host: Optional[str] = '0.0.0.0', 
         install_requirements: Optional[bool] = False, 
         log_config: Optional[str] = None, 
         metrics: Optional[bool] = False, 
@@ -852,8 +851,8 @@ def add(
         no_reduce: Optional[bool] = False, 
         output_array_type: Optional[str] = None, 
         polling: Optional[str] = 'ANY', 
-        port: Optional[Union[int, List[int]]] = None, 
-        port_monitoring: Optional[str] = None, 
+        port: Optional[int] = None, 
+        port_monitoring: Optional[int] = None, 
         py_modules: Optional[List[str]] = None, 
         quiet: Optional[bool] = False, 
         quiet_error: Optional[bool] = False, 
@@ -907,7 +906,6 @@ def add(
               - To specify more parameters, use `--gpus device=[YOUR-GPU-DEVICE-ID],runtime=nvidia,capabilities=display
         :param grpc_metadata: The metadata to be passed to the gRPC request.
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
         :param install_requirements: If set, install `requirements.txt` in the Hub Executor bundle to local
         :param log_config: The YAML config of the logger used in this object.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
@@ -938,7 +936,7 @@ def add(
               Define per Endpoint:
               JSON dict, {endpoint: PollingType}
               {'/custom': 'ALL', '/search': 'ANY', '*': 'ANY'}
-        :param port: The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param port: The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports. Then, every resulting address will be considered as one replica of the Executor.
         :param port_monitoring: The port on which the prometheus server is exposed, default is a random port between [49152, 65535]
         :param py_modules: The customized python modules need to be imported before loading the executor
           
@@ -1053,7 +1051,6 @@ def add(
               - To specify more parameters, use `--gpus device=[YOUR-GPU-DEVICE-ID],runtime=nvidia,capabilities=display
         :param grpc_metadata: The metadata to be passed to the gRPC request.
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
         :param install_requirements: If set, install `requirements.txt` in the Hub Executor bundle to local
         :param log_config: The YAML config of the logger used in this object.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
@@ -1084,7 +1081,7 @@ def add(
               Define per Endpoint:
               JSON dict, {endpoint: PollingType}
               {'/custom': 'ALL', '/search': 'ANY', '*': 'ANY'}
-        :param port: The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param port: The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports. Then, every resulting address will be considered as one replica of the Executor.
         :param port_monitoring: The port on which the prometheus server is exposed, default is a random port between [49152, 65535]
         :param py_modules: The customized python modules need to be imported before loading the executor
           
@@ -1207,11 +1204,6 @@ def add(
 
         args.noblock_on_start = True
 
-        port = kwargs.get('port', None)
-        if not port:
-            port = helper.random_port()
-            args.port = port
-
         if len(needs) > 1 and args.external and args.no_reduce:
             raise ValueError(
                 'External Executors with multiple needs have to do auto reduce.'
@@ -1252,8 +1244,8 @@ def config_gateway(
         name: Optional[str] = 'gateway', 
         no_crud_endpoints: Optional[bool] = False, 
         no_debug_endpoints: Optional[bool] = False, 
-        port: Optional[Union[int, List[int]]] = None, 
-        port_monitoring: Optional[str] = None, 
+        port: Optional[int] = None, 
+        port_monitoring: Optional[int] = None, 
         prefetch: Optional[int] = 1000, 
         protocol: Optional[Union[str, List[str]]] = ['GRPC'], 
         proxy: Optional[bool] = False, 
@@ -1297,7 +1289,7 @@ def config_gateway(
         :param graph_conditions: Dictionary stating which filtering conditions each Executor in the graph requires to receive Documents.
         :param graph_description: Routing graph for the gateway
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param host: The host address of the runtime, by default it is 0.0.0.0.
         :param log_config: The YAML config of the logger used in this object.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
         :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
@@ -1392,7 +1384,7 @@ def config_gateway(
         :param graph_conditions: Dictionary stating which filtering conditions each Executor in the graph requires to receive Documents.
         :param graph_description: Routing graph for the gateway
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
-        :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
+        :param host: The host address of the runtime, by default it is 0.0.0.0.
         :param log_config: The YAML config of the logger used in this object.
         :param metrics: If set, the sdk implementation of the OpenTelemetry metrics will be available for default monitoring and custom measurements. Otherwise a no-op implementation will be provided.
         :param metrics_exporter_host: If tracing is enabled, this hostname will be used to configure the metrics exporter agent.
diff --git a/jina/orchestrate/pods/factory.py b/jina/orchestrate/pods/factory.py
index 1642c7533d6c0..9117e3e3390fb 100644
--- a/jina/orchestrate/pods/factory.py
+++ b/jina/orchestrate/pods/factory.py
@@ -27,6 +27,7 @@ def build_pod(args: 'Namespace') -> Type['BasePod']:
         :return: the created BaseDeployment
         """
         # copy to update but forward original
+            
         cargs = deepcopy(args)
 
         if is_valid_huburi(cargs.uses):
diff --git a/jina/parsers/helper.py b/jina/parsers/helper.py
index b7a93fe92dd2f..3291eaaff4be9 100644
--- a/jina/parsers/helper.py
+++ b/jina/parsers/helper.py
@@ -311,10 +311,10 @@ def __call__(self, parser, args, values, option_string=None):
         :param values: the values to add to the parser
         :param option_string: inherited, not used
         """
-        if isinstance(values, list):
-            d = [_port_to_int(port) for port in values]
-        elif isinstance(values, str):
-            d = _port_to_int(values)
+        d = []
+        for value in values:
+            value = value.split(',')
+            d.extend([_port_to_int(port) for port in value])
         setattr(args, self.dest, d)
 
 
@@ -328,4 +328,23 @@ def _port_to_int(port):
         return port
 
 
+class CastHostAction(argparse.Action):
+    """argparse action to cast a list of values to int"""
+
+    def __call__(self, parser, args, values, option_string=None):
+        """
+        call the CastHostAction
+
+
+        .. # noqa: DAR401
+        :param parser: the parser
+        :param args: args to initialize the values
+        :param values: the values to add to the parser
+        :param option_string: inherited, not used
+        """
+        d = []
+        for value in values:
+            d.extend(value.split(','))
+        setattr(args, self.dest, d)
+
 _chf = _ColoredHelpFormatter
diff --git a/jina/parsers/orchestrate/pod.py b/jina/parsers/orchestrate/pod.py
index 2370cfcc48225..8d353ff67389b 100644
--- a/jina/parsers/orchestrate/pod.py
+++ b/jina/parsers/orchestrate/pod.py
@@ -124,7 +124,7 @@ def mixin_pod_runtime_args_parser(arg_group, pod_type='worker'):
     """
     port_description = (
         'The port for input data to bind to, default is a random port between [49152, 65535]. '
-        'In the case of an external Executor (`--external` or `external=True`) this can be a list of ports, separated by commas. '
+        'In the case of an external Executor (`--external` or `external=True`) this can be a list of ports. '
         'Then, every resulting address will be considered as one replica of the Executor.'
     )
 
@@ -133,7 +133,8 @@ def mixin_pod_runtime_args_parser(arg_group, pod_type='worker'):
             '--port',
             '--port-in',
             type=str,
-            default=helper.random_port(),
+            nargs='+',
+            default=[helper.random_port()],
             action=CastToIntAction,
             help=port_description,
         )
@@ -171,7 +172,9 @@ def mixin_pod_runtime_args_parser(arg_group, pod_type='worker'):
     arg_group.add_argument(
         '--port-monitoring',
         type=str,
-        default=str(helper.random_port()),
+        nargs='+',
+        default=[helper.random_port()],
+        action=CastToIntAction,
         dest='port_monitoring',
         help=f'The port on which the prometheus server is exposed, default is a random port between [49152, 65535]',
     )
diff --git a/jina/parsers/orchestrate/runtimes/remote.py b/jina/parsers/orchestrate/runtimes/remote.py
index 1798031803b35..7af5d27669ccf 100644
--- a/jina/parsers/orchestrate/runtimes/remote.py
+++ b/jina/parsers/orchestrate/runtimes/remote.py
@@ -1,7 +1,7 @@
 """Argparser module for remote runtime"""
 
 from jina import __default_host__, helper
-from jina.parsers.helper import KVAppendAction, add_arg_group
+from jina.parsers.helper import CastHostAction, KVAppendAction, add_arg_group
 from jina.parsers.orchestrate.runtimes.runtime import mixin_base_runtime_parser
 
 
@@ -10,7 +10,17 @@ def mixin_remote_runtime_parser(parser):
     :param parser: the parser
     """
     gp = add_arg_group(parser, title='RemoteRuntime')
-    _add_host(gp)
+
+    gp.add_argument(
+        '--host',
+        '--host-in',
+        nargs='+',
+        default=[__default_host__],
+        action=CastHostAction,
+        help=f'The host of the Gateway, which the client should connect to, by default it is {__default_host__}.'
+        ' In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts. '
+        ' Then, every resulting address will be considered as one replica of the Executor.',
+    )
 
 
 def mixin_client_gateway_parser(parser):
@@ -18,9 +28,16 @@ def mixin_client_gateway_parser(parser):
     :param parser: the parser
     """
     gp = add_arg_group(parser, title='ClientGateway')
-    _add_host(gp)
     _add_proxy(gp)
 
+    gp.add_argument(
+        '--host',
+        '--host-in',
+        type=str,
+        default=__default_host__,
+        help=f'The host of the Gateway, which the client should connect to, by default it is {__default_host__}.',
+    )
+
     gp.add_argument(
         '--port',
         type=int,
@@ -171,9 +188,7 @@ def _add_host(arg_group):
         '--host-in',
         type=str,
         default=__default_host__,
-        help=f'The host address of the runtime, by default it is {__default_host__}.'
-        ' In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas.'
-        ' Then, every resulting address will be considered as one replica of the Executor.',
+        help=f'The host address of the runtime, by default it is {__default_host__}.',
     )
 
 
diff --git a/jina/serve/runtimes/head/__init__.py b/jina/serve/runtimes/head/__init__.py
index 8c97dc4443bba..8655be824e746 100644
--- a/jina/serve/runtimes/head/__init__.py
+++ b/jina/serve/runtimes/head/__init__.py
@@ -158,7 +158,7 @@ async def async_setup(self):
                 service, health_pb2.HealthCheckResponse.SERVING
             )
         reflection.enable_server_reflection(service_names, self._grpc_server)
-
+     
         bind_addr = f'{self.args.host}:{self.args.port}'
         self._grpc_server.add_insecure_port(bind_addr)
         self.logger.debug(f'start listening on {bind_addr}')
diff --git a/jina/serve/runtimes/worker/__init__.py b/jina/serve/runtimes/worker/__init__.py
index 5964dbecdc468..0980fc4097815 100644
--- a/jina/serve/runtimes/worker/__init__.py
+++ b/jina/serve/runtimes/worker/__init__.py
@@ -142,7 +142,7 @@ async def _async_setup_grpc_server(self):
             self._health_servicer, self._grpc_server
         )
 
-        reflection.enable_server_reflection(service_names, self._grpc_server)
+        reflection.enable_server_reflection(service_names, self._grpc_server)     
         bind_addr = f'{self.args.host}:{self.args.port}'
         self.logger.debug(f'start listening on {bind_addr}')
         self._grpc_server.add_insecure_port(bind_addr)
diff --git a/jina_cli/api.py b/jina_cli/api.py
index 2085fa9435568..8538822ae8087 100644
--- a/jina_cli/api.py
+++ b/jina_cli/api.py
@@ -74,6 +74,10 @@ def executor(args: 'Namespace'):
 
     :returns: return the same as `pod` or `worker_runtime`
     """
+    args.host = args.host[0]
+    args.port = args.port[0]
+    args.port_monitoring = args.port_monitoring[0]
+        
     if args.native:
         return executor_native(args)
     else:
@@ -103,6 +107,7 @@ def gateway(args: 'Namespace'):
     """
     from jina.serve.runtimes import get_runtime
 
+    args.port_monitoring = args.port_monitoring[0]
     _update_gateway_args(args)
 
     runtime_cls = get_runtime('GatewayRuntime')
diff --git a/jina_cli/autocomplete.py b/jina_cli/autocomplete.py
index 63b4aed83e444..4290b319afd96 100644
--- a/jina_cli/autocomplete.py
+++ b/jina_cli/autocomplete.py
@@ -367,9 +367,9 @@
         ],
         'client': [
             '--help',
+            '--proxy',
             '--host',
             '--host-in',
-            '--proxy',
             '--port',
             '--tls',
             '--asyncio',
diff --git a/jina_cli/export.py b/jina_cli/export.py
index 58f7f88f4f625..26eff27ee6bbd 100644
--- a/jina_cli/export.py
+++ b/jina_cli/export.py
@@ -79,7 +79,7 @@ def _export_parser_args(parser_fn, type_as_str: bool = False, **kwargs):
             elif isinstance(a, KVAppendAction):
                 ddd['type'] = dict
             elif isinstance(a, CastToIntAction):
-                ddd['type'] = Union[int, List[int]]
+                ddd['type'] = int
             else:
                 ddd['type'] = a.type
             if ddd['choices']:
diff --git a/tests/helper.py b/tests/helper.py
index 6ed60415fba2a..615a9647a452d 100644
--- a/tests/helper.py
+++ b/tests/helper.py
@@ -1,6 +1,6 @@
 from docarray import DocumentArray
-
 from jina import Executor, requests
+from jina.parsers import set_pod_parser
 
 
 class ProcessExecutor(Executor):
@@ -23,3 +23,14 @@ def _validate_custom_gateway_process(port, text, expected):
 
     resp = requests.get(f'http://127.0.0.1:{port}/stream?text={text}').json()
     assert resp == expected
+
+# set_pod_parser returns a parser for worker runtime, which expects list of ports (because external executors
+# can provide multiple ports and hosts). However this parser is not compatible with ContainerPod, Pod and worker runtime.
+# Should we add a seperate parser for Pod?
+def _generate_args(cli_split: list=[]):
+    args = set_pod_parser().parse_args(cli_split)
+    args.host = args.host[0]
+    args.port = args.port[0]
+    args.port_monitoring = args.port_monitoring[0]
+    
+    return args
diff --git a/tests/integration/deployments/test_deployment.py b/tests/integration/deployments/test_deployment.py
index 4e2ae5b768209..c7a480037a8e8 100644
--- a/tests/integration/deployments/test_deployment.py
+++ b/tests/integration/deployments/test_deployment.py
@@ -323,7 +323,7 @@ def _create_regular_deployment(
     replicas=None,
 ):
     args = set_deployment_parser().parse_args([])
-    args.port = port
+    args.port = [port]
     args.name = name
     if shards:
         args.shards = shards
diff --git a/tests/integration/distributed-replicas/test_distributed_replicas.py b/tests/integration/distributed-replicas/test_distributed_replicas.py
index 59e56d4df383d..0bcc5a3c778ed 100644
--- a/tests/integration/distributed-replicas/test_distributed_replicas.py
+++ b/tests/integration/distributed-replicas/test_distributed_replicas.py
@@ -66,8 +66,9 @@ def _external_deployment_args_docker(num_shards, port=None):
     return set_deployment_parser().parse_args(args)
 
 
-@pytest.mark.parametrize('hosts', ['localhost,localhost', 'localhost'])
-def test_distributed_replicas(input_docs, hosts):
+@pytest.mark.parametrize('hosts', ['localhost,localhost', ['localhost','localhost'], 'localhost'])
+@pytest.mark.parametrize('ports', [lambda port1,port2: f'{port1},{port2}', lambda port1,port2: [port1,port2]])
+def test_distributed_replicas(input_docs, hosts, ports):
     port1, port2 = random_port(), random_port()
     args1, args2 = _external_deployment_args(
         num_shards=1, port=port1
@@ -77,7 +78,7 @@ def test_distributed_replicas(input_docs, hosts):
     with depl1, depl2:
         flow = Flow().add(
             host=hosts,
-            port=f'{port1},{port2}',
+            port=ports(port1, port2),
             external=True,
         )
         with flow:
@@ -87,18 +88,19 @@ def test_distributed_replicas(input_docs, hosts):
         assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
 
 
-def test_distributed_replicas_host_parsing(input_docs):
+@pytest.mark.parametrize('hosts', [lambda port1,port2: f'localhost:{port1},localhost:{port2}', lambda port1,port2: [f'localhost:{port1}',f'localhost:{port2}']])
+@pytest.mark.parametrize('ports', [lambda port1,port2: f'{port1},{port2}', lambda port1,port2: [port1,port2], lambda port1,port2: None])
+def test_distributed_replicas_host_parsing(input_docs, hosts, ports):
     port1, port2 = random_port(), random_port()
     args1, args2 = _external_deployment_args(
         num_shards=1, port=port1
     ), _external_deployment_args(num_shards=1, port=port2)
     depl1 = Deployment(args1)
     depl2 = Deployment(args2)
-    hosts = f'localhost:{port1},localhost:{port2}'
     with depl1, depl2:
         flow = Flow().add(
-            host=hosts,
-            port=f'{port1},{port2}',
+            host=hosts(port1, port2),
+            port=ports(port1, port2),
             external=True,
         )
         with flow:
@@ -108,8 +110,9 @@ def test_distributed_replicas_host_parsing(input_docs):
         assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
 
 
-@pytest.mark.parametrize('hosts', ['localhost,localhost', 'localhost'])
-def test_distributed_replicas_docker(input_docs, hosts, replica_docker_image_built):
+@pytest.mark.parametrize('hosts', ['localhost,localhost', ['localhost','localhost'], 'localhost'])
+@pytest.mark.parametrize('ports', [lambda port1,port2: f'{port1},{port2}', lambda port1,port2: [port1,port2]])
+def test_distributed_replicas_docker(input_docs, hosts, ports, replica_docker_image_built):
     port1, port2 = random_port(), random_port()
     args1, args2 = _external_deployment_args_docker(
         num_shards=1, port=port1
@@ -119,7 +122,7 @@ def test_distributed_replicas_docker(input_docs, hosts, replica_docker_image_bui
     with depl1, depl2:
         flow = Flow().add(
             host=hosts,
-            port=f'{port1},{port2}',
+            port=ports(port1, port2),
             external=True,
         )
         with flow:
@@ -127,3 +130,14 @@ def test_distributed_replicas_docker(input_docs, hosts, replica_docker_image_bui
 
         depl1_id = resp[0].tags['uuid']
         assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
+
+# TODO add external=True and multiple shards and replicas (raise Error or sth)
+# def test_multiple_shards_and_replicas():
+#     port1, port2 = random_port(), random_port()
+#     with pytest.raises(ValueError):
+#         flow = Flow().add(
+#             host=['localhost', 'localhost'],
+#             port=f'{port1},{port2}',
+#             external=True,
+#             shards=2,
+#         )
\ No newline at end of file
diff --git a/tests/integration/gateway_streamer/test_gateway_streamer.py b/tests/integration/gateway_streamer/test_gateway_streamer.py
index 0f6f8f4a07ceb..6c2369f96a442 100644
--- a/tests/integration/gateway_streamer/test_gateway_streamer.py
+++ b/tests/integration/gateway_streamer/test_gateway_streamer.py
@@ -1,13 +1,13 @@
 import multiprocessing
 
 import pytest
-
 from jina import DocumentArray, Executor, requests
-from jina.parsers import set_pod_parser
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 from jina.serve.streamer import GatewayStreamer
 
+from tests.helper import _generate_args
+
 
 class StreamerTestExecutor(Executor):
     @requests
@@ -18,7 +18,7 @@ def foo(self, docs, parameters, **kwargs):
 
 
 def _create_worker_runtime(port, name=''):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     args.uses = 'StreamerTestExecutor'
diff --git a/tests/integration/issues/github_2371/test_pod.py b/tests/integration/issues/github_2371/test_pod.py
index fb28c15c3d1f3..fdd9a44f090ba 100644
--- a/tests/integration/issues/github_2371/test_pod.py
+++ b/tests/integration/issues/github_2371/test_pod.py
@@ -1,9 +1,9 @@
-from jina.parsers import set_pod_parser
 from jina.orchestrate.pods.factory import PodFactory
+from tests.helper import _generate_args
 
 
 def test_pod_instantiate_start_same_context():
-    arg = set_pod_parser().parse_args([])
+    arg = _generate_args()
     pod_args = [arg, arg]
 
     for args in pod_args:
@@ -13,7 +13,7 @@ def test_pod_instantiate_start_same_context():
 
 
 def test_pod_instantiate_start_different_context():
-    arg = set_pod_parser().parse_args([])
+    arg = _generate_args()
     pod_args = [arg, arg]
     pods = []
     for args in pod_args:
diff --git a/tests/integration/monitoring/test_manual_flow.py b/tests/integration/monitoring/test_manual_flow.py
index 183bb7a2caa3e..27bc039c24f36 100644
--- a/tests/integration/monitoring/test_manual_flow.py
+++ b/tests/integration/monitoring/test_manual_flow.py
@@ -4,16 +4,17 @@
 import pytest
 import requests as req
 from docarray import Document, DocumentArray
-
 from jina import Executor, Flow, requests
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.gateway import GatewayRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 
 def _create_worker_runtime(port, name='', executor=None, port_monitoring=None):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
 
@@ -31,8 +32,7 @@ def _create_worker_runtime(port, name='', executor=None, port_monitoring=None):
 def _create_gateway_runtime(
     graph_description, pod_addresses, port, port_monitoring, protocol, retries=-1
 ):
-    with GatewayRuntime(
-        set_gateway_parser().parse_args(
+    args = set_gateway_parser().parse_args(
             [
                 '--graph-description',
                 graph_description,
@@ -49,6 +49,9 @@ def _create_gateway_runtime(
                 protocol,
             ]
         )
+    args.port_monitoring = port_monitoring
+    with GatewayRuntime(
+        args
     ) as runtime:
         runtime.run_forever()
 
diff --git a/tests/integration/pods/container/head-runtime/runtime.py b/tests/integration/pods/container/head-runtime/runtime.py
index d3e6b64322adf..8b33176d8fe25 100644
--- a/tests/integration/pods/container/head-runtime/runtime.py
+++ b/tests/integration/pods/container/head-runtime/runtime.py
@@ -6,6 +6,8 @@
 
 def run(*args, **kwargs):
     runtime_args = set_pod_parser().parse_args(args)
+    runtime_args.host = runtime_args.host[0]
+    runtime_args.port = runtime_args.port[0]
     with HeadRuntime(runtime_args) as runtime:
         runtime.run_forever()
 
diff --git a/tests/integration/pods/container/test_pod.py b/tests/integration/pods/container/test_pod.py
index 3ddc69b6ae57a..6e8875875f453 100644
--- a/tests/integration/pods/container/test_pod.py
+++ b/tests/integration/pods/container/test_pod.py
@@ -4,16 +4,17 @@
 import time
 
 import pytest
-
 from jina import Client, Document
 from jina.enums import PodRoleType, PollingType
 from jina.helper import random_port
 from jina.orchestrate.pods import Pod
 from jina.orchestrate.pods.container import ContainerPod
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve.runtimes.head import HeadRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 
 
@@ -102,7 +103,7 @@ async def test_pods_trivial_topology(
 
 
 def _create_worker_pod(port):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = 'worker'
     args.uses = 'docker://worker-runtime'
@@ -110,7 +111,7 @@ def _create_worker_pod(port):
 
 
 def _create_head_pod(port, connection_list_dict):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = 'head'
     args.pod_role = PodRoleType.HEAD
diff --git a/tests/integration/pods/test_pod.py b/tests/integration/pods/test_pod.py
index 29e3273cd46a7..ff18da2b6760c 100644
--- a/tests/integration/pods/test_pod.py
+++ b/tests/integration/pods/test_pod.py
@@ -5,18 +5,17 @@
 from collections import defaultdict
 
 import pytest
-
 from jina import Client, Document, Executor, requests
 from jina.enums import PodRoleType, PollingType
 from jina.orchestrate.pods import Pod
-from jina.parsers import set_gateway_parser, set_pod_parser
-from jina.resources.health_check.gateway import (
-    check_health_http,
-    check_health_websocket,
-)
+from jina.parsers import set_gateway_parser
+from jina.resources.health_check.gateway import (check_health_http,
+                                                 check_health_websocket)
 from jina.resources.health_check.pod import check_health_pod
 from jina.serve.networking import GrpcConnectionPool
 
+from tests.helper import _generate_args
+
 
 @pytest.mark.asyncio
 # test gateway, head and worker pod by creating them manually in the most simple configuration
@@ -514,7 +513,7 @@ async def _start_create_pod(pod, port_generator, type='worker', executor=None):
 
 
 def _create_worker_pod(port, name='', executor=None):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     args.no_block_on_start = True
@@ -531,7 +530,7 @@ def _create_head_pod(
     uses_before=None,
     uses_after=None,
 ):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     args.pod_cls = 'HeadRuntime'
diff --git a/tests/integration/runtime_signal_handling/test_runtime_captures_signals.py b/tests/integration/runtime_signal_handling/test_runtime_captures_signals.py
index 06d3f33b2c25b..2f731f1683f8e 100644
--- a/tests/integration/runtime_signal_handling/test_runtime_captures_signals.py
+++ b/tests/integration/runtime_signal_handling/test_runtime_captures_signals.py
@@ -4,13 +4,14 @@
 import time
 
 import pytest
-
 from jina import Document, DocumentArray, Executor, requests
 from jina.clients.request import request_generator
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve.networking import GrpcConnectionPool
 from jina_cli.api import executor_native, gateway
 
+from tests.helper import _generate_args
+
 
 class DummyExecutor(Executor):
     def __init__(self, dir=None, *args, **kwargs):
@@ -42,7 +43,7 @@ def _create_test_data_message():
 def test_executor_runtimes(signal, tmpdir):
     import time
 
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
 
     def run(args):
 
diff --git a/tests/integration/runtimes/test_gateway_dry_run.py b/tests/integration/runtimes/test_gateway_dry_run.py
index 5fbed86b33fd9..5494518dabe7d 100644
--- a/tests/integration/runtimes/test_gateway_dry_run.py
+++ b/tests/integration/runtimes/test_gateway_dry_run.py
@@ -1,16 +1,17 @@
 import multiprocessing
 
 import pytest
-
 from jina import Client
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.gateway import GatewayRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 
 def _create_worker_runtime(port, name='', executor=None):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     if executor:
diff --git a/tests/integration/runtimes/test_jina_info.py b/tests/integration/runtimes/test_jina_info.py
index b4b0d808cf0e3..50d4e248a8a58 100644
--- a/tests/integration/runtimes/test_jina_info.py
+++ b/tests/integration/runtimes/test_jina_info.py
@@ -4,18 +4,18 @@
 import grpc
 import pytest
 import requests
-
 from jina import __jina_env__, __version__
-from jina.parsers import set_pod_parser
 from jina.proto import jina_pb2, jina_pb2_grpc
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 from .test_runtimes import _create_gateway_runtime, _create_head_runtime
 
 
 def _create_worker_runtime(port, name='', executor=None):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     if executor:
diff --git a/tests/integration/runtimes/test_network_failures.py b/tests/integration/runtimes/test_network_failures.py
index 0620fe357714e..bd4f149dc4ca1 100644
--- a/tests/integration/runtimes/test_network_failures.py
+++ b/tests/integration/runtimes/test_network_failures.py
@@ -4,13 +4,14 @@
 
 import pytest
 from docarray import DocumentArray
-
 from jina import Client, Executor, requests
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.gateway import GatewayRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 from .test_runtimes import _create_gateway_runtime, _create_head_runtime
 
 
@@ -26,7 +27,7 @@ def foo(self, docs, *args, **kwargs):
 
 
 def _create_worker_runtime(port, name='', executor=None):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.uses = 'DummyExec'
     args.name = name
diff --git a/tests/integration/runtimes/test_runtimes.py b/tests/integration/runtimes/test_runtimes.py
index 9c12bf1902114..aeeb2700d133d 100644
--- a/tests/integration/runtimes/test_runtimes.py
+++ b/tests/integration/runtimes/test_runtimes.py
@@ -7,11 +7,10 @@
 
 import grpc
 import pytest
-
 from jina import Client, Document, DocumentArray, Executor, requests
 from jina.clients.request import request_generator
 from jina.enums import PollingType
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.proto import jina_pb2_grpc
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
@@ -19,6 +18,8 @@
 from jina.serve.runtimes.head import HeadRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 
 @pytest.mark.asyncio
 # test gateway, head and worker runtime by creating them manually in the most simple configuration
@@ -683,7 +684,7 @@ async def _create_worker(pod, port_generator, type='worker', executor=None):
 
 
 def _create_worker_runtime(port, name='', executor=None):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     if executor:
@@ -701,7 +702,7 @@ def _create_head_runtime(
     uses_after=None,
     retries=-1,
 ):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.port = port
     args.name = name
     args.retries = retries
diff --git a/tests/unit/orchestrate/deployments/config/test_docker_compose_pod_config.py b/tests/unit/orchestrate/deployments/config/test_docker_compose_pod_config.py
index b209495af9a38..4a7bcb54d3186 100644
--- a/tests/unit/orchestrate/deployments/config/test_docker_compose_pod_config.py
+++ b/tests/unit/orchestrate/deployments/config/test_docker_compose_pod_config.py
@@ -62,6 +62,7 @@ def test_parse_args(
         args_list.extend(['--uses-metas', uses_metas])
     args = set_deployment_parser().parse_args(args_list)
     deployment_config = DockerComposeConfig(args)
+    args.host = args.host[0]
 
     assert namespace_equal(
         deployment_config.services_args['head_service'],
@@ -244,6 +245,7 @@ def test_parse_args_custom_executor(shards: int, replicas: int):
         ]
     )
     deployment_config = DockerComposeConfig(args)
+    args.host = args.host[0]
 
     if shards > 1:
         assert (
@@ -344,6 +346,7 @@ def test_parse_args_custom_executor(shards: int, replicas: int):
 def test_worker_services(name: str, shards: str):
     args = set_deployment_parser().parse_args(['--name', name, '--shards', shards])
     deployment_config = DockerComposeConfig(args)
+    args.host = args.host[0]
 
     actual_services = deployment_config.worker_services
 
diff --git a/tests/unit/orchestrate/deployments/config/test_k8s_deployment_config.py b/tests/unit/orchestrate/deployments/config/test_k8s_deployment_config.py
index 31a0b7fc7b4e6..365e808f66825 100644
--- a/tests/unit/orchestrate/deployments/config/test_k8s_deployment_config.py
+++ b/tests/unit/orchestrate/deployments/config/test_k8s_deployment_config.py
@@ -53,6 +53,7 @@ def test_parse_args(
         args_list.extend(['--uses-metas', uses_metas])
     args = set_deployment_parser().parse_args(args_list)
     deployment_config = K8sDeploymentConfig(args, 'default-namespace')
+    args.host = args.host[0]
 
     if shards > 1:
         assert namespace_equal(
@@ -166,6 +167,7 @@ def test_parse_args_custom_executor(shards: int):
         ]
     )
     deployment_config = K8sDeploymentConfig(args, 'default-namespace')
+    args.host = args.host[0]
 
     if shards > 1:
         assert (
@@ -237,6 +239,7 @@ def test_deployments(name: str, shards: str, gpus):
         ['--name', name, '--shards', shards, '--gpus', gpus]
     )
     deployment_config = K8sDeploymentConfig(args, 'ns')
+    args.host = args.host[0]
 
     if name != 'gateway' and int(shards) > int(1):
         head_deployment = deployment_config.head_deployment
diff --git a/tests/unit/orchestrate/deployments/test_deployments.py b/tests/unit/orchestrate/deployments/test_deployments.py
index b218b0fcf5e4a..27f847f92f242 100644
--- a/tests/unit/orchestrate/deployments/test_deployments.py
+++ b/tests/unit/orchestrate/deployments/test_deployments.py
@@ -530,7 +530,7 @@ def test_pod_remote_pod_replicas_host(num_shards, num_replicas):
             __default_host__,
         ]
     )
-    assert args.host == __default_host__
+    assert args.host == [__default_host__]
     with Deployment(args) as pod:
         assert pod.num_pods == num_shards * num_replicas + (1 if num_shards > 1 else 0)
         pod_args = dict(pod.pod_args['pods'])
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow.py b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
index ee7863a801e5c..ee00334d6185d 100644
--- a/tests/unit/orchestrate/flow/flow-construct/test_flow.py
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
@@ -351,10 +351,12 @@ def test_return_results_sync_flow(protocol, on_done):
 @pytest.mark.parametrize(
     'input,expected_host,expected_port',
     [
-        ('0.0.0.0', '0.0.0.0', None),
-        ('0.0.0.0:12345', '0.0.0.0', '12345'),
-        ('123.124.125.0:45678', '123.124.125.0', '45678'),
-        ('api.jina.ai:45678', 'api.jina.ai', '45678'),
+        ('0.0.0.0', ['0.0.0.0'], None),
+        ('0.0.0.0:12345', ['0.0.0.0'], ['12345']),
+        ('123.124.125.0:45678', ['123.124.125.0'], ['45678']),
+        ('api.jina.ai:45678', ['api.jina.ai'], ['45678']),
+        (['api.jina.ai','123.124.125.0'], ['api.jina.ai','123.124.125.0'], None),
+        (['api.jina.ai:12345','123.124.125.0:45678'], ['api.jina.ai','123.124.125.0'], ['12345','45678']),
     ],
 )
 def test_flow_host_expose_shortcut(input, expected_host, expected_port):
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py b/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py
index 715dafd940fae..2cbcbd4e37214 100644
--- a/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py
@@ -69,6 +69,7 @@ def test_load_dump_load(tmpdir):
     'yaml_file', ['yaml/flow-gateway.yml', 'yaml/flow-gateway-api.yml']
 )
 def test_load_modify_dump_load(tmpdir, yaml_file):
+    # TODO: later fix loading ports from flow yaml
     f: Flow = Flow.load_config(yaml_file)
     # assert vars inside `with`
     assert f._kwargs['name'] == 'abc'
@@ -77,7 +78,7 @@ def test_load_modify_dump_load(tmpdir, yaml_file):
     # assert executor args
     assert f._deployment_nodes['custom1'].args.uses == 'jinahub://CustomExecutor1'
     assert f._deployment_nodes['custom2'].args.uses == 'CustomExecutor2'
-    assert int(f._deployment_nodes['custom2'].args.port) == 23456
+    assert f._deployment_nodes['custom2'].args.port == [23456]
 
     # change args inside the gateway configuration
     f.port = 12346
@@ -97,7 +98,7 @@ def test_load_modify_dump_load(tmpdir, yaml_file):
     # assert args modified in code
     assert f1.port == 12346
     assert f1.protocol == GatewayProtocolType.WEBSOCKET
-    assert int(f1._deployment_nodes['custom2'].args.port) == 23457
+    assert f1._deployment_nodes['custom2'].args.port == [23457]
 
 
 def test_dump_load_build(monkeypatch):
@@ -125,15 +126,15 @@ def test_dump_load_build(monkeypatch):
     # these were passed by the user
     assert f.port == f1.port
     assert f.protocol == f1.protocol
-    assert int(f['executor1'].args.port) == int(f1['executor1'].args.port)
+    assert f['executor1'].args.port == f1['executor1'].args.port
     assert f['executor2'].args.host == f1['executor2'].args.host
     # this was set during `load_config`
-    assert int(f['executor2'].args.port) == int(f1['executor2'].args.port)
+    assert f['executor2'].args.port == f1['executor2'].args.port
 
     monkeypatch.setenv('JINA_FULL_CLI', 'true')
     f2: Flow = Flow.load_config(JAML.dump(f)).build()
     # these were passed by the user
-    assert int(f.port) == int(f2.port)
+    assert f.port == f2.port
     # validate gateway args (set during build)
     assert f['gateway'].args.port == f2['gateway'].args.port
 
@@ -141,14 +142,14 @@ def test_dump_load_build(monkeypatch):
 def test_load_flow_with_port():
     f = Flow.load_config('yaml/test-flow-port.yml')
     with f:
-        assert int(f.port) == 12345
+        assert f.port == 12345
 
 
 def test_load_flow_from_cli():
     a = set_flow_parser().parse_args(['--uses', 'yaml/test-flow-port.yml'])
     f = Flow.load_config(a.uses)
     with f:
-        assert int(f.port) == 12345
+        assert f.port == 12345
 
 
 def test_load_flow_from_yaml():
diff --git a/tests/unit/orchestrate/pods/container/test_container_pod.py b/tests/unit/orchestrate/pods/container/test_container_pod.py
index 01656320e8939..8dd75acb86e24 100644
--- a/tests/unit/orchestrate/pods/container/test_container_pod.py
+++ b/tests/unit/orchestrate/pods/container/test_container_pod.py
@@ -2,13 +2,14 @@
 import time
 
 import pytest
-
 from jina import Flow, __cache_path__
 from jina.excepts import RuntimeFailToStart
 from jina.helper import random_port
 from jina.orchestrate.pods.container import ContainerPod
-from jina.parsers import set_gateway_parser, set_pod_parser
-from tests.helper import _validate_dummy_custom_gateway_response
+from jina.parsers import set_gateway_parser
+
+from tests.helper import (_generate_args,
+                          _validate_dummy_custom_gateway_response)
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 
@@ -42,8 +43,7 @@ def dummy_exec_docker_image_built():
 def test_container_pod_pass_envs(env_checker_docker_image_built):
     import docker
 
-    with ContainerPod(
-        set_pod_parser().parse_args(
+    args = _generate_args(
             [
                 '--uses',
                 'docker://env-checker',
@@ -53,6 +53,8 @@ def test_container_pod_pass_envs(env_checker_docker_image_built):
                 'key2=value2',
             ]
         )
+    with ContainerPod(
+        args
     ) as pod:
         container = pod._container
         status = container.status
@@ -98,7 +100,7 @@ def test_container_pod_volume_setting(
 
     default_workspace = __cache_path__
 
-    with ContainerPod(set_pod_parser().parse_args(pod_args)) as pod:
+    with ContainerPod(_generate_args(pod_args)) as pod:
         container = pod._container
         source = container.attrs['Mounts'][0]['Source']
         destination = container.attrs['Mounts'][0]['Destination']
@@ -135,7 +137,7 @@ def fail_start_docker_image_built():
 def test_failing_executor(fail_start_docker_image_built):
     import docker
 
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         [
             '--uses',
             'docker://fail-start',
@@ -222,7 +224,7 @@ def images(self):
             return {}
 
     monkeypatch.setattr(docker, 'from_env', MockClient)
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         [
             '--uses',
             'docker://jinahub/pod',
@@ -290,4 +292,4 @@ def test_container_pod_with_flow_custom_gateway(
     with flow:
         _validate_dummy_custom_gateway_response(
             flow.port, {'arg1': 'hello', 'arg2': 'world', 'arg3': 'default-arg3'}
-        )
+        )
\ No newline at end of file
diff --git a/tests/unit/orchestrate/pods/test_pod.py b/tests/unit/orchestrate/pods/test_pod.py
index 05591af4be4f1..2c273576261ae 100644
--- a/tests/unit/orchestrate/pods/test_pod.py
+++ b/tests/unit/orchestrate/pods/test_pod.py
@@ -2,13 +2,14 @@
 import time
 
 import pytest
-
 from jina.excepts import RuntimeFailToStart
 from jina.orchestrate.pods import Pod
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve import runtimes
 from jina.serve.executors import BaseExecutor
 
+from tests.helper import _generate_args
+
 
 @pytest.fixture()
 def fake_env():
@@ -29,7 +30,7 @@ def __init__(self, *args, **kwargs):
 
 def test_pod_runtime_env_setting(fake_env):
     with Pod(
-        set_pod_parser().parse_args(
+        _generate_args(
             [
                 '--uses',
                 'EnvChecker1',
@@ -100,7 +101,7 @@ def test_gateway_runtimes(protocol, expected):
     ['WorkerRuntime', 'HeadRuntime'],
 )
 def test_non_gateway_runtimes(runtime_cls):
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         [
             '--runtime-cls',
             runtime_cls,
@@ -118,7 +119,7 @@ def __init__(self, *args, **kwargs):
 
 
 def test_failing_executor():
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         [
             '--uses',
             'RaisingExecutor',
@@ -156,7 +157,7 @@ def test_failing_gateway_runtimes(protocol, expected):
 
 
 def test_failing_head():
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         [
             '--runtime-cls',
             'HeadRuntime',
@@ -189,7 +190,7 @@ def run_forever(self):
         'get_runtime',
         lambda *args, **kwargs: SlowFakeRuntime,
     )
-    pod = Pod(set_pod_parser().parse_args(['--noblock-on-start']))
+    pod = Pod(_generate_args(['--noblock-on-start']))
     pod.start()
     pod.close()
 
@@ -214,6 +215,6 @@ def run_forever(self):
         'get_runtime',
         lambda *args, **kwargs: SlowFakeRuntime,
     )
-    pod = Pod(set_pod_parser().parse_args(['--noblock-on-start']))
+    pod = Pod(_generate_args(['--noblock-on-start']))
     pod.start()
     pod.close()
diff --git a/tests/unit/parsers/peapods/runtimes/test_port_protocol_parser.py b/tests/unit/parsers/peapods/runtimes/test_port_protocol_parser.py
index 5323f255c2e4f..f07c2694ee8ed 100644
--- a/tests/unit/parsers/peapods/runtimes/test_port_protocol_parser.py
+++ b/tests/unit/parsers/peapods/runtimes/test_port_protocol_parser.py
@@ -36,6 +36,7 @@ def test_multiple_port_protocol_gateway_kwargs(
     [
         (['12345'], [12345]),
         (['12345', '12344'], [12345, 12344]),
+        (['12345, 12344'], [12345, 12344]),
     ],
 )
 @pytest.mark.parametrize(
@@ -56,9 +57,43 @@ def test_multiple_port_protocol_gateway_args_list(
     assert args.protocol == expected_protocol
 
 
-def test_pod_port_cast():
-    args = set_pod_parser().parse_args(['--port', '12345'])
-    assert args.port == 12345
+@pytest.mark.parametrize(
+    'port,expected_port',
+    [
+        (['12345'], [12345]),
+        (['12345', '12344'], [12345, 12344]),
+        (['12345, 12344'], [12345, 12344]),
+    ],
+)
+def test_pod_port_cast(port, expected_port):
+    args = set_pod_parser().parse_args(['--port'] + port)
+    assert args.port == expected_port
+
+
+def test_pod_port_default():
+    args = set_pod_parser().parse_args([])
+    assert isinstance(args.port, list)
+    assert len(args.port) == 1
+
+
+@pytest.mark.parametrize(
+    'host,expected_host',
+    [
+        (['localhost'], ['localhost']),
+        (['0.0.0.0,localhost'], ['0.0.0.0', 'localhost']),
+        (['0.0.0.0,localhost', '127.0.0.1'], ['0.0.0.0', 'localhost', '127.0.0.1']),
+    ],
+)
+def test_pod_host_cast(host, expected_host):
+    args = set_pod_parser().parse_args(['--host'] + host)
+    assert args.host == expected_host
+
+
+def test_pod_host_default():
+    from jina import __default_host__
+
+    args = set_pod_parser().parse_args([])
+    assert args.host == [__default_host__]
 
 
 def test_default_port_protocol_gateway():
diff --git a/tests/unit/serve/executors/test_executor.py b/tests/unit/serve/executors/test_executor.py
index dbfe7c8f096a5..2f8952551e67a 100644
--- a/tests/unit/serve/executors/test_executor.py
+++ b/tests/unit/serve/executors/test_executor.py
@@ -8,17 +8,17 @@
 import pytest
 import yaml
 from docarray import Document, DocumentArray
-
 from jina import Client, Executor, Flow, __cache_path__, requests
 from jina.clients.request import request_generator
 from jina.excepts import RuntimeFailToStart
 from jina.helper import random_port
-from jina.parsers import set_pod_parser
 from jina.serve.executors.metas import get_default_metas
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 
+from tests.helper import _generate_args
+
 
 class WorkspaceExec(Executor):
     @requests
@@ -505,7 +505,7 @@ def foo(self, docs: DocumentArray, **kwargs):
                 doc.text = 'BlockingExecutor'
             return docs
 
-    args = set_pod_parser().parse_args(['--uses', 'BlockingExecutor'])
+    args = _generate_args(['--uses', 'BlockingExecutor'])
 
     cancel_event = multiprocessing.Event()
 
diff --git a/tests/unit/serve/gateway/test_gateway.py b/tests/unit/serve/gateway/test_gateway.py
index 5b58010d41679..c7f4041b93d1d 100644
--- a/tests/unit/serve/gateway/test_gateway.py
+++ b/tests/unit/serve/gateway/test_gateway.py
@@ -4,15 +4,13 @@
 import time
 
 import pytest
-
 from jina.helper import random_port
-from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.parsers import set_gateway_parser
 from jina.serve.runtimes.gateway import GatewayRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
-from tests.helper import (
-    _validate_custom_gateway_process,
-    _validate_dummy_custom_gateway_response,
-)
+
+from tests.helper import (_generate_args, _validate_custom_gateway_process,
+                          _validate_dummy_custom_gateway_response)
 from tests.unit.yaml.dummy_gateway import DummyGateway
 from tests.unit.yaml.dummy_gateway_get_streamer import DummyGatewayGetStreamer
 
@@ -62,7 +60,7 @@ def _start_gateway_runtime(uses, uses_with, worker_port):
 
 
 def _create_worker_runtime(port, uses):
-    args = set_pod_parser().parse_args(['--uses', uses, '--port', str(port)])
+    args = _generate_args(['--uses', uses, '--port', str(port)])
 
     with WorkerRuntime(args) as runtime:
         runtime.run_forever()
diff --git a/tests/unit/serve/instrumentation/test_networking_histograms.py b/tests/unit/serve/instrumentation/test_networking_histograms.py
index f0573d0fd8f9e..e8fac976b10d7 100644
--- a/tests/unit/serve/instrumentation/test_networking_histograms.py
+++ b/tests/unit/serve/instrumentation/test_networking_histograms.py
@@ -59,7 +59,7 @@ def test_recording_methods(metrics_setup: Tuple[InMemoryMetricReader, Meter]):
         metric_reader.get_metrics_data().resource_metrics[0].scope_metrics[0].metrics
     )
     data_points_sums: Dict[str, HistogramDataPoint] = {
-        hist.name: next(hist.data.data_points).sum for hist in histogram_metrics
+        hist.name: next(iter(hist.data.data_points)).sum for hist in histogram_metrics
     }
     assert data_points_sums == {
         'request_time': 10,
diff --git a/tests/unit/serve/runtimes/head/test_head_runtime.py b/tests/unit/serve/runtimes/head/test_head_runtime.py
index 5a836aa7412f5..c8ff4de7b6a3d 100644
--- a/tests/unit/serve/runtimes/head/test_head_runtime.py
+++ b/tests/unit/serve/runtimes/head/test_head_runtime.py
@@ -7,11 +7,9 @@
 
 import grpc
 import pytest
-
 from jina import Document, DocumentArray
 from jina.clients.request import request_generator
 from jina.enums import PollingType
-from jina.parsers import set_pod_parser
 from jina.proto import jina_pb2_grpc
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
@@ -19,9 +17,11 @@
 from jina.types.request import Request
 from jina.types.request.data import DataRequest
 
+from tests.helper import _generate_args
+
 
 def test_regular_data_case():
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.polling = PollingType.ANY
     connection_list_dict = {0: [f'fake_ip:8080']}
     args.connection_list = json.dumps(connection_list_dict)
@@ -45,9 +45,9 @@ def test_regular_data_case():
 @pytest.mark.parametrize('disable_reduce', [False, True])
 def test_message_merging(disable_reduce):
     if not disable_reduce:
-        args = set_pod_parser().parse_args([])
+        args = _generate_args()
     else:
-        args = set_pod_parser().parse_args(['--no-reduce'])
+        args = _generate_args(['--no-reduce'])
     args.polling = PollingType.ALL
     connection_list_dict = {0: [f'ip1:8080'], 1: [f'ip2:8080'], 2: [f'ip3:8080']}
     args.connection_list = json.dumps(connection_list_dict)
@@ -67,7 +67,7 @@ def test_message_merging(disable_reduce):
 
 
 def test_uses_before_uses_after():
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.polling = PollingType.ALL
     args.uses_before_address = 'fake_address'
     args.uses_after_address = 'fake_address'
@@ -104,7 +104,7 @@ def decompress(self):
         decompress,
     )
 
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     args.polling = PollingType.ANY
     connection_list_dict = {0: [f'fake_ip:8080']}
     args.connection_list = json.dumps(connection_list_dict)
@@ -129,16 +129,14 @@ def decompress(self):
 
 @pytest.mark.parametrize('polling', ['any', 'all'])
 def test_dynamic_polling(polling):
-    args = set_pod_parser().parse_args(
-        [
+    args = _generate_args([
             '--polling',
             json.dumps(
                 {'/any': PollingType.ANY, '/all': PollingType.ALL, '*': polling}
             ),
             '--shards',
             str(2),
-        ]
-    )
+        ])
 
     connection_list_dict = {0: [f'fake_ip:8080'], 1: [f'fake_ip:8080']}
     args.connection_list = json.dumps(connection_list_dict)
@@ -174,7 +172,7 @@ def test_dynamic_polling(polling):
 
 @pytest.mark.parametrize('polling', ['any', 'all'])
 def test_base_polling(polling):
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         [
             '--polling',
             polling,
@@ -215,7 +213,7 @@ def test_base_polling(polling):
 
 @pytest.mark.asyncio
 async def test_head_runtime_reflection():
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
     cancel_event, handle_queue, runtime_thread = _create_runtime(args)
 
     assert AsyncNewLoopRuntime.wait_for_ready_or_shutdown(
@@ -239,7 +237,7 @@ async def test_head_runtime_reflection():
 
 
 def test_timeout_behaviour():
-    args = set_pod_parser().parse_args(['--timeout-send', '100'])
+    args = _generate_args(['--timeout-send', '100'])
     args.polling = PollingType.ANY
     connection_list_dict = {0: [f'fake_ip:8080']}
     args.connection_list = json.dumps(connection_list_dict)
@@ -339,4 +337,4 @@ def _queue_length_copy(queue: 'multiprocessing.Manager().Queue'):
     while not c_queue.empty():
         c_queue.get()
         length += 1
-    return length
+    return length
\ No newline at end of file
diff --git a/tests/unit/serve/runtimes/worker/test_worker_runtime.py b/tests/unit/serve/runtimes/worker/test_worker_runtime.py
index e520e53462ffc..15e19ea1e7faa 100644
--- a/tests/unit/serve/runtimes/worker/test_worker_runtime.py
+++ b/tests/unit/serve/runtimes/worker/test_worker_runtime.py
@@ -8,22 +8,22 @@
 import grpc
 import pytest
 import requests as req
-
 from docarray import Document
 from jina import DocumentArray, Executor, requests
 from jina.clients.request import request_generator
-from jina.parsers import set_pod_parser
 from jina.proto import jina_pb2, jina_pb2_grpc
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
 from jina.serve.runtimes.worker import WorkerRuntime
 from jina.serve.runtimes.worker.request_handling import WorkerRequestHandler
 
+from tests.helper import _generate_args
+
 
 @pytest.mark.slow
 @pytest.mark.timeout(5)
 def test_worker_runtime():
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
 
     cancel_event = multiprocessing.Event()
 
@@ -93,7 +93,7 @@ def foo(self, docs, **kwargs):
 @pytest.mark.asyncio
 @pytest.mark.parametrize('uses', ['AsyncSlowNewDocsExecutor', 'SlowNewDocsExecutor'])
 async def test_worker_runtime_slow_async_exec(uses):
-    args = set_pod_parser().parse_args(['--uses', uses])
+    args = _generate_args(['--uses', uses])
 
     cancel_event = multiprocessing.Event()
 
@@ -146,7 +146,7 @@ async def task_wrapper():
 @pytest.mark.slow
 @pytest.mark.timeout(10)
 def test_error_in_worker_runtime(monkeypatch):
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
 
     cancel_event = multiprocessing.Event()
 
@@ -204,7 +204,7 @@ def foo(self, docs, **kwargs):
 @pytest.mark.asyncio
 @pytest.mark.skip
 async def test_worker_runtime_slow_init_exec():
-    args = set_pod_parser().parse_args(['--uses', 'SlowInitExecutor'])
+    args = _generate_args(['--uses', 'SlowInitExecutor'])
 
     cancel_event = multiprocessing.Event()
 
@@ -257,7 +257,7 @@ def start_runtime(args, cancel_event):
 
 @pytest.mark.asyncio
 async def test_worker_runtime_reflection():
-    args = set_pod_parser().parse_args([])
+    args = _generate_args()
 
     cancel_event = multiprocessing.Event()
 
@@ -320,7 +320,7 @@ def process_2(self, docs):
             ...
 
     port = port_generator()
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         ['--monitoring', '--port-monitoring', str(port), '--uses', 'DummyExecutor']
     )
 
@@ -388,7 +388,7 @@ def process_2(self, docs):
             ...
 
     port = port_generator()
-    args = set_pod_parser().parse_args(
+    args = _generate_args(
         ['--monitoring', '--port-monitoring', str(port), '--uses', 'DummyExecutor']
     )
 
@@ -433,7 +433,7 @@ def start_runtime(args, cancel_event):
 @pytest.mark.slow
 @pytest.mark.timeout(10)
 async def test_error_in_worker_runtime_with_exit_on_exceptions(monkeypatch):
-    args = set_pod_parser().parse_args(['--exit-on-exceptions', 'RuntimeError'])
+    args = _generate_args(['--exit-on-exceptions', 'RuntimeError'])
 
     cancel_event = multiprocessing.Event()
 
@@ -468,4 +468,4 @@ def start_runtime(args, cancel_event):
     cancel_event.set()
     runtime_thread.join()
 
-    assert not AsyncNewLoopRuntime.is_ready(target)
+    assert not AsyncNewLoopRuntime.is_ready(target)
\ No newline at end of file
diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py
index 9f94fe07755ba..6e1d255169743 100644
--- a/tests/unit/test_cli.py
+++ b/tests/unit/test_cli.py
@@ -3,16 +3,17 @@
 import subprocess
 
 import pytest
-
 from jina.checker import NetworkChecker
 from jina.jaml import JAML
 from jina.orchestrate.pods.factory import PodFactory
-from jina.parsers import set_deployment_parser, set_pod_parser
+from jina.parsers import set_deployment_parser
 from jina.parsers.ping import set_ping_parser
 from jina_cli.autocomplete import ac_table
 from jina_cli.export import api_to_dict
 from jina_cli.lookup import _build_lookup_table, lookup_and_print
 
+from tests.helper import _generate_args
+
 
 def test_export_api(tmpdir):
     with open(tmpdir / 'test.yml', 'w', encoding='utf8') as fp:
@@ -96,7 +97,7 @@ def test_parse_env_map():
 
 @pytest.mark.slow
 def test_ping():
-    a1 = set_pod_parser().parse_args([])
+    a1 = _generate_args()
     a2 = set_ping_parser().parse_args(['executor', f'0.0.0.0:{a1.port}'])
 
     a3 = set_ping_parser().parse_args(
diff --git a/tests/unit/test_helper.py b/tests/unit/test_helper.py
index 210c52d9b8318..e51b3d878e239 100644
--- a/tests/unit/test_helper.py
+++ b/tests/unit/test_helper.py
@@ -9,8 +9,6 @@
 from jina.clients.helper import _safe_callback, pprint_routes
 from jina.excepts import BadClientCallback, NotSupportedError
 from jina.helper import (
-    _parse_hosts,
-    _parse_ports,
     cached_property,
     convert_tuple_to_list,
     deprecated_alias,
@@ -19,7 +17,6 @@
     get_ci_vendor,
     is_port_free,
     is_yaml_filepath,
-    make_iterable,
     parse_arg,
     random_port,
     reset_ports,
@@ -364,35 +361,6 @@ async def dummy():
     assert first_fd_count == end_fd_count
 
 
-@pytest.mark.parametrize(
-    'port, output',
-    [('8080', 8080), ('1,2,6', [1, 2, 6])],
-)
-def test_parse_port(port, output):
-    assert _parse_ports(port) == output
-
-
-@pytest.mark.parametrize(
-    'host, output',
-    [('1234', '1234'), ('1,2,6', ['1', '2', '6'])],
-)
-def test_parse_host(host, output):
-    assert _parse_hosts(host) == output
-
-
-@pytest.mark.parametrize(
-    'o, expected',
-    [
-        (1, [1]),
-        ('a', ['a']),
-        (bytes('hello', 'utf-8'), [bytes('hello', 'utf-8')]),
-        ([1, 2], [1, 2]),
-    ],
-)
-def test_make_iterable(o, expected):
-    assert make_iterable(o) == expected
-
-
 @pytest.mark.parametrize(
     'input, expected',
     [


Code review comments:
diff_hunk: @@ -824,40 +817,52 @@ def _set_pod_args(args: Namespace) -> Dict[int, List[Namespace]]:
                     _args.name = f'{replica_id}'
 
                 # the gateway needs to respect the assigned port
-                if args.deployment_role == DeploymentRoleType.GATEWAY or args.external:
-                    _args.port = args.port
-                elif shards == 1 and replicas == 1:
-                    _args.port = args.port
-                    _args.port_monitoring = args.port_monitoring
-
-                elif shards == 1:
-                    _args.port_monitoring = (
-                        helper.random_port()
-                        if replica_id >= len(args.all_port_monitoring)
-                        else args.all_port_monitoring[replica_id]
-                    )
-                    # if there are no shards/replicas, we dont need to distribute ports randomly
-                    # we should rather use the pre assigned one
-                    _args.port = helper.random_port()
-                elif shards > 1:
-                    port_monitoring_index = (
-                        replica_id + replicas * shard_id + 1
-                    )  # the first index is for the head
-                    _args.port_monitoring = (
-                        helper.random_port()
-                        if port_monitoring_index >= len(args.all_port_monitoring)
-                        else args.all_port_monitoring[
-                            port_monitoring_index
-                        ]  # we skip the head port here
-                    )
-                    _args.port = helper.random_port()
+                if self.args.deployment_role == DeploymentRoleType.GATEWAY:
+                    _args.port = self.args.port
+                
+                elif not self.external:
+                    if shards == 1 and replicas == 1:
+                        _args.port = self.args.port[0]
+                        _args.port_monitoring = self.args.port_monitoring
+
+                    elif shards == 1:
+                        _args.port_monitoring = (
+                            helper.random_port()
+                            if replica_id >= len(self.args.all_port_monitoring)
+                            else self.args.all_port_monitoring[replica_id]
+                        )
+                        # if there are no shards/replicas, we dont need to distribute ports randomly
+                        # we should rather use the pre assigned one
+                        _args.port = helper.random_port()
+                    elif shards > 1:
+                        port_monitoring_index = (
+                            replica_id + replicas * shard_id + 1
+                        )  # the first index is for the head
+                        _args.port_monitoring = (
+                            helper.random_port()
+                            if port_monitoring_index >= len(self.args.all_port_monitoring)
+                            else self.args.all_port_monitoring[
+                                port_monitoring_index
+                            ]  # we skip the head port here
+                        )
+                        _args.port = helper.random_port()
+                    else:
+                        _args.port = helper.random_port()
+                        _args.port_monitoring = helper.random_port()
+                
                 else:
-                    _args.port = helper.random_port()
-                    _args.port_monitoring = helper.random_port()
+                    # if shards > 1:
path: jina/orchestrate/deployments/__init__.py
text: @user1:
remove commented code

diff_hunk: @@ -119,11 +122,22 @@ def test_distributed_replicas_docker(input_docs, hosts, replica_docker_image_bui
     with depl1, depl2:
         flow = Flow().add(
             host=hosts,
-            port=f'{port1},{port2}',
+            port=ports(port1, port2),
             external=True,
         )
         with flow:
             resp = flow.index(inputs=input_docs, request_size=2)
 
         depl1_id = resp[0].tags['uuid']
         assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
+
+# TODO add external=True and multiple shards and replicas (raise Error or sth)
+# def test_multiple_shards_and_replicas():
path: tests/integration/distributed-replicas/test_distributed_replicas.py
text: @user1:
if this is a TODO, then put this as a draft and not as ready to go

diff_hunk: @@ -66,8 +66,9 @@ def _external_deployment_args_docker(num_shards, port=None):
     return set_deployment_parser().parse_args(args)
 
 
-@pytest.mark.parametrize('hosts', ['localhost,localhost', 'localhost'])
-def test_distributed_replicas(input_docs, hosts):
+@pytest.mark.parametrize('hosts', ['localhost,localhost', ['localhost','localhost'], 'localhost'])
+@pytest.mark.parametrize('ports', [lambda port1,port2: f'{port1},{port2}', lambda port1,port2: [port1,port2]])
path: tests/integration/distributed-replicas/test_distributed_replicas.py
text: @user1:
there is no need to use lambda here

@author:
is there another way instead of lambda? `port1` and `port2` are generated in the test function

diff_hunk: @@ -87,18 +88,19 @@ def test_distributed_replicas(input_docs, hosts):
         assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
 
 
-def test_distributed_replicas_host_parsing(input_docs):
+@pytest.mark.parametrize('hosts', [lambda port1,port2: f'localhost:{port1},localhost:{port2}', lambda port1,port2: [f'localhost:{port1}',f'localhost:{port2}']])
+@pytest.mark.parametrize('ports', [lambda port1,port2: f'{port1},{port2}', lambda port1,port2: [port1,port2], lambda port1,port2: None])
path: tests/integration/distributed-replicas/test_distributed_replicas.py
text: @user1:
no need for lambdas