Repository: jina-ai/serve
Pull request: feat: deployment yaml
closes: https://github.com/jina-ai/jina/issues/5552


TODOs:
- [x] Separate deployment parser and executor parser
- [x] remove some arguments from executor
- [x] Add jaml parser for Deployment under jaml/parsers/deployment
- [x] Flow.add() should be able to decide whether to call Deployment.load_config(uses=deployment_uses) or Deployment(uses=executor_uses) ==> rely on jtype to infer this, by default, infer as executor yaml
- [x] Add schema for deployment
- [x] Add Deployment.load/save_config
- [ ] Documentation
- [ ] tests:
    - [x] Json schema tests
    - [x] Yaml parser tests
    - [x] Deployment from yaml (embedded executor config + nested executor yaml)
    - [x] Flow with add deployment yaml
    - [x] Flow from yaml with deployment yaml

PS: can get inspired from #5342 and #5189


Problem statement:
Create Deployment YAML separate from Executor YAML
create deployment yaml separate from Executor yaml

* add Deployment.load_config

A deployment can use an Executor YAML like so:
 Deployment(uses='executor.yaml', replicas=...)

But a deployment YAML is used like so:
Deployment.load_config(uses='deployment.yaml')

Possible usages of executors inside Flow YAML:
1. executor config is embedded in config
```yaml
 jtype: flow
 executors:
   - jtype: Deployment
     uses:
         jtype: MyExecutor
         py_modules: executor.py
     replicas: 2
```

2. old usage for backward compatibility:
```yaml
 jtype: flow
 executors:
    - uses: executor.yaml
      replicas: 12
```
3. ~~simply `deployment.yaml`~~
```yaml
 jtype: flow
 executors:
    - uses: deployment.yaml
      replicas: 2
      shard: 3
```
instead check if this is possible:
```yaml
 jtype: flow
 executors/deployments:
   - deployment.yaml
   - uses: executor.yaml
      replicas: 12
   - jtype: Deployment
     uses:
         jtype: MyExecutor
         py_modules: executor.py
     replicas: 2
```


Deployment args include protocol, shard, replicas, expose-gateway

Executor args include the rest

Hints from issue discussion:

Commit message: test: test flow with deployment

Patch to review:
diff --git a/jina/jaml/__init__.py b/jina/jaml/__init__.py
index f09bb5e69fa43..77ec56eef14a8 100644
--- a/jina/jaml/__init__.py
+++ b/jina/jaml/__init__.py
@@ -740,6 +740,7 @@ def _delitem(
                     else _extra_search_paths,
                 )
 
+            from jina.orchestrate.deployments import Deployment
             from jina.orchestrate.flow.base import Flow
 
             if issubclass(cls, Flow):
@@ -759,6 +760,15 @@ def _delitem(
                     include_unknown_tags=False,
                     jtype_whitelist=('Flow',),
                 )
+            elif issubclass(cls, Deployment):
+                no_tag_yml['with']['extra_search_paths'] = (
+                    no_tag_yml['with'].get('extra_search_paths') or []
+                ) + (extra_search_paths or [])
+                tag_yml = JAML.unescape(
+                    JAML.dump(no_tag_yml),
+                    include_unknown_tags=False,
+                    jtype_whitelist=('Deployment',),
+                )
             else:
                 # revert yaml's tag and load again, this time with substitution
                 tag_yml = JAML.unescape(JAML.dump(no_tag_yml))
diff --git a/jina/jaml/parsers/__init__.py b/jina/jaml/parsers/__init__.py
index 8eee1a74d2900..99dd8caba4fbe 100644
--- a/jina/jaml/parsers/__init__.py
+++ b/jina/jaml/parsers/__init__.py
@@ -1,6 +1,7 @@
 import warnings
 from typing import List, Optional, Type
 
+from jina import Deployment
 from jina.excepts import BadYAMLVersion
 from jina.jaml import JAMLCompatible
 from jina.jaml.parsers.base import VersionedYAMLParser
@@ -22,6 +23,8 @@ def _get_all_parser(cls: Type['JAMLCompatible']):
         return _get_exec_parser()
     elif issubclass(cls, BaseGateway):
         return _get_gateway_parser()
+    elif issubclass(cls, Deployment):
+        return _get_deployment_parser()
     else:
         raise NotImplementedError(f'No parser exists for cls {cls.__name__}')
 
