Repository: getindata/kedro-kubeflow
Pull request #227 (2023-02-27 14:08:46)

Release 0.7.4
Bump version and CHANGELOG for next release.


Patch:
diff --git a/.bumpversion.cfg b/.bumpversion.cfg
index aca72f2..05fac45 100644
--- a/.bumpversion.cfg
+++ b/.bumpversion.cfg
@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.7.3
+current_version = 0.7.4
 
 [bumpversion:file:pyproject.toml]
 
diff --git a/.copier-answers.yml b/.copier-answers.yml
index cd704af..06362a1 100644
--- a/.copier-answers.yml
+++ b/.copier-answers.yml
@@ -8,7 +8,7 @@ description: Kedro plugin with Kubeflow Pipelines support
 docs_url: https://kedro-kubeflow.readthedocs.io/
 full_name: Kedro Kubeflow Plugin
 github_url: https://github.com/getindata/kedro-kubeflow
-initial_version: 0.7.3
+initial_version: 0.7.4
 keywords:
 - kedro-plugin
 - kubeflow
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 451e296..5f242bd 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,6 +1,6 @@
 repos:
 - repo: https://github.com/pycqa/isort
-  rev: 5.10.1
+  rev: 5.12.0
   hooks:
   - id: isort
     args: ["--profile", "black", "--line-length=79"]
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8353ef4..fcf594b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+## [0.7.4] - 2023-02-27
+
+-   Removed field validation from resources configuration field - now it can take any custom parameters such as "nvidia.com/gpu":1
+-   Added support for kedro namespaces for parameters
+
 ## [0.7.3] - 2022-09-23
 
 -   Fixed plugin config provider so it respects environment provided by the user
@@ -167,7 +172,9 @@
 -   Method to schedule runs for most recent version of given pipeline `kedro kubeflow schedule` 
 -   Shortcut to open UI for pipelines using `kedro kubeflow ui` 
 
-[Unreleased]: https://github.com/getindata/kedro-kubeflow/compare/0.7.3...HEAD
+[Unreleased]: https://github.com/getindata/kedro-kubeflow/compare/0.7.4...HEAD
+
+[0.7.4]: https://github.com/getindata/kedro-kubeflow/compare/0.7.3...0.7.4
 
 [0.7.3]: https://github.com/getindata/kedro-kubeflow/compare/0.7.2...0.7.3
 
diff --git a/kedro_kubeflow/__init__.py b/kedro_kubeflow/__init__.py
index 4910b9e..ed9d4d8 100644
--- a/kedro_kubeflow/__init__.py
+++ b/kedro_kubeflow/__init__.py
@@ -1,1 +1,1 @@
-__version__ = "0.7.3"
+__version__ = "0.7.4"
diff --git a/kedro_kubeflow/config.py b/kedro_kubeflow/config.py
index 534eeee..612f3a3 100644
--- a/kedro_kubeflow/config.py
+++ b/kedro_kubeflow/config.py
@@ -178,9 +178,13 @@ class DefaultConfigDict(defaultdict):
         )
 
 
-class ResourceConfig(BaseModel):
-    cpu: Optional[str]
-    memory: Optional[str]
+class ResourceConfig(dict):
+    def __getitem__(self, key):
+        defaults: dict = super().get("__default__")
+        this: dict = super().get(key, {})
+        updated_defaults = defaults.copy()
+        updated_defaults.update(this)
+        return updated_defaults
 
 
 class TolerationConfig(BaseModel):
@@ -286,9 +290,15 @@ class RunConfig(BaseModel):
 
     @validator("resources", always=True)
     def _validate_resources(cls, value):
-        return RunConfig._create_default_dict_with(
-            value, ResourceConfig(cpu="500m", memory="1024Mi")
+        default = ResourceConfig(
+            {"__default__": {"cpu": "500m", "memory": "1024Mi"}}
         )
+        if isinstance(value, dict):
+            default.update(value)
+        elif value is not None:
+            logger.error(f"Unknown type for resource config {type(value)}")
+            raise TypeError(f"Unknown type for resource config {type(value)}")
+        return default
 
     @validator("retry_policy", always=True)
     def _validate_retry_policy(cls, value):
diff --git a/kedro_kubeflow/generators/one_pod_pipeline_generator.py b/kedro_kubeflow/generators/one_pod_pipeline_generator.py
index 0f0a594..8300f67 100644
--- a/kedro_kubeflow/generators/one_pod_pipeline_generator.py
+++ b/kedro_kubeflow/generators/one_pod_pipeline_generator.py
@@ -12,6 +12,7 @@ from .utils import (
     customize_op,
     is_local_fs,
     maybe_add_params,
+    merge_namespaced_params_to_dict,
 )
 
 
@@ -26,8 +27,10 @@ class OnePodPipelineGenerator(object):
         self.catalog = context.config_loader.get("catalog*")
 
     def generate_pipeline(self, pipeline, image, image_pull_policy):
+        merged_params = merge_namespaced_params_to_dict(self.context.params)
+
         @dsl.pipeline(self.project_name, self.run_config.description)
-        @maybe_add_params(self.context.params)
+        @maybe_add_params(merged_params)
         def convert_kedro_pipeline_to_kfp() -> None:
             dsl.get_pipeline_conf().set_ttl_seconds_after_finished(
                 self.run_config.ttl
@@ -39,13 +42,16 @@ class OnePodPipelineGenerator(object):
                 self.run_config,
                 self.context,
             ):
-                self._build_kfp_op(pipeline, image, image_pull_policy)
+                self._build_kfp_op(
+                    pipeline, merged_params, image, image_pull_policy
+                )
 
         return convert_kedro_pipeline_to_kfp
 
     def _build_kfp_op(
         self,
         pipeline,
+        params,
         image,
         image_pull_policy,
     ) -> dsl.ContainerOp:
@@ -59,9 +65,7 @@ class OnePodPipelineGenerator(object):
                 f"--pipeline {pipeline} "
                 f"--config config.yaml"
             ),
