Repository: getindata/kedro-kubeflow
Pull request #267 (2024-06-25 10:16:43)

E2e tests fixes after 0.19 updates
#### Description

`describe the purpose of the change here` 

Resolves `<issue nr here>`

##### PR Checklist
- [ ] Tests added
- [ ] [Changelog](CHANGELOG.md) updated 



Patch:
diff --git a/.gitignore b/.gitignore
index 6007e80..afaa55f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -173,4 +173,4 @@ terraform/terraform.tfstate
 
 .idea
 conf/azure/credentials.yml
-
+.vscode
diff --git a/kedro_kubeflow/cli.py b/kedro_kubeflow/cli.py
index a0c5e3c..78a8cd0 100644
--- a/kedro_kubeflow/cli.py
+++ b/kedro_kubeflow/cli.py
@@ -276,8 +276,8 @@ def init(ctx, kfp_url: str, with_github_actions: bool):
     context_helper = ctx.obj["context_helper"]
     project_name = context_helper.context.project_path.name
     if with_github_actions:
-        image = f"gcr.io/${{google_project_id}}/{project_name}:${{commit_id}}"
-        run_name = f"{project_name}:${{commit_id}}"
+        image = f"gcr.io/${{oc.env:KEDRO_CONFIG_GOOGLE_PROJECT_ID}}/{project_name}:${{oc.env:KEDRO_CONFIG_COMMIT_ID, unknown-commit}}"  # noqa: E501
+        run_name = f"{project_name}:${{oc.env:KEDRO_CONFIG_COMMIT_ID, unknown-commit}}"
     else:
         image = project_name
         run_name = project_name
@@ -287,9 +287,13 @@ def init(ctx, kfp_url: str, with_github_actions: bool):
     with open(config_path, "w") as f:
         f.write(sample_config)
 
-    click.echo(f"Configuration generated in {config_path}")
+    click.echo(f"Configuration generated in {config_path}.")
 
     if with_github_actions:
+        click.echo(
+            "Make sure to update settings.py to add custom"
+            " resolver for environment variables (oc.env). to src/.../settings.py:"
+        )
         PluginConfig.initialize_github_actions(
             project_name,
             where=Path.cwd(),
diff --git a/kedro_kubeflow/config.py b/kedro_kubeflow/config.py
index 2c521cf..b771f68 100644
--- a/kedro_kubeflow/config.py
+++ b/kedro_kubeflow/config.py
@@ -51,12 +51,12 @@ run_config:
   # This sets the caching option for pipeline using
   # execution_options.caching_strategy.max_cache_staleness
   # See https://en.wikipedia.org/wiki/ISO_8601 in section 'Duration'
-  #max_cache_staleness: P0D
+  # max_cache_staleness: P0D
 
   # Set to false to disable kfp artifacts exposal
   # This setting can be useful if you don't want to store
   # intermediate results in the MLMD
-  #store_kedro_outputs_as_kfp_artifacts: True
+  # store_kedro_outputs_as_kfp_artifacts: True
 
   # Strategy used to generate Kubeflow pipeline nodes from Kedro nodes
   # Available strategies:
@@ -70,14 +70,14 @@ run_config:
   #           This strategy mitigate potential performance issues with `none` strategy
   #           but at the cost of degraded user experience within Kubeflow UI: a graph
   #           is collapsed to one node.
-  #node_merge_strategy: none
+  # node_merge_strategy: none
 
   # Optional volume specification
   volume:
 
     # Storage class - use null (or no value) to use the default storage
     # class deployed on the Kubernetes cluster
-    storageclass: # default
+    storageclass: null # default
 
     # The size of the volume that is created. Applicable for some storage
     # classes
@@ -86,7 +86,7 @@ run_config:
     # Access mode of the volume used to exchange data. ReadWriteMany is
     # preferred, but it is not supported on some environements (like GKE)
     # Default value: ReadWriteOnce
-    #access_modes: [ReadWriteMany]
+    # access_modes: [ReadWriteMany]
 
     # Flag indicating if the data-volume-init step (copying raw data to the
     # fresh volume) should be skipped
diff --git a/kedro_kubeflow/context_helper.py b/kedro_kubeflow/context_helper.py
index b1376d0..26d5111 100644
--- a/kedro_kubeflow/context_helper.py
+++ b/kedro_kubeflow/context_helper.py
@@ -1,10 +1,15 @@
 import os
-from functools import lru_cache
+from functools import cached_property, lru_cache
 from typing import Any, Dict
 
 from kedro import __version__ as kedro_version
-from kedro.config import OmegaConfigLoader
+from kedro.config import (
+    AbstractConfigLoader,
+    MissingConfigException,
+    OmegaConfigLoader,
+)
 from kedro.framework.session import KedroSession
+from omegaconf import DictConfig, OmegaConf
 from omegaconf.resolvers import oc
 from semver import VersionInfo
 
@@ -27,7 +32,7 @@ class EnvTemplatedConfigLoader(OmegaConfigLoader):
         config_patterns: Dict[str, Any] = None,
         *,
         base_env: str = "base",
-        default_run_env: str = "local"
+        default_run_env: str = "local",
     ):
         import warnings  # TODO remove this class
 