@@ -38,6 +41,12 @@ def _get_exec_parser():
     return [ExecutorLegacyParser], ExecutorLegacyParser
 
 
+def _get_deployment_parser():
+    from jina.jaml.parsers.deployment.legacy import DeploymentLegacyParser
+
+    return [DeploymentLegacyParser], DeploymentLegacyParser
+
+
 def _get_gateway_parser():
     from jina.jaml.parsers.gateway.legacy import GatewayLegacyParser
 
diff --git a/jina/jaml/parsers/deployment/__init__.py b/jina/jaml/parsers/deployment/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/jina/jaml/parsers/deployment/legacy.py b/jina/jaml/parsers/deployment/legacy.py
new file mode 100644
index 0000000000000..3892868e4f21c
--- /dev/null
+++ b/jina/jaml/parsers/deployment/legacy.py
@@ -0,0 +1,57 @@
+from typing import TYPE_CHECKING, Any, Dict, Optional, Type
+
+from jina.helper import ArgNamespace
+from jina.jaml.parsers.base import BaseLegacyParser
+from jina.jaml.parsers.flow.v1 import _get_taboo
+from jina.parsers import set_deployment_parser
+
+if TYPE_CHECKING:
+    from jina.orchestrate.deployments import Deployment
+
+
+class DeploymentLegacyParser(BaseLegacyParser):
+    """Legacy parser for gateway."""
+
+    def parse(
+        self,
+        cls: Type['Deployment'],
+        data: Dict,
+        runtime_args: Optional[Dict[str, Any]] = None,
+    ) -> 'Deployment':
+        """
+        :param cls: target class type to parse into, must be a :class:`JAMLCompatible` type
+        :param data: deployment yaml file loaded as python dict
+        :param runtime_args: Optional runtime_args to be directly passed without being parsed into a yaml config
+        :return: the Deployment YAML parser given the syntax version number
+        """
+        cls._init_from_yaml = True
+        # tmp_p = {kk: expand_env_var(vv) for kk, vv in data.get('with', {}).items()}
+
+        obj = cls(
+            **data.get('with', {}),
+            runtime_args=runtime_args,
+        )
+        cls._init_from_yaml = False
+
+        obj.is_updated = False
+        return obj
+
+    def dump(self, data: 'Deployment') -> Dict:
+        """
+        :param data: versioned deployment object
+        :return: the dictionary given a versioned deployment object
+        """
+        r = {}
+        r['with'] = {}
+        parser = set_deployment_parser()
+        non_default_kw = ArgNamespace.get_non_defaults_args(data.args, parser)
+        for t in _get_taboo(parser):
+            if t in non_default_kw:
+                non_default_kw.pop(t)
+
+        if non_default_kw:
+            r['with'].update(non_default_kw)
+        if data._gateway_kwargs:
+            r['with'].update(data._gateway_kwargs)
+
+        return r
diff --git a/jina/jaml/parsers/flow/v1.py b/jina/jaml/parsers/flow/v1.py
index a6cdc923d0693..98512eff941e2 100644
--- a/jina/jaml/parsers/flow/v1.py
+++ b/jina/jaml/parsers/flow/v1.py
@@ -2,6 +2,7 @@
 import os
 from typing import Any, Dict, Optional
 
+from jina import Deployment
 from jina.helper import GATEWAY_NAME, ArgNamespace, expand_env_var
 from jina.jaml.parsers.base import VersionedYAMLParser
 from jina.orchestrate.flow.base import Flow