-            arguments=create_arguments_from_parameters(
-                self.context.params.keys()
-            ),
+            arguments=create_arguments_from_parameters(params.keys()),
             container_kwargs={"env": create_container_environment()},
             file_outputs={
                 output: f"/home/kedro/{self.catalog[output]['filepath']}"
diff --git a/kedro_kubeflow/generators/pod_per_node_pipeline_generator.py b/kedro_kubeflow/generators/pod_per_node_pipeline_generator.py
index 6508dfe..bf0eb09 100644
--- a/kedro_kubeflow/generators/pod_per_node_pipeline_generator.py
+++ b/kedro_kubeflow/generators/pod_per_node_pipeline_generator.py
@@ -15,6 +15,7 @@ from .utils import (
     customize_op,
     is_local_fs,
     maybe_add_params,
+    merge_namespaced_params_to_dict,
 )
 
 
@@ -36,11 +37,13 @@ class PodPerNodePipelineGenerator(object):
                 )
 
     def generate_pipeline(self, pipeline, image, image_pull_policy):
+        merged_params = merge_namespaced_params_to_dict(self.context.params)
+
         @dsl.pipeline(
             name=self.project_name,
             description=self.run_config.description,
         )
-        @maybe_add_params(self.context.params)
+        @maybe_add_params(merged_params)
         def convert_kedro_pipeline_to_kfp() -> None:
             """Convert from a Kedro pipeline into a kfp container graph."""
 
@@ -58,7 +61,11 @@ class PodPerNodePipelineGenerator(object):
                 self.context,
             ):
                 kfp_ops = self._build_kfp_ops(
-                    pipeline, node_dependencies, image, image_pull_policy
+                    pipeline,
+                    merged_params,
+                    node_dependencies,
+                    image,
+                    image_pull_policy,
                 )
 
                 self.configure_max_cache_staleness(kfp_ops)
@@ -71,6 +78,7 @@ class PodPerNodePipelineGenerator(object):
     def _build_kfp_ops(
         self,
         pipeline,
+        params,
         node_dependencies: Dict[Node, Set[Node]],
         image,
         image_pull_policy,
@@ -129,9 +137,7 @@ class PodPerNodePipelineGenerator(object):
                         f"--node {node.name} "
                         f"--config config.yaml"
                     ),
