Repository: getindata/kedro-kubeflow
Pull request #64 (2021-08-11 11:36:13)

Support for Dex-secured Kubeflow clusters
#### Description

This PR provides 3 features, initially developed by @kromanow94 (https://github.com/kromanow94/kedro-kubeflow):

-   Ability to access KFP API behind Dex+authservice authentication
-   Support for multi-user KFP setup (experiment namespace passed via `run-once` or `schedule`)
-   New config param: `max_cache_staleness` to avoid caching KFP steps if required

##### PR Checklist
- [x] Tests added
- [x] [Changelog](CHANGELOG.md) updated 



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4952357..5a89ccb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,9 @@
 
 -   Support for kedro-mlflow>=0.7
 -   Support of Google Vertex AI Pipelines (EXPERIMENTAL)
+-   Ability to access KFP API behind Dex+authservice authentication
+-   Support for multi-user KFP setup (experiment namespace passed via `run-once` or `schedule`)
+-   New config param: `max_cache_staleness` to avoid caching KFP steps if required
 
 ## [0.3.1] - 2021-05-25
 
diff --git a/docs/source/03_getting_started/05_authentication.md b/docs/source/03_getting_started/05_authentication.md
new file mode 100644
index 0000000..8bb1ad7
--- /dev/null
+++ b/docs/source/03_getting_started/05_authentication.md
@@ -0,0 +1,21 @@
+# Authenticating to Kubeflow Pipelines API
+
+Plugin supports 2 ways of authenticating to Kubeflow Pipelines API:
+
+## 1. KFP behind IAP proxy on Google Cloud
+
+It's already described in [GCP AI Platform support](02_gcp.md) chapter.
+
+## 2. KFP behind Dex+authservice
+
+Dex is the recommended authentication mechanism for on-premise Kubeflow clusters. The usual setup looks in a way that:
+
+* [oidc-autheservice](https://github.com/arrikto/oidc-authservice) redirect unauthenticated users to Dex,
+* [Dex](https://github.com/dexidp/dex) authenticates user in remote system, like LDAP or OpenID and also acts as OpenID provider,
+* oidc-autheservice asks Dex for token and creates the session used across entire Kubeflow.
+
+In order to use `kedro-kubeflow` behind Dex-secured clusters, use the following manual:
+
+1. Setup [staticPassword](https://github.com/dexidp/dex/blob/b79d9a84bc0c35e13a9d5141e95b641af0f81c8f/cmd/dex/config_test.go#L105) authentication method and add a user that you're going to use as CI/CD account.
+2. Point your Kedro project to `/pipeline` API on Kubeflow, for example: `https://kubeflow.local/pipeline`
+3. Set environment variables `DEX_USERNAME` and `DEX_PASSWORD` before calling `kedro kubeflow`
diff --git a/docs/source/03_getting_started/index.rst b/docs/source/03_getting_started/index.rst
index 2027467..1270849 100644
--- a/docs/source/03_getting_started/index.rst
+++ b/docs/source/03_getting_started/index.rst
@@ -8,3 +8,4 @@ Getting started
    Google Cloud Platform support <02_gcp.md>
    Kedro-Mlflow integration <03_mlflow.md>
    Continuous Deployment <04_continuous_deployment.md>
+   Authenticating to Kubeflow Pipelines API <05_authentication.md>
diff --git a/kedro_kubeflow/auth.py b/kedro_kubeflow/auth.py
index a33677e..36ad561 100644
--- a/kedro_kubeflow/auth.py
+++ b/kedro_kubeflow/auth.py
@@ -1,7 +1,13 @@
 import logging
 import os
+import re
+from urllib.parse import urlsplit, urlunsplit
+
+import requests
 
 IAP_CLIENT_ID = "IAP_CLIENT_ID"
+DEX_USERNAME = "DEX_USERNAME"
+DEX_PASSWORD = "DEX_PASSWORD"
 
 
 class AuthHandler(object):
@@ -42,3 +48,38 @@ class AuthHandler(object):
             self.log.error("Failed to obtain IAP access token. " + str(e))
         finally:
             return jwt_token
+
+    def obtain_dex_authservice_session(self, kfp_api):
+        if DEX_USERNAME not in os.environ or DEX_PASSWORD not in os.environ:
+            self.log.debug(
+                "Skipping DEX authentication due to missing env variables"
+            )
+            return None
+
+        s = requests.Session()
+        r = s.get(kfp_api)
+        form_relative_url = re.search(
+            '/dex/auth/local\\?req=([^"]*)', r.text
+        ).group(0)
+
+        kfp_url_parts = urlsplit(kfp_api)
+        form_absolute_url = urlunsplit(
+            [
+                kfp_url_parts.scheme,
+                kfp_url_parts.netloc,
+                form_relative_url,
+                None,
+                None,
+            ]
+        )
+
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+        }
+        data = {
+            "login": os.environ[DEX_USERNAME],
+            "password": os.environ[DEX_PASSWORD],
+        }
+
+        s.post(form_absolute_url, headers=headers, data=data)
+        return s.cookies.get_dict()["authservice_session"]
diff --git a/kedro_kubeflow/cli.py b/kedro_kubeflow/cli.py
index b3bca03..fd9eff3 100644
--- a/kedro_kubeflow/cli.py
+++ b/kedro_kubeflow/cli.py
@@ -29,7 +29,10 @@ def commands():
 def kubeflow_group(ctx, metadata, env):
     """Interact with Kubeflow Pipelines"""
     ctx.ensure_object(dict)
-    ctx.obj["context_helper"] = ContextHelper.init(metadata, env)
+    ctx.obj["context_helper"] = ContextHelper.init(
+        metadata,
+        env,
+    )
 
 
 @kubeflow_group.command()
@@ -55,8 +58,17 @@ def list_pipelines(ctx):
     help="Name of pipeline to run",
     default="__default__",
 )
+@click.option(
+    "-en",
+    "--experiment-namespace",
+    "experiment_namespace",
+    type=str,
+    default=None,
+    help="Namespace where pipeline experiment run should be deployed to. Not needed "
+    "if provided experiment name already exists.",
+)
 @click.pass_context
-def run_once(ctx, image: str, pipeline: str):
+def run_once(ctx, image: str, pipeline: str, experiment_namespace: str):
     """Deploy pipeline as a single run within given experiment.
     Config can be specified in kubeflow.yml as well."""
     context_helper = ctx.obj["context_helper"]
@@ -66,6 +78,7 @@ def run_once(ctx, image: str, pipeline: str):
         pipeline=pipeline,
         image=image if image else config.image,
         experiment_name=config.experiment_name,
+        experiment_namespace=experiment_namespace,
         run_name=config.run_name,
         wait=config.wait_for_completion,
         image_pull_policy=config.image_pull_policy,
@@ -159,14 +172,27 @@ def upload_pipeline(ctx, image, pipeline) -> None:
     type=str,
     help="Name of experiment associated with this run.",
 )
+@click.option(
+    "-en",
+    "--experiment-namespace",
+    "experiment_namespace",
+    type=str,
+    default=None,
+    help="Namespace where pipeline experiment run should be deployed to. Not needed "
+    "if provided experiment name already exists.",
+)
 @click.pass_context
-def schedule(ctx, experiment_name: str, cron_expression: str):
+def schedule(
+    ctx, experiment_namespace: str, experiment_name: str, cron_expression: str
+):
     """Schedules recurring execution of latest version of the pipeline"""
     context_helper = ctx.obj["context_helper"]
     config = context_helper.config.run_config
     experiment = experiment_name if experiment_name else config.experiment_name
 
-    context_helper.kfp_client.schedule(experiment, cron_expression)
+    context_helper.kfp_client.schedule(
+        experiment, experiment_namespace, cron_expression
+    )
 
 
 @kubeflow_group.command()
diff --git a/kedro_kubeflow/config.py b/kedro_kubeflow/config.py
index 7d8ed74..4c6e540 100644
--- a/kedro_kubeflow/config.py
+++ b/kedro_kubeflow/config.py
@@ -35,6 +35,11 @@ run_config:
   # volume after pipeline finishes) [in seconds]. Default: 1 week
   ttl: 604800
 
+  # 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
+
   # Optional volume specification (only for non vertex-ai)
   volume:
 
@@ -190,6 +195,10 @@ class RunConfig(Config):
     def wait_for_completion(self):
         return bool(self._get_or_default("wait_for_completion", False))
 
+    @property
+    def max_cache_staleness(self):
+        return str(self._get_or_default("max_cache_staleness", None))
+
     @property
     def ttl(self):
         return int(self._get_or_default("ttl", 3600 * 24 * 7))
diff --git a/kedro_kubeflow/context_helper.py b/kedro_kubeflow/context_helper.py
index 4274bbb..c010836 100644
--- a/kedro_kubeflow/context_helper.py
+++ b/kedro_kubeflow/context_helper.py
@@ -11,7 +11,11 @@ class ContextHelper(object):
 
     CONFIG_FILE_PATTERN = "kubeflow*"
 
-    def __init__(self, metadata, env):
+    def __init__(
+        self,
+        metadata,
+        env,
+    ):
         self._metadata = metadata
         self._env = env
 
@@ -48,15 +52,28 @@ class ContextHelper(object):
         else:
             from .kfpclient import KubeflowClient
 
-            return KubeflowClient(self.config, self.project_name, self.context)
+            return KubeflowClient(
+                self.config,
+                self.project_name,
+                self.context,
+            )
 
     @staticmethod
-    def init(metadata, env):
+    def init(
+        metadata,
+        env,
+    ):
         version = VersionInfo.parse(kedro_version)
         if version.match(">=0.17.0"):
-            return ContextHelper(metadata, env)
+            return ContextHelper(
+                metadata,
+                env,
+            )
         else:
-            return ContextHelper16(metadata, env)
+            return ContextHelper16(
+                metadata,
+                env,
+            )
 
 
 class ContextHelper16(ContextHelper):
diff --git a/kedro_kubeflow/generator.py b/kedro_kubeflow/generator.py
index aeccb39..f66811d 100644
--- a/kedro_kubeflow/generator.py
+++ b/kedro_kubeflow/generator.py
@@ -8,6 +8,7 @@ from typing import Dict, Set
 import kubernetes.client as k8s
 from kedro.pipeline.node import Node
 from kfp import dsl
+from kfp.compiler._k8s_helper import sanitize_k8s_name
 
 from .auth import IAP_CLIENT_ID
 from .utils import clean_name, is_mlflow_enabled
@@ -41,6 +42,13 @@ class PipelineGenerator(object):
         self.run_config = config.run_config
         self.catalog = context.config_loader.get("catalog*")
 
+    def configure_max_cache_staleness(self, kfp_ops):
+        if self.run_config.max_cache_staleness not in [None, ""]:
+            for _, op in kfp_ops.items():
+                op.execution_options.caching_strategy.max_cache_staleness = (
+                    self.run_config.max_cache_staleness
+                )
+
     def generate_pipeline(self, pipeline, image, image_pull_policy):
         @dsl.pipeline(
             name=self.project_name,
@@ -55,17 +63,19 @@ class PipelineGenerator(object):
             node_dependencies = self.context.pipelines.get(
                 pipeline
             ).node_dependencies
-            with self._create_pipeline_exit_handler():
+            with self._create_pipeline_exit_handler(pipeline):
                 kfp_ops = self._build_kfp_ops(
-                    node_dependencies, image, image_pull_policy
+                    pipeline, node_dependencies, image, image_pull_policy
                 )
+
+                self.configure_max_cache_staleness(kfp_ops)
                 for node, dependencies in node_dependencies.items():
                     for dependency in dependencies:
                         kfp_ops[node.name].after(kfp_ops[dependency.name])
 
         return convert_kedro_pipeline_to_kfp
 
-    def _create_pipeline_exit_handler(self):
+    def _create_pipeline_exit_handler(self, pipeline):
         enable_volume_cleaning = (
             self.run_config.volume is not None
             and not self.run_config.volume.keep
@@ -74,25 +84,32 @@ class PipelineGenerator(object):
         if not enable_volume_cleaning:
             return contextlib.nullcontext()
 
-        return dsl.ExitHandler(
-            dsl.ContainerOp(
-                name="schedule-volume-termination",
-                image="gcr.io/cloud-builders/kubectl",
-                command=[
-                    "kubectl",
-                    "delete",
-                    "pvc",
-                    "{{workflow.name}}-data-volume",
-                    "--wait=false",
-                    "--ignore-not-found",
-                    "--output",
-                    "name",
-                ],
-            )
+        exit_container_op = dsl.ContainerOp(
+            name="schedule-volume-termination",
+            image="gcr.io/cloud-builders/kubectl",
+            command=[
+                "kubectl",
+                "delete",
+                "pvc",
+                "{{workflow.name}}-"
+                + sanitize_k8s_name(f"{pipeline}-data-volume"),
+                "--wait=false",
+                "--ignore-not-found",
+                "--output",
+                "name",
+            ],
         )
 
+        if self.run_config.max_cache_staleness not in [None, ""]:
+            exit_container_op.execution_options.caching_strategy.max_cache_staleness = (
+                self.run_config.max_cache_staleness
+            )
+
+        return dsl.ExitHandler(exit_container_op)
+
     def _build_kfp_ops(
         self,
+        pipeline,
         node_dependencies: Dict[Node, Set[Node]],
         image,
         image_pull_policy,
@@ -101,7 +118,9 @@ class PipelineGenerator(object):
         kfp_ops = {}
 
         node_volumes = (
-            self._setup_volumes(image, image_pull_policy)
+            self._setup_volumes(
+                f"{pipeline}-data-volume", image, image_pull_policy
+            )
             if self.run_config.volume is not None
             else {}
         )
@@ -159,6 +178,8 @@ class PipelineGenerator(object):
                         "run",
                         "--params",
                         params,
+                        "--pipeline",
+                        pipeline,
                         "--node",
                         node.name,
                     ],
@@ -185,14 +206,21 @@ class PipelineGenerator(object):
             )
         return op
 
-    def _setup_volumes(self, image, image_pull_policy):
+    def _setup_volumes(self, volume_name, image, image_pull_policy):
         vop = dsl.VolumeOp(
             name="data-volume-create",
-            resource_name="data-volume",
+            resource_name=volume_name,
             size=self.run_config.volume.size,
             modes=self.run_config.volume.access_modes,
             storage_class=self.run_config.volume.storageclass,
         )
+
+        if self.run_config.max_cache_staleness not in [None, ""]:
+            vop.add_pod_annotation(
+                "pipelines.kubeflow.org/max_cache_staleness",
+                self.run_config.max_cache_staleness,
+            )
+
         if self.run_config.volume.skip_init:
             return {"/home/kedro/data": vop.volume}
         else:
diff --git a/kedro_kubeflow/kfpclient.py b/kedro_kubeflow/kfpclient.py
index 8d2789f..d845ca1 100644
--- a/kedro_kubeflow/kfpclient.py
+++ b/kedro_kubeflow/kfpclient.py
@@ -19,9 +19,20 @@ class KubeflowClient(object):
     log = logging.getLogger(__name__)
 
     def __init__(self, config, project_name, context):
+        client_params = {}
         token = AuthHandler().obtain_id_token()
+        if token is not None:
+            client_params = {"existing_token": token}
+        dex_authservice_session = AuthHandler().obtain_dex_authservice_session(
+            kfp_api=config.host,
+        )
+        if dex_authservice_session is not None:
+            client_params = {
+                "cookies": f"authservice_session={dex_authservice_session}"
+            }
         self.host = config.host
-        self.client = Client(self.host, existing_token=token)
+        self.client = Client(host=self.host, **client_params)
+
         self.project_name = project_name
         self.pipeline_description = config.run_config.description
         self.generator = PipelineGenerator(config, project_name, context)
@@ -37,6 +48,7 @@ class KubeflowClient(object):
         pipeline,
         image,
         experiment_name,
+        experiment_namespace,
         run_name,
         wait,
         image_pull_policy="IfNotPresent",
@@ -47,6 +59,7 @@ class KubeflowClient(object):
             ),
             arguments={},
             experiment_name=experiment_name,
+            namespace=experiment_namespace,
             run_name=run_name,
         )
 
@@ -126,23 +139,28 @@ class KubeflowClient(object):
             )
             return (pipeline.id, pipeline.default_version.id)
 