@@ -64,6 +69,7 @@ class EnvTemplatedConfigLoader(OmegaConfigLoader):
 class ContextHelper(object):
 
     CONFIG_FILE_PATTERN = "kubeflow*"
+    CONFIG_KEY = "kubeflow"
 
     def __init__(self, metadata, env):
         self._metadata = metadata
@@ -76,7 +82,7 @@ class ContextHelper(object):
     @property
     @lru_cache()
     def session(self):
-        return KedroSession.create(self._metadata.package_name, env=self._env)
+        return KedroSession.create(self._metadata.project_path, env=self._env)
 
     @property
     def env(self):
@@ -86,14 +92,46 @@ class ContextHelper(object):
     def context(self):
         return self.session.load_context()
 
-    @property
-    @lru_cache()
+    @cached_property
     def config(self) -> PluginConfig:
-        raw = EnvTemplatedConfigLoader(
-            self.context.config_loader.conf_source,
-            env=self._env,
-        ).get(self.CONFIG_FILE_PATTERN)
-        return PluginConfig(**raw)
+        cl: AbstractConfigLoader = self.context.config_loader
+        try:
+            if self.CONFIG_KEY not in cl.config_patterns.keys():
+                cl.config_patterns.update(
+                    {
+                        self.CONFIG_KEY: [
+                            self.CONFIG_FILE_PATTERN,
+                            f"{self.CONFIG_FILE_PATTERN}/**",
+                        ]
+                    }
+                )
+            kubeflow_config = self._ensure_obj_is_dict(cl.get(self.CONFIG_KEY))
+        except MissingConfigException:
+            if not isinstance(cl, OmegaConfigLoader):
+                raise ValueError(
+                    f"You're using a custom config loader: {cl.__class__.__qualname__}{os.linesep}"
+                    f"you need to add the {self.CONFIG_KEY} config to it.{os.linesep}"
+                    f"Make sure you add {self.CONFIG_FILE_PATTERN} to config_pattern in CONFIG_LOADER_ARGS "
+                    f"in the settings.py file.{os.linesep}"
+                    """Example:
+CONFIG_LOADER_ARGS = {
+    # other args
+    "config_patterns": {"kubeflow": ["kubeflow*"]}
+}
+                    """.strip()
+                )
+            else:
+                raise ValueError(
+                    "Missing kubeflow.yml files in configuration. " "Make sure that you configure your project first"
+                )
+        return PluginConfig.parse_obj(kubeflow_config)
+
+    def _ensure_obj_is_dict(self, obj):
+        if isinstance(obj, DictConfig):
+            obj = OmegaConf.to_container(obj)
+        elif isinstance(obj, dict) and any(isinstance(v, DictConfig) for v in obj.values()):
+            obj = {k: (OmegaConf.to_container(v) if isinstance(v, DictConfig) else v) for k, v in obj.items()}
+        return obj
 
     @property
     @lru_cache()
diff --git a/poetry.lock b/poetry.lock
index c19bfb6..ad5cc5a 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -3573,13 +3573,13 @@ files = [
 
 [[package]]
 name = "setuptools"
-version = "70.1.0"
+version = "70.1.1"
 description = "Easily download, build, install, upgrade, and uninstall Python packages"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "setuptools-70.1.0-py3-none-any.whl", hash = "sha256:d9b8b771455a97c8a9f3ab3448ebe0b29b5e105f1228bba41028be116985a267"},