@@ -69,15 +70,25 @@ def parse(
 
         pp = data.get('executors', data.get('deployments', []))
         for deployments in pp:
-            p_deployment_attr = {
-                kk: expand_env_var(vv) for kk, vv in deployments.items()
-            }
-            # in v1 YAML, flow is an optional argument
-            if p_deployment_attr.get('name', None) != GATEWAY_NAME:
-                # ignore gateway when reading, it will be added during build()
-                method = p_deployment_attr.get('method', 'add')
-                # support methods: add, needs, inspect
-                getattr(obj, method)(**p_deployment_attr, copy_flow=False)
+            if isinstance(deployments, str):
+                dep = Deployment.load_config(deployments)
+                getattr(obj, 'add')(dep)
+            elif (
+                isinstance(deployments, dict)
+                and deployments.get('jtype') == 'Deployment'
+            ):
+                dep = Deployment.load_config(deployments)
+                getattr(obj, 'add')(dep)
+            else:
+                p_deployment_attr = {
+                    kk: expand_env_var(vv) for kk, vv in deployments.items()
+                }
+                # in v1 YAML, flow is an optional argument
+                if p_deployment_attr.get('name', None) != GATEWAY_NAME:
+                    # ignore gateway when reading, it will be added during build()
+                    method = p_deployment_attr.get('method', 'add')
+                    # support methods: add, needs, inspect
+                    getattr(obj, method)(**p_deployment_attr, copy_flow=False)
         gateway = data.get(GATEWAY_NAME, {})
         if gateway:
             gateway_attr = {kk: expand_env_var(vv) for kk, vv in gateway.items()}
diff --git a/jina/jaml/parsers/gateway/legacy.py b/jina/jaml/parsers/gateway/legacy.py
index 8177b2a5f5515..2bbc1cc93d97e 100644
--- a/jina/jaml/parsers/gateway/legacy.py
+++ b/jina/jaml/parsers/gateway/legacy.py
@@ -15,9 +15,9 @@ def parse(
     ) -> 'BaseGateway':
         """
         :param cls: target class type to parse into, must be a :class:`JAMLCompatible` type
-        :param data: flow yaml file loaded as python dict
+        :param data: gateway yaml file loaded as python dict
         :param runtime_args: Optional runtime_args to be directly passed without being parsed into a yaml config
-        :return: the Flow YAML parser given the syntax version number
+        :return: the Gateway YAML parser given the syntax version number
         """
         from jina.logging.predefined import default_logger
 
@@ -40,7 +40,7 @@ def parse(
     def dump(self, data: 'BaseGateway') -> Dict:
         """
         :param data: versioned gateway object
-        :return: the dictionary given a versioned flow object
+        :return: the dictionary given a versioned gateway object
         """
         a = {k: v for k, v in data._init_kwargs_dict.items()}
         r = {}
diff --git a/jina/orchestrate/deployments/__init__.py b/jina/orchestrate/deployments/__init__.py
index 88192521474ee..bc962c5642904 100644
--- a/jina/orchestrate/deployments/__init__.py
+++ b/jina/orchestrate/deployments/__init__.py
@@ -8,6 +8,7 @@
 import time
 from argparse import Namespace
 from collections import defaultdict
+from contextlib import ExitStack
 from itertools import cycle
 from typing import TYPE_CHECKING, Dict, List, Optional, Set, Type, Union, overload
 
@@ -27,6 +28,7 @@
 from jina.enums import DeploymentRoleType, GatewayProtocolType, PodRoleType, PollingType
 from jina.helper import ArgNamespace, parse_host_scheme, random_port
 from jina.importer import ImportExtensions
+from jina.jaml import JAMLCompatible
 from jina.logging.logger import JinaLogger
 from jina.orchestrate.deployments.install_requirements_helper import (
     _get_package_path_from_uses,
@@ -47,7 +49,13 @@
     from jina.serve.executors import BaseExecutor
 
 
-class Deployment(PostMixin, BaseOrchestrator):
+class DeploymentType(type(ExitStack), type(JAMLCompatible)):
+    """Type of Deployment, metaclass of :class:`Deployment`"""
+
+    pass
+
+
+class Deployment(JAMLCompatible, PostMixin, BaseOrchestrator, metaclass=DeploymentType):
     """A Deployment is an immutable set of pods, which run in replicas. They share the same input and output socket.
     Internally, the pods can run with the process/thread backend. They can be also run in their own containers
     :param args: arguments parsed from the CLI
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 9eda2659120b2..ac350221a8325 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -1052,6 +1052,7 @@ def add(
     @allowed_levels([FlowBuildLevel.EMPTY])
     def add(
         self,
+        deployment: Union[str, Deployment] = None,
         **kwargs,
     ) -> Union['Flow', 'AsyncFlow']:
         # implementation_stub_inject_start_add
@@ -1208,47 +1209,51 @@ def add(
             op_flow, deployment_name, needs, connect_to_last_deployment=True
         )
 
-        # set the kwargs inherit from `Flow(kwargs1=..., kwargs2=)`
-        for key, value in op_flow._common_kwargs.items():
+        if deployment is None:
 
-            # do not inherit from all the argument from the flow and respect EXECUTOR_ARGS_BLACKLIST
-            if key not in kwargs and key not in EXECUTOR_ARGS_BLACKLIST:
-                kwargs[key] = value
+            # set the kwargs inherit from `Flow(kwargs1=..., kwargs2=)`
+            for key, value in op_flow._common_kwargs.items():
 
-        # update kwargs of this Deployment
-        kwargs.update(
-            dict(
-                name=deployment_name,
-                deployment_role=deployment_role,
+                # do not inherit from all the argument from the flow and respect EXECUTOR_ARGS_BLACKLIST
+                if key not in kwargs and key not in EXECUTOR_ARGS_BLACKLIST:
+                    kwargs[key] = value
+
+            # update kwargs of this Deployment
+            kwargs.update(
+                dict(
+                    name=deployment_name,
+                    deployment_role=deployment_role,
+                )
             )
-        )
-        parser = set_deployment_parser()
-        if deployment_role == DeploymentRoleType.GATEWAY:
-            parser = set_gateway_parser()
+            parser = set_deployment_parser()
+            if deployment_role == DeploymentRoleType.GATEWAY:
+                parser = set_gateway_parser()
 
-        args = ArgNamespace.kwargs2namespace(
-            kwargs, parser, True, fallback_parsers=FALLBACK_PARSERS
-        )
+            args = ArgNamespace.kwargs2namespace(
+                kwargs, parser, True, fallback_parsers=FALLBACK_PARSERS
+            )
 
-        # deployment workspace if not set then derive from flow workspace
-        if args.workspace:
-            args.workspace = os.path.abspath(args.workspace)
-        else:
-            args.workspace = self.workspace
+            # deployment workspace if not set then derive from flow workspace
+            if args.workspace:
+                args.workspace = os.path.abspath(args.workspace)
+            else:
+                args.workspace = self.workspace
 
-        args.noblock_on_start = True
+            args.noblock_on_start = True
 
-        if len(needs) > 1 and args.external and args.no_reduce:
-            raise ValueError(
-                'External Executors with multiple needs have to do auto reduce.'
-            )
+            if len(needs) > 1 and args.external and args.no_reduce:
+                raise ValueError(
+                    'External Executors with multiple needs have to do auto reduce.'
+                )
+            deployment = Deployment(args, needs, include_gateway=False)
+        elif isinstance(deployment, str):
+            deployment = Deployment.load_config(deployment)
 
-        op_flow._deployment_nodes[deployment_name] = Deployment(
-            args, needs, include_gateway=False
-        )
+        op_flow._deployment_nodes[deployment_name] = deployment
 
-        if not args.floating:
-            op_flow._last_deployment = deployment_name
+        # TODO: uncomment later
+        # if not args.floating:
+        #     op_flow._last_deployment = deployment_name
 
         return op_flow
 
diff --git a/tests/integration/deployments/deployment-embedded-executor-config.yml b/tests/integration/deployments/deployment-embedded-executor-config.yml
new file mode 100644
index 0000000000000..0f4db4bb716fa
--- /dev/null
+++ b/tests/integration/deployments/deployment-embedded-executor-config.yml
@@ -0,0 +1,10 @@
+jtype: Deployment
+with:
+  uses:
+    jtype: DummyExecutor
+    py_modules:
+      - executor.py
+    with:
+      arg: world
+  replicas: 2
+  shards: 3
\ No newline at end of file
diff --git a/tests/integration/deployments/deployment-nested-executor-config.yml b/tests/integration/deployments/deployment-nested-executor-config.yml
new file mode 100644
index 0000000000000..b37d6ecd44231
--- /dev/null
+++ b/tests/integration/deployments/deployment-nested-executor-config.yml
@@ -0,0 +1,5 @@
+jtype: Deployment
+with:
+  uses: executor.yml
+  replicas: 3
+  shards: 2
\ No newline at end of file
diff --git a/tests/integration/deployments/deployment-overridden-executor-config.yml b/tests/integration/deployments/deployment-overridden-executor-config.yml
new file mode 100644
index 0000000000000..07dedcbde0fec
--- /dev/null
+++ b/tests/integration/deployments/deployment-overridden-executor-config.yml
@@ -0,0 +1,9 @@
+jtype: Deployment
+with:
+  uses: DummyExecutor
+  py_modules:
+      - executor.py
+  uses_with:
+    arg: helloworld
+  replicas: 3
+  shards: 3
diff --git a/tests/integration/deployments/executor.py b/tests/integration/deployments/executor.py
index 855d1143c595e..2fb60f8c78e1f 100644
--- a/tests/integration/deployments/executor.py
+++ b/tests/integration/deployments/executor.py
@@ -4,6 +4,11 @@
 
 
 class DummyExecutor(Executor):
+    def __init__(self, arg='hello', **kwargs):
+        super().__init__(**kwargs)
+        self.arg = arg
+
     @requests
-    def foo(self, **kwargs):
-        pass
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = self.arg
diff --git a/tests/integration/deployments/test_deployment.py b/tests/integration/deployments/test_deployment.py
index ef51a254229c9..82ecd08ca81d4 100644
--- a/tests/integration/deployments/test_deployment.py
+++ b/tests/integration/deployments/test_deployment.py
@@ -387,6 +387,27 @@ def test_deployment_uses(uses):
         pass
 
 
+@pytest.mark.parametrize(
+    'config_file,expected_replicas,expected_shards,expected_text',
+    [
+        ('deployment-nested-executor-config.yml', 3, 2, 'hello'),
+        ('deployment-embedded-executor-config.yml', 2, 3, 'world'),
+        ('deployment-overridden-executor-config.yml', 3, 3, 'helloworld'),
+    ],
+)
+def test_deployment_load_config(
+    config_file, expected_replicas, expected_shards, expected_text
+):
+    depl = Deployment.load_config(config_file)
+
+    with depl:
+        assert depl.args.replicas == expected_replicas
+        assert depl.args.shards == expected_shards
+        docs = depl.post(on='/', inputs=DocumentArray.empty(5))
+        assert len(docs) == 5
+        assert all(doc.text == expected_text for doc in docs)
+
+
 class MyServeExec(Executor):
     @requests
     def foo(self, docs, **kwargs):
diff --git a/tests/unit/jaml/test_type_parse.py b/tests/unit/jaml/test_type_parse.py
index 4de33662c5e4a..9a8dcf839875b 100644
--- a/tests/unit/jaml/test_type_parse.py
+++ b/tests/unit/jaml/test_type_parse.py
@@ -4,7 +4,7 @@
 import pytest
 import yaml
 
-from jina import Flow, requests
+from jina import Deployment, Flow, requests
 from jina.constants import __default_executor__
 from jina.excepts import BadConfigSource
 from jina.jaml import JAML, JAMLCompatible
@@ -205,7 +205,6 @@ def test_exception_invalid_yaml():
 
 def test_jtype(tmpdir):
     flow_path = os.path.join(tmpdir, 'flow.yml')
-    exec_path = os.path.join(tmpdir, 'exec.yml')
 
     f = Flow()
     f.save_config(flow_path)
@@ -213,6 +212,9 @@ def test_jtype(tmpdir):
         conf = yaml.safe_load(file)
         assert 'jtype' in conf
         assert conf['jtype'] == 'Flow'
+    assert type(Flow.load_config(flow_path)) == Flow
+
+    exec_path = os.path.join(tmpdir, 'exec.yml')
 
     e = BaseExecutor()
     e.save_config(exec_path)
@@ -222,7 +224,24 @@ def test_jtype(tmpdir):
         assert conf['jtype'] == 'BaseExecutor'
 
     assert type(BaseExecutor.load_config(exec_path)) == BaseExecutor
-    assert type(Flow.load_config(flow_path)) == Flow
+
+    dep_path = os.path.join(tmpdir, 'dep.yml')
+
+    dep = Deployment(uses='YourExecutor', port=12345, replicas=3, shards=2)
+    dep.save_config(dep_path)
+    with open(dep_path, 'r') as file:
+        conf = yaml.safe_load(file)
+        assert 'jtype' in conf
+        assert conf['jtype'] == 'Deployment'
+        assert conf['with']['shards'] == 2
+        assert conf['with']['replicas'] == 3
+        assert conf['with']['port'] == 12345
+
+    loaded_deployment = Deployment.load_config(dep_path)
+    assert type(loaded_deployment) == Deployment
+    assert loaded_deployment.port == [12345]
+    assert loaded_deployment.args.shards == 2
+    assert loaded_deployment.args.replicas == 3
 
 
 def test_load_dataclass_executor():
diff --git a/tests/unit/orchestrate/deployments/test_load_config.py b/tests/unit/orchestrate/deployments/test_load_config.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow_deployment.py b/tests/unit/orchestrate/flow/flow-construct/test_flow_deployment.py
new file mode 100644
index 0000000000000..b3c184342f21d
--- /dev/null
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow_deployment.py
@@ -0,0 +1,25 @@
+import os
+
+import pytest
+from docarray import DocumentArray
+
+from jina import Flow
+
+cur_dir = os.path.dirname(os.path.abspath(__file__))
+_deployment_yaml_path = os.path.join(cur_dir, '../../../yaml/test-deployment.yml')
+_deployment_yaml_with_exec_config_path = os.path.join(
+    cur_dir, '../../../yaml/test-deployment-exec-config.yml'
+)
+
+
+@pytest.mark.parametrize(
+    'deployment_config',
+    [_deployment_yaml_path, _deployment_yaml_with_exec_config_path],
+)
+def test_flow_deployment(deployment_config):
+
+    flow = Flow().add(deployment=deployment_config)
+    with flow:
+        docs = flow.post(on='/', inputs=DocumentArray.empty(3))
+        assert len(docs) == 3
+        assert all([doc.text == 'indexed' for doc in docs])
diff --git a/tests/unit/test_yamlparser.py b/tests/unit/test_yamlparser.py
index cd4c9a686994e..b75f8f12b5cf5 100644
--- a/tests/unit/test_yamlparser.py
+++ b/tests/unit/test_yamlparser.py
@@ -3,11 +3,11 @@
 import pytest
 import yaml
 
+from jina import Deployment, Gateway
 from jina.constants import __default_executor__, __default_host__
 from jina.helper import expand_dict, expand_env_var
 from jina.jaml import JAML
 from jina.serve.executors import BaseExecutor
-from jina import Gateway
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 
@@ -205,3 +205,17 @@ def test_load_gateway_override_with(yaml_file, gateway_name):
         assert gateway.arg1 == 'arg1'
         assert gateway.arg2 == 'arg2'
         assert gateway.arg3 == 'arg3'
+
+
+@pytest.mark.parametrize(
+    'yaml_file,expected_replicas,expected_shards,expected_uses',
+    [
+        ('test-deployment.yml', 2, 3, 'BaseExecutor'),
+        ('test-deployment-exec-config.yml', 3, 2, 'yaml/dummy_ext_exec_success.yml'),
+    ],
+)
+def test_load_deployment(yaml_file, expected_replicas, expected_shards, expected_uses):
+    with Deployment.load_config(f'yaml/{yaml_file}') as dep:
+        assert dep.args.replicas == expected_replicas
+        assert dep.args.shards == expected_shards
+        assert dep.args.uses == expected_uses
diff --git a/tests/unit/yaml/dummy_exec.py b/tests/unit/yaml/dummy_exec.py
index acc7bb8c2f6dc..897f6a0afa565 100644
--- a/tests/unit/yaml/dummy_exec.py
+++ b/tests/unit/yaml/dummy_exec.py
@@ -1,5 +1,11 @@
+from docarray import DocumentArray
+
+from jina import requests
 from jina.serve.executors import BaseExecutor
 
 
 class DummyExternalIndexer(BaseExecutor):
-    pass
+    @requests
+    def index(self, docs: DocumentArray, **kwargs):
+        for doc in docs:
+            doc.text = 'indexed'
diff --git a/tests/unit/yaml/test-deployment-exec-config.yml b/tests/unit/yaml/test-deployment-exec-config.yml
new file mode 100644
index 0000000000000..c89d1acd307d7
--- /dev/null
+++ b/tests/unit/yaml/test-deployment-exec-config.yml
@@ -0,0 +1,7 @@
+jtype: Deployment
+with:
+  replicas: 3
+  shards: 2
+  uses: yaml/dummy_ext_exec_success.yml
+
+
diff --git a/tests/unit/yaml/test-deployment.yml b/tests/unit/yaml/test-deployment.yml
new file mode 100644
index 0000000000000..c3e946662dc0f
--- /dev/null
+++ b/tests/unit/yaml/test-deployment.yml
@@ -0,0 +1,8 @@
+jtype: Deployment
+with:
+  replicas: 2
+  shards: 3
+  uses: DummyExternalIndexer
+  py_modules:
+    - dummy_exec.py
+


Code review comments:
diff_hunk: @@ -1,6 +1,7 @@
 import warnings
 from typing import List, Optional, Type
 
+from jina import Deployment
path: jina/jaml/parsers/__init__.py
text: @user1:
can we avoid importing directly from `jina`? Import from its own module