-    def _ensure_experiment_exists(self, experiment_name):
+    def _ensure_experiment_exists(self, experiment_name, experiment_namespace):
         try:
             experiment = self.client.get_experiment(
-                experiment_name=experiment_name
+                experiment_name=experiment_name,
+                namespace=experiment_namespace,
             )
             self.log.info(f"Existing experiment found: {experiment.id}")
         except ValueError as e:
             if not str(e).startswith("No experiment is found"):
                 raise
 
-            experiment = self.client.create_experiment(experiment_name)
+            experiment = self.client.create_experiment(
+                experiment_name, namespace=experiment_namespace
+            )
             self.log.info(f"New experiment created: {experiment.id}")
 
         return experiment.id
 
-    def schedule(self, experiment_name, cron_expression):
-        experiment_id = self._ensure_experiment_exists(experiment_name)
+    def schedule(self, experiment_name, experiment_namespace, cron_expression):
+        experiment_id = self._ensure_experiment_exists(
+            experiment_name, experiment_namespace
+        )
         pipeline_id = self._get_pipeline_id(self.project_name)
         self._disable_runs(experiment_id, pipeline_id)
         self.client.create_recurring_run(
diff --git a/setup.py b/setup.py
index 9187cd6..4e990b4 100644
--- a/setup.py
+++ b/setup.py
@@ -22,6 +22,7 @@ EXTRA_REQUIRE = {
         "pytest-cov>=2.8.0, <3.0.0",
         "tox==3.23.1",
         "pre-commit==2.9.3",
+        "responses>=0.13.4",
     ],
     "docs": [
         "sphinx==3.4.2",


Test patch:
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 57a37c0..8f140db 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -2,6 +2,7 @@ import os
 import unittest
 from unittest.mock import patch
 
+import responses
 from google.auth.exceptions import DefaultCredentialsError
 
 from kedro_kubeflow.auth import AuthHandler
@@ -55,3 +56,60 @@ class TestAuthHandler(unittest.TestCase):
 
         # then
         assert token == "TOKEN"
+
+    def test_should_skip_dex_auth_if_env_is_not_set(self):
+        # given
+        # no env set
+
+        # when
+        session = AuthHandler().obtain_dex_authservice_session(None)
+
+        # then
+        assert session is None
+
+    def test_should_skip_dex_auth_if_env_is_incomplete(self):
+        # given
+        os.environ["DEX_USERNAME"] = "user@example.com"
+        # no password set
+
+        # when
+        session = AuthHandler().obtain_dex_authservice_session(None)
+
+        # then
+        assert session is None
+
+    def tearDown(self):
+        if "DEX_USERNAME" in os.environ:
+            del os.environ["DEX_USERNAME"]
+        if "DEX_PASSWORD" in os.environ:
+            del os.environ["DEX_PASSWORD"]
+        if "IAP_CLIENT_ID" in os.environ:
+            del os.environ["IAP_CLIENT_ID"]
+
+    @responses.activate
+    def test_should_get_cookie_from_dex_secured_system(self):
+        # given
+        os.environ["DEX_USERNAME"] = "user@example.com"
+        os.environ["DEX_PASSWORD"] = "pa$$"
+        responses.add(
+            responses.GET,
+            "https://kubeflow.local/pipeline",
+            body='<a href="/dex/auth/local?req=qjrrnpg3hngdu6odii3hcmfae" target="_self"',
+        )
+        responses.add(
+            responses.POST,
+            "https://kubeflow.local/dex/auth/local?req=qjrrnpg3hngdu6odii3hcmfae",
+            headers={"Set-cookie": "authservice_session=sessionID"},
+        )
+
+        # when
+        session = AuthHandler().obtain_dex_authservice_session(
+            "https://kubeflow.local/pipeline"
+        )
+
+        # then
+        assert session == "sessionID"
+        assert (
+            responses.calls[1].request.body
+            == "login=user%40example.com&password=pa%24%24"
+        )
diff --git a/tests/test_cli.py b/tests/test_cli.py
index ee026ef..070eb1a 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -57,7 +57,16 @@ class TestPluginCLI(unittest.TestCase):
         runner = CliRunner()
 
         result = runner.invoke(
-            run_once, ["-i", "new_img", "-p", "new_pipe"], obj=config
+            run_once,
+            [
+                "-i",
+                "new_img",
+                "-p",
+                "new_pipe",
+                "--experiment-namespace",
+                "my-ns",
+            ],
+            obj=config,
         )
 
         assert result.exit_code == 0
@@ -68,6 +77,7 @@ class TestPluginCLI(unittest.TestCase):
             pipeline="new_pipe",
             run_name="test run",
             wait=True,
+            experiment_namespace="my-ns",
         )
 
     @patch("webbrowser.open_new_tab")
@@ -127,7 +137,7 @@ class TestPluginCLI(unittest.TestCase):
 
         assert result.exit_code == 0
         context_helper.kfp_client.schedule.assert_called_with(
-            "test_experiment", "* * *"
+            "test_experiment", None, "* * *"
         )
 
     @patch.object(Path, "cwd")
diff --git a/tests/test_generator.py b/tests/test_generator.py
index 7ef49e0..7d2d72f 100644
--- a/tests/test_generator.py
+++ b/tests/test_generator.py
@@ -204,6 +204,8 @@ class TestGenerator(unittest.TestCase):
                 "--params",
                 "param1:{{pipelineparam:op=;name=param1}},"
                 "param2:{{pipelineparam:op=;name=param2}}",
+                "--pipeline",
+                "pipeline",
                 "--node",
                 node_name,
             ]
