Repository: getindata/kedro-kubeflow (python)

Issue:
Expose output dataset as artifacts
So they can be downloaded from Pipelines UI

Pull request:
Output artifacts exposed in the Pipelines UI
Resolves #51 

##### PR Checklist
- [x] Tests added
- [x] [Changelog](CHANGELOG.md) updated 


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 921297f..14eb24f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@
 - Support to inject Kedro pipeline parameters for the run
 - Ability to specify resources allocation for the nodes
 - Possibility to configure the pipeline description in the config file
+- The registered output artifacts are not exposed in the Pipelines UI
 
 ## [0.2.0] - 2021-01-18
 
diff --git a/kedro_kubeflow/generator.py b/kedro_kubeflow/generator.py
index bee228c..21ccb39 100644
--- a/kedro_kubeflow/generator.py
+++ b/kedro_kubeflow/generator.py
@@ -40,6 +40,7 @@ class PipelineGenerator(object):
         self.volume_meta = config.run_config.volume
         self.resources = config.run_config.resources
         self.description = config.run_config.description
+        self.catalog = context.config_loader.get("catalog*")
 
     def generate_pipeline(self, pipeline, image, image_pull_policy):
         @dsl.pipeline(
@@ -134,6 +135,12 @@ class PipelineGenerator(object):
                     ],
                     pvolumes=node_volumes,
                     container_kwargs=kwargs,
+                    file_outputs={
+                        output: "/home/kedro/"
+                        + self.catalog[output]["filepath"]
+                        for output in node.outputs
+                        if output in self.catalog
+                    },
                 ),
                 image_pull_policy,
             )


Test patch:
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 04aaf02..57a37c0 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -43,3 +43,15 @@ class TestAuthHandler(unittest.TestCase):
                 in cm.output[0]
             )
             assert token is None
+
+    @patch("google.oauth2.id_token.fetch_id_token")
+    def test_should_provide_valid_token(self, fetch_id_token_mock):
+        # given
+        os.environ["IAP_CLIENT_ID"] = "unittest-client-id"
+        fetch_id_token_mock.return_value = "TOKEN"
+
+        # when
+        token = AuthHandler().obtain_id_token()
+
+        # then
+        assert token == "TOKEN"
diff --git a/tests/test_generator.py b/tests/test_generator.py
index df7c3f5..9c02257 100644
--- a/tests/test_generator.py
+++ b/tests/test_generator.py
@@ -2,6 +2,7 @@
 
 import unittest
 from inspect import signature
+from unittest.mock import MagicMock
 
 import kfp
 from kedro.pipeline import Pipeline, node
@@ -257,13 +258,42 @@ class TestGenerator(unittest.TestCase):
         # then
         assert pipeline._component_description == "DESC"
 
-    def create_generator(self, config={}, params={}):
+    def test_artifact_registration(self):
+        # given
+        self.create_generator(
+            catalog={
+                "B": {
+                    "type": "pandas.CSVDataSet",
+                    "filepath": "data/02_intermediate/b.csv",
+                }
+            }
+        )
+
+        # when
+        pipeline = self.generator_under_test.generate_pipeline(
+            "pipeline", "unittest-image", "Always"
+        )
+        with kfp.dsl.Pipeline(None) as dsl_pipeline:
+            pipeline()
+
+        # then
+        outputs1 = dsl_pipeline.ops["node1"].file_outputs
+        assert len(outputs1) == 1
+        assert "B" in outputs1
+        assert 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
+
+    def create_generator(self, config={}, params={}, catalog={}):
         project_name = "my-awesome-project"
+        config_loader = MagicMock()
+        config_loader.get.return_value = catalog
         context = type(
             "obj",
             (object,),
             {
                 "params": params,
+                "config_loader": config_loader,
                 "pipelines": {
                     "pipeline": Pipeline(
                         [
diff --git a/tests/test_kfpclient.py b/tests/test_kfpclient.py
index 4b9950c..b21ce86 100644
--- a/tests/test_kfpclient.py
+++ b/tests/test_kfpclient.py
@@ -5,7 +5,7 @@ import unittest
 from tempfile import NamedTemporaryFile
 from unittest.mock import patch
 
-from kedro.pipeline import Pipeline
+from kfp import dsl
 
 from kedro_kubeflow.config import PluginConfig
 from kedro_kubeflow.kfpclient import KubeflowClient
@@ -132,14 +132,17 @@ class TestKubeflowClient(unittest.TestCase):
             with open(f.name) as yamlfile:
                 assert "generateName: my-awesome-project-" in yamlfile.read()
 
-    @patch("google.oauth2.id_token.fetch_id_token")
+    @patch("kedro_kubeflow.kfpclient.AuthHandler")
+    @patch("kedro_kubeflow.kfpclient.PipelineGenerator")
     @patch("kedro_kubeflow.kfpclient.Client")
     def test_should_use_jwt_token_in_kfp_client(
-        self, kfp_client_mock, fetch_id_token_mock
+        self, kfp_client_mock, pipeline_generator_mock, auth_handler_mock
     ):
         # given
         os.environ["IAP_CLIENT_ID"] = "unittest-client-id"
-        fetch_id_token_mock.return_value = "unittest-token"
+        auth_handler_mock.return_value.obtain_id_token.return_value = (
+            "unittest-token"
+        )
 
         # when
         self.client_under_test = KubeflowClient(
@@ -282,25 +285,26 @@ class TestKubeflowClient(unittest.TestCase):
         self.kfp_client_mock.pipeline_uploads.upload_pipeline.assert_not_called()
         self.kfp_client_mock.pipeline_uploads.upload_pipeline_version.assert_called()
 
+    @patch("kedro_kubeflow.kfpclient.PipelineGenerator")
     @patch("kedro_kubeflow.kfpclient.Client")
-    def create_client(self, config, kfp_client_mock):
+    def create_client(self, config, kfp_client_mock, pipeline_generator_mock):
         project_name = "my-awesome-project"
-        context = type(
-            "obj",
-            (object,),
-            {
-                "params": {},
-                "pipelines": {"pipeline": Pipeline([])},
-            },
-        )
         self.client_under_test = KubeflowClient(
             PluginConfig({"host": "http://unittest", "run_config": config}),
             project_name,
-            context,
+            None,  # context,
         )
         self.client_under_test.client = kfp_client_mock
         self.kfp_client_mock = self.client_under_test.client
 
+        @dsl.pipeline(name=project_name)
+        def empty_pipeline():
+            pass
+
+        self.client_under_test.generator.generate_pipeline.return_value = (
+            empty_pipeline
+        )
+
     def mock_mlflow(self, enabled=False):
         def fakeimport(name, *args, **kw):
             if not enabled and name == "mlflow":