-                    arguments=create_arguments_from_parameters(
-                        self.context.params.keys()
-                    ),
+                    arguments=create_arguments_from_parameters(params.keys()),
                     pvolumes=node_volumes,
                     container_kwargs={"env": nodes_env},
                     file_outputs={
diff --git a/kedro_kubeflow/generators/utils.py b/kedro_kubeflow/generators/utils.py
index c3bedf3..cb52c19 100644
--- a/kedro_kubeflow/generators/utils.py
+++ b/kedro_kubeflow/generators/utils.py
@@ -2,7 +2,7 @@ import contextlib
 import itertools
 import json
 import os
-from functools import wraps
+from functools import reduce, wraps
 from inspect import Parameter, signature
 
 import fsspec
@@ -40,6 +40,49 @@ def maybe_add_params(kedro_parameters):
     return decorator
 
 
+def merge_namespaced_params_to_dict(kedro_parameters):
+    def dict_merge(dct, merge_dct):
+        """Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
+        updating only top-level keys, dict_merge recurses down into dicts nested
+        to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
+        ``dct``.
+        :param dct: dict onto which the merge is executed
+        :param merge_dct: dct merged into dct
+        :return: None
+        """
+        for k, v in merge_dct.items():
+            if (
+                k in dct
+                and isinstance(dct[k], dict)
+                and isinstance(merge_dct[k], dict)
+            ):  # noqa
+                dict_merge(dct[k], merge_dct[k])
+            else:
+                dct[k] = merge_dct[k]
+
+        return dct
+
+    normalized_params = []
+    for key, param in kedro_parameters.items():
+        if "." in key:
+            nested_keys = key.split(".")
+            top_key = nested_keys.pop(0)
+            bottom_key = nested_keys.pop()
+            inner_dict = {bottom_key: param}
+            for nested_key in nested_keys[-1:]:
+                inner_dict = {nested_key: inner_dict}
+
+            inner_dict = {top_key: inner_dict}
+            normalized_params.append(inner_dict)
+        else:
+            normalized_params.append({key: param})
+
+    if normalized_params:
+        return reduce(dict_merge, normalized_params)
+    else:
+        return dict()
+
+
 def create_container_environment():
     env_vars = [
         k8s.V1EnvVar(
@@ -131,7 +174,7 @@ def customize_op(op, image_pull_policy, run_config: RunConfig):
             k8s.V1SecurityContext(run_as_user=run_config.volume.owner)
         )
 
-    resources = run_config.resources[op.name].dict(exclude_none=True)
+    resources = run_config.resources[op.name]
     op.container.resources = k8s.V1ResourceRequirements(
         limits=resources,
         requests=resources,
diff --git a/pyproject.toml b/pyproject.toml
index 834ad06..d4839bf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "kedro-kubeflow"
-version = "0.7.3"
+version = "0.7.4"
 description = "Kedro plugin with Kubeflow Pipelines support"
 readme = "README.md"
 authors = ['Mateusz Pytel <mateusz.pytel@getindata.com>', 'Mariusz Strzelecki <mariusz.strzelecki@getindata.com>']
diff --git a/sonar-project.properties b/sonar-project.properties
index 5be3a5b..45f1b08 100644
--- a/sonar-project.properties
+++ b/sonar-project.properties
@@ -6,7 +6,7 @@ sonar.tests=tests/
 sonar.python.coverage.reportPaths=coverage.xml
 sonar.python.version=3.9
 
-sonar.projectVersion=0.7.3
+sonar.projectVersion=0.7.4
 sonar.projectDescription=Kedro plugin with Kubeflow Pipelines support
 sonar.links.homepage=https://kedro-kubeflow.readthedocs.io/
 sonar.links.ci=https://github.com/getindata/kedro-kubeflow/actions


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index bcf8a61..1dd14ab 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -40,7 +40,7 @@ jobs:
 
     - name: Test with tox
       run: |
-        pip install tox-pip-version tox-gh-actions
+        pip install tox-pip-version tox-gh-actions "tox<4.0.0"
         tox -v
 
     - name: Store coverage reports
diff --git a/tests/e2e/catalog.yml b/tests/e2e/catalog.yml
index 19778a1..1ddd0f8 100644
--- a/tests/e2e/catalog.yml
+++ b/tests/e2e/catalog.yml
@@ -20,64 +20,38 @@ model_input_table:
   filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/03_primary/model_input_table.csv
   layer: primary
 
-data_processing.preprocessed_companies:
+preprocessed_companies:
   type: pandas.CSVDataSet
   filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/02_intermediate/preprocessed_companies.csv
   layer: intermediate
 
-data_processing.preprocessed_shuttles:
+preprocessed_shuttles:
   type: pandas.CSVDataSet
   filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/02_intermediate/preprocessed_shuttles.csv
   layer: intermediate
 
-data_science.active_modelling_pipeline.X_train:
+X_train:
   type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/active_modelling_pipeline/05_model_input/X_train.pickle
+  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/05_model_input/X_train.pickle
   layer: model_input
 
-data_science.active_modelling_pipeline.y_train:
+y_train:
   type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/active_modelling_pipeline/05_model_input/y_train.pickle
+  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/05_model_input/y_train.pickle
   layer: model_input
 
-data_science.active_modelling_pipeline.X_test:
+X_test:
   type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/active_modelling_pipeline/05_model_input/X_test.pickle
+  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/05_model_input/X_test.pickle
   layer: model_input
 
-data_science.active_modelling_pipeline.y_test:
+y_test:
   type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/active_modelling_pipeline/05_model_input/y_test.pickle
+  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/05_model_input/y_test.pickle
   layer: model_input
 
-data_science.active_modelling_pipeline.regressor:
+regressor:
   type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/active_modelling_pipeline/06_models/regressor.pickle
-  versioned: true
-  layer: models
-
-data_science.candidate_modelling_pipeline.X_train:
-  type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/candidate_modelling_pipeline/05_model_input/X_train.pickle
-  layer: model_input
-
-data_science.candidate_modelling_pipeline.y_train:
-  type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/candidate_modelling_pipeline/05_model_input/y_train.pickle
-  layer: model_input
-
-data_science.candidate_modelling_pipeline.X_test:
-  type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/candidate_modelling_pipeline/05_model_input/X_test.pickle
-  layer: model_input
-
-data_science.candidate_modelling_pipeline.y_test:
-  type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/candidate_modelling_pipeline/05_model_input/y_test.pickle
-  layer: model_input
-
-data_science.candidate_modelling_pipeline.regressor:
-  type: pickle.PickleDataSet
-  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/candidate_modelling_pipeline/06_models/regressor.pickle
+  filepath: gs://gid-ml-ops-sandbox-plugin-tests/kube_${run_id}/06_models/regressor.pickle
   versioned: true
   layer: models
diff --git a/tests/test_config.py b/tests/test_config.py
index bd33360..d66e2cf 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -28,27 +28,28 @@ run_config:
 class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
     def test_plugin_config(self):
         cfg = PluginConfig(**yaml.safe_load(CONFIG_YAML))
-        assert cfg.host == "https://example.com"
-        assert cfg.run_config.image == "gcr.io/project-image/test"
-        assert cfg.run_config.image_pull_policy == "Always"
-        assert cfg.run_config.experiment_name == "Test Experiment"
-        assert cfg.run_config.run_name == "test run"
-        assert cfg.run_config.scheduled_run_name == "scheduled run"
-        assert cfg.run_config.wait_for_completion
-        assert cfg.run_config.volume.storageclass == "default"
-        assert cfg.run_config.volume.size == "3Gi"
-        assert cfg.run_config.volume.keep is True
-        assert cfg.run_config.volume.access_modes == ["ReadWriteOnce"]
-        assert cfg.run_config.resources["node1"] is not None
-        assert cfg.run_config.description == "My awesome pipeline"
-        assert cfg.run_config.ttl == 300
+        self.assertEqual(cfg.host, "https://example.com")
+        self.assertEqual(cfg.run_config.image, "gcr.io/project-image/test")
+        self.assertEqual(cfg.run_config.image_pull_policy, "Always")
+        self.assertEqual(cfg.run_config.experiment_name, "Test Experiment")
+        self.assertEqual(cfg.run_config.run_name, "test run")
+        self.assertEqual(cfg.run_config.scheduled_run_name, "scheduled run")
+        self.assertTrue(cfg.run_config.wait_for_completion)
+        self.assertEqual(cfg.run_config.volume.storageclass, "default")
+        self.assertEqual(cfg.run_config.volume.size, "3Gi")
+        self.assertTrue(cfg.run_config.volume.keep)
+        self.assertEqual(cfg.run_config.volume.access_modes, ["ReadWriteOnce"])
+        self.assertIsNotNone(cfg.run_config.resources["node1"])
+        self.assertIsNotNone(cfg.run_config.resources["__default__"])
+        self.assertEqual(cfg.run_config.description, "My awesome pipeline")
+        self.assertEqual(cfg.run_config.ttl, 300)
 
     def test_defaults(self):
         cfg = PluginConfig(**self.minimal_config())
-        assert cfg.run_config.image_pull_policy == "IfNotPresent"
+        self.assertEqual(cfg.run_config.image_pull_policy, "IfNotPresent")
         assert cfg.run_config.description is None
         SECONDS_IN_ONE_WEEK = 3600 * 24 * 7
-        assert cfg.run_config.ttl == SECONDS_IN_ONE_WEEK
+        self.assertEqual(cfg.run_config.ttl, SECONDS_IN_ONE_WEEK)
         assert cfg.run_config.volume is None
 
     def test_missing_required_config(self):
@@ -61,8 +62,31 @@ class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
                 {"run_config": {"resources": {"__default__": {"cpu": "100m"}}}}
             )
         )
-        assert cfg.run_config.resources["node2"].cpu == "100m"
-        assert cfg.run_config.resources["node3"].cpu == "100m"
+        self.assertEqual(cfg.run_config.resources["node2"]["cpu"], "100m")
+        self.assertEqual(cfg.run_config.resources["node3"]["cpu"], "100m")
+
+    def test_resources_gpu_label(self):
+        cfg = PluginConfig(
+            **self.minimal_config(
+                {
+                    "run_config": {
+                        "resources": {
+                            "__default__": {
+                                "cpu": "100m",
+                                "nvidia.com/gpu": "1",
+                                "nvidia.com/tpu": "1",
+                            }
+                        }
+                    }
+                }
+            )
+        )
+        self.assertEqual(
+            cfg.run_config.resources["__default__"]["nvidia.com/gpu"], "1"
+        )
+        self.assertEqual(
+            cfg.run_config.resources["node3"]["nvidia.com/tpu"], "1"
+        )
 
     def test_resources_no_default(self):
         cfg = PluginConfig(
@@ -70,10 +94,10 @@ class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
                 {"run_config": {"resources": {"node2": {"cpu": "100m"}}}}
             )
         )