-    {file = "setuptools-70.1.0.tar.gz", hash = "sha256:01a1e793faa5bd89abc851fa15d0a0db26f160890c7102cd8dce643e886b47f5"},
+    {file = "setuptools-70.1.1-py3-none-any.whl", hash = "sha256:a58a8fde0541dab0419750bcc521fbdf8585f6e5cb41909df3a472ef7b81ca95"},
+    {file = "setuptools-70.1.1.tar.gz", hash = "sha256:937a48c7cdb7a21eb53cd7f9b59e525503aa8abaf3584c730dc5f7a5bec3a650"},
 ]
 
 [package.extras]
diff --git a/pyproject.toml b/pyproject.toml
index 92c29e3..4762117 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,8 +39,9 @@ known_third_party = ["click", "google", "kedro", "kfp", "kubernetes", "tabulate"
 [tool.poetry.dependencies]
 python = ">=3.8,<3.12"
 kedro = ">0.18.4, <=0.19.6"
+# kedro-viz = { version = "<8", optional = true} # because of pydantic conflict
 click = ">=8.0.4"
-kfp = ">=1.8.12,<2.0"
+kfp = ">=1.8.12,<2.0" # blocking pydantic upgrade
 tabulate = ">=0.8.7"
 semver = "~=2.10"
 fsspec = ">=2021.4"


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index de8c10a..0b11cff 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -132,12 +132,14 @@ jobs:
       - name: Init and update configuration
         working-directory: ./spaceflights
         run: |
+          sed -i "s/\(CONFIG_LOADER_ARGS.*\)$/from omegaconf.resolvers import oc\n\1\n      \"custom_resolvers\": { \"oc.env\": oc.env },/g" src/spaceflights/settings.py
           kedro docker init
           kedro kubeflow init http://localhost:9000
           echo "!data/01_raw" >> .dockerignore
-          mv ../tests/e2e/catalog.yml conf/base/catalog.yml
+          cp ../tests/e2e/catalog.yml conf/base/catalog.yml
           rm -f conf/base/kubeflow.y*
-          mv ../tests/e2e/kubeflow.yml conf/base/kubeflow.yml
+          cp ../tests/e2e/kubeflow.yml conf/base/kubeflow.yml
+          
 
       - name: Prepare docker env
         uses: docker/setup-buildx-action@v1
diff --git a/tests/common.py b/tests/common.py
index 40baca8..a494850 100644
--- a/tests/common.py
+++ b/tests/common.py
@@ -8,6 +8,7 @@ class MinimalConfigMixin:
                 "image": "asd",
                 "experiment_name": "exp",
                 "run_name": "unit tests",
+                "node_merge_strategy": "none",
             },
             "host": "localhost:8080",
         }
diff --git a/tests/e2e/kubeflow.yml b/tests/e2e/kubeflow.yml
index 65a3e61..bf3da74 100644
--- a/tests/e2e/kubeflow.yml
+++ b/tests/e2e/kubeflow.yml
@@ -1,12 +1,12 @@
 
 # Base url of the Kubeflow Pipelines, should include the schema (http/https)