diff --git a/tests/test_kfpclient.py b/tests/test_kfpclient.py
index b21ce86..521ecf9 100644
--- a/tests/test_kfpclient.py
+++ b/tests/test_kfpclient.py
@@ -86,6 +86,7 @@ class TestKubeflowClient(unittest.TestCase):
             image="unittest-image",
             experiment_name="experiment",
             wait=False,
+            experiment_namespace="exp_namespace",
         )
 
         # then
@@ -99,6 +100,7 @@ class TestKubeflowClient(unittest.TestCase):
             "arguments": {},
             "experiment_name": "experiment",
             "run_name": "unittest",
+            "namespace": "exp_namespace",
         }
 
     def test_should_run_pipeline_and_wait(self):
@@ -115,6 +117,7 @@ class TestKubeflowClient(unittest.TestCase):
             image="unittest-image",
             experiment_name="experiment",
             wait=True,
+            experiment_namespace=None,
         )
 
         # then
@@ -143,6 +146,33 @@ class TestKubeflowClient(unittest.TestCase):
         auth_handler_mock.return_value.obtain_id_token.return_value = (
             "unittest-token"
         )
+        auth_handler_mock.return_value.obtain_dex_authservice_session.return_value = (
+            None
+        )
+
+        # when
+        self.client_under_test = KubeflowClient(
+            PluginConfig({"host": "http://unittest", "run_config": {}}),
+            None,
+            None,
+        )
+
+        # then
+        kfp_client_mock.assert_called_with(
+            host="http://unittest", existing_token="unittest-token"
+        )
+
+    @patch("kedro_kubeflow.kfpclient.AuthHandler")
+    @patch("kedro_kubeflow.kfpclient.PipelineGenerator")
+    @patch("kedro_kubeflow.kfpclient.Client")
+    def test_should_use_dex_session_in_kfp_client(
+        self, kfp_client_mock, pipeline_generator_mock, auth_handler_mock
+    ):
+        # given
+        auth_handler_mock.return_value.obtain_id_token.return_value = None
+        auth_handler_mock.return_value.obtain_dex_authservice_session.return_value = (
+            "session_id"
+        )
 
         # when
         self.client_under_test = KubeflowClient(
@@ -153,7 +183,7 @@ class TestKubeflowClient(unittest.TestCase):
 
         # then
         kfp_client_mock.assert_called_with(
-            "http://unittest", existing_token="unittest-token"
+            host="http://unittest", cookies="authservice_session=session_id"
         )
 
     def test_should_schedule_pipeline(self):
@@ -170,6 +200,7 @@ class TestKubeflowClient(unittest.TestCase):
         self.client_under_test.schedule(
             experiment_name="EXPERIMENT",
             cron_expression="0 * * * * *",
+            experiment_namespace=None,
         )
 
         # then
@@ -199,6 +230,7 @@ class TestKubeflowClient(unittest.TestCase):
         self.client_under_test.schedule(
             experiment_name="EXPERIMENT",
             cron_expression="0 * * * * *",
+            experiment_namespace=None,
         )
 
         # then
@@ -228,6 +260,7 @@ class TestKubeflowClient(unittest.TestCase):
         self.client_under_test.schedule(
             experiment_name="EXPERIMENT",
             cron_expression="0 * * * * *",
+            experiment_namespace=None,
         )
 
         # then