diff --git a/sdk/python/kfp/kubeflow_client/__init__.py b/sdk/python/kfp/kubeflow_client/__init__.py new file mode 100644 index 00000000000..968292d7133 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/__init__.py @@ -0,0 +1,45 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Kubeflow PipelinesClient — simplified, name-first KFP interface. + +Can be used directly via ``from kfp.kubeflow_client import +PipelinesClient`` or through the Kubeflow SDK re-export at +``kubeflow.pipelines``. +""" + +from kfp.kubeflow_client import constants +from kfp.kubeflow_client.api.pipelines_client import PipelinesClient +from kfp.kubeflow_client.backends.kubernetes import KubernetesBackendConfig +from kfp.kubeflow_client.types import Experiment +from kfp.kubeflow_client.types import ListExperimentsResponse +from kfp.kubeflow_client.types import ListPipelinesResponse +from kfp.kubeflow_client.types import ListPipelineVersionsResponse +from kfp.kubeflow_client.types import ListRunsResponse +from kfp.kubeflow_client.types import Pipeline +from kfp.kubeflow_client.types import PipelineVersion +from kfp.kubeflow_client.types import Run + +__all__ = [ + 'constants', + 'Experiment', + 'KubernetesBackendConfig', + 'ListExperimentsResponse', + 'ListPipelinesResponse', + 'ListPipelineVersionsResponse', + 'ListRunsResponse', + 'Pipeline', + 'PipelinesClient', + 'PipelineVersion', + 'Run', +] diff --git a/sdk/python/kfp/kubeflow_client/api/__init__.py b/sdk/python/kfp/kubeflow_client/api/__init__.py new file mode 100644 index 00000000000..4b777c68394 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/api/__init__.py @@ -0,0 +1,18 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Client API package.""" + +from kfp.kubeflow_client.api.pipelines_client import PipelinesClient + +__all__ = ['PipelinesClient'] diff --git a/sdk/python/kfp/kubeflow_client/api/pipelines_client.py b/sdk/python/kfp/kubeflow_client/api/pipelines_client.py new file mode 100644 index 00000000000..b72e4f75bee --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/api/pipelines_client.py @@ -0,0 +1,432 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""PipelinesClient — simplified, name-first client for Kubeflow Pipelines. + +This module provides a streamlined interface over the KFP backend API, +designed for re-export by the Kubeflow SDK at ``kubeflow.pipelines``. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any, TYPE_CHECKING + +if TYPE_CHECKING: + from kfp.client import Client + +from kfp.kubeflow_client.backends.kubernetes import KubernetesBackend +from kfp.kubeflow_client.backends.kubernetes import KubernetesBackendConfig +from kfp.kubeflow_client.types import Experiment +from kfp.kubeflow_client.types import ListExperimentsResponse +from kfp.kubeflow_client.types import ListPipelinesResponse +from kfp.kubeflow_client.types import ListPipelineVersionsResponse +from kfp.kubeflow_client.types import ListRunsResponse +from kfp.kubeflow_client.types import Pipeline +from kfp.kubeflow_client.types import PipelineVersion +from kfp.kubeflow_client.types import Run + +__all__ = ['PipelinesClient'] + + +class PipelinesClient: + """Simplified, name-first client for Kubeflow Pipelines. + + Provides the core author → compile → upload → run → monitor workflow + from a single import. Designed to be re-exported by the Kubeflow SDK + at ``kubeflow.pipelines.PipelinesClient``. + + Args: + backend_config: Connection parameters for the KFP API server. + When ``None``, uses ``KubernetesBackendConfig()`` (zero-arg + construction with auto-discovery). + """ + + def __init__( + self, + backend_config: KubernetesBackendConfig | None = None, + ) -> None: + if backend_config is None: + backend_config = KubernetesBackendConfig() + self._backend = KubernetesBackend(backend_config) + self._kfp_client_instance = None + + # ------------------------------------------------------------------ + # Pipeline operations + # ------------------------------------------------------------------ + + def upload_pipeline( + self, + pipeline: Callable | str, + *, + name: str | None = None, + version: str | None = None, + description: str | None = None, + ) -> PipelineVersion: + """Upload a pipeline (or new version) to the server. + + Handles callable functions, file paths, new pipelines, and new + versions through a single unified method. + + Note: + When creating a **new** pipeline with an explicit ``version`` + name, the server auto-generates the first version name during + upload. A best-effort rename is attempted afterward; if the rename + fails (e.g. due to permissions), a warning is logged and the + returned version retains the server-generated name. + + Args: + pipeline: A ``@dsl.pipeline``-decorated function or a path to a + compiled pipeline YAML file. + name: Display name for the pipeline. If omitted, auto-generated + from the function's ``@dsl.pipeline(name=...)`` value or + the filename without extension. + version: Version label. If omitted, auto-generated. Calling + ``upload_pipeline`` again with the same ``name`` and no + explicit ``version`` creates a new version each time. + description: Pipeline description. + + Returns: + A ``PipelineVersion`` object representing the uploaded version. + """ + return self._backend.upload_pipeline( + pipeline, name=name, version_name=version, description=description) + + def get_pipeline(self, name: str) -> Pipeline: + """Get a pipeline by name. + + Args: + name: Pipeline display name. + + Returns: + A ``Pipeline`` object. + + Raises: + ValueError: If no pipeline matches or multiple pipelines match. + """ + return self._backend.get_pipeline(name) + + def get_pipeline_version( + self, + name: str, + version: str, + ) -> PipelineVersion: + """Get a specific pipeline version by pipeline name and version name. + + Args: + name: Pipeline display name. + version: Version display name. + + Returns: + A ``PipelineVersion`` object. + + Raises: + ValueError: If the pipeline or version is not found. + """ + return self._backend.get_pipeline_version(name, version) + + def list_pipelines( + self, + *, + page_token: str = '', + page_size: int = 10, + ) -> ListPipelinesResponse: + """List pipelines available on the server. + + Args: + page_token: Token for obtaining the next page. + page_size: Number of results per page. + + Returns: + A ``ListPipelinesResponse`` with ``.pipelines`` and + ``.next_page_token``. + """ + return self._backend.list_pipelines( + page_token=page_token, page_size=page_size) + + def list_pipeline_versions( + self, + name: str, + *, + page_token: str = '', + page_size: int = 10, + ) -> ListPipelineVersionsResponse: + """List versions of a pipeline by name. + + Args: + name: Pipeline display name. + page_token: Token for obtaining the next page. + page_size: Number of results per page. + + Returns: + A ``ListPipelineVersionsResponse`` with ``.pipeline_versions`` + and ``.next_page_token``. + """ + return self._backend.list_pipeline_versions( + name, page_token=page_token, page_size=page_size) + + def delete_pipeline( + self, + name: str, + *, + version: str | None = None, + force: bool = False, + ) -> None: + """Delete a pipeline or a specific pipeline version. + + Args: + name: Pipeline display name. + version: If provided, delete only this version. If ``None``, + delete the entire pipeline and all versions. + force: When deleting an entire pipeline, required if the pipeline + has more than one version. Ignored when ``version`` is set. + + Raises: + ValueError: If the pipeline has multiple versions and + ``force=False``. + """ + self._backend.delete_pipeline(name, version=version, force=force) + + # ------------------------------------------------------------------ + # Run operations + # ------------------------------------------------------------------ + + def run( + self, + pipeline: str | Callable | Pipeline | PipelineVersion, + *, + params: dict[str, Any] | None = None, + name: str | None = None, + experiment: str | None = None, + version: str | None = None, + ) -> Run: + """Run a pipeline. + + Supports multiple input types: + - A pipeline name (``str`` without file extension): resolves the + uploaded pipeline on the server. + - A path to a compiled YAML file (``str`` ending in ``.yaml``/ + ``.yml``): compile-and-submit inline, no upload. + - A ``@dsl.pipeline``-decorated callable: compile-and-submit inline. + - A ``Pipeline`` or ``PipelineVersion`` object (from + ``get_pipeline``/``upload_pipeline``). + + Note: + String inputs are classified as file paths when they end in + ``.yaml`` or ``.yml``. If you have an uploaded pipeline whose + display name ends with such an extension, pass the ``Pipeline`` + object from ``get_pipeline()`` instead. + + Args: + pipeline: Pipeline to run (see above). + params: Pipeline parameters as a dict. + name: Run display name. Auto-generated if omitted. + experiment: Experiment name. If ``None``, the server's default + experiment is used. If provided and the experiment does not + exist, raises ``ValueError``. + version: Pipeline version name (used when ``pipeline`` is a + name string or a ``Pipeline`` object). Uses latest version + if omitted. + + Returns: + A ``Run`` object. + """ + return self._backend.run( + pipeline, + params=params, + name=name, + experiment=experiment, + version=version, + ) + + def get_run(self, run_id: str) -> Run: + """Get a run by ID. + + Args: + run_id: The run identifier. + + Returns: + A ``Run`` object. + """ + return self._backend.get_run(run_id) + + def list_runs( + self, + *, + pipeline: str | None = None, + experiment: str | None = None, + status: str | None = None, + page_token: str = '', + page_size: int = 10, + ) -> ListRunsResponse: + """List runs, optionally filtered by pipeline, experiment, or status. + + Note: + The ``pipeline`` filter is applied client-side because the KFP + v2beta1 API does not support server-side filtering by pipeline ID. + When used, the returned page may contain fewer items than + ``page_size`` — including zero items with a non-empty + ``next_page_token`` if all runs on that server page belong to + other pipelines. Callers should continue paginating until + ``next_page_token`` is empty. + + Args: + pipeline: Filter by pipeline display name (client-side). + experiment: Filter by experiment display name. + status: Filter by run state (e.g. ``"succeeded"``). + page_token: Token for obtaining the next page. + page_size: Number of results per page. + + Returns: + A ``ListRunsResponse`` with ``.runs`` and ``.next_page_token``. + """ + return self._backend.list_runs( + pipeline=pipeline, + experiment=experiment, + status=status, + page_token=page_token, + page_size=page_size, + ) + + def wait_for_run_status( + self, + run: str | Run, + *, + status: set[str] | None = None, + timeout: int | None = 600, + polling_interval: int = 5, + callbacks: list[Callable[[Run], None]] | None = None, + ) -> Run: + """Wait for a run to reach a target state. + + Args: + run: A ``Run`` object or a run ID string. + status: Set of states to wait for. Defaults to + ``{constants.RUN_COMPLETE}`` (``"succeeded"``). The wait + always exits immediately on any terminal state regardless + of this parameter. + timeout: Maximum seconds to wait. Defaults to 600 (10 minutes). + Pass ``None`` to wait indefinitely. + polling_interval: Seconds between status checks. + callbacks: Called with the final ``Run`` object when the wait + ends (on any stop condition). + + Returns: + The ``Run`` object at the time the wait concluded. + + Raises: + TimeoutError: If ``timeout`` expires before reaching a stop + condition. + """ + return self._backend.wait_for_run_status( + run, + status=status, + timeout=timeout, + polling_interval=polling_interval, + callbacks=callbacks, + ) + + # ------------------------------------------------------------------ + # Experiment operations + # ------------------------------------------------------------------ + + def create_experiment( + self, + name: str, + *, + description: str | None = None, + ) -> Experiment: + """Create a new experiment. + + If an experiment with the given name already exists, returns it. + + Args: + name: Experiment display name. + description: Experiment description. + + Returns: + An ``Experiment`` object. + """ + return self._backend.create_experiment(name, description=description) + + def get_experiment(self, name: str) -> Experiment: + """Get an experiment by name. + + Args: + name: Experiment display name. + + Returns: + An ``Experiment`` object. + + Raises: + ValueError: If no experiment with that name is found. + """ + return self._backend.get_experiment(name) + + def list_experiments( + self, + *, + page_token: str = '', + page_size: int = 10, + ) -> ListExperimentsResponse: + """List experiments. + + Args: + page_token: Token for obtaining the next page. + page_size: Number of results per page. + + Returns: + A ``ListExperimentsResponse`` with ``.experiments`` and + ``.next_page_token``. + """ + return self._backend.list_experiments( + page_token=page_token, page_size=page_size) + + def delete_experiment(self, name: str) -> None: + """Delete an experiment by name. + + Args: + name: Experiment display name. + + Raises: + ValueError: If no experiment with that name is found. + """ + self._backend.delete_experiment(name) + + # ------------------------------------------------------------------ + # Escape hatch + # ------------------------------------------------------------------ + + @property + def kfp_client(self) -> Client: + """Access the underlying ``kfp.Client`` for advanced operations. + + Lazily constructed on first access, sharing connection + parameters from ``KubernetesBackendConfig``. + """ + if self._kfp_client_instance is None: + from kfp.client import Client + config = self._backend.config + kwargs: dict[str, Any] = {} + if config.base_url: + kwargs['host'] = config.base_url + if config.user_token: + kwargs['existing_token'] = config.user_token + if config.namespace: + kwargs['namespace'] = config.namespace + if config.custom_ca: + kwargs['ssl_ca_cert'] = config.custom_ca + if config.is_secure is not None: + kwargs['verify_ssl'] = config.is_secure + self._kfp_client_instance = Client(**kwargs) + return self._kfp_client_instance diff --git a/sdk/python/kfp/kubeflow_client/api/pipelines_client_test.py b/sdk/python/kfp/kubeflow_client/api/pipelines_client_test.py new file mode 100644 index 00000000000..1906965fc7a --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/api/pipelines_client_test.py @@ -0,0 +1,463 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for PipelinesClient (delegation to backend).""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field +from typing import Any +from unittest.mock import Mock +from unittest.mock import patch + +from kfp.kubeflow_client import constants +from kfp.kubeflow_client.api.pipelines_client import PipelinesClient +from kfp.kubeflow_client.backends.kubernetes.types import \ + KubernetesBackendConfig +import kfp_server_api +import pytest + +_AUTH_MODULE = 'kfp.kubeflow_client.backends.kubernetes.auth' +_BACKEND_MODULE = 'kfp.kubeflow_client.backends.kubernetes.backend' + +SUCCESS = 'success' +FAILED = 'failed' + + +@dataclass +class TestCase: + """A single test scenario for parametrized tests.""" + + name: str + expected_status: str = SUCCESS + config: dict[str, Any] = field(default_factory=dict) + expected_output: Any | None = None + expected_error: type[Exception] | None = None + expected_error_match: str | None = None + __test__ = False + + +# ------------------------------------------------------------------ +# Fixtures +# ------------------------------------------------------------------ + + +@pytest.fixture +def client(): + with patch(f'{_AUTH_MODULE}.apply_in_cluster_credentials'), \ + patch(f'{_BACKEND_MODULE}.KubernetesBackend.verify_backend'): + return PipelinesClient( + backend_config=KubernetesBackendConfig( + base_url='http://localhost:8888', + namespace='test-ns', + )) + + +# ------------------------------------------------------------------ +# test_upload_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend upload_pipeline', + expected_output={'pipeline_version_id': 'vid-1'}, + ), + ], + ids=lambda tc: tc.name) +def test_upload_pipeline(client, test_case): + mock_version = Mock(pipeline_version_id='vid-1') + with patch.object( + client._backend, 'upload_pipeline', + return_value=mock_version) as mock_upload: + result = client.upload_pipeline( + lambda: None, name='my-pipe', version='v1', description='desc') + mock_upload.assert_called_once() + call_kwargs = mock_upload.call_args[1] + assert call_kwargs['name'] == 'my-pipe' + assert call_kwargs['version_name'] == 'v1' + assert call_kwargs['description'] == 'desc' + assert result.pipeline_version_id == test_case.expected_output[ + 'pipeline_version_id'] + + +# ------------------------------------------------------------------ +# test_get_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'pipeline_id': 'pid-1'}, + ), + TestCase( + name='not found raises ValueError', + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline not found', + ), + ], + ids=lambda tc: tc.name) +def test_get_pipeline(client, test_case): + if test_case.expected_status == SUCCESS: + mock_pipeline = Mock(pipeline_id='pid-1') + with patch.object( + client._backend, 'get_pipeline', + return_value=mock_pipeline) as mock_get: + result = client.get_pipeline('my-pipe') + mock_get.assert_called_once_with('my-pipe') + assert result.pipeline_id == test_case.expected_output[ + 'pipeline_id'] + else: + with patch.object( + client._backend, + 'get_pipeline', + side_effect=ValueError('Pipeline not found')): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + client.get_pipeline('nonexistent') + + +# ------------------------------------------------------------------ +# test_get_pipeline_version +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'display_name': 'v1'}, + ), + TestCase( + name='not found raises ValueError', + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='version not found', + ), + ], + ids=lambda tc: tc.name) +def test_get_pipeline_version(client, test_case): + if test_case.expected_status == SUCCESS: + mock_pv = Mock(display_name='v1') + with patch.object( + client._backend, 'get_pipeline_version', + return_value=mock_pv) as mock_get: + result = client.get_pipeline_version('my-pipe', 'v1') + mock_get.assert_called_once_with('my-pipe', 'v1') + assert result.display_name == test_case.expected_output[ + 'display_name'] + else: + with patch.object( + client._backend, + 'get_pipeline_version', + side_effect=ValueError('version not found')): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + client.get_pipeline_version('my-pipe', 'bad-ver') + + +# ------------------------------------------------------------------ +# test_list_pipelines +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'next_page_token': 'tok'}, + ), + ], + ids=lambda tc: tc.name) +def test_list_pipelines(client, test_case): + mock_response = Mock(pipelines=[Mock()], next_page_token='tok') + with patch.object( + client._backend, 'list_pipelines', + return_value=mock_response) as mock_list: + result = client.list_pipelines(page_size=5) + mock_list.assert_called_once_with(page_token='', page_size=5) + assert result.next_page_token == test_case.expected_output[ + 'next_page_token'] + + +# ------------------------------------------------------------------ +# test_list_pipeline_versions +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'count': 2}, + ), + ], + ids=lambda tc: tc.name) +def test_list_pipeline_versions(client, test_case): + mock_resp = Mock(pipeline_versions=[Mock(), Mock()], next_page_token='t2') + with patch.object( + client._backend, 'list_pipeline_versions', + return_value=mock_resp) as mock_list: + result = client.list_pipeline_versions('my-pipe') + mock_list.assert_called_once_with( + 'my-pipe', page_token='', page_size=10) + assert len( + result.pipeline_versions) == test_case.expected_output['count'] + + +# ------------------------------------------------------------------ +# test_delete_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + config={ + 'version': 'v1', + 'force': True + }, + ), + ], + ids=lambda tc: tc.name) +def test_delete_pipeline(client, test_case): + with patch.object(client._backend, 'delete_pipeline') as mock_del: + client.delete_pipeline('my-pipe', version='v1', force=True) + mock_del.assert_called_once_with('my-pipe', version='v1', force=True) + + +# ------------------------------------------------------------------ +# test_run +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend run', + expected_output={'run_id': 'r-1'}, + ), + ], + ids=lambda tc: tc.name) +def test_run(client, test_case): + mock_run = Mock(run_id='r-1') + with patch.object( + client._backend, 'run', return_value=mock_run) as mock_backend_run: + result = client.run( + 'my-pipe', params={'x': '1'}, name='run-1', version='v2') + mock_backend_run.assert_called_once_with( + 'my-pipe', + params={'x': '1'}, + name='run-1', + experiment=None, + version='v2', + ) + assert result.run_id == test_case.expected_output['run_id'] + + +# ------------------------------------------------------------------ +# test_get_run +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'run_id': 'r-1'}, + ), + ], + ids=lambda tc: tc.name) +def test_get_run(client, test_case): + mock_run = Mock(run_id='r-1') + with patch.object( + client._backend, 'get_run', return_value=mock_run) as mock_get: + result = client.get_run('r-1') + mock_get.assert_called_once_with('r-1') + assert result.run_id == test_case.expected_output['run_id'] + + +# ------------------------------------------------------------------ +# test_list_runs +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'count': 1}, + ), + ], + ids=lambda tc: tc.name) +def test_list_runs(client, test_case): + mock_response = Mock(runs=[Mock(run_id='r-1')], next_page_token='t2') + with patch.object( + client._backend, 'list_runs', + return_value=mock_response) as mock_list: + result = client.list_runs( + pipeline='my-pipe', status='succeeded', page_size=20) + mock_list.assert_called_once_with( + pipeline='my-pipe', + experiment=None, + status='succeeded', + page_token='', + page_size=20, + ) + assert len(result.runs) == test_case.expected_output['count'] + + +# ------------------------------------------------------------------ +# test_wait_for_run_status +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'state': 'SUCCEEDED'}, + ), + ], + ids=lambda tc: tc.name) +def test_wait_for_run_status(client, test_case): + mock_run = Mock(run_id='r-1', state='SUCCEEDED') + with patch.object( + client._backend, 'wait_for_run_status', + return_value=mock_run) as mock_wait: + result = client.wait_for_run_status( + 'r-1', status={constants.RUN_COMPLETE}, timeout=30) + mock_wait.assert_called_once_with( + 'r-1', + status={constants.RUN_COMPLETE}, + timeout=30, + polling_interval=5, + callbacks=None, + ) + assert result.state == test_case.expected_output['state'] + + +# ------------------------------------------------------------------ +# test_create_experiment +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'experiment_id': 'exp-1'}, + ), + ], + ids=lambda tc: tc.name) +def test_create_experiment(client, test_case): + mock_exp = Mock(experiment_id='exp-1') + with patch.object( + client._backend, 'create_experiment', + return_value=mock_exp) as mock_create: + result = client.create_experiment('my-exp', description='desc') + mock_create.assert_called_once_with('my-exp', description='desc') + assert result.experiment_id == test_case.expected_output[ + 'experiment_id'] + + +# ------------------------------------------------------------------ +# test_get_experiment +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'experiment_id': 'exp-1'}, + ), + ], + ids=lambda tc: tc.name) +def test_get_experiment(client, test_case): + mock_exp = Mock(experiment_id='exp-1') + with patch.object( + client._backend, 'get_experiment', + return_value=mock_exp) as mock_get: + result = client.get_experiment('my-exp') + mock_get.assert_called_once_with('my-exp') + assert result.experiment_id == test_case.expected_output[ + 'experiment_id'] + + +# ------------------------------------------------------------------ +# test_list_experiments +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='delegates to backend', + expected_output={'next_page_token': 'tok2'}, + ), + ], + ids=lambda tc: tc.name) +def test_list_experiments(client, test_case): + mock_response = Mock( + experiments=[Mock(experiment_id='e-1')], next_page_token='tok2') + with patch.object( + client._backend, 'list_experiments', + return_value=mock_response) as mock_list: + result = client.list_experiments(page_size=5) + mock_list.assert_called_once_with(page_token='', page_size=5) + assert result.next_page_token == test_case.expected_output[ + 'next_page_token'] + + +# ------------------------------------------------------------------ +# test_delete_experiment +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase(name='delegates to backend',), + ], + ids=lambda tc: tc.name) +def test_delete_experiment(client, test_case): + with patch.object(client._backend, 'delete_experiment') as mock_del: + client.delete_experiment('my-exp') + mock_del.assert_called_once_with('my-exp') + + +# ------------------------------------------------------------------ +# test_kfp_client +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase(name='creates instance once (cached property)',), + ], + ids=lambda tc: tc.name) +def test_kfp_client(client, test_case): + with patch('kfp.client.Client') as MockClient: + MockClient.return_value = Mock() + first = client.kfp_client + second = client.kfp_client + assert first is second + MockClient.assert_called_once() diff --git a/sdk/python/kfp/kubeflow_client/backends/__init__.py b/sdk/python/kfp/kubeflow_client/backends/__init__.py new file mode 100644 index 00000000000..a266fb61dc7 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/__init__.py @@ -0,0 +1,13 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/__init__.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/__init__.py new file mode 100644 index 00000000000..fa9714ef1a4 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/__init__.py @@ -0,0 +1,20 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Kubernetes backend package.""" + +from kfp.kubeflow_client.backends.kubernetes.backend import KubernetesBackend +from kfp.kubeflow_client.backends.kubernetes.types import \ + KubernetesBackendConfig + +__all__ = ['KubernetesBackend', 'KubernetesBackendConfig'] diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/auth.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/auth.py new file mode 100644 index 00000000000..fc1a9d19011 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/auth.py @@ -0,0 +1,69 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Authentication helpers for the Kubernetes backend.""" + +from __future__ import annotations + +import logging +import os + +from kfp.kubeflow_client.backends.kubernetes import constants +import kfp_server_api + +logger = logging.getLogger(__name__) + + +def apply_in_cluster_credentials( + api_config: kfp_server_api.Configuration,) -> None: + """Apply default in-cluster service account credentials.""" + token_path = os.environ.get(constants.TOKEN_PATH_ENV) + if not token_path: + if os.path.exists(constants.KFP_SA_TOKEN_PATH): + token_path = constants.KFP_SA_TOKEN_PATH + elif os.path.exists(constants.K8S_SA_TOKEN_PATH): + token_path = constants.K8S_SA_TOKEN_PATH + logger.debug( + 'Kubeflow pipelines token path missing; using Kubernetes ' + 'service account token at %s for API auth.', token_path) + + if not token_path: + logger.debug( + 'No in-cluster token file found; skipping credential setup.') + return + + try: + from kfp.client.set_volume_credentials import \ + ServiceAccountTokenVolumeCredentials + except ImportError: + logger.debug( + 'In-cluster credential module not available.', exc_info=True) + return + try: + credentials = ServiceAccountTokenVolumeCredentials(path=token_path) + credentials.refresh_api_key_hook(api_config) + api_config.api_key_prefix['authorization'] = 'Bearer' + api_config.refresh_api_key_hook = credentials.refresh_api_key_hook + except FileNotFoundError: + logger.warning( + 'Token file not found; proceeding without authentication.', + exc_info=True) + + +def refresh_credentials(api_config: kfp_server_api.Configuration,) -> None: + """Refresh the API token using the configured refresh hook.""" + if api_config.refresh_api_key_hook is not None: + api_config.refresh_api_key_hook(api_config) + else: + raise RuntimeError('Token expired but no refresh hook is configured. ' + 'Re-create the client with a fresh token.') diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/backend.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/backend.py new file mode 100644 index 00000000000..2882a384b2c --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/backend.py @@ -0,0 +1,1105 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Kubernetes backend for PipelinesClient.""" + +from __future__ import annotations + +from collections.abc import Callable +import datetime +import json +import logging +import os +import shutil +import tempfile +import time +from typing import Any +import warnings + +from google.protobuf import json_format +from kfp import compiler +from kfp.kubeflow_client import constants +from kfp.kubeflow_client.backends.kubernetes import auth +from kfp.kubeflow_client.backends.kubernetes import \ + constants as backend_constants +from kfp.kubeflow_client.backends.kubernetes import utils +from kfp.kubeflow_client.backends.kubernetes.types import \ + KubernetesBackendConfig +from kfp.kubeflow_client.types import Experiment +from kfp.kubeflow_client.types import ListExperimentsResponse +from kfp.kubeflow_client.types import ListPipelinesResponse +from kfp.kubeflow_client.types import ListPipelineVersionsResponse +from kfp.kubeflow_client.types import ListRunsResponse +from kfp.kubeflow_client.types import Pipeline +from kfp.kubeflow_client.types import PipelineVersion +from kfp.kubeflow_client.types import Run +from kfp.pipeline_spec import pipeline_spec_pb2 +import kfp_server_api +import yaml + +logger = logging.getLogger(__name__) + +_VALID_UPLOAD_EXTENSIONS = ('.yaml', '.yml', '.tar.gz', '.tgz', '.zip') + + +class KubernetesBackend: + """Kubernetes backend providing API connectivity for PipelinesClient. + + Manages the ``kfp_server_api`` configuration, service API instances, + namespace resolution, and credential lifecycle. + + Args: + config: Connection parameters for the KFP API server. + """ + + def __init__(self, config: KubernetesBackendConfig) -> None: + self._config = config + self._namespace: str | None = None + + self._api_config = self._build_api_configuration(config) + api_client = kfp_server_api.ApiClient(self._api_config) + + self._pipelines_api = kfp_server_api.PipelineServiceApi(api_client) + self._run_api = kfp_server_api.RunServiceApi(api_client) + self._experiment_api = kfp_server_api.ExperimentServiceApi(api_client) + self._upload_api = kfp_server_api.PipelineUploadServiceApi(api_client) + self._healthz_api = kfp_server_api.HealthzServiceApi(api_client) + + self.verify_backend() + + @property + def config(self) -> KubernetesBackendConfig: + """The backend configuration.""" + return self._config + + @property + def api_config(self) -> kfp_server_api.Configuration: + """The underlying kfp_server_api configuration.""" + return self._api_config + + @property + def pipelines_api(self) -> kfp_server_api.PipelineServiceApi: + """Pipeline service API instance.""" + return self._pipelines_api + + @property + def run_api(self) -> kfp_server_api.RunServiceApi: + """Run service API instance.""" + return self._run_api + + @property + def experiment_api(self) -> kfp_server_api.ExperimentServiceApi: + """Experiment service API instance.""" + return self._experiment_api + + @property + def upload_api(self) -> kfp_server_api.PipelineUploadServiceApi: + """Pipeline upload service API instance.""" + return self._upload_api + + @property + def namespace(self) -> str: + """Resolved target namespace (cached after first resolution).""" + if self._namespace is None: + self._namespace = utils.resolve_namespace(self._config.namespace) + return self._namespace + + def refresh_credentials(self) -> None: + """Refresh the API token using the configured refresh hook.""" + auth.refresh_credentials(self._api_config) + + def verify_backend(self) -> None: + """Verify that the KFP API server is reachable. + + Logs a warning on failure but never raises, matching the + TrainerClient.verify_backend() pattern. + """ + try: + self._healthz_api.healthz_service_get_healthz() + except Exception as e: + logger.warning( + 'KFP API server is not reachable at %s: %s. ' + 'Requests will fail until the server becomes available.', + self._api_config.host, e) + + # ------------------------------------------------------------------ + # Pipeline operations + # ------------------------------------------------------------------ + + def get_pipeline(self, name: str) -> Pipeline: + """Get a pipeline by name.""" + pipeline_id = self._get_pipeline_id_by_name(name) + if pipeline_id is None: + raise ValueError(f'Pipeline not found: {name!r}. ' + 'Use list_pipelines() to see available pipelines.') + return self._pipelines_api.pipeline_service_get_pipeline( + pipeline_id=pipeline_id) + + def get_pipeline_version( + self, + name: str, + version: str, + ) -> PipelineVersion: + """Get a specific pipeline version by pipeline name and version + name.""" + pipeline = self.get_pipeline(name) + version_id = self._get_version_id_by_name(pipeline.pipeline_id, version) + if version_id is None: + raise ValueError(f'Pipeline version not found: {version!r} ' + f'for pipeline {name!r}.') + return self._pipelines_api.pipeline_service_get_pipeline_version( + pipeline_id=pipeline.pipeline_id, + pipeline_version_id=version_id, + ) + + def list_pipelines( + self, + *, + page_token: str = '', + page_size: int = 10, + ) -> ListPipelinesResponse: + """List pipelines available on the server.""" + return self._pipelines_api.pipeline_service_list_pipelines( + namespace=self.namespace, + page_token=page_token, + page_size=page_size, + ) + + def list_pipeline_versions( + self, + name: str, + *, + page_token: str = '', + page_size: int = 10, + ) -> ListPipelineVersionsResponse: + """List versions of a pipeline by name.""" + pipeline_id = self._get_pipeline_id_by_name(name) + if pipeline_id is None: + raise ValueError(f'Pipeline not found: {name!r}. ' + 'Use list_pipelines() to see available pipelines.') + return self._pipelines_api.pipeline_service_list_pipeline_versions( + pipeline_id=pipeline_id, + page_token=page_token, + page_size=page_size, + ) + + def delete_pipeline( + self, + name: str, + *, + version: str | None = None, + force: bool = False, + ) -> None: + """Delete a pipeline or a specific pipeline version.""" + pipeline_id = self._get_pipeline_id_by_name(name) + if pipeline_id is None: + raise ValueError(f'Pipeline not found: {name!r}. ' + 'Use list_pipelines() to see available pipelines.') + + if version is not None: + version_id = self._get_version_id_by_name(pipeline_id, version) + if version_id is None: + raise ValueError(f'Pipeline version not found: {version!r} ' + f'for pipeline {name!r}.') + self._pipelines_api.pipeline_service_delete_pipeline_version( + pipeline_id=pipeline_id, + pipeline_version_id=version_id, + ) + return + + if not force: + versions_response = ( + self._pipelines_api.pipeline_service_list_pipeline_versions( + pipeline_id=pipeline_id, + page_size=2, + )) + versions = versions_response.pipeline_versions or [] + if len(versions) > 1: + raise ValueError( + f'Pipeline {name!r} has multiple versions. ' + 'Use force=True to delete the pipeline and all versions, ' + 'or specify version= to delete a single version.') + + self._pipelines_api.pipeline_service_delete_pipeline( + pipeline_id=pipeline_id, cascade=True) + + def upload_pipeline( + self, + pipeline: Callable | str, + *, + name: str | None = None, + version_name: str | None = None, + description: str | None = None, + ) -> PipelineVersion: + """Upload a pipeline (or new version) to the server.""" + package_path, temp_dir = self._resolve_pipeline_to_file(pipeline) + + try: + if name is None: + name = self._infer_pipeline_name(pipeline, package_path) + self._validate_pipeline_name(name) + + existing_pipeline_id = self._get_pipeline_id_by_name(name) + + if existing_pipeline_id is not None: + return self._upload_version( + package_path, + pipeline_id=existing_pipeline_id, + version_name=version_name, + description=description, + ) + else: + return self._upload_new_pipeline( + package_path, + name=name, + version_name=version_name, + description=description, + ) + finally: + if temp_dir is not None: + shutil.rmtree(temp_dir, ignore_errors=True) + + # ------------------------------------------------------------------ + # Run operations + # ------------------------------------------------------------------ + + def run( + self, + pipeline: str | Callable | Pipeline | PipelineVersion, + *, + params: dict[str, Any] | None = None, + name: str | None = None, + experiment: str | None = None, + version: str | None = None, + ) -> Run: + """Run a pipeline with full dispatch logic.""" + run_name = name or self._generate_run_name(pipeline) + + version_is_usable = ( + isinstance(pipeline, Pipeline) or + (isinstance(pipeline, str) and not self._is_yaml_path(pipeline) and + not self._is_archive_path(pipeline))) + if version is not None and not version_is_usable: + logger.warning( + 'The version parameter is ignored when pipeline is not a ' + 'name string (got %s).', + type(pipeline).__name__ + if not isinstance(pipeline, str) else repr(pipeline)) + + if isinstance(pipeline, PipelineVersion): + return self.run_from_version( + pipeline_id=pipeline.pipeline_id, + version_id=pipeline.pipeline_version_id, + params=params, + run_name=run_name, + experiment=experiment, + ) + + if isinstance(pipeline, Pipeline): + return self.run_pipeline( + pipeline=pipeline, + version=version, + params=params, + run_name=run_name, + experiment=experiment, + ) + + if callable(pipeline): + return self._run_inline( + pipeline_callable=pipeline, + params=params, + run_name=run_name, + experiment=experiment, + ) + + if isinstance(pipeline, str) and self._is_yaml_path(pipeline): + return self.run_from_file( + file_path=pipeline, + params=params, + run_name=run_name, + experiment=experiment, + ) + + if isinstance(pipeline, str) and self._is_archive_path(pipeline): + raise ValueError( + f'Archive files ({pipeline!r}) cannot be used for inline ' + 'runs. Use upload_pipeline() first, then run by name.') + + if isinstance(pipeline, str): + return self.run_by_name( + pipeline_name=pipeline, + version_name=version, + params=params, + run_name=run_name, + experiment=experiment, + ) + + raise ValueError(f'Unsupported pipeline type: {type(pipeline)!r}. ' + 'Expected a pipeline name, file path, callable, ' + 'Pipeline, or PipelineVersion.') + + def get_run(self, run_id: str) -> Run: + """Get a run by ID.""" + return self._run_api.run_service_get_run(run_id=run_id) + + def list_runs( + self, + *, + pipeline: str | None = None, + experiment: str | None = None, + status: str | None = None, + page_token: str = '', + page_size: int = 10, + ) -> ListRunsResponse: + """List runs, optionally filtered by pipeline, experiment, or + status.""" + pipeline_id = None + if pipeline is not None: + pipeline_id = self._get_pipeline_id_by_name(pipeline) + if pipeline_id is None: + raise ValueError(f'Pipeline not found: {pipeline!r}.') + + experiment_id = None + if experiment is not None: + experiment_id = self._get_experiment_id_by_name(experiment) + if experiment_id is None: + raise ValueError(f'Experiment not found: {experiment!r}. ' + 'Use create_experiment() first.') + + filter_predicates = [] + if status is not None: + filter_predicates.append({ + 'operation': 'EQUALS', + 'key': 'state', + 'stringValue': status.upper(), + }) + + filter_str = None + if filter_predicates: + filter_str = json.dumps({'predicates': filter_predicates}) + + response = self._run_api.run_service_list_runs( + namespace=self.namespace, + experiment_id=experiment_id or '', + page_token=page_token, + page_size=page_size, + filter=filter_str, + ) + + if pipeline_id is not None and response.runs: + original_count = len(response.runs) + response.runs = [ + run for run in response.runs + if (run.pipeline_version_reference and + run.pipeline_version_reference.pipeline_id == pipeline_id) + ] + filtered_count = original_count - len(response.runs) + if filtered_count > 0: + logger.info( + 'Client-side pipeline filter removed %d of %d runs.', + filtered_count, original_count) + + return response + + def wait_for_run_status( + self, + run: str | Run, + *, + status: set[str] | None = None, + timeout: int | None = 600, + polling_interval: int = 5, + callbacks: list[Callable[[Run], None]] | None = None, + ) -> Run: + """Wait for a run to reach a target state. + + Args: + run: A Run object or a run ID string. + status: Set of states to wait for. + timeout: Maximum time to wait, in seconds. Pass ``None`` to + wait indefinitely. + polling_interval: Time between status checks, in seconds. + callbacks: Called with the final Run object when the wait ends. + """ + if status is None: + status = {constants.RUN_COMPLETE} + target_states = {state.lower() for state in status} + + run_id = run.run_id if isinstance(run, Run) else run + start_time = time.monotonic() + first_poll_succeeded = False + max_auth_retries = 2 + auth_retries = 0 + + while True: + try: + run_response = self._run_api.run_service_get_run(run_id=run_id) + first_poll_succeeded = True + auth_retries = 0 + except kfp_server_api.ApiException as api_error: + if (first_poll_succeeded and api_error.status == 401 and + auth_retries < max_auth_retries): + auth_retries += 1 + logger.info( + 'Access token expired, refreshing ' + '(attempt %d/%d)...', auth_retries, max_auth_retries) + self.refresh_credentials() + continue + raise + + current_state = (run_response.state or '').lower() + + if current_state in target_states: + self._invoke_callbacks(callbacks, run_response) + return run_response + + if current_state in constants.TERMINAL_STATES: + logger.info( + 'Run %s reached terminal state %r before target %s.', + run_id, current_state, target_states) + self._invoke_callbacks(callbacks, run_response) + return run_response + + if timeout is not None: + elapsed = time.monotonic() - start_time + if elapsed >= timeout: + self._invoke_callbacks(callbacks, run_response) + raise TimeoutError(f'Run {run_id} did not reach state ' + f'{target_states} within {timeout}s. ' + f'Current state: {current_state!r}.') + + time.sleep(polling_interval) + + def run_by_name( + self, + pipeline_name: str, + version_name: str | None, + params: dict[str, Any] | None, + run_name: str, + experiment: str | None, + ) -> Run: + """Run an uploaded pipeline by name.""" + experiment_id = self._resolve_experiment_id(experiment) + pipeline_id = self._get_pipeline_id_by_name(pipeline_name) + if pipeline_id is None: + raise ValueError(f'Pipeline not found: {pipeline_name!r}. ' + 'Upload it first with upload_pipeline().') + + if version_name: + version_id = self._get_version_id_by_name(pipeline_id, version_name) + if version_id is None: + raise ValueError( + f'Pipeline version not found: {version_name!r} ' + f'for pipeline {pipeline_name!r}.') + else: + version_id = self._get_latest_version_id(pipeline_id) + + return self._run_from_version_reference( + pipeline_id=pipeline_id, + version_id=version_id, + params=params, + run_name=run_name, + experiment_id=experiment_id, + ) + + def run_pipeline( + self, + pipeline: Pipeline, + version: str | None, + params: dict[str, Any] | None, + run_name: str, + experiment: str | None, + ) -> Run: + """Run a pipeline from a Pipeline object.""" + experiment_id = self._resolve_experiment_id(experiment) + if version: + version_id = self._get_version_id_by_name(pipeline.pipeline_id, + version) + if version_id is None: + raise ValueError(f'Pipeline version not found: {version!r} ' + f'for pipeline {pipeline.display_name!r}.') + else: + version_id = self._get_latest_version_id(pipeline.pipeline_id) + return self._run_from_version_reference( + pipeline_id=pipeline.pipeline_id, + version_id=version_id, + params=params, + run_name=run_name, + experiment_id=experiment_id, + ) + + def run_from_version( + self, + pipeline_id: str, + version_id: str, + params: dict[str, Any] | None, + run_name: str, + experiment: str | None, + ) -> Run: + """Run a pipeline from a direct version reference.""" + experiment_id = self._resolve_experiment_id(experiment) + return self._run_from_version_reference( + pipeline_id=pipeline_id, + version_id=version_id, + params=params, + run_name=run_name, + experiment_id=experiment_id, + ) + + def run_from_file( + self, + file_path: str, + params: dict[str, Any] | None, + run_name: str, + experiment: str | None, + ) -> Run: + """Submit a run from a compiled pipeline YAML file.""" + experiment_id = self._resolve_experiment_id(experiment) + if not os.path.isfile(file_path): + raise ValueError(f'Pipeline file not found: {file_path}') + pipeline_spec_dict = self._load_pipeline_spec(file_path) + runtime_config = kfp_server_api.V2beta1RuntimeConfig( + parameters=params or {},) + run_body = kfp_server_api.V2beta1Run( + display_name=run_name, + experiment_id=experiment_id, + pipeline_spec=pipeline_spec_dict, + runtime_config=runtime_config, + ) + return self._run_api.run_service_create_run(run=run_body) + + # ------------------------------------------------------------------ + # Experiment operations + # ------------------------------------------------------------------ + + def create_experiment( + self, + name: str, + *, + description: str | None = None, + ) -> Experiment: + """Create a new experiment. + + Returns existing if name matches. + """ + existing_id = self._get_experiment_id_by_name(name) + if existing_id is not None: + if description: + logger.warning( + 'Experiment %r already exists; provided description will ' + 'not be applied.', name) + return self._experiment_api.experiment_service_get_experiment( + experiment_id=existing_id) + + experiment_body = kfp_server_api.V2beta1Experiment( + display_name=name, + description=description, + namespace=self.namespace, + ) + return self._experiment_api.experiment_service_create_experiment( + experiment=experiment_body) + + def get_experiment(self, name: str) -> Experiment: + """Get an experiment by name.""" + experiment_id = self._get_experiment_id_by_name(name) + if experiment_id is None: + raise ValueError(f'Experiment not found: {name!r}. ' + 'Use create_experiment() to create one.') + return self._experiment_api.experiment_service_get_experiment( + experiment_id=experiment_id) + + def list_experiments( + self, + *, + page_token: str = '', + page_size: int = 10, + ) -> ListExperimentsResponse: + """List experiments.""" + return self._experiment_api.experiment_service_list_experiments( + namespace=self.namespace, + page_token=page_token, + page_size=page_size, + ) + + def delete_experiment(self, name: str) -> None: + """Delete an experiment by name.""" + experiment_id = self._get_experiment_id_by_name(name) + if experiment_id is None: + raise ValueError(f'Experiment not found: {name!r}.') + self._experiment_api.experiment_service_delete_experiment( + experiment_id=experiment_id) + + # ------------------------------------------------------------------ + # Private helpers — name resolution + # ------------------------------------------------------------------ + + @staticmethod + def _equals_filter(key: str, value: str) -> str: + """Build a JSON filter string for an equality predicate.""" + return json.dumps({ + 'predicates': [{ + 'operation': 'EQUALS', + 'key': key, + 'stringValue': value, + }] + }) + + def _get_pipeline_id_by_name(self, name: str) -> str | None: + """Resolve a pipeline display name to its ID.""" + result = self._pipelines_api.pipeline_service_list_pipelines( + namespace=self.namespace, + filter=self._equals_filter('display_name', name)) + pipelines = result.pipelines or [] + if len(pipelines) == 0: + return None + if len(pipelines) == 1: + return pipelines[0].pipeline_id + pipeline_ids = [p.pipeline_id for p in pipelines] + raise ValueError( + f'Multiple pipelines found with name {name!r}: {pipeline_ids}. ' + 'Use kfp.Client directly to operate by pipeline ID.') + + def _get_version_id_by_name( + self, + pipeline_id: str, + version_name: str, + ) -> str | None: + """Resolve a version display name to its ID within a pipeline.""" + result = self._pipelines_api.pipeline_service_list_pipeline_versions( + pipeline_id=pipeline_id, + filter=self._equals_filter('display_name', version_name), + ) + versions = result.pipeline_versions or [] + if len(versions) == 0: + return None + if len(versions) == 1: + return versions[0].pipeline_version_id + version_ids = [v.pipeline_version_id for v in versions] + raise ValueError(f'Multiple versions found with name {version_name!r}: ' + f'{version_ids}.') + + def _get_latest_version_id(self, pipeline_id: str) -> str: + """Get the latest (most recently created) version ID for a pipeline.""" + result = self._pipelines_api.pipeline_service_list_pipeline_versions( + pipeline_id=pipeline_id, + page_size=1, + sort_by='created_at desc', + ) + versions = result.pipeline_versions or [] + if not versions: + raise ValueError(f'Pipeline {pipeline_id!r} has no versions.') + return versions[0].pipeline_version_id + + def _get_experiment_id_by_name(self, name: str) -> str | None: + """Resolve an experiment display name to its ID.""" + result = self._experiment_api.experiment_service_list_experiments( + namespace=self.namespace, + filter=self._equals_filter('display_name', name), + ) + experiments = result.experiments or [] + if len(experiments) == 0: + return None + if len(experiments) == 1: + return experiments[0].experiment_id + experiment_ids = [e.experiment_id for e in experiments] + raise ValueError(f'Multiple experiments found with name {name!r}: ' + f'{experiment_ids}.') + + # ------------------------------------------------------------------ + # Private helpers — pipeline upload + # ------------------------------------------------------------------ + + def _upload_new_pipeline( + self, + package_path: str, + *, + name: str, + version_name: str | None, + description: str | None, + ) -> PipelineVersion: + """Upload a brand-new pipeline and return its first version.""" + upload_kwargs: dict[str, Any] = { + 'name': name, + 'namespace': self.namespace, + } + if description: + upload_kwargs['description'] = description + + pipeline_response = self._upload_api.upload_pipeline( + package_path, **upload_kwargs) + + versions_response = ( + self._pipelines_api.pipeline_service_list_pipeline_versions( + pipeline_id=pipeline_response.pipeline_id, + page_size=1, + sort_by='created_at desc', + )) + versions = versions_response.pipeline_versions or [] + if not versions: + raise RuntimeError( + f'Pipeline {name!r} was uploaded but no version was ' + 'created by the server. This is unexpected.') + + first_version = versions[0] + + if version_name and first_version.display_name != version_name: + try: + update_body = kfp_server_api.V2beta1PipelineVersion( + pipeline_id=first_version.pipeline_id, + pipeline_version_id=first_version.pipeline_version_id, + display_name=version_name, + ) + self._pipelines_api.pipeline_service_update_pipeline_version( + pipeline_version_pipeline_id=first_version.pipeline_id, + pipeline_version_pipeline_version_id=( + first_version.pipeline_version_id), + pipeline_version=update_body, + ) + first_version.display_name = version_name + except kfp_server_api.ApiException as e: + if e.status in (401, 403): + warnings.warn( + f'Could not rename the first pipeline version to ' + f'{version_name!r} (HTTP {e.status}): insufficient ' + f'permissions. The version was uploaded successfully ' + f'but retains its server-generated name.', + stacklevel=2, + ) + else: + warnings.warn( + f'Could not rename the first pipeline version to ' + f'{version_name!r} (HTTP {e.status}). The version ' + f'was uploaded successfully but retains its ' + f'server-generated name.', + stacklevel=2, + ) + + return first_version + + def _upload_version( + self, + package_path: str, + *, + pipeline_id: str, + version_name: str | None, + description: str | None, + ) -> PipelineVersion: + """Upload a new version to an existing pipeline.""" + if not version_name: + version_name = ( + datetime.datetime.now().astimezone().strftime( + '%Y-%m-%d %H-%M-%S')) + upload_kwargs: dict[str, Any] = { + 'pipelineid': pipeline_id, + 'name': version_name, + } + if description: + upload_kwargs['description'] = description + + return self._upload_api.upload_pipeline_version(package_path, + **upload_kwargs) + + # ------------------------------------------------------------------ + # Private helpers — run creation + # ------------------------------------------------------------------ + + def _resolve_experiment_id( + self, + experiment: str | None, + ) -> str | None: + """Resolve an experiment name to its ID. + + Returns None when experiment is None, letting the server use its + default experiment. + """ + if experiment is None: + return None + experiment_id = self._get_experiment_id_by_name(experiment) + if experiment_id is None: + raise ValueError(f'Experiment not found: {experiment!r}. ' + 'Use create_experiment() first.') + return experiment_id + + def _run_from_version_reference( + self, + pipeline_id: str, + version_id: str, + params: dict[str, Any] | None, + run_name: str, + experiment_id: str | None, + ) -> Run: + """Create a run from a pipeline version reference (ID-based).""" + runtime_config = kfp_server_api.V2beta1RuntimeConfig( + parameters=params or {},) + pipeline_version_reference = ( + kfp_server_api.V2beta1PipelineVersionReference( + pipeline_id=pipeline_id, + pipeline_version_id=version_id, + )) + run_body = kfp_server_api.V2beta1Run( + display_name=run_name, + experiment_id=experiment_id, + pipeline_version_reference=pipeline_version_reference, + runtime_config=runtime_config, + ) + return self._run_api.run_service_create_run(run=run_body) + + def _load_pipeline_spec(self, file_path: str) -> dict: + """Load a pipeline spec from a YAML file and return as dict.""" + with open(file_path, 'r') as f: + try: + docs = list(yaml.safe_load_all(f)) + except yaml.YAMLError as e: + raise ValueError( + f'Failed to parse pipeline YAML at {file_path!r}: {e}' + ) from e + + if not docs or not docs[0]: + raise ValueError( + f'Pipeline file is empty or contains no valid YAML ' + f'documents: {file_path}') + pipeline_spec_dict = docs[0] + if len(docs) > 2: + raise ValueError( + f'Expected at most 2 YAML documents (pipeline spec + ' + f'platform spec), got {len(docs)} in {file_path!r}.') + platform_spec_dict = docs[1] if len(docs) > 1 and docs[1] else {} + + pipeline_spec = json_format.ParseDict( + pipeline_spec_dict, + pipeline_spec_pb2.PipelineSpec(), + ignore_unknown_fields=True) + platform_spec = json_format.ParseDict( + platform_spec_dict, + pipeline_spec_pb2.PlatformSpec(), + ignore_unknown_fields=True) + + if platform_spec == pipeline_spec_pb2.PlatformSpec(): + return json_format.MessageToDict(pipeline_spec) + return { + 'pipeline_spec': json_format.MessageToDict(pipeline_spec), + 'platform_spec': json_format.MessageToDict(platform_spec), + } + + @staticmethod + def _invoke_callbacks( + callbacks: list[Callable[[Run], None]] | None, + run: Run, + ) -> None: + """Invoke user-provided callbacks with the final run state.""" + if not callbacks: + return + for callback in callbacks: + try: + callback(run) + except Exception as error: + raise RuntimeError( + f'Callback {callback!r} raised an exception: ' + f'{error}') from error + + def _run_inline( + self, + pipeline_callable: Callable, + params: dict[str, Any] | None, + run_name: str, + experiment: str | None, + ) -> Run: + """Compile a callable and submit inline (no upload).""" + package_path, temp_dir = self._resolve_pipeline_to_file( + pipeline_callable) + try: + return self.run_from_file( + file_path=package_path, + params=params, + run_name=run_name, + experiment=experiment, + ) + finally: + if temp_dir is not None: + shutil.rmtree(temp_dir, ignore_errors=True) + + # ------------------------------------------------------------------ + # Private helpers — SDK-level preprocessing + # ------------------------------------------------------------------ + + def _resolve_pipeline_to_file( + self, + pipeline: Callable | str, + ) -> tuple[str, str | None]: + """Resolve a pipeline source to a compiled YAML file path. + + Returns: + A tuple of (package_path, temp_dir). ``temp_dir`` is ``None`` + when the input is a user-provided file path (no cleanup needed). + """ + if callable(pipeline): + temp_dir = tempfile.mkdtemp() + package_path = os.path.join(temp_dir, 'pipeline.yaml') + try: + compiler.Compiler().compile( + pipeline_func=pipeline, + package_path=package_path, + ) + except Exception as error: + shutil.rmtree(temp_dir, ignore_errors=True) + raise ValueError( + f'Failed to compile pipeline: {error}') from error + return package_path, temp_dir + if isinstance(pipeline, str): + if not os.path.isfile(pipeline): + raise ValueError(f'Pipeline file not found: {pipeline!r}.') + if not any( + pipeline.endswith(ext) for ext in _VALID_UPLOAD_EXTENSIONS): + raise ValueError(f'Unsupported file type: {pipeline!r}. ' + f'Expected one of: ' + f'{", ".join(_VALID_UPLOAD_EXTENSIONS)}') + return pipeline, None + raise ValueError( + f'Expected a callable or file path, got {type(pipeline)!r}.') + + def _infer_pipeline_name( + self, + pipeline: Callable | str, + package_path: str, + ) -> str: + """Infer a pipeline name from the source.""" + if callable(pipeline) and hasattr(pipeline, 'name') and pipeline.name: + return pipeline.name + if callable(pipeline) and hasattr(pipeline, + '__name__') and pipeline.__name__: + return pipeline.__name__.replace('_', '-') + + name_from_spec = self._read_pipeline_name_from_yaml(package_path) + if name_from_spec: + return name_from_spec + + if isinstance(pipeline, str): + basename = os.path.basename(pipeline) + return self._strip_pipeline_extension(basename) + return 'pipeline' + + @staticmethod + def _validate_pipeline_name(name: str) -> None: + """Validate that a pipeline name is non-empty.""" + if not name or name.isspace(): + raise ValueError( + 'Invalid pipeline name. Pipeline name cannot be empty ' + 'or contain only whitespace.') + + @staticmethod + def _read_pipeline_name_from_yaml(package_path: str,) -> str | None: + """Try to extract the pipeline name from a compiled YAML file.""" + try: + if not package_path.endswith(('.yaml', '.yml')): + return None + with open(package_path, 'r') as f: + doc = yaml.safe_load(f) + if isinstance(doc, dict): + pipeline_info = doc.get('pipelineInfo', {}) + name = pipeline_info.get('name') + if name and isinstance(name, str) and not name.isspace(): + return name + except (OSError, yaml.YAMLError): + logger.debug( + 'Could not read pipeline name from %s.', + package_path, + exc_info=True) + return None + + # ------------------------------------------------------------------ + # Private helpers — utilities + # ------------------------------------------------------------------ + + @staticmethod + def _is_yaml_path(value: str) -> bool: + """Check if a string looks like a YAML pipeline file path.""" + return value.endswith('.yaml') or value.endswith('.yml') + + @staticmethod + def _is_archive_path(value: str) -> bool: + """Check if a string looks like an archive pipeline file path.""" + return (value.endswith('.tar.gz') or value.endswith('.tgz') or + value.endswith('.zip')) + + @staticmethod + def _strip_pipeline_extension(filename: str) -> str: + """Strip known pipeline file extensions including compound ones.""" + for suffix in ('.tar.gz', '.tgz', '.zip', '.yaml', '.yml'): + if filename.endswith(suffix): + return filename[:-len(suffix)] + return filename + + @staticmethod + def _generate_run_name( + pipeline: str | Callable | Pipeline | PipelineVersion,) -> str: + """Generate a default run display name.""" + timestamp = ( + datetime.datetime.now().astimezone().strftime('%Y-%m-%d %H-%M-%S')) + if callable(pipeline): + base_name = getattr(pipeline, 'name', None) + if base_name is None: + base_name = getattr(pipeline, '__name__', 'pipeline') + return f'{base_name} {timestamp}' + if isinstance(pipeline, str): + if (KubernetesBackend._is_yaml_path(pipeline) or + KubernetesBackend._is_archive_path(pipeline)): + basename = os.path.basename(pipeline) + name = KubernetesBackend._strip_pipeline_extension(basename) + return f'{name} {timestamp}' + return f'{pipeline} {timestamp}' + if isinstance(pipeline, (Pipeline, PipelineVersion)): + display_name = getattr(pipeline, 'display_name', 'pipeline') + return f'{display_name} {timestamp}' + return f'pipeline {timestamp}' + + # ------------------------------------------------------------------ + # Private helpers — configuration + # ------------------------------------------------------------------ + + def _build_api_configuration( + self, + config: KubernetesBackendConfig, + ) -> kfp_server_api.Configuration: + """Build a kfp_server_api.Configuration from + KubernetesBackendConfig.""" + api_config = kfp_server_api.Configuration() + + if config.custom_ca: + api_config.ssl_ca_cert = config.custom_ca + else: + system_ca = utils.detect_system_ca_bundle() + if system_ca: + api_config.ssl_ca_cert = system_ca + + if config.base_url: + host = config.base_url + if not (host.startswith('http://') or host.startswith('https://')): + logger.warning('No scheme in base_url %r, defaulting to https.', + config.base_url) + host = 'https://' + host + api_config.host = host.rstrip('/') + else: + api_config.host = utils.discover_host( + backend_constants.DEFAULT_NAMESPACE) + + if config.is_secure is not None: + api_config.verify_ssl = config.is_secure + elif api_config.host: + api_config.verify_ssl = api_config.host.startswith('https') + + if config.user_token: + api_config.api_key['authorization'] = config.user_token + api_config.api_key_prefix['authorization'] = 'Bearer' + else: + auth.apply_in_cluster_credentials(api_config) + + return api_config diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/backend_test.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/backend_test.py new file mode 100644 index 00000000000..d7729ac66d5 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/backend_test.py @@ -0,0 +1,2548 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for the Kubernetes backend.""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field +import json +import logging +import os +import tempfile +from typing import Any +from unittest.mock import MagicMock +from unittest.mock import Mock +from unittest.mock import mock_open +from unittest.mock import patch +import warnings + +from kfp.kubeflow_client import constants +from kfp.kubeflow_client.backends.kubernetes import auth +from kfp.kubeflow_client.backends.kubernetes import utils +from kfp.kubeflow_client.backends.kubernetes.backend import KubernetesBackend +from kfp.kubeflow_client.backends.kubernetes.types import \ + KubernetesBackendConfig +import kfp_server_api +import pytest + +_AUTH_MODULE = 'kfp.kubeflow_client.backends.kubernetes.auth' +_BACKEND_MODULE = 'kfp.kubeflow_client.backends.kubernetes.backend' + +SUCCESS = 'success' +FAILED = 'failed' + + +@dataclass +class TestCase: + """A single test scenario for parametrized tests.""" + + name: str + expected_status: str = SUCCESS + config: dict[str, Any] = field(default_factory=dict) + expected_output: Any | None = None + expected_error: type[Exception] | None = None + expected_error_match: str | None = None + __test__ = False + + +# ------------------------------------------------------------------ +# Fixtures +# ------------------------------------------------------------------ + + +@pytest.fixture +def make_backend(): + """Factory fixture to create a KubernetesBackend with mocked auth.""" + + def _make(base_url='http://localhost:8888', namespace='test-ns', **kwargs): + with patch(f'{_AUTH_MODULE}.apply_in_cluster_credentials'), \ + patch(f'{_BACKEND_MODULE}.KubernetesBackend.verify_backend'): + return KubernetesBackend( + KubernetesBackendConfig( + base_url=base_url, namespace=namespace, **kwargs)) + + return _make + + +@pytest.fixture +def backend(make_backend): + """Convenience fixture for a default backend instance.""" + return make_backend() + + +# ------------------------------------------------------------------ +# test_backend_config_repr +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize('test_case', [ + TestCase( + name='token is masked in repr', + config={ + 'base_url': 'http://example.com', + 'user_token': 'super-secret-token', + 'namespace': 'ns', + }, + expected_output={ + 'must_not_contain': 'super-secret-token', + 'must_contain': "'***'", + }, + ), + TestCase( + name='no token shows None', + config={ + 'base_url': 'http://example.com', + }, + expected_output={ + 'must_contain': 'user_token=None', + }, + ), +]) +def test_backend_config_repr(test_case): + """Test KubernetesBackendConfig.__repr__ across scenarios.""" + cfg = KubernetesBackendConfig(**test_case.config) + rep = repr(cfg) + + if 'must_not_contain' in test_case.expected_output: + assert test_case.expected_output['must_not_contain'] not in rep + assert test_case.expected_output['must_contain'] in rep + + +# ------------------------------------------------------------------ +# test_build_api_configuration +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize('test_case', [ + TestCase( + name='explicit http base_url sets host directly', + config={'base_url': 'http://my-host:9999'}, + expected_output={'host': 'http://my-host:9999'}, + ), + TestCase( + name='base_url without scheme defaults to https', + config={'base_url': 'my-host:9999'}, + expected_output={'host': 'https://my-host:9999'}, + ), + TestCase( + name='https base_url sets host correctly', + config={'base_url': 'https://secure.example.com'}, + expected_output={'host': 'https://secure.example.com'}, + ), + TestCase( + name='https url sets verify_ssl true', + config={'base_url': 'https://secure.example.com'}, + expected_output={'verify_ssl': True}, + ), + TestCase( + name='http url sets verify_ssl false', + config={'base_url': 'http://insecure.example.com'}, + expected_output={'verify_ssl': False}, + ), + TestCase( + name='is_secure overrides scheme-based verify_ssl', + config={ + 'base_url': 'http://insecure.example.com', + 'is_secure': True + }, + expected_output={'verify_ssl': True}, + ), + TestCase( + name='custom_ca sets ssl_ca_cert', + config={ + 'base_url': 'https://host', + 'custom_ca': '/path/to/ca.crt' + }, + expected_output={'ssl_ca_cert': '/path/to/ca.crt'}, + ), + TestCase( + name='user_token sets api_key and prefix', + config={ + 'base_url': 'http://host', + 'user_token': 'my-token' + }, + expected_output={ + 'api_key': 'my-token', + 'api_key_prefix': 'Bearer', + }, + ), +]) +def test_build_api_configuration(make_backend, test_case): + """Test KubernetesBackend API configuration from config parameters.""" + backend = make_backend(**test_case.config) + + for key, expected in test_case.expected_output.items(): + if key == 'host': + assert backend.api_config.host == expected + elif key == 'verify_ssl': + assert backend.api_config.verify_ssl == expected + elif key == 'ssl_ca_cert': + assert backend.api_config.ssl_ca_cert == expected + elif key == 'api_key': + assert backend.api_config.api_key['authorization'] == expected + elif key == 'api_key_prefix': + assert backend.api_config.api_key_prefix[ + 'authorization'] == expected + + +def test_build_api_configuration_uses_default_namespace_for_discovery(): + """Verify discover_host is called with DEFAULT_NAMESPACE, not user ns.""" + with patch(f'{_AUTH_MODULE}.apply_in_cluster_credentials'), \ + patch(f'{_BACKEND_MODULE}.KubernetesBackend.verify_backend'), \ + patch(f'{_BACKEND_MODULE}.utils.discover_host', + return_value='http://ml-pipeline.kubeflow.svc.cluster.local:8888' + ) as mock_discover: + KubernetesBackend( + KubernetesBackendConfig(base_url=None, namespace='user-profile-ns')) + mock_discover.assert_called_once_with('kubeflow') + + +def test_build_api_configuration_applies_system_ca_when_no_custom_ca(): + """Verify system CA bundle is used when custom_ca is not provided.""" + with patch(f'{_AUTH_MODULE}.apply_in_cluster_credentials'), \ + patch(f'{_BACKEND_MODULE}.KubernetesBackend.verify_backend'), \ + patch(f'{_BACKEND_MODULE}.utils.detect_system_ca_bundle', + return_value='/etc/pki/tls/certs/ca-bundle.crt'): + backend = KubernetesBackend( + KubernetesBackendConfig(base_url='https://host', namespace='ns')) + assert backend.api_config.ssl_ca_cert == \ + '/etc/pki/tls/certs/ca-bundle.crt' + + +def test_build_api_configuration_custom_ca_overrides_system_ca(): + """Verify custom_ca takes precedence over system CA bundle.""" + with patch(f'{_AUTH_MODULE}.apply_in_cluster_credentials'), \ + patch(f'{_BACKEND_MODULE}.KubernetesBackend.verify_backend'), \ + patch(f'{_BACKEND_MODULE}.utils.detect_system_ca_bundle', + return_value='/etc/pki/tls/certs/ca-bundle.crt'): + backend = KubernetesBackend( + KubernetesBackendConfig( + base_url='https://host', + namespace='ns', + custom_ca='/my/custom/ca.crt')) + assert backend.api_config.ssl_ca_cert == '/my/custom/ca.crt' + + +# ------------------------------------------------------------------ +# test_resolve_namespace +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize('test_case', [ + TestCase( + name='explicit namespace returned as-is', + config={'namespace': 'explicit-ns'}, + expected_output='explicit-ns', + ), + TestCase( + name='in-cluster namespace file read', + config={ + 'namespace': None, + 'mock_file': 'file-namespace\n' + }, + expected_output='file-namespace', + ), + TestCase( + name='default fallback to kubeflow', + config={ + 'namespace': None, + 'mock_file': None + }, + expected_output='kubeflow', + ), +]) +def test_resolve_namespace(test_case): + """Test utils.resolve_namespace across discovery scenarios.""" + ns = test_case.config['namespace'] + mock_file_content = test_case.config.get('mock_file') + + if ns is not None: + assert utils.resolve_namespace(ns) == test_case.expected_output + elif mock_file_content is not None: + with patch('builtins.open', mock_open(read_data=mock_file_content)): + assert utils.resolve_namespace(None) == test_case.expected_output + else: + original_open = open + + def mock_open_fn(path, *args, **kwargs): + if 'serviceaccount' in str(path): + raise FileNotFoundError + return original_open(path, *args, **kwargs) + + import kubernetes + mock_k8s = MagicMock() + mock_k8s.config.ConfigException = kubernetes.config.ConfigException + mock_k8s.config.list_kube_config_contexts.side_effect = ( + FileNotFoundError) + with patch('builtins.open', side_effect=mock_open_fn): + with patch.dict('sys.modules', {'kubernetes': mock_k8s}): + assert utils.resolve_namespace( + None) == test_case.expected_output + + +# ------------------------------------------------------------------ +# test_apply_in_cluster_credentials +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize('test_case', [ + TestCase( + name='no token files skips silently', + expected_output={'authorization_absent': True}, + ), +]) +def test_apply_in_cluster_credentials(test_case): + """Test auth.apply_in_cluster_credentials behavior.""" + api_config = kfp_server_api.Configuration() + with patch('os.path.exists', return_value=False): + with patch('os.environ.get', return_value=None): + auth.apply_in_cluster_credentials(api_config) + assert 'authorization' not in api_config.api_key + + +# ------------------------------------------------------------------ +# test_detect_system_ca_bundle +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='SSL_CERT_FILE env var takes priority', + config={'env': { + 'SSL_CERT_FILE': '/custom/ca.crt' + }}, + expected_output='/custom/ca.crt', + ), + TestCase( + name='REQUESTS_CA_BUNDLE used when SSL_CERT_FILE absent', + config={'env': { + 'REQUESTS_CA_BUNDLE': '/bundle/ca.pem' + }}, + expected_output='/bundle/ca.pem', + ), + TestCase( + name='OpenSSL default cafile used when env vars absent', + config={'mode': 'openssl_default'}, + expected_output='/openssl/default/ca.pem', + ), + TestCase( + name='falls back to common OS path', + config={'mode': 'os_path'}, + expected_output='/etc/pki/tls/certs/ca-bundle.crt', + ), + TestCase( + name='returns None when nothing found', + config={'mode': 'nothing'}, + expected_output=None, + ), + ], + ids=lambda tc: tc.name) +def test_detect_system_ca_bundle(test_case): + """Test utils.detect_system_ca_bundle across detection scenarios.""" + env_vars = test_case.config.get('env') + mode = test_case.config.get('mode') + result = None + + if env_vars: + env_clean = { + k: v + for k, v in os.environ.items() + if k not in ('SSL_CERT_FILE', 'REQUESTS_CA_BUNDLE') + } + env_clean.update(env_vars) + with patch.dict(os.environ, env_clean, clear=True): + with patch('os.path.isfile', return_value=True): + result = utils.detect_system_ca_bundle() + elif mode == 'openssl_default': + env_clean = { + k: v + for k, v in os.environ.items() + if k not in ('SSL_CERT_FILE', 'REQUESTS_CA_BUNDLE') + } + with patch.dict(os.environ, env_clean, clear=True): + with patch('ssl.get_default_verify_paths') as mock_ssl: + mock_ssl.return_value = MagicMock( + cafile='/openssl/default/ca.pem') + with patch('os.path.isfile', return_value=True): + result = utils.detect_system_ca_bundle() + elif mode == 'os_path': + env_clean = { + k: v + for k, v in os.environ.items() + if k not in ('SSL_CERT_FILE', 'REQUESTS_CA_BUNDLE') + } + with patch.dict(os.environ, env_clean, clear=True): + with patch('ssl.get_default_verify_paths') as mock_ssl: + mock_ssl.return_value = MagicMock(cafile=None) + with patch( + 'os.path.isfile', + side_effect=lambda p: p == + '/etc/pki/tls/certs/ca-bundle.crt'): + result = utils.detect_system_ca_bundle() + elif mode == 'nothing': + env_clean = { + k: v + for k, v in os.environ.items() + if k not in ('SSL_CERT_FILE', 'REQUESTS_CA_BUNDLE') + } + with patch.dict(os.environ, env_clean, clear=True): + with patch('ssl.get_default_verify_paths') as mock_ssl: + mock_ssl.return_value = MagicMock(cafile=None) + with patch('os.path.isfile', return_value=False): + result = utils.detect_system_ca_bundle() + else: + pytest.fail(f'Unhandled test config: {test_case.config}') + + assert result == test_case.expected_output + + +# ------------------------------------------------------------------ +# test_discover_host +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize('test_case', [ + TestCase( + name='env var takes priority', + config={'env': { + 'KF_PIPELINES_ENDPOINT': 'http://my-endpoint' + }}, + expected_output='http://my-endpoint', + ), + TestCase( + name='env var without scheme adds https', + config={'env': { + 'KF_PIPELINES_ENDPOINT': 'my-endpoint:8080' + }}, + expected_output='https://my-endpoint:8080', + ), + TestCase( + name='in-cluster returns DNS name', + config={'mode': 'in_cluster'}, + expected_output=('http://ml-pipeline.test-ns.svc.cluster.local:8888'), + ), + TestCase( + name='kube proxy fallback', + config={'mode': 'kube_proxy'}, + expected_output=( + 'http://localhost:8001/' + 'api/v1/namespaces/test-ns/services/ml-pipeline:http/proxy/'), + ), + TestCase( + name='no kubernetes package in-cluster falls back to DNS', + config={'mode': 'no_k8s_in_cluster'}, + expected_output='http://ml-pipeline.test-ns.svc.cluster.local:8888', + ), + TestCase( + name='no kubernetes package not in-cluster raises ValueError', + config={'mode': 'no_k8s_not_in_cluster'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Could not auto-discover KFP endpoint', + ), +]) +def test_discover_host(test_case): + """Test utils.discover_host across discovery scenarios.""" + env_vars = test_case.config.get('env') + mode = test_case.config.get('mode') + + if env_vars: + with patch.dict(os.environ, env_vars): + result = utils.discover_host('test-ns') + elif mode == 'in_cluster': + env_clean = { + k: v for k, v in os.environ.items() if k != 'KF_PIPELINES_ENDPOINT' + } + import kubernetes + mock_k8s = MagicMock() + mock_k8s.config.ConfigException = kubernetes.config.ConfigException + mock_k8s.config.load_incluster_config.return_value = None + with patch.dict(os.environ, env_clean, clear=True): + with patch.dict('sys.modules', {'kubernetes': mock_k8s}): + result = utils.discover_host('test-ns') + elif mode == 'kube_proxy': + env_clean = { + k: v for k, v in os.environ.items() if k != 'KF_PIPELINES_ENDPOINT' + } + import kubernetes + mock_k8s = MagicMock() + mock_k8s.config.ConfigException = kubernetes.config.ConfigException + mock_k8s.config.load_incluster_config.side_effect = ( + kubernetes.config.ConfigException('no incluster')) + mock_k8s_client_config = MagicMock() + mock_k8s_client_config.host = 'http://localhost:8001' + mock_k8s.client.Configuration.return_value = mock_k8s_client_config + mock_k8s.config.load_kube_config.return_value = None + with patch.dict(os.environ, env_clean, clear=True): + with patch.dict('sys.modules', {'kubernetes': mock_k8s}): + result = utils.discover_host('test-ns') + elif mode == 'no_k8s_in_cluster': + env_clean = { + k: v for k, v in os.environ.items() if k != 'KF_PIPELINES_ENDPOINT' + } + with patch.dict(os.environ, env_clean, clear=True): + with patch.dict('sys.modules', {'kubernetes': None}): + with patch('os.path.exists', return_value=True): + result = utils.discover_host('test-ns') + elif mode == 'no_k8s_not_in_cluster': + env_clean = { + k: v for k, v in os.environ.items() if k != 'KF_PIPELINES_ENDPOINT' + } + with patch.dict(os.environ, env_clean, clear=True): + with patch.dict('sys.modules', {'kubernetes': None}): + with patch('os.path.exists', return_value=False): + with pytest.raises( + ValueError, match=test_case.expected_error_match): + utils.discover_host('test-ns') + return + + assert result == test_case.expected_output + + +# ------------------------------------------------------------------ +# test_get_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='success', + config={ + 'pipeline_id': 'pid-1', + 'display_name': 'my-pipe' + }, + expected_output={'pipeline_id': 'pid-1'}, + ), + TestCase( + name='not found raises ValueError', + config={'pipelines': []}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline not found', + ), + ], + ids=lambda tc: tc.name) +def test_get_pipeline(backend, test_case): + if test_case.expected_status == SUCCESS: + mock_pipeline = Mock( + pipeline_id=test_case.config['pipeline_id'], + display_name=test_case.config['display_name'], + ) + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipelines', + return_value=Mock(pipelines=[mock_pipeline])): + with patch.object( + backend.pipelines_api, + 'pipeline_service_get_pipeline', + return_value=mock_pipeline) as mock_get: + result = backend.get_pipeline('my-pipe') + mock_get.assert_called_once_with(pipeline_id='pid-1') + assert result.pipeline_id == test_case.expected_output[ + 'pipeline_id'] + else: + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipelines', + return_value=Mock(pipelines=[])): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.get_pipeline('nonexistent') + + +# ------------------------------------------------------------------ +# test_get_pipeline_version +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='success returns version', + config={ + 'pipeline_name': 'my-pipe', + 'version_name': 'v1' + }, + expected_output={'display_name': 'v1'}, + ), + TestCase( + name='pipeline not found raises ValueError', + config={ + 'pipeline_name': 'ghost-pipe', + 'version_name': 'v1' + }, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline not found', + ), + TestCase( + name='version not found raises ValueError', + config={ + 'pipeline_name': 'my-pipe', + 'version_name': 'bad-ver' + }, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline version not found', + ), + ], + ids=lambda tc: tc.name) +def test_get_pipeline_version(backend, test_case): + pipeline_name = test_case.config['pipeline_name'] + version_name = test_case.config['version_name'] + + if test_case.expected_status == SUCCESS: + mock_pipeline = Mock(pipeline_id='pid-1', display_name=pipeline_name) + mock_version = Mock( + pipeline_version_id='vid-1', display_name=version_name) + with patch.object(backend, 'get_pipeline', return_value=mock_pipeline): + with patch.object( + backend, '_get_version_id_by_name', return_value='vid-1'): + with patch.object( + backend.pipelines_api, + 'pipeline_service_get_pipeline_version', + return_value=mock_version) as mock_get: + result = backend.get_pipeline_version( + pipeline_name, version_name) + mock_get.assert_called_once_with( + pipeline_id='pid-1', pipeline_version_id='vid-1') + assert result.display_name == test_case.expected_output[ + 'display_name'] + elif 'Pipeline not found' in (test_case.expected_error_match or ''): + with patch.object( + backend, + 'get_pipeline', + side_effect=ValueError( + f'Pipeline not found: {pipeline_name!r}.')): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.get_pipeline_version(pipeline_name, version_name) + else: + mock_pipeline = Mock(pipeline_id='pid-1', display_name=pipeline_name) + with patch.object(backend, 'get_pipeline', return_value=mock_pipeline): + with patch.object( + backend, '_get_version_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.get_pipeline_version(pipeline_name, version_name) + + +# ------------------------------------------------------------------ +# test_list_pipelines +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='returns paginated response', + config={'page_size': 5}, + expected_output={ + 'count': 1, + 'next_page_token': 'tok' + }, + ), + ], + ids=lambda tc: tc.name) +def test_list_pipelines(backend, test_case): + mock_response = Mock(pipelines=[Mock()], next_page_token='tok') + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipelines', + return_value=mock_response) as mock_list: + result = backend.list_pipelines(page_size=test_case.config['page_size']) + mock_list.assert_called_once_with( + namespace='test-ns', + page_token='', + page_size=test_case.config['page_size'], + ) + assert len(result.pipelines) == test_case.expected_output['count'] + assert result.next_page_token == test_case.expected_output[ + 'next_page_token'] + + +# ------------------------------------------------------------------ +# test_get_run +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='returns run by id', + config={'run_id': 'r-1'}, + expected_output={ + 'run_id': 'r-1', + 'state': 'SUCCEEDED' + }, + ), + ], + ids=lambda tc: tc.name) +def test_get_run(backend, test_case): + mock_run = Mock(run_id=test_case.config['run_id'], state='SUCCEEDED') + with patch.object( + backend.run_api, 'run_service_get_run', + return_value=mock_run) as mock_get: + result = backend.get_run(test_case.config['run_id']) + mock_get.assert_called_once_with(run_id=test_case.config['run_id']) + assert result.run_id == test_case.expected_output['run_id'] + assert result.state == test_case.expected_output['state'] + + +# ------------------------------------------------------------------ +# test_list_pipeline_versions +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='success returns versions', + config={'pipeline_id': 'pid-1'}, + expected_output={'count': 2}, + ), + TestCase( + name='pipeline not found raises ValueError', + config={'pipeline_id': None}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline not found', + ), + ], + ids=lambda tc: tc.name) +def test_list_pipeline_versions(backend, test_case): + if test_case.expected_status == SUCCESS: + mock_resp = Mock( + pipeline_versions=[Mock(), Mock()], next_page_token='t2') + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=mock_resp) as mock_list: + result = backend.list_pipeline_versions('my-pipe') + mock_list.assert_called_once_with( + pipeline_id='pid-1', page_token='', page_size=10) + assert len(result.pipeline_versions + ) == test_case.expected_output['count'] + else: + with patch.object( + backend, '_get_pipeline_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.list_pipeline_versions('ghost-pipe') + + +# ------------------------------------------------------------------ +# test_delete_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='single version no force deletes pipeline', + config={'scenario': 'single_version_no_force'}, + ), + TestCase( + name='multiple versions no force raises', + config={'scenario': 'multi_version_no_force'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='multiple versions', + ), + TestCase( + name='multiple versions with force deletes pipeline', + config={'scenario': 'multi_version_force'}, + ), + TestCase( + name='specific version deletes version only', + config={'scenario': 'specific_version'}, + ), + TestCase( + name='nonexistent version raises', + config={'scenario': 'nonexistent_version'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline version not found', + ), + TestCase( + name='pipeline not found raises', + config={'scenario': 'not_found'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline not found', + ), + ], + ids=lambda tc: tc.name) +def test_delete_pipeline(backend, test_case): + scenario = test_case.config['scenario'] + + if scenario == 'single_version_no_force': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[Mock()])): + with patch.object( + backend.pipelines_api, + 'pipeline_service_delete_pipeline') as mock_del: + backend.delete_pipeline('my-pipe', force=False) + mock_del.assert_called_once_with( + pipeline_id='pid-1', cascade=True) + + elif scenario == 'multi_version_no_force': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[Mock(), Mock()])): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.delete_pipeline('my-pipe', force=False) + + elif scenario == 'multi_version_force': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object(backend.pipelines_api, + 'pipeline_service_delete_pipeline') as mock_del: + backend.delete_pipeline('my-pipe', force=True) + mock_del.assert_called_once_with( + pipeline_id='pid-1', cascade=True) + + elif scenario == 'specific_version': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object( + backend, '_get_version_id_by_name', return_value='vid-1'): + with patch.object( + backend.pipelines_api, + 'pipeline_service_delete_pipeline_version', + ) as mock_del: + backend.delete_pipeline('my-pipe', version='v1') + mock_del.assert_called_once_with( + pipeline_id='pid-1', pipeline_version_id='vid-1') + + elif scenario == 'nonexistent_version': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object( + backend, '_get_version_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.delete_pipeline('my-pipe', version='bad-ver') + + elif scenario == 'not_found': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.delete_pipeline('ghost-pipe') + + +# ------------------------------------------------------------------ +# test_upload_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='from callable compiles and cleans up', + config={'scenario': 'from_callable'}, + expected_output={'pipeline_version_id': 'vid-1'}, + ), + TestCase( + name='to existing pipeline creates version', + config={'scenario': 'to_existing'}, + expected_output={'pipeline_version_id': 'vid-2'}, + ), + ], + ids=lambda tc: tc.name) +def test_upload_pipeline(backend, test_case): + scenario = test_case.config['scenario'] + + if scenario == 'from_callable': + mock_version = Mock(pipeline_version_id='vid-1') + with patch.object( + backend, + '_resolve_pipeline_to_file', + return_value=('/tmp/pipe.yaml', '/tmp/tmpdir')): + with patch.object( + backend, '_infer_pipeline_name', return_value='my-pipe'): + with patch.object( + backend, '_get_pipeline_id_by_name', return_value=None): + with patch.object( + backend, + '_upload_new_pipeline', + return_value=mock_version): + with patch('shutil.rmtree') as mock_rm: + result = backend.upload_pipeline( + lambda: None, name='my-pipe') + mock_rm.assert_called_once_with( + '/tmp/tmpdir', ignore_errors=True) + assert result.pipeline_version_id == test_case.expected_output[ + 'pipeline_version_id'] + + elif scenario == 'to_existing': + mock_version = Mock(pipeline_version_id='vid-2') + with patch.object( + backend, + '_resolve_pipeline_to_file', + return_value=('/tmp/pipe.yaml', None)): + with patch.object( + backend, '_infer_pipeline_name', + return_value='existing-pipe'): + with patch.object( + backend, '_get_pipeline_id_by_name', + return_value='pid-1'): + with patch.object( + backend, '_upload_version', + return_value=mock_version) as mock_uv: + result = backend.upload_pipeline( + '/tmp/pipe.yaml', + name='existing-pipe', + version_name='v2', + ) + mock_uv.assert_called_once_with( + '/tmp/pipe.yaml', + pipeline_id='pid-1', + version_name='v2', + description=None, + ) + assert result.pipeline_version_id == test_case.expected_output[ + 'pipeline_version_id'] + + +# ------------------------------------------------------------------ +# test_run +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='callable dispatches to _run_inline', + config={ + 'input_type': 'callable', + 'kwargs': { + 'params': { + 'x': '1' + }, + 'name': 'my-run' + }, + }, + expected_output={'run_id': 'r-1'}, + ), + TestCase( + name='string name delegates to run_by_name', + config={ + 'input_type': 'string', + 'input_value': 'my-pipe', + 'kwargs': { + 'name': 'run-1' + }, + }, + expected_output={'run_id': 'r-2'}, + ), + TestCase( + name='string name with version', + config={ + 'input_type': 'string', + 'input_value': 'my-pipe', + 'kwargs': { + 'version': 'v2', + 'name': 'run-2' + }, + }, + expected_output={'run_id': 'r-3'}, + ), + TestCase( + name='name nonexistent version raises', + config={ + 'input_type': 'string_version_not_found', + 'input_value': 'my-pipe', + 'kwargs': { + 'version': 'bad' + }, + }, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='version not found', + ), + TestCase( + name='archive file raises', + config={ + 'input_type': 'archive_path', + 'input_value': '/path/to/file.tar.gz', + }, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Archive files', + ), + TestCase( + name='pipeline version object', + config={ + 'input_type': 'pipeline_version_obj', + 'kwargs': { + 'name': 'run-pv' + }, + }, + expected_output={'run_id': 'r-4'}, + ), + TestCase( + name='pipeline object', + config={ + 'input_type': 'pipeline_obj', + 'kwargs': { + 'name': 'run-pipe' + }, + }, + expected_output={'run_id': 'r-5'}, + ), + TestCase( + name='yaml path dispatches to run_from_file', + config={ + 'input_type': 'yaml_path', + 'input_value': '/tmp/my-pipeline.yaml', + 'kwargs': { + 'name': 'yaml-run' + }, + }, + expected_output={'run_id': 'r-6'}, + ), + TestCase( + name='callable with version logs warning', + config={ + 'input_type': 'callable_with_version', + 'kwargs': { + 'version': 'v2', + 'name': 'run-w1' + }, + 'expected_warning': 'version parameter is ignored', + }, + expected_output={'run_id': 'r-warn-1'}, + ), + TestCase( + name='yaml path with version logs warning', + config={ + 'input_type': 'yaml_with_version', + 'input_value': '/tmp/pipe.yaml', + 'kwargs': { + 'version': 'v2', + 'name': 'run-w2' + }, + 'expected_warning': 'version parameter is ignored', + }, + expected_output={'run_id': 'r-warn-2'}, + ), + TestCase( + name='pipeline version obj with version logs warning', + config={ + 'input_type': 'pv_obj_with_version', + 'kwargs': { + 'version': 'v2', + 'name': 'run-w3' + }, + 'expected_warning': 'version parameter is ignored', + }, + expected_output={'run_id': 'r-warn-3'}, + ), + TestCase( + name='pipeline obj with version delegates to run_pipeline', + config={ + 'input_type': 'pipeline_obj_with_version', + 'kwargs': { + 'version': 'v2', + 'name': 'run-ver' + }, + }, + expected_output={'run_id': 'r-ver'}, + ), + TestCase( + name='unsupported type raises', + config={'input_type': 'unsupported'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Unsupported pipeline', + ), + ], + ids=lambda tc: tc.name) +def test_run(backend, test_case, caplog): + input_type = test_case.config['input_type'] + kwargs = test_case.config.get('kwargs', {}) + + if input_type == 'callable': + mock_run = Mock(run_id='r-1', state='PENDING') + + def fake_pipeline(): + pass + + fake_pipeline.name = 'test-pipe' + with patch.object( + backend, '_run_inline', return_value=mock_run) as mock_inline: + result = backend.run(fake_pipeline, **kwargs) + mock_inline.assert_called_once() + assert result.run_id == test_case.expected_output['run_id'] + + elif input_type == 'string': + input_value = test_case.config['input_value'] + version = kwargs.get('version') + run_name = kwargs['name'] + run_id = test_case.expected_output['run_id'] + mock_run = Mock(run_id=run_id) + with patch.object( + backend, 'run_by_name', + return_value=mock_run) as mock_run_by_name: + result = backend.run(input_value, **kwargs) + mock_run_by_name.assert_called_once_with( + pipeline_name=input_value, + version_name=version, + params=None, + run_name=run_name, + experiment=None, + ) + assert result.run_id == run_id + + elif input_type == 'string_version_not_found': + input_value = test_case.config['input_value'] + with patch.object( + backend, 'run_by_name', + side_effect=ValueError('version not found')): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.run(input_value, **kwargs) + + elif input_type == 'archive_path': + input_value = test_case.config['input_value'] + with pytest.raises( + test_case.expected_error, match=test_case.expected_error_match): + backend.run(input_value) + + elif input_type == 'pipeline_version_obj': + pv = Mock( + spec=kfp_server_api.V2beta1PipelineVersion, + pipeline_id='pid-1', + pipeline_version_id='vid-1', + ) + mock_run = Mock(run_id='r-4') + with patch.object( + backend, 'run_from_version', return_value=mock_run) as mock_ref: + result = backend.run(pv, **kwargs) + mock_ref.assert_called_once_with( + pipeline_id='pid-1', + version_id='vid-1', + params=None, + run_name='run-pv', + experiment=None, + ) + assert result.run_id == test_case.expected_output['run_id'] + + elif input_type == 'pipeline_obj': + pipeline_obj = Mock( + spec=kfp_server_api.V2beta1Pipeline, pipeline_id='pid-1') + mock_run = Mock(run_id='r-5') + with patch.object( + backend, 'run_pipeline', + return_value=mock_run) as mock_run_pipeline: + result = backend.run(pipeline_obj, **kwargs) + mock_run_pipeline.assert_called_once_with( + pipeline=pipeline_obj, + version=None, + params=None, + run_name='run-pipe', + experiment=None, + ) + assert result.run_id == test_case.expected_output['run_id'] + + elif input_type == 'yaml_path': + input_value = test_case.config['input_value'] + mock_run = Mock(run_id='r-6') + with patch.object( + backend, 'run_from_file', return_value=mock_run) as mock_file: + result = backend.run(input_value, **kwargs) + mock_file.assert_called_once_with( + file_path=input_value, + params=None, + run_name='yaml-run', + experiment=None, + ) + assert result.run_id == test_case.expected_output['run_id'] + + elif input_type == 'callable_with_version': + mock_run = Mock(run_id='r-warn-1') + pipeline_fn = Mock(__name__='my_pipe') + with patch.object(backend, '_run_inline', return_value=mock_run): + with caplog.at_level( + logging.WARNING, + logger='kfp.kubeflow_client.backends.kubernetes.backend'): + backend.run(pipeline_fn, **kwargs) + assert test_case.config['expected_warning'] in caplog.text + + elif input_type == 'yaml_with_version': + input_value = test_case.config['input_value'] + mock_run = Mock(run_id='r-warn-2') + with patch.object(backend, 'run_from_file', return_value=mock_run): + with caplog.at_level( + logging.WARNING, + logger='kfp.kubeflow_client.backends.kubernetes.backend'): + backend.run(input_value, **kwargs) + assert test_case.config['expected_warning'] in caplog.text + + elif input_type == 'pv_obj_with_version': + mock_run = Mock(run_id='r-warn-3') + pv_obj = Mock( + spec=kfp_server_api.V2beta1PipelineVersion, + pipeline_id='pid-1', + pipeline_version_id='vid-1', + ) + with patch.object(backend, 'run_from_version', return_value=mock_run): + with caplog.at_level( + logging.WARNING, + logger='kfp.kubeflow_client.backends.kubernetes.backend'): + backend.run(pv_obj, **kwargs) + assert test_case.config['expected_warning'] in caplog.text + + elif input_type == 'pipeline_obj_with_version': + pipeline_obj = Mock( + spec=kfp_server_api.V2beta1Pipeline, + pipeline_id='pid-1', + display_name='my-pipe', + ) + mock_run = Mock(run_id='r-ver') + with patch.object( + backend, 'run_pipeline', + return_value=mock_run) as mock_run_pipeline: + result = backend.run(pipeline_obj, **kwargs) + mock_run_pipeline.assert_called_once_with( + pipeline=pipeline_obj, + version='v2', + params=None, + run_name='run-ver', + experiment=None, + ) + assert result.run_id == test_case.expected_output['run_id'] + + elif input_type == 'unsupported': + with pytest.raises( + test_case.expected_error, match=test_case.expected_error_match): + backend.run(12345) + + +# ------------------------------------------------------------------ +# test_list_runs +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='no filters', + config={'scenario': 'no_filters'}, + expected_output={ + 'count': 1, + 'next_page_token': 't2' + }, + ), + TestCase( + name='pipeline filter removes non-matching runs', + config={'scenario': 'pipeline_filter'}, + expected_output={ + 'count': 1, + 'run_id': 'r-1' + }, + ), + TestCase( + name='pipeline not found raises', + config={'scenario': 'pipeline_not_found'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline not found', + ), + TestCase( + name='experiment not found raises', + config={'scenario': 'experiment_not_found'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Experiment not found', + ), + TestCase( + name='status filter passed to server', + config={'scenario': 'status_filter'}, + expected_output={'status_value': 'SUCCEEDED'}, + ), + ], + ids=lambda tc: tc.name) +def test_list_runs(backend, test_case): + scenario = test_case.config['scenario'] + + if scenario == 'no_filters': + mock_response = Mock(runs=[Mock(run_id='r-1')], next_page_token='t2') + with patch.object( + backend.run_api, + 'run_service_list_runs', + return_value=mock_response) as mock_list: + result = backend.list_runs(page_size=20) + mock_list.assert_called_once_with( + namespace='test-ns', + experiment_id='', + page_token='', + page_size=20, + filter=None, + ) + assert len(result.runs) == test_case.expected_output['count'] + assert result.next_page_token == test_case.expected_output[ + 'next_page_token'] + + elif scenario == 'pipeline_filter': + run_match = Mock( + run_id='r-1', + pipeline_version_reference=Mock(pipeline_id='pid-1'), + ) + run_other = Mock( + run_id='r-2', + pipeline_version_reference=Mock(pipeline_id='pid-other'), + ) + response = Mock(runs=[run_match, run_other], next_page_token='') + with patch.object( + backend, '_get_pipeline_id_by_name', return_value='pid-1'): + with patch.object( + backend.run_api, 'run_service_list_runs', + return_value=response): + result = backend.list_runs(pipeline='my-pipe') + assert len(result.runs) == test_case.expected_output['count'] + assert result.runs[0].run_id == test_case.expected_output[ + 'run_id'] + + elif scenario == 'pipeline_not_found': + with patch.object( + backend, '_get_pipeline_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.list_runs(pipeline='bad-pipe') + + elif scenario == 'experiment_not_found': + with patch.object( + backend, '_get_experiment_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.list_runs(experiment='bad-exp') + + elif scenario == 'status_filter': + response = Mock(runs=[], next_page_token='') + with patch.object( + backend.run_api, 'run_service_list_runs', + return_value=response) as mock_list: + backend.list_runs(status='succeeded') + call_kwargs = mock_list.call_args[1] + filter_dict = json.loads(call_kwargs['filter']) + assert (filter_dict['predicates'][0]['stringValue'] == + test_case.expected_output['status_value']) + + +# ------------------------------------------------------------------ +# test_wait_for_run_status +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='returns on terminal state', + config={ + 'side_effects': [Mock(run_id='r-1', state='SUCCEEDED')], + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0 + }, + }, + expected_output={'state': 'SUCCEEDED'}, + ), + TestCase( + name='raises timeout error', + config={ + 'side_effects': [Mock(run_id='r-1', state='RUNNING')], + 'kwargs': { + 'timeout': 0, + 'polling_interval': 0 + }, + }, + expected_status=FAILED, + expected_error=TimeoutError, + ), + TestCase( + name='timeout None waits indefinitely until terminal state', + config={ + 'side_effects': [ + Mock(run_id='r-1', state='RUNNING'), + Mock(run_id='r-1', state='RUNNING'), + Mock(run_id='r-1', state='SUCCEEDED'), + ], + 'kwargs': { + 'timeout': None, + 'polling_interval': 0 + }, + }, + expected_output={'state': 'SUCCEEDED'}, + ), + TestCase( + name='callbacks are invoked', + config={ + 'side_effects': [Mock(run_id='r-1', state='SUCCEEDED')], + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0, + }, + 'use_callback': True, + }, + expected_output={'callback_called': True}, + ), + TestCase( + name='accepts run object', + config={ + 'use_run_object': True, + 'side_effects': [Mock(run_id='r-1', state='SUCCEEDED')], + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0 + }, + }, + expected_output={'run_id': 'r-1'}, + ), + TestCase( + name='terminal state not in target still returns', + config={ + 'side_effects': [Mock(run_id='r-1', state='FAILED')], + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0 + }, + }, + expected_output={'state': 'FAILED'}, + ), + TestCase( + name='token refresh on 401', + config={ + 'side_effects': 'auth_refresh', + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0, + }, + }, + expected_output={ + 'state': 'SUCCEEDED', + 'refresh_called': True + }, + ), + TestCase( + name='auth retry exhaustion raises', + config={ + 'side_effects': 'auth_exhaustion', + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0, + }, + }, + expected_status=FAILED, + expected_error=kfp_server_api.ApiException, + expected_output={'status_code': 401}, + ), + TestCase( + name='first poll 401 raises without refresh', + config={ + 'side_effects': 'first_poll_401', + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0, + }, + }, + expected_status=FAILED, + expected_error=kfp_server_api.ApiException, + expected_output={ + 'status_code': 401, + 'refresh_not_called': True + }, + ), + TestCase( + name='non-401 api exception propagates immediately', + config={ + 'side_effects': 'non_401', + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0, + }, + }, + expected_status=FAILED, + expected_error=kfp_server_api.ApiException, + expected_output={'status_code': 500}, + ), + TestCase( + name='callbacks fire on timeout', + config={ + 'side_effects': [Mock(run_id='r-1', state='RUNNING')], + 'kwargs': { + 'timeout': 0, + 'polling_interval': 0 + }, + 'use_callback': True, + }, + expected_status=FAILED, + expected_error=TimeoutError, + expected_output={'callback_called': True}, + ), + TestCase( + name='callbacks fire on terminal non-target state', + config={ + 'side_effects': [Mock(run_id='r-1', state='FAILED')], + 'kwargs': { + 'status': {constants.RUN_COMPLETE}, + 'polling_interval': 0, + }, + 'use_callback': True, + }, + expected_output={ + 'state': 'FAILED', + 'callback_called': True + }, + ), + ], + ids=lambda tc: tc.name) +def test_wait_for_run_status(backend, test_case): + config = test_case.config + kwargs = config['kwargs'] + side_effects = config['side_effects'] + use_callback = config.get('use_callback', False) + use_run_object = config.get('use_run_object', False) + + callback = Mock() if use_callback else None + if use_callback: + kwargs = {**kwargs, 'callbacks': [callback]} + + run_input = 'r-1' + if use_run_object: + run_input = Mock(run_id='r-1', state='SUCCEEDED') + + if side_effects == 'auth_refresh': + mock_run_ok = Mock(run_id='r-1', state='SUCCEEDED') + mock_side_effects = [ + Mock(run_id='r-1', state='RUNNING'), + kfp_server_api.ApiException(status=401), + mock_run_ok, + ] + with patch.object( + backend.run_api, + 'run_service_get_run', + side_effect=mock_side_effects): + with patch.object(backend, 'refresh_credentials') as mock_ref: + result = backend.wait_for_run_status(run_input, **kwargs) + mock_ref.assert_called_once() + assert result.state == test_case.expected_output['state'] + + elif side_effects == 'auth_exhaustion': + mock_side_effects = [ + Mock(run_id='r-1', state='RUNNING'), + kfp_server_api.ApiException(status=401), + kfp_server_api.ApiException(status=401), + kfp_server_api.ApiException(status=401), + ] + with patch.object( + backend.run_api, + 'run_service_get_run', + side_effect=mock_side_effects): + with patch.object(backend, 'refresh_credentials'): + with pytest.raises(test_case.expected_error) as exc_info: + backend.wait_for_run_status(run_input, **kwargs) + assert exc_info.value.status == test_case.expected_output[ + 'status_code'] + + elif side_effects == 'first_poll_401': + with patch.object( + backend.run_api, + 'run_service_get_run', + side_effect=kfp_server_api.ApiException(status=401)): + with patch.object(backend, 'refresh_credentials') as mock_ref: + with pytest.raises(test_case.expected_error) as exc_info: + backend.wait_for_run_status(run_input, **kwargs) + assert exc_info.value.status == test_case.expected_output[ + 'status_code'] + mock_ref.assert_not_called() + + elif side_effects == 'non_401': + mock_side_effects = [ + Mock(run_id='r-1', state='RUNNING'), + kfp_server_api.ApiException(status=500), + ] + with patch.object( + backend.run_api, + 'run_service_get_run', + side_effect=mock_side_effects): + with pytest.raises(test_case.expected_error) as exc_info: + backend.wait_for_run_status(run_input, **kwargs) + assert exc_info.value.status == test_case.expected_output[ + 'status_code'] + + elif isinstance(side_effects, list): + with patch.object( + backend.run_api, + 'run_service_get_run', + side_effect=side_effects if len(side_effects) > 1 else None, + return_value=side_effects[0] + if len(side_effects) == 1 else None): + if test_case.expected_status == FAILED: + with pytest.raises(test_case.expected_error): + backend.wait_for_run_status(run_input, **kwargs) + if use_callback: + callback.assert_called_once_with(side_effects[0]) + else: + result = backend.wait_for_run_status(run_input, **kwargs) + if 'state' in (test_case.expected_output or {}): + assert result.state == test_case.expected_output['state'] + if 'run_id' in (test_case.expected_output or {}): + assert result.run_id == test_case.expected_output['run_id'] + if use_callback: + callback.assert_called_once_with(side_effects[0]) + + +# ------------------------------------------------------------------ +# test_create_experiment +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='new experiment', + config={'existing_id': None}, + expected_output={'experiment_id': 'exp-1'}, + ), + TestCase( + name='idempotent returns existing', + config={'existing_id': 'exp-1'}, + expected_output={'experiment_id': 'exp-1'}, + ), + ], + ids=lambda tc: tc.name) +def test_create_experiment(backend, test_case): + existing_id = test_case.config['existing_id'] + + if existing_id is None: + mock_exp = Mock(experiment_id='exp-1', display_name='my-exp') + with patch.object( + backend, '_get_experiment_id_by_name', return_value=None): + with patch.object( + backend.experiment_api, + 'experiment_service_create_experiment', + return_value=mock_exp) as mock_create: + result = backend.create_experiment('my-exp') + mock_create.assert_called_once() + assert result.experiment_id == test_case.expected_output[ + 'experiment_id'] + else: + existing = Mock(experiment_id='exp-1', display_name='my-exp') + with patch.object( + backend, '_get_experiment_id_by_name', return_value='exp-1'): + with patch.object( + backend.experiment_api, + 'experiment_service_get_experiment', + return_value=existing): + result = backend.create_experiment('my-exp') + assert result.experiment_id == test_case.expected_output[ + 'experiment_id'] + + +# ------------------------------------------------------------------ +# test_get_experiment +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='success', + config={'existing_id': 'exp-1'}, + expected_output={'experiment_id': 'exp-1'}, + ), + TestCase( + name='not found raises', + config={'existing_id': None}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Experiment not found', + ), + ], + ids=lambda tc: tc.name) +def test_get_experiment(backend, test_case): + if test_case.expected_status == SUCCESS: + mock_exp = Mock(experiment_id='exp-1', display_name='my-exp') + with patch.object( + backend, '_get_experiment_id_by_name', return_value='exp-1'): + with patch.object( + backend.experiment_api, + 'experiment_service_get_experiment', + return_value=mock_exp): + result = backend.get_experiment('my-exp') + assert result.experiment_id == test_case.expected_output[ + 'experiment_id'] + else: + with patch.object( + backend, '_get_experiment_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.get_experiment('nonexistent') + + +# ------------------------------------------------------------------ +# test_list_experiments +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='returns paginated response', + expected_output={'next_page_token': 'tok2'}, + ), + ], + ids=lambda tc: tc.name) +def test_list_experiments(backend, test_case): + mock_response = Mock( + experiments=[Mock(experiment_id='e-1')], next_page_token='tok2') + with patch.object( + backend.experiment_api, + 'experiment_service_list_experiments', + return_value=mock_response) as mock_list: + result = backend.list_experiments(page_size=5) + mock_list.assert_called_once_with( + namespace='test-ns', page_token='', page_size=5) + assert result.next_page_token == test_case.expected_output[ + 'next_page_token'] + + +# ------------------------------------------------------------------ +# test_delete_experiment +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='success', + config={'existing_id': 'exp-1'}, + ), + TestCase( + name='not found raises', + config={'existing_id': None}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Experiment not found', + ), + ], + ids=lambda tc: tc.name) +def test_delete_experiment(backend, test_case): + if test_case.expected_status == SUCCESS: + with patch.object( + backend, '_get_experiment_id_by_name', return_value='exp-1'): + with patch.object( + backend.experiment_api, + 'experiment_service_delete_experiment') as mock_del: + backend.delete_experiment('my-exp') + mock_del.assert_called_once_with(experiment_id='exp-1') + else: + with patch.object( + backend, '_get_experiment_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend.delete_experiment('bad-exp') + + +# ------------------------------------------------------------------ +# test__infer_pipeline_name +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='dsl name attribute', + config={ + 'source': 'dsl_name', + 'dsl_name': 'dsl-assigned-name' + }, + expected_output='dsl-assigned-name', + ), + TestCase( + name='function name my_pipeline', + config={ + 'source': 'func_name', + 'func_name': 'my_pipeline' + }, + expected_output='my-pipeline', + ), + TestCase( + name='function name hello_world_pipe', + config={ + 'source': 'func_name', + 'func_name': 'hello_world_pipe' + }, + expected_output='hello-world-pipe', + ), + TestCase( + name='function name already-dashed', + config={ + 'source': 'func_name', + 'func_name': 'already-dashed' + }, + expected_output='already-dashed', + ), + TestCase( + name='file path fallback', + config={'source': 'file_path'}, + expected_output='my-pipeline', + ), + ], + ids=lambda tc: tc.name) +def test__infer_pipeline_name(backend, test_case): + source = test_case.config['source'] + + if source == 'dsl_name': + func = Mock() + func.name = test_case.config['dsl_name'] + result = backend._infer_pipeline_name(func, '/tmp/f.yaml') + assert result == test_case.expected_output + + elif source == 'func_name': + func = Mock() + func.name = None + func.__name__ = test_case.config['func_name'] + func.__class__ = type('function', (), {}) + with patch('os.path.isfile', return_value=False): + result = backend._infer_pipeline_name(func, '/tmp/f.yaml') + assert result == test_case.expected_output + + elif source == 'file_path': + result = backend._infer_pipeline_name('/tmp/my-pipeline.yaml', + '/tmp/my-pipeline.yaml') + assert result == test_case.expected_output + + +# ------------------------------------------------------------------ +# test__is_yaml_path +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='pipeline.yaml', + config={'path': 'pipeline.yaml'}, + expected_output=True), + TestCase( + name='pipeline.yml', + config={'path': 'pipeline.yml'}, + expected_output=True), + TestCase( + name='path/to/spec.yaml', + config={'path': 'path/to/spec.yaml'}, + expected_output=True), + TestCase( + name='pipeline.tar.gz', + config={'path': 'pipeline.tar.gz'}, + expected_output=False), + TestCase( + name='pipeline.zip', + config={'path': 'pipeline.zip'}, + expected_output=False), + TestCase( + name='pipeline-name', + config={'path': 'pipeline-name'}, + expected_output=False), + TestCase( + name='pipeline.tgz', + config={'path': 'pipeline.tgz'}, + expected_output=False), + ], + ids=lambda tc: tc.name) +def test__is_yaml_path(test_case): + assert KubernetesBackend._is_yaml_path( + test_case.config['path']) == test_case.expected_output + + +# ------------------------------------------------------------------ +# test__validate_pipeline_name +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='valid name passes', + config={'name': 'my-pipeline'}, + ), + TestCase( + name='empty raises', + config={'name': ''}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='cannot be empty', + ), + TestCase( + name='whitespace raises', + config={'name': ' '}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='cannot be empty', + ), + ], + ids=lambda tc: tc.name) +def test__validate_pipeline_name(test_case): + if test_case.expected_status == SUCCESS: + KubernetesBackend._validate_pipeline_name(test_case.config['name']) + else: + with pytest.raises( + test_case.expected_error, match=test_case.expected_error_match): + KubernetesBackend._validate_pipeline_name(test_case.config['name']) + + +# ------------------------------------------------------------------ +# test__read_pipeline_name_from_yaml +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='reads name from valid pipelineInfo', + config={'content': 'pipelineInfo:\n name: my-pipe\nroot: {}\n'}, + expected_output='my-pipe', + ), + TestCase( + name='returns None for non-yaml extension', + config={ + 'content': 'pipelineInfo:\n name: my-pipe\n', + 'suffix': '.json' + }, + expected_output=None, + ), + TestCase( + name='returns None for missing pipelineInfo', + config={'content': 'root:\n dag: {}\n'}, + expected_output=None, + ), + TestCase( + name='returns None for empty name', + config={'content': 'pipelineInfo:\n name: ""\n'}, + expected_output=None, + ), + TestCase( + name='returns None for whitespace-only name', + config={'content': 'pipelineInfo:\n name: " "\n'}, + expected_output=None, + ), + ], + ids=lambda tc: tc.name) +def test__read_pipeline_name_from_yaml(test_case): + suffix = test_case.config.get('suffix', '.yaml') + with tempfile.NamedTemporaryFile( + mode='w', suffix=suffix, delete=False) as f: + f.write(test_case.config['content']) + f.flush() + try: + result = KubernetesBackend._read_pipeline_name_from_yaml(f.name) + assert result == test_case.expected_output + finally: + os.unlink(f.name) + + +# ------------------------------------------------------------------ +# test__generate_run_name +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='from callable', + config={'source': 'callable'}, + expected_output='my-pipe ', + ), + TestCase( + name='from string', + config={'source': 'string'}, + expected_output='some-pipeline ', + ), + TestCase( + name='from yaml path', + config={'source': 'yaml_path'}, + expected_output='train ', + ), + TestCase( + name='from pipeline object', + config={'source': 'pipeline_object'}, + expected_output='uploaded-pipe ', + ), + ], + ids=lambda tc: tc.name) +def test__generate_run_name(test_case): + source = test_case.config['source'] + + if source == 'callable': + func = Mock() + func.name = 'my-pipe' + result = KubernetesBackend._generate_run_name(func) + elif source == 'string': + result = KubernetesBackend._generate_run_name('some-pipeline') + elif source == 'yaml_path': + result = KubernetesBackend._generate_run_name('/tmp/train.yaml') + elif source == 'pipeline_object': + pipe = kfp_server_api.V2beta1Pipeline(display_name='uploaded-pipe') + result = KubernetesBackend._generate_run_name(pipe) + + assert result.startswith(test_case.expected_output) + + +# ------------------------------------------------------------------ +# test__resolve_pipeline_to_file +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='file not found raises', + config={'scenario': 'file_not_found'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Pipeline file not found', + ), + TestCase( + name='unsupported extension raises', + config={'scenario': 'unsupported_extension'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Unsupported file type', + ), + TestCase( + name='valid yaml returns path no cleanup', + config={'scenario': 'valid_yaml'}, + expected_output={ + 'path': '/tmp/my-pipeline.yaml', + 'temp_dir': None + }, + ), + TestCase( + name='valid tar.gz returns path no cleanup', + config={'scenario': 'valid_tar_gz'}, + expected_output={ + 'path': '/tmp/my-pipeline.tar.gz', + 'temp_dir': None + }, + ), + TestCase( + name='callable compilation failure cleans temp dir', + config={'scenario': 'callable_failure'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Failed to compile pipeline', + ), + TestCase( + name='callable compilation success returns temp path', + config={'scenario': 'callable_success'}, + expected_output={ + 'path': '/tmp/fake-tmpdir/pipeline.yaml', + 'temp_dir': '/tmp/fake-tmpdir', + }, + ), + TestCase( + name='non callable non string raises', + config={'scenario': 'non_callable_non_string'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Expected a callable', + ), + ], + ids=lambda tc: tc.name) +def test__resolve_pipeline_to_file(backend, test_case): + scenario = test_case.config['scenario'] + + if scenario == 'file_not_found': + with patch('os.path.isfile', return_value=False): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._resolve_pipeline_to_file('/no/such/file.yaml') + + elif scenario == 'unsupported_extension': + with patch('os.path.isfile', return_value=True): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._resolve_pipeline_to_file('/tmp/pipeline.json') + + elif scenario == 'valid_yaml': + with patch('os.path.isfile', return_value=True): + path, temp_dir = backend._resolve_pipeline_to_file( + '/tmp/my-pipeline.yaml') + assert path == test_case.expected_output['path'] + assert temp_dir == test_case.expected_output['temp_dir'] + + elif scenario == 'valid_tar_gz': + with patch('os.path.isfile', return_value=True): + path, temp_dir = backend._resolve_pipeline_to_file( + '/tmp/my-pipeline.tar.gz') + assert path == test_case.expected_output['path'] + assert temp_dir == test_case.expected_output['temp_dir'] + + elif scenario == 'callable_failure': + + def bad_pipeline(): + pass + + with patch('tempfile.mkdtemp', return_value='/tmp/fake-tmpdir'): + with patch( + 'kfp.compiler.Compiler.compile', + side_effect=RuntimeError('compile failed')): + with patch('shutil.rmtree') as mock_rm: + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._resolve_pipeline_to_file(bad_pipeline) + mock_rm.assert_called_once_with( + '/tmp/fake-tmpdir', ignore_errors=True) + + elif scenario == 'callable_success': + + def good_pipeline(): + pass + + with patch('tempfile.mkdtemp', return_value='/tmp/fake-tmpdir'): + with patch('kfp.compiler.Compiler.compile'): + path, temp_dir = backend._resolve_pipeline_to_file( + good_pipeline) + assert path == test_case.expected_output['path'] + assert temp_dir == test_case.expected_output['temp_dir'] + + elif scenario == 'non_callable_non_string': + with pytest.raises( + test_case.expected_error, match=test_case.expected_error_match): + backend._resolve_pipeline_to_file(12345) + + +# ------------------------------------------------------------------ +# test__upload_new_pipeline +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='rename version success', + config={'scenario': 'rename_success'}, + expected_output={'display_name': 'v1'}, + ), + TestCase( + name='rename version failure warns', + config={'scenario': 'rename_failure'}, + expected_output={'display_name': 'auto-generated-name'}, + ), + TestCase( + name='rename version 403 warns insufficient permissions', + config={'scenario': 'rename_failure_403'}, + expected_output={'display_name': 'auto-generated-name'}, + ), + TestCase( + name='no version created raises RuntimeError', + config={'scenario': 'no_version'}, + expected_status=FAILED, + expected_error=RuntimeError, + expected_error_match='no version was created', + ), + ], + ids=lambda tc: tc.name) +def test__upload_new_pipeline(backend, test_case): + scenario = test_case.config['scenario'] + + if scenario == 'rename_success': + mock_pipeline = Mock(pipeline_id='pid-1') + mock_version = Mock( + pipeline_id='pid-1', + pipeline_version_id='vid-1', + display_name='auto-generated-name', + ) + with patch.object( + backend.upload_api, 'upload_pipeline', + return_value=mock_pipeline): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[mock_version])): + with patch.object( + backend.pipelines_api, + 'pipeline_service_update_pipeline_version', + create=True, + ) as mock_update: + result = backend._upload_new_pipeline( + '/tmp/pipe.yaml', + name='my-pipe', + version_name='v1', + description=None, + ) + mock_update.assert_called_once() + assert result.display_name == test_case.expected_output[ + 'display_name'] + + elif scenario == 'rename_failure': + mock_pipeline = Mock(pipeline_id='pid-1') + mock_version = Mock( + pipeline_id='pid-1', + pipeline_version_id='vid-1', + display_name='auto-generated-name', + ) + with patch.object( + backend.upload_api, 'upload_pipeline', + return_value=mock_pipeline): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[mock_version])): + with patch.object( + backend.pipelines_api, + 'pipeline_service_update_pipeline_version', + create=True, + side_effect=kfp_server_api.ApiException( + status=404, reason='Not Found')): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + result = backend._upload_new_pipeline( + '/tmp/pipe.yaml', + name='my-pipe', + version_name='v1', + description=None, + ) + assert len(caught) == 1 + assert 'Could not rename' in str(caught[0].message) + assert result.display_name == test_case.expected_output[ + 'display_name'] + + elif scenario == 'rename_failure_403': + mock_pipeline = Mock(pipeline_id='pid-1') + mock_version = Mock( + pipeline_id='pid-1', + pipeline_version_id='vid-1', + display_name='auto-generated-name', + ) + with patch.object( + backend.upload_api, 'upload_pipeline', + return_value=mock_pipeline): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[mock_version])): + with patch.object( + backend.pipelines_api, + 'pipeline_service_update_pipeline_version', + create=True, + side_effect=kfp_server_api.ApiException( + status=403, reason='Forbidden')): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + result = backend._upload_new_pipeline( + '/tmp/pipe.yaml', + name='my-pipe', + version_name='v1', + description=None, + ) + assert len(caught) == 1 + assert 'insufficient permissions' in str(caught[0].message) + assert result.display_name == test_case.expected_output[ + 'display_name'] + + elif scenario == 'no_version': + mock_pipeline = Mock(pipeline_id='pid-1') + with patch.object( + backend.upload_api, 'upload_pipeline', + return_value=mock_pipeline): + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[])): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._upload_new_pipeline( + '/tmp/pipe.yaml', + name='my-pipe', + version_name=None, + description=None, + ) + + +# ------------------------------------------------------------------ +# test__load_pipeline_spec +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='valid pipeline spec', + config={'scenario': 'valid'}, + expected_output={'contains': 'pipelineInfo'}, + ), + TestCase( + name='empty file raises', + config={'scenario': 'empty'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='empty', + ), + TestCase( + name='multi doc yaml with platform spec', + config={'scenario': 'multi_doc'}, + expected_output={ + 'contains_keys': ['pipeline_spec', 'platform_spec'] + }, + ), + ], + ids=lambda tc: tc.name) +def test__load_pipeline_spec(backend, test_case): + scenario = test_case.config['scenario'] + + if scenario == 'valid': + spec_content = ('pipelineInfo:\n' + ' name: test-pipeline\n' + 'root:\n' + ' dag: {}\n' + 'schemaVersion: 2.1.0\n' + 'sdkVersion: kfp-2.0.0\n') + with tempfile.NamedTemporaryFile( + mode='w', suffix='.yaml', delete=False) as f: + f.write(spec_content) + f.flush() + try: + result = backend._load_pipeline_spec(f.name) + assert test_case.expected_output['contains'] in result + finally: + os.unlink(f.name) + + elif scenario == 'empty': + with tempfile.NamedTemporaryFile( + mode='w', suffix='.yaml', delete=False) as f: + f.write('') + f.flush() + try: + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._load_pipeline_spec(f.name) + finally: + os.unlink(f.name) + + elif scenario == 'multi_doc': + spec_content = ('pipelineInfo:\n' + ' name: test-pipeline\n' + 'root:\n' + ' dag: {}\n' + 'schemaVersion: 2.1.0\n' + 'sdkVersion: kfp-2.0.0\n' + '---\n' + 'platforms:\n' + ' kubernetes:\n' + ' deploymentSpec: {}\n') + with tempfile.NamedTemporaryFile( + mode='w', suffix='.yaml', delete=False) as f: + f.write(spec_content) + f.flush() + try: + result = backend._load_pipeline_spec(f.name) + for key in test_case.expected_output['contains_keys']: + assert key in result + finally: + os.unlink(f.name) + + +# ------------------------------------------------------------------ +# test__invoke_callbacks +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='wraps exception in RuntimeError', + expected_status=FAILED, + expected_error=RuntimeError, + expected_error_match='callback boom', + ), + ], + ids=lambda tc: tc.name) +def test__invoke_callbacks(test_case): + + def bad_callback(run): + raise ValueError('callback boom') + + mock_run = Mock(run_id='r-1', state='SUCCEEDED') + with pytest.raises( + test_case.expected_error, match=test_case.expected_error_match): + KubernetesBackend._invoke_callbacks([bad_callback], mock_run) + + +# ------------------------------------------------------------------ +# test__get_latest_version_id +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='returns latest version id', + config={'has_versions': True}, + expected_output='vid-latest', + ), + TestCase( + name='no versions raises', + config={'has_versions': False}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='has no versions', + ), + ], + ids=lambda tc: tc.name) +def test__get_latest_version_id(backend, test_case): + if test_case.config['has_versions']: + mock_version = Mock( + pipeline_version_id='vid-latest', + created_at='2026-01-01T00:00:00Z', + ) + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[mock_version])): + result = backend._get_latest_version_id('pid-1') + assert result == test_case.expected_output + else: + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipeline_versions', + return_value=Mock(pipeline_versions=[])): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._get_latest_version_id('pid-1') + + +# ------------------------------------------------------------------ +# test__equals_filter +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='produces correct filter JSON', + config={ + 'key': 'display_name', + 'value': 'test' + }, + expected_output=( + '{"predicates": [{"operation": "EQUALS",' + ' "key": "display_name", "stringValue": "test"}]}'), + ), + ], + ids=lambda tc: tc.name) +def test__equals_filter(backend, test_case): + result = backend._equals_filter(test_case.config['key'], + test_case.config['value']) + assert json.loads(result) == json.loads(test_case.expected_output) + + +# ------------------------------------------------------------------ +# test__get_pipeline_id_by_name +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='multiple matches raises', + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Multiple pipelines', + ), + ], + ids=lambda tc: tc.name) +def test__get_pipeline_id_by_name(backend, test_case): + pipelines = [ + Mock(pipeline_id='pid-1'), + Mock(pipeline_id='pid-2'), + ] + with patch.object( + backend.pipelines_api, + 'pipeline_service_list_pipelines', + return_value=Mock(pipelines=pipelines)): + with pytest.raises( + test_case.expected_error, match=test_case.expected_error_match): + backend._get_pipeline_id_by_name('dup-name') + + +# ------------------------------------------------------------------ +# test__resolve_experiment_id +# ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + 'test_case', [ + TestCase( + name='returns None when not specified', + config={'experiment': None}, + expected_output=None, + ), + TestCase( + name='named not found raises', + config={'experiment': 'nonexistent'}, + expected_status=FAILED, + expected_error=ValueError, + expected_error_match='Experiment not found', + ), + ], + ids=lambda tc: tc.name) +def test__resolve_experiment_id(backend, test_case): + experiment = test_case.config['experiment'] + + if test_case.expected_status == SUCCESS: + result = backend._resolve_experiment_id(experiment) + assert result == test_case.expected_output + else: + with patch.object( + backend, '_get_experiment_id_by_name', return_value=None): + with pytest.raises( + test_case.expected_error, + match=test_case.expected_error_match): + backend._resolve_experiment_id(experiment) diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/constants.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/constants.py new file mode 100644 index 00000000000..58bc01922db --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/constants.py @@ -0,0 +1,24 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Constants for the Kubernetes backend.""" + +IN_CLUSTER_DNS_NAME = 'http://ml-pipeline.{}.svc.cluster.local:8888' +KUBE_PROXY_PATH = 'api/v1/namespaces/{}/services/ml-pipeline:http/proxy/' +NAMESPACE_PATH = '/var/run/secrets/kubernetes.io/serviceaccount/namespace' +DEFAULT_NAMESPACE = 'kubeflow' + +KFP_SA_TOKEN_PATH = '/var/run/secrets/kubeflow/pipelines/token' +K8S_SA_TOKEN_PATH = '/var/run/secrets/kubernetes.io/serviceaccount/token' +TOKEN_PATH_ENV = 'KF_PIPELINES_SA_TOKEN_PATH' +ENDPOINT_ENV = 'KF_PIPELINES_ENDPOINT' diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/types.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/types.py new file mode 100644 index 00000000000..d3529917c15 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/types.py @@ -0,0 +1,53 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Type definitions for the Kubernetes backend.""" + +from __future__ import annotations + +import dataclasses + + +@dataclasses.dataclass +class KubernetesBackendConfig: + """Connection configuration for the KFP API server. + + Args: + base_url: KFP API server URL including scheme and port + (e.g. ``https://ml-pipeline.example.com:8080``). If omitted, + auto-discovered following kfp.Client conventions (in-cluster DNS + or kubeconfig proxy). + user_token: Bearer token for authentication. + is_secure: Whether to verify TLS certificates (controls + ``verify_ssl`` on the underlying HTTP client). Does not control + whether the connection uses TLS — that is determined by the URL + scheme. Inferred from scheme if omitted (``True`` for https, + ``False`` for http). + custom_ca: Path to PEM-encoded root certificates. + namespace: Kubernetes namespace. If omitted, auto-detected. + """ + + base_url: str | None = None + user_token: str | None = None + is_secure: bool | None = None + custom_ca: str | None = None + namespace: str | None = None + + def __repr__(self) -> str: + token_display = '***' if self.user_token else None + return (f'KubernetesBackendConfig(' + f'base_url={self.base_url!r}, ' + f'user_token={token_display!r}, ' + f'is_secure={self.is_secure!r}, ' + f'custom_ca={self.custom_ca!r}, ' + f'namespace={self.namespace!r})') diff --git a/sdk/python/kfp/kubeflow_client/backends/kubernetes/utils.py b/sdk/python/kfp/kubeflow_client/backends/kubernetes/utils.py new file mode 100644 index 00000000000..c3638aa7382 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/backends/kubernetes/utils.py @@ -0,0 +1,159 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Utility helpers for the Kubernetes backend.""" + +from __future__ import annotations + +import logging +import os +import ssl + +from kfp.kubeflow_client.backends.kubernetes import constants + +logger = logging.getLogger(__name__) + +_COMMON_CA_BUNDLE_PATHS = ( + '/etc/ssl/certs/ca-certificates.crt', + '/etc/pki/tls/certs/ca-bundle.crt', + '/etc/ssl/ca-bundle.pem', + '/etc/ssl/cert.pem', +) + + +def detect_system_ca_bundle() -> str | None: + """Best-effort detection of a system CA certificate bundle. + + Resolution order: + 1. ``SSL_CERT_FILE`` environment variable. + 2. ``REQUESTS_CA_BUNDLE`` environment variable. + 3. OpenSSL default CA file (via :func:`ssl.get_default_verify_paths`). + 4. Common OS bundle paths (Debian, RHEL, openSUSE, macOS/Alpine). + + Returns: + Absolute path to a CA bundle file, or ``None`` if no bundle was found. + """ + for env_var in ('SSL_CERT_FILE', 'REQUESTS_CA_BUNDLE'): + path = os.environ.get(env_var) + if path and os.path.isfile(path): + logger.debug('System CA bundle from %s: %s', env_var, path) + return path + + try: + defaults = ssl.get_default_verify_paths() + if defaults.cafile and os.path.isfile(defaults.cafile): + logger.debug('System CA bundle from OpenSSL defaults: %s', + defaults.cafile) + return defaults.cafile + except Exception: # pylint: disable=broad-except + logger.debug('ssl.get_default_verify_paths() failed.', exc_info=True) + + for path in _COMMON_CA_BUNDLE_PATHS: + if os.path.isfile(path): + logger.debug('System CA bundle found at: %s', path) + return path + + return None + + +def discover_host(namespace: str) -> str: + """Auto-discover the KFP API server endpoint.""" + endpoint_from_env = os.environ.get(constants.ENDPOINT_ENV) + if endpoint_from_env: + host = endpoint_from_env.rstrip('/') + if not (host.startswith('http://') or host.startswith('https://')): + logger.warning( + 'No scheme in KF_PIPELINES_ENDPOINT %r, defaulting ' + 'to https.', endpoint_from_env) + host = 'https://' + host + return host + + try: + import kubernetes as k8s + except ImportError: + if os.path.exists(constants.NAMESPACE_PATH): + return constants.IN_CLUSTER_DNS_NAME.format(namespace) + raise ValueError( + 'Could not auto-discover KFP endpoint: the kubernetes package ' + 'is not installed and no in-cluster environment was detected. ' + 'Set base_url in KubernetesBackendConfig or the ' + 'KF_PIPELINES_ENDPOINT environment variable.') + + try: + k8s.config.load_incluster_config() + return constants.IN_CLUSTER_DNS_NAME.format(namespace) + except (k8s.config.ConfigException, FileNotFoundError): + logger.debug('In-cluster config not available.', exc_info=True) + + # Only the host URL is extracted from kubeconfig; kubeconfig auth + # credentials are not applied to the API configuration. This matches + # kfp.Client behavior and assumes kubectl proxy handles auth. + try: + k8s_config = k8s.client.Configuration() + k8s.config.load_kube_config(client_configuration=k8s_config) + if k8s_config.host: + return (k8s_config.host.rstrip('/') + '/' + + constants.KUBE_PROXY_PATH.format(namespace)) + except (k8s.config.ConfigException, FileNotFoundError): + logger.debug('Kubeconfig not available.', exc_info=True) + + fallback = constants.IN_CLUSTER_DNS_NAME.format(namespace) + logger.warning( + 'Could not detect KFP endpoint via in-cluster config or ' + 'kubeconfig. Falling back to %s. Set base_url in ' + 'KubernetesBackendConfig or KF_PIPELINES_ENDPOINT to override.', + fallback) + return fallback + + +def resolve_namespace(configured_namespace: str | None) -> str: + """Return the configured namespace, auto-detecting if needed. + + Resolution order: + 1. Explicitly configured via ``KubernetesBackendConfig.namespace``. + 2. In-cluster: ``/var/run/secrets/kubernetes.io/serviceaccount/namespace``. + 3. Out-of-cluster: namespace from the current kubeconfig context. + 4. Fallback: ``"kubeflow"``. + + Note: Does not read ~/.config/kfp/context.json (used by kfp.Client's + set_user_namespace). This will be implemented in further phases. + """ + if configured_namespace: + return configured_namespace + + try: + with open(constants.NAMESPACE_PATH, 'r') as f: + return f.read().strip() + except FileNotFoundError: + pass + + try: + import kubernetes as k8s + except ImportError: + logger.debug('kubernetes package not installed.') + else: + try: + _, active_context = k8s.config.list_kube_config_contexts() + namespace = active_context.get('context', {}).get('namespace') + if namespace: + logger.debug('Namespace resolved from kubeconfig context: %r.', + namespace) + return namespace + except (k8s.config.ConfigException, FileNotFoundError): + logger.debug( + 'Could not read namespace from kubeconfig.', exc_info=True) + + logger.debug( + 'Namespace not resolved from cluster or kubeconfig; ' + 'using default %r.', constants.DEFAULT_NAMESPACE) + return constants.DEFAULT_NAMESPACE diff --git a/sdk/python/kfp/kubeflow_client/constants.py b/sdk/python/kfp/kubeflow_client/constants.py new file mode 100644 index 00000000000..5cf50822d77 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/constants.py @@ -0,0 +1,45 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Run state constants for PipelinesClient.""" + +__all__ = [ + 'RUN_SUCCEEDED', + 'RUN_FAILED', + 'RUN_SKIPPED', + 'RUN_CANCELED', + 'RUN_CANCELING', + 'RUN_RUNNING', + 'RUN_PENDING', + 'RUN_PAUSED', + 'RUN_COMPLETE', + 'TERMINAL_STATES', +] + +RUN_SUCCEEDED = 'succeeded' +RUN_FAILED = 'failed' +RUN_SKIPPED = 'skipped' +RUN_CANCELED = 'canceled' +RUN_CANCELING = 'canceling' +RUN_RUNNING = 'running' +RUN_PENDING = 'pending' +RUN_PAUSED = 'paused' + +RUN_COMPLETE = RUN_SUCCEEDED + +TERMINAL_STATES = frozenset({ + RUN_SUCCEEDED, + RUN_FAILED, + RUN_SKIPPED, + RUN_CANCELED, +}) diff --git a/sdk/python/kfp/kubeflow_client/types.py b/sdk/python/kfp/kubeflow_client/types.py new file mode 100644 index 00000000000..b12a378beb4 --- /dev/null +++ b/sdk/python/kfp/kubeflow_client/types.py @@ -0,0 +1,42 @@ +# Copyright The Kubeflow Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Type aliases for PipelinesClient. + +These aliases provide clean names over the auto-generated kfp_server_api +model classes used by the KFP backend API. +""" + +import kfp_server_api + +__all__ = [ + 'Pipeline', + 'PipelineVersion', + 'Run', + 'Experiment', + 'ListPipelinesResponse', + 'ListPipelineVersionsResponse', + 'ListRunsResponse', + 'ListExperimentsResponse', +] + +Pipeline = kfp_server_api.V2beta1Pipeline +PipelineVersion = kfp_server_api.V2beta1PipelineVersion +Run = kfp_server_api.V2beta1Run +Experiment = kfp_server_api.V2beta1Experiment + +ListPipelinesResponse = kfp_server_api.V2beta1ListPipelinesResponse +ListPipelineVersionsResponse = ( + kfp_server_api.V2beta1ListPipelineVersionsResponse) +ListRunsResponse = kfp_server_api.V2beta1ListRunsResponse +ListExperimentsResponse = kfp_server_api.V2beta1ListExperimentsResponse