-host: ${kubeflow_hostname}
+host: ${oc.env:KEDRO_CONFIG_KUBEFLOW_HOSTNAME,http://localhost:9000}
 
 # Configuration used to run the pipeline
 run_config:
 
   # Name of the image to run as the pipeline steps
-  image: gcr.io/gid-ml-ops-sandbox/kedro-kubeflow-e2e:${commit_id|latest}
+  image: gcr.io/gid-ml-ops-sandbox/kedro-kubeflow-e2e:${oc.env:KEDRO_CONFIG_COMMIT_ID, latest}
 
   # Pull policy to be used for the steps. Use Always if you push the images
   # on the same tag, or Never if you use only local images
@@ -19,7 +19,7 @@ run_config:
   run_name: kedro-kubeflow-e2e
 
   # Name of the scheduled run, templated with the schedule parameters
-  scheduled_run_name: kedro-kubeflow-e2e
+  scheduled_run_name: kedro-kubeflow-e2e-${oc.env:KEDRO_CONFIG_RUN_ID,unknown}
 
   # Optional pipeline description
   #description: "Very Important Pipeline"
diff --git a/tests/test_context_helper.py b/tests/test_context_helper.py
index d066016..6cc0ac7 100644
--- a/tests/test_context_helper.py
+++ b/tests/test_context_helper.py
@@ -1,8 +1,11 @@
+import json
 import os
 import unittest
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from unittest.mock import MagicMock, Mock, patch
 
+import yaml
 from kedro.framework.session import KedroSession
 
 from kedro_kubeflow.config import PluginConfig
@@ -32,7 +35,7 @@ class TestContextHelper(unittest.TestCase, MinimalConfigMixin):
 
     def test_context(self):
         metadata = Mock()
-        metadata.package_name = "test_package"
+        metadata.project_path = "test_package"
         kedro_session = MagicMock(KedroSession)
         kedro_session.load_context.return_value = "sample_context"
 
@@ -42,18 +45,56 @@ class TestContextHelper(unittest.TestCase, MinimalConfigMixin):
             assert helper.context == "sample_context"
             create.assert_called_with("test_package", env="test")
 
+    # def test_config(self):
+    #     metadata = Mock()
+    #     metadata.package_name = "test_package"
+    #     session = MagicMock()
+    #     cfg = self.minimal_config()
+    #     session.load_context().config_loader.return_value = cfg.dict()
+    #     with patch.object(KedroSession, "create", return_value=session), patch(
+    #         "kedro_kubeflow.context_helper.EnvTemplatedConfigLoader"
+    #     ) as config_loader:
+    #         config_loader.return_value.get.return_value =
+    #         helper = ContextHelper.init(metadata, "test")
+    #         assert helper.config == PluginConfig(**self.minimal_config())
+    #         assert config_loader.call_args.kwargs["env"] == "test"
+
     def test_config(self):
         metadata = Mock()
         metadata.package_name = "test_package"
-        context = MagicMock()
-        context.config_loader.return_value.get.return_value = ["one", "two"]
-        with patch.object(KedroSession, "create", context), patch(
-            "kedro_kubeflow.context_helper.EnvTemplatedConfigLoader"
-        ) as config_loader:
-            config_loader.return_value.get.return_value = self.minimal_config()
+        session = MagicMock()
+        cfg = PluginConfig.parse_obj(self.minimal_config())
+        session.load_context().config_loader.get.return_value = cfg.dict()
+        with patch.object(KedroSession, "create", return_value=session):
             helper = ContextHelper.init(metadata, "test")
-            assert helper.config == PluginConfig(**self.minimal_config())
-            assert config_loader.call_args.kwargs["env"] == "test"
+            assert helper.config == cfg
+
+    # TODO debug and fix omegaconf test
+    def test_config_with_omegaconf(self):
+        from kedro.config import OmegaConfigLoader
+
+        with TemporaryDirectory() as tmp_dir_raw:
+            tmp_dir = Path(tmp_dir_raw)
+            (tmp_dir / "conf" / "base").mkdir(parents=True, exist_ok=False)
+            (conf_dir := tmp_dir / "conf" / "local").mkdir(parents=True, exist_ok=False)
+            cfg = PluginConfig.parse_obj(self.minimal_config())
+            (conf_dir / "kubeflow.yml").write_text(
+                yaml.dump(json.loads(json.dumps(cfg.dict()))), None
+            )  # because enums are not
+
+            metadata = Mock()
+            metadata.package_name = "test_package"
+            for config_pattern in [{"kubeflow": ["kubeflow*"]}]:
+                session = MagicMock()
+                session.load_context().config_loader = OmegaConfigLoader(
+                    str(tmp_dir / "conf"),
+                    config_patterns=config_pattern,
+                    default_run_env="local",
+                    base_env="local",
+                )
+                with patch.object(KedroSession, "create", return_value=session):
+                    helper = ContextHelper.init(metadata, "test")
+                    assert helper.config == cfg
 
 
 class TestEnvTemplatedConfigLoader(unittest.TestCase):