-        assert cfg.run_config.resources["node2"].cpu == "100m"
+        self.assertEqual(cfg.run_config.resources["node2"]["cpu"], "100m")
         self.assertDictEqual(
-            cfg.run_config.resources["node3"].dict(),
-            cfg.run_config.resources["__default__"].dict(),
+            cfg.run_config.resources["node3"],
+            cfg.run_config.resources["__default__"],
         )
 
     def test_resources_default_and_node_specific(self):
@@ -90,14 +114,14 @@ class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
             )
         )
         self.assertDictEqual(
-            cfg.run_config.resources["node2"].dict(),
+            cfg.run_config.resources["node2"],
             {
                 "cpu": "100m",
                 "memory": "64Mi",
             },
         )
         self.assertDictEqual(
-            cfg.run_config.resources["node3"].dict(),
+            cfg.run_config.resources["node3"],
             {
                 "cpu": "200m",
                 "memory": "64Mi",
@@ -149,13 +173,15 @@ class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
             cfg.run_config.tolerations["node2"][0].dict(), toleration_config[0]
         )
 
-        assert (
+        self.assertEqual(
             isinstance(cfg.run_config.tolerations["node2"], list)
-            and len(cfg.run_config.tolerations["node2"]) == 1
+            and len(cfg.run_config.tolerations["node2"]),
+            1,
         )
-        assert (
+        self.assertEqual(
             isinstance(cfg.run_config.tolerations["node3"], list)
-            and len(cfg.run_config.tolerations["node3"]) == 0
+            and len(cfg.run_config.tolerations["node3"]),
+            0,
         )
 
     def test_tolerations_default_and_node_specific(self):
@@ -206,8 +232,8 @@ class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
         cfg = PluginConfig(
             **self.minimal_config({"run_config": {"run_name": "some run"}})
         )
-        assert cfg.run_config.run_name == "some run"
-        assert cfg.run_config.scheduled_run_name == "some run"
+        self.assertEqual(cfg.run_config.run_name, "some run")
+        self.assertEqual(cfg.run_config.scheduled_run_name, "some run")
 
     def test_retry_policy_default_and_node_specific(self):
         cfg = PluginConfig(
@@ -275,4 +301,4 @@ class TestPluginConfig(unittest.TestCase, MinimalConfigMixin):
             },
         )
 
-        assert cfg.run_config.retry_policy["node2"] is None
+        self.assertIsNone(cfg.run_config.retry_policy["node2"])
diff --git a/tests/test_generator_utils.py b/tests/test_generator_utils.py
index 41442d5..c162176 100644
--- a/tests/test_generator_utils.py
+++ b/tests/test_generator_utils.py
@@ -2,7 +2,10 @@
 
 import unittest
 
-from kedro_kubeflow.generators.utils import is_local_fs
+from kedro_kubeflow.generators.utils import (
+    is_local_fs,
+    merge_namespaced_params_to_dict,
+)
 
 
 def gcsfs_is_missing():
@@ -19,3 +22,34 @@ class TestGeneratorUtils(unittest.TestCase):
     def test_is_local(self):
         assert is_local_fs("data/test/file.txt") is True
         assert is_local_fs("gs://test-bucket/file.txt") is False
+
+    def test_namespaced_params_merged_into_dict_properly(self):
+        # given
+        params = {
+            "outer_namespace.inner_namespace1.param1": "outer_namespace.inner_namespace1.param1_v",
+            "outer_namespace.inner_namespace1.param2": "outer_namespace.inner_namespace1.param2_v",
+            "outer_namespace.inner_namespace2.param1": "outer_namespace.inner_namespace2.param1_v",
+            "outer_namespace.inner_namespace2.param2": "outer_namespace.inner_namespace2.param2_v",
+            "outer_namespace.param": "outer_namespace.param",
+            "param1": 42,
+        }
+
+        # when
+        result = merge_namespaced_params_to_dict(params)
+
+        # expect
+        expected = {
+            "outer_namespace": {
+                "inner_namespace1": {
+                    "param1": "outer_namespace.inner_namespace1.param1_v",
+                    "param2": "outer_namespace.inner_namespace1.param2_v",
+                },
+                "inner_namespace2": {
+                    "param1": "outer_namespace.inner_namespace2.param1_v",
+                    "param2": "outer_namespace.inner_namespace2.param2_v",
+                },
+                "param": "outer_namespace.param",
+            },
+            "param1": 42,
+        }
+        assert result == expected
diff --git a/tests/test_one_pod_pipeline_generator.py b/tests/test_one_pod_pipeline_generator.py
index 0f3c2ca..526d76b 100644
--- a/tests/test_one_pod_pipeline_generator.py
+++ b/tests/test_one_pod_pipeline_generator.py
@@ -83,6 +83,92 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 "{{pipelineparam:op=;name=param3}}",
             ]
 
+    def test_should_support_nested_params_and_inject_them_to_the_node(self):
+        # given
+        self.create_generator(
+            params={
+                "param1": {"nested1": {"nested2": 1, "nested3": 2}},
+                "param2": 42,
+                "param3": datetime.date(2022, 2, 24),
+            }
+        )
+
+        # when
+        with patch(
+            "kedro.framework.project.pipelines",
+            new=self.pipelines_under_test,
+        ):
+            with kfp.dsl.Pipeline(None) as dsl_pipeline:
+                pipeline = self.generator_under_test.generate_pipeline(
+                    "pipeline", "unittest-image", "Always"
+                )
+                default_params = signature(pipeline).parameters
+                pipeline()
+
+            # then
+            assert len(default_params) == 3
+            assert default_params["param1"].default == {
+                "nested1": {"nested2": 1, "nested3": 2}
+            }
+            assert default_params["param2"].default == 42
+            assert default_params["param3"].default == "2022-02-24"
+            assert dsl_pipeline.ops["pipeline"].container.args[1:] == [
+                "param1",
+                "{{pipelineparam:op=;name=param1}}",
+                "param2",
+                "{{pipelineparam:op=;name=param2}}",
+                "param3",
+                "{{pipelineparam:op=;name=param3}}",
+            ]
+
+    def test_should_support_namespaced_params_and_inject_them_to_the_node(
+        self,
+    ):
+        # given
+        self.create_generator(
+            params={
+                "outer_namespace.inner_namespace1.param1": "outer_namespace.inner_namespace1.param1_v",
+                "outer_namespace.inner_namespace1.param2": "outer_namespace.inner_namespace1.param2_v",
+                "outer_namespace.inner_namespace2.param1": "outer_namespace.inner_namespace2.param1_v",
+                "outer_namespace.inner_namespace2.param2": "outer_namespace.inner_namespace2.param2_v",
+                "outer_namespace.param": "outer_namespace.param",
+                "param1": 42,
+            }
+        )
+
+        # when
+        with patch(
+            "kedro.framework.project.pipelines",
+            new=self.pipelines_under_test,
+        ):
+            with kfp.dsl.Pipeline(None) as dsl_pipeline:
+                pipeline = self.generator_under_test.generate_pipeline(
+                    "pipeline", "unittest-image", "Always"
+                )
+                default_params = signature(pipeline).parameters
+                pipeline()
+
+            # then
+            assert len(default_params) == 2
+            assert default_params["outer_namespace"].default == {
+                "inner_namespace1": {
+                    "param1": "outer_namespace.inner_namespace1.param1_v",
+                    "param2": "outer_namespace.inner_namespace1.param2_v",
+                },
+                "inner_namespace2": {
+                    "param1": "outer_namespace.inner_namespace2.param1_v",
+                    "param2": "outer_namespace.inner_namespace2.param2_v",
+                },
+                "param": "outer_namespace.param",
+            }
+            assert default_params["param1"].default == 42
+            assert dsl_pipeline.ops["pipeline"].container.args[1:] == [
+                "outer_namespace",
+                "{{pipelineparam:op=;name=outer_namespace}}",
+                "param1",
+                "{{pipelineparam:op=;name=param1}}",
+            ]
+
     def test_should_use_default_resources_spec_if_not_requested(self):
         # given
         self.create_generator(config={})
diff --git a/tests/test_pod_per_node_pipeline_generator.py b/tests/test_pod_per_node_pipeline_generator.py
index 2b56f55..1a0e2cc 100644
--- a/tests/test_pod_per_node_pipeline_generator.py
+++ b/tests/test_pod_per_node_pipeline_generator.py
@@ -36,8 +36,8 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             # then
-            assert (
-                dsl_pipeline.ops["node1"].container.image == "unittest-image"
+            self.assertEqual(
+                dsl_pipeline.ops["node1"].container.image, "unittest-image"
             )
             assert (
                 dsl_pipeline.ops["node1"].container.image_pull_policy
@@ -60,7 +60,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             # then
-            assert len(dsl_pipeline.ops) == 5
+            self.assertEqual(len(dsl_pipeline.ops), 6)
             assert "on-exit" in dsl_pipeline.ops
             assert (
                 dsl_pipeline.ops["on-exit"]
@@ -73,18 +73,20 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
             volume_spec = dsl_pipeline.ops[
                 "data-volume-create"
             ].k8s_resource.spec
-            assert volume_spec.resources.requests["storage"] == "1Gi"
-            assert volume_spec.access_modes == ["ReadWriteOnce"]
+            self.assertEqual(volume_spec.resources.requests["storage"], "1Gi")
+            self.assertEqual(volume_spec.access_modes, ["ReadWriteOnce"])
             assert volume_spec.storage_class_name is None
             volume_init_spec = dsl_pipeline.ops["data-volume-init"].container
-            assert volume_init_spec.image == "unittest-image"
-            assert volume_init_spec.image_pull_policy == "IfNotPresent"
-            assert volume_init_spec.security_context.run_as_user == 0
+            self.assertEqual(volume_init_spec.image, "unittest-image")
+            self.assertEqual(
+                volume_init_spec.image_pull_policy, "IfNotPresent"
+            )
+            self.assertEqual(volume_init_spec.security_context.run_as_user, 0)
             assert volume_init_spec.args[0].startswith("cp --verbose -r")
             for node_name in ["data-volume-init", "node1", "node2"]:
                 volumes = dsl_pipeline.ops[node_name].container.volume_mounts
-                assert len(volumes) == 1
-                assert volumes[0].name == "data-volume-create"
+                self.assertEqual(len(volumes), 1)
+                self.assertEqual(volumes[0].name, "data-volume-create")
                 assert (
                     dsl_pipeline.ops[
                         node_name
@@ -172,14 +174,14 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             # then
-            assert len(dsl_pipeline.ops) == 5
+            self.assertEqual(len(dsl_pipeline.ops), 6)
             assert "on-exit" in dsl_pipeline.ops
             volume_spec = dsl_pipeline.ops[
                 "data-volume-create"
             ].k8s_resource.spec
-            assert volume_spec.resources.requests["storage"] == "1Mi"
-            assert volume_spec.access_modes == ["ReadWriteOnce"]
-            assert volume_spec.storage_class_name == "nfs"
+            self.assertEqual(volume_spec.resources.requests["storage"], "1Mi")
+            self.assertEqual(volume_spec.access_modes, ["ReadWriteOnce"])
+            self.assertEqual(volume_spec.storage_class_name, "nfs")
 
     def test_should_change_effective_user_if_to_volume_owner(self):
         # given
@@ -207,7 +209,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
 
             # then
             volume_init_spec = dsl_pipeline.ops["data-volume-init"].container
-            assert volume_init_spec.security_context.run_as_user == 47
+            self.assertEqual(volume_init_spec.security_context.run_as_user, 47)
             for node_name in ["data-volume-init", "node1", "node2"]:
                 assert (
                     dsl_pipeline.ops[
@@ -233,16 +235,19 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             # then
-            assert len(dsl_pipeline.ops) == 3
+            self.assertEqual(len(dsl_pipeline.ops), 4)
             init_step = dsl_pipeline.ops["mlflow-start-run"].container
-            assert init_step.image == "unittest-image"
-            assert init_step.args == [
-                "kubeflow",
-                "--env",
-                "unittests",
-                "mlflow-start",
-                "{{workflow.uid}}",
-            ]
+            self.assertEqual(init_step.image, "unittest-image")
+            self.assertEqual(
+                init_step.args,
+                [
+                    "kubeflow",
+                    "--env",
+                    "unittests",
+                    "mlflow-start",
+                    "{{workflow.uid}}",
+                ],
+            )
             assert "MLFLOW_RUN_ID" not in {e.name for e in init_step.env}
             for node_name in ["node1", "node2"]:
                 env = {
@@ -271,14 +276,14 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             # then
-            assert len(dsl_pipeline.ops) == 4
+            self.assertEqual(len(dsl_pipeline.ops), 5)
             assert "data-volume-create" in dsl_pipeline.ops
             assert "on-exit" in dsl_pipeline.ops
             assert "data-volume-init" not in dsl_pipeline.ops
             for node_name in ["node1", "node2"]:
                 volumes = dsl_pipeline.ops[node_name].container.volume_mounts
-                assert len(volumes) == 1
-                assert volumes[0].name == "data-volume-create"
+                self.assertEqual(len(volumes), 1)
+                self.assertEqual(volumes[0].name, "data-volume-create")
 
     def test_should_support_params_and_inject_them_to_the_nodes(self):
         # given
@@ -297,19 +302,106 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             # then
-            assert len(default_params) == 2
-            assert default_params["param1"].default == 0.3
-            assert default_params["param2"].default == 42
+            self.assertEqual(len(default_params), 2)
+            self.assertEqual(default_params["param1"].default, 0.3)
+            self.assertEqual(default_params["param2"].default, 42)
             for node_name in ["node1", "node2"]:
                 args = dsl_pipeline.ops[node_name].container.args
-                assert args == [
-                    "_",
+                self.assertEqual(
+                    args,
+                    [
+                        "_",
+                        "param1",
+                        "{{pipelineparam:op=;name=param1}}",
+                        "param2",
+                        "{{pipelineparam:op=;name=param2}}",
+                    ],
+                )
+
+    def test_should_support_nested_params_and_inject_them_to_the_node(self):
+        # given
+        self.create_generator(
+            params={
+                "param1": {"nested1": {"nested2": 1, "nested3": 2}},
+                "param2": 42,
+            }
+        )
+
+        # when
+        with patch(
+            "kedro.framework.project.pipelines",
+            new=self.pipelines_under_test,
+        ):
+            with kfp.dsl.Pipeline(None) as dsl_pipeline:
+                pipeline = self.generator_under_test.generate_pipeline(
+                    "pipeline", "unittest-image", "Always"
+                )
+                default_params = signature(pipeline).parameters
+                pipeline()
+
+            # then
+            assert len(default_params) == 2
+            assert default_params["param1"].default == {
+                "nested1": {"nested2": 1, "nested3": 2}
+            }
+            assert default_params["param2"].default == 42
+            for node_name in ["node1", "node2", "node3"]:
+                assert dsl_pipeline.ops[node_name].container.args[1:] == [
                     "param1",
                     "{{pipelineparam:op=;name=param1}}",
                     "param2",
                     "{{pipelineparam:op=;name=param2}}",
                 ]
 
+    def test_should_support_namespaced_params_and_inject_them_to_the_node(
+        self,
+    ):
+        # given
+        self.create_generator(
+            params={
+                "outer_namespace.inner_namespace1.param1": "outer_namespace.inner_namespace1.param1_v",
+                "outer_namespace.inner_namespace1.param2": "outer_namespace.inner_namespace1.param2_v",
+                "outer_namespace.inner_namespace2.param1": "outer_namespace.inner_namespace2.param1_v",
+                "outer_namespace.inner_namespace2.param2": "outer_namespace.inner_namespace2.param2_v",
+                "outer_namespace.param": "outer_namespace.param",
+                "param1": 42,
+            }
+        )
+
+        # when
+        with patch(
+            "kedro.framework.project.pipelines",
+            new=self.pipelines_under_test,
+        ):
+            with kfp.dsl.Pipeline(None) as dsl_pipeline:
+                pipeline = self.generator_under_test.generate_pipeline(
+                    "pipeline", "unittest-image", "Always"
+                )
+                default_params = signature(pipeline).parameters
+                pipeline()
+
+            # then
+            assert len(default_params) == 2
+            assert default_params["outer_namespace"].default == {
+                "inner_namespace1": {
+                    "param1": "outer_namespace.inner_namespace1.param1_v",
+                    "param2": "outer_namespace.inner_namespace1.param2_v",
+                },
+                "inner_namespace2": {
+                    "param1": "outer_namespace.inner_namespace2.param1_v",
+                    "param2": "outer_namespace.inner_namespace2.param2_v",
+                },
+                "param": "outer_namespace.param",
+            }
+            assert default_params["param1"].default == 42
+            for node_name in ["node1", "node2", "node3"]:
+                assert dsl_pipeline.ops[node_name].container.args[1:] == [
+                    "outer_namespace",
+                    "{{pipelineparam:op=;name=outer_namespace}}",
+                    "param1",
+                    "{{pipelineparam:op=;name=param1}}",
+                ]
+
     def test_should_fallbackto_default_resources_spec_if_not_requested(self):
         # given
         self.create_generator(config={})
@@ -337,6 +429,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 "resources": {
                     "__default__": {"cpu": "100m"},
                     "node1": {"cpu": "400m", "memory": "64Gi"},
+                    "node3": {"memory": "32Gi", "nvidia.com/gpu": "1"},
                 }
             }
         )
@@ -355,10 +448,21 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
             # then
             node1_spec = dsl_pipeline.ops["node1"].container.resources
             node2_spec = dsl_pipeline.ops["node2"].container.resources
-            assert node1_spec.limits == {"cpu": "400m", "memory": "64Gi"}
-            assert node1_spec.requests == {"cpu": "400m", "memory": "64Gi"}
-            assert node2_spec.limits == {"cpu": "100m"}
-            assert node2_spec.requests == {"cpu": "100m"}
+            node3_spec = dsl_pipeline.ops["node3"].container.resources
+            for spec, result in zip(
+                (node1_spec, node2_spec, node3_spec),
+                (
+                    {"cpu": "400m", "memory": "64Gi"},
+                    {"cpu": "100m"},
+                    {"cpu": "100m", "memory": "32Gi", "nvidia.com/gpu": "1"},
+                ),
+            ):
+                for precise_spec in (spec.limits, spec.requests):
+                    self.assertDictEqual(precise_spec, result)
+            # self.assertEqual(node1_spec.limits , {"cpu": "400m", "memory": "64Gi"})
+            # self.assertEqual(node1_spec.requests , {"cpu": "400m", "memory": "64Gi"})
+            # self.assertEqual(node2_spec.limits , {"cpu": "100m"})
+            # self.assertEqual(node2_spec.requests , {"cpu": "100m"})
 
     def test_can_add_extra_volumes(self):
         self.create_generator(
@@ -391,7 +495,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 pipeline()
 
             volume_mounts = dsl_pipeline.ops["node1"].container.volume_mounts
-            assert len(volume_mounts) == 1
+            self.assertEqual(len(volume_mounts), 1)
 
     def test_should_not_add_retry_policy_if_not_requested(self):
         # given
@@ -411,7 +515,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
             # then
             for node_name in ["node1", "node2"]:
                 op = dsl_pipeline.ops[node_name]
-                assert op.num_retries == 0
+                self.assertEqual(op.num_retries, 0)
                 assert op.retry_policy is None
                 assert op.backoff_factor is None
                 assert op.backoff_duration is None
@@ -449,16 +553,16 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
 
             # then
             op1 = dsl_pipeline.ops["node1"]
-            assert op1.num_retries == 100
-            assert op1.retry_policy == "Always"
-            assert op1.backoff_factor == 1
-            assert op1.backoff_duration == "5m"
+            self.assertEqual(op1.num_retries, 100)
+            self.assertEqual(op1.retry_policy, "Always")
+            self.assertEqual(op1.backoff_factor, 1)
+            self.assertEqual(op1.backoff_duration, "5m")
             assert op1.backoff_max_duration is None
             op2 = dsl_pipeline.ops["node2"]
-            assert op2.num_retries == 4
-            assert op2.retry_policy == "Always"
-            assert op2.backoff_factor == 2
-            assert op2.backoff_duration == "60s"
+            self.assertEqual(op2.num_retries, 4)
+            self.assertEqual(op2.retry_policy, "Always")
+            self.assertEqual(op2.backoff_factor, 2)
+            self.assertEqual(op2.backoff_duration, "60s")
             assert op2.backoff_max_duration is None
 
     def test_should_add_max_cache_staleness(self):
@@ -494,7 +598,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
             )
 
             # then
-            assert pipeline._component_description == "DESC"
+            self.assertEqual(pipeline._component_description, "DESC")
 
     def test_artifact_registration(self):
         # given
@@ -520,11 +624,15 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
 
             # then
             outputs1 = dsl_pipeline.ops["node1"].file_outputs
-            assert len(outputs1) == 1
+            self.assertEqual(len(outputs1), 1)
             assert "B" in outputs1
-            assert outputs1["B"] == "/home/kedro/data/02_intermediate/b.csv"
+            self.assertEqual(
+                outputs1["B"], "/home/kedro/data/02_intermediate/b.csv"
+            )
             outputs2 = dsl_pipeline.ops["node2"].file_outputs
-            assert len(outputs2) == 0  # output "C" is missing in the catalog
+            self.assertEqual(
+                len(outputs2), 0
+            )  # output "C" is missing in the catalog)
 
     def test_should_skip_artifact_registration_if_requested(self):
         # given
@@ -551,7 +659,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
 
             # then
             outputs1 = dsl_pipeline.ops["node1"].file_outputs
-            assert len(outputs1) == 0
+            self.assertEqual(len(outputs1), 0)
 
     def test_should_skip_volume_removal_if_requested(self):
         # given
@@ -596,7 +704,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                         for e in dsl_pipeline.ops[node_name].container.env
                     }
                     assert "KEDRO_CONFIG_MY_KEY" in env_values
-                    assert env_values["KEDRO_CONFIG_MY_KEY"] == "42"
+                    self.assertEqual(env_values["KEDRO_CONFIG_MY_KEY"], "42")
                     assert "SOME_VALUE" not in env_values
         finally:
             del os.environ["KEDRO_CONFIG_MY_KEY"]
@@ -620,6 +728,7 @@ class TestGenerator(unittest.TestCase, MinimalConfigMixin):
                 [
                     node(identity, "A", "B", name="node1"),
                     node(identity, "B", "C", name="node2"),
+                    node(identity, "B", "D", name="node3"),
                 ]
             )
         }