Add Pytest & Example Pipeline Validation with CI Workflow (#11)

Signed-off-by: VaniHaripriya <[email protected]>

Author: VaniHaripriya

.github/workflows/component-pipeline-tests.yml

@@ -0,0 +1,60 @@
+---
+name: Run Tests and Validate Example Pipelines for Updated Components
+
+on:
+  pull_request:
+    paths:
+      - "components/**"
+      - "pipelines/**"
+      - "scripts/tests/**"
+      - "scripts/validate_examples/**"
+      - ".github/workflows/component-pipeline-tests.yml"
+  push:
+    branches:
+      - main
+      - release-*
+    paths:
+      - "components/**"
+      - "pipelines/**"
+      - "scripts/tests/**"
+      - "scripts/validate_examples/**"
+      - ".github/workflows/component-pipeline-tests.yml"
+
+jobs:
+  targeted-tests:
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python CI
+        uses: ./.github/actions/setup-python-ci
+        with:
+          python-version: "3.11"
+
+      - name: Detect changed files
+        id: changes
+        uses: ./.github/actions/detect-changed-assets
+
+      - name: Run targeted pytest suites
+        if: ${{ steps.changes.outputs.has-changes == 'true' }}
+        run: |
+          echo "Running pytest for the following paths:"
+          PATHS="${{ steps.changes.outputs.changed-components }} ${{ steps.changes.outputs.changed-pipelines }}"
+          echo "$PATHS"
+          uv run python -m scripts.tests.run_component_tests $PATHS
+
+      - name: Validate example pipelines
+        if: ${{ steps.changes.outputs.has-changes == 'true' }}
+        run: |
+          echo "Validating example pipelines for:"
+          PATHS="${{ steps.changes.outputs.changed-components }} ${{ steps.changes.outputs.changed-pipelines }}"
+          echo "$PATHS"
+          uv run python -m scripts.validate_examples $PATHS
+
+      - name: No component changes detected
+        if: ${{ steps.changes.outputs.has-changes != 'true' }}
+        run: echo "No components or pipelines changed. Skipping targeted tests."

docs/BESTPRACTICES.md

@@ -1,11 +1,36 @@
 # Component Development Best Practices
 
-*This guide is under development. Please check back soon for comprehensive best practices for developing Kubeflow
-Pipelines components.*
+The sections below capture the minimum expectations for component and pipeline
+authors. Additional guidance will be added over time as the repository matures.
 
-## Coming Soon
+## Writing Component/Pipeline Smoke Tests
 
-This document will cover:
+- Keep test dependencies limited to `pytest` and the Python standard library.
+- Place tests in a `tests/` folder that sits next to the component or pipeline
+  (for example `components/training/my_component/tests/`).
+- Aim for fast smoke coverage (import checks, signature validation, lightweight
+  golden tests) that finishes well under the two-minute timeout enforced by
+  CI.
+- Run `python scripts/tests/run_component_tests.py <path-to-component>` locally
+  to execute only the suites you are touching. The helper automatically enforces
+  the `pytest-timeout` guardrail used in CI.
+
+## Validating Example Pipelines
+
+- Keep example pipelines in an `example_pipelines.py` module beside the asset.
+  Every pipeline exported from the module must be decorated with
+  `@dsl.pipeline`.
+- The CI workflow imports each module by running
+  `python -m scripts.validate_examples <paths>` for
+  every changed component or pipeline and compiles the matching sample pipelines
+  using `kfp.compiler`. Run the same command locally (with or without explicit
+  paths) before submitting a PR to catch syntax or dependency regressions early.
+- Keep the examples dependency-light; they should compile without extra
+  installations beyond the repo's base tooling (`kfp`, `pytest`, stdlib).
+
+## Additional Topics (Coming Soon)
+
+Future revisions of this guide will capture:
 
 - Component design patterns
 - Performance optimization

pyproject.toml

@@ -30,6 +30,7 @@ test = [
 ]
 ci = [
     "pytest",
+    "pytest-timeout",
     "docstring-parser",
     "jinja2",
     "semver",

scripts/__init__.py

@@ -0,0 +1 @@
+"""Utility scripts for components and pipelines."""

scripts/tests/__init__.py

@@ -0,0 +1 @@
+"""Pytest discovery and execution utilities for component and pipeline tests."""

scripts/tests/run_component_tests.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Pytest discovery helper for Kubeflow components and pipelines.
+
+This script discovers `tests/` directories under the provided component or
+pipeline paths and runs pytest with a two-minute timeout per test.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+import warnings
+from pathlib import Path
+from typing import List, Sequence
+
+import pytest
+
+from ..utils import get_repo_root, normalize_targets
+
+REPO_ROOT = get_repo_root()
+TIMEOUT_SECONDS = 120
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments.
+
+    Returns:
+        Parsed command-line arguments.
+    """
+    parser = argparse.ArgumentParser(
+        description=(
+            "Discover tests/ directories for the specified components or pipelines "
+            "and execute pytest with a two-minute timeout per test."
+        )
+    )
+    parser.add_argument(
+        "paths",
+        metavar="PATH",
+        nargs="*",
+        help=(
+            "Component or pipeline directories (or files within them) to test. "
+            "If omitted, all components and pipelines are scanned."
+        ),
+    )
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=TIMEOUT_SECONDS,
+        help="Per-test timeout in seconds (default: 120).",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Pass the -vv flag to pytest for more detailed output.",
+    )
+    return parser.parse_args()
+
+
+def discover_test_dirs(targets: Sequence[Path]) -> List[Path]:
+    """Discover tests/ directories under the given targets.
+
+    Args:
+        targets: Sequence of component or pipeline paths to search.
+
+    Returns:
+        List of discovered tests/ directory paths.
+    """
+    discovered: List[Path] = []
+
+    for target in targets:
+        search_root = target if target.is_dir() else target.parent
+        if not search_root.exists():
+            continue
+
+        direct = search_root / "tests"
+        if direct.is_dir() and _is_member_of_pipeline_or_component(direct):
+            if direct not in discovered:
+                discovered.append(direct)
+
+    return discovered
+
+
+def _is_member_of_pipeline_or_component(candidate: Path) -> bool:
+    """Check if a path is within components/ or pipelines/ directory.
+
+    Args:
+        candidate: Path to check.
+
+    Returns:
+        True if the path is within components/ or pipelines/, False otherwise.
+    """
+    try:
+        relative = candidate.relative_to(REPO_ROOT)
+    except ValueError:
+        warnings.warn(
+            f"Unable to determine relative path for {candidate} " f"relative to repo root {REPO_ROOT}. Skipping.",
+        )
+        return False
+
+    return relative.parts and relative.parts[0] in {"components", "pipelines"}
+
+
+def build_pytest_args(
+    test_dirs: Sequence[Path],
+    timeout_seconds: int,
+    verbose: bool,
+) -> List[str]:
+    """Build pytest command-line arguments.
+
+    Args:
+        test_dirs: Directories containing tests to run.
+        timeout_seconds: Per-test timeout in seconds.
+        verbose: Whether to enable verbose pytest output.
+
+    Returns:
+        List of pytest command-line arguments.
+    """
+    args: List[str] = [
+        f"--timeout={timeout_seconds}",
+        "--timeout-method=signal",
+    ]
+    if verbose:
+        args.append("-vv")
+
+    args.extend(str(directory) for directory in test_dirs)
+    return args
+
+
+def main() -> int:
+    """Main entry point for running component/pipeline tests.
+
+    Returns:
+        Exit code (0 for success, non-zero for failure).
+    """
+    args = parse_args()
+    targets = normalize_targets(args.paths)
+    test_dirs = discover_test_dirs(targets)
+
+    if not test_dirs:
+        print("No tests/ directories found under the supplied paths. Nothing to do.")
+        return 0
+
+    relative_dirs = ", ".join(str(directory.relative_to(REPO_ROOT)) for directory in test_dirs)
+    print(f"Running pytest for: {relative_dirs}")
+
+    pytest_args = build_pytest_args(
+        test_dirs=test_dirs,
+        timeout_seconds=args.timeout,
+        verbose=args.verbose,
+    )
+
+    exit_code = pytest.main(pytest_args)
+    if exit_code == 0:
+        print("✅ Pytest completed successfully.")
+    else:
+        print("❌ Pytest reported failures. See log above for details.")
+
+    return exit_code
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/utils/__init__.py

@@ -0,0 +1,19 @@
+"""Shared utilities for scripts."""
+
+from .parsing import (
+    find_functions_with_decorator,
+    find_pipeline_functions,
+)
+from .paths import (
+    get_default_targets,
+    get_repo_root,
+    normalize_targets,
+)
+
+__all__ = [
+    "find_functions_with_decorator",
+    "find_pipeline_functions",
+    "get_default_targets",
+    "get_repo_root",
+    "normalize_targets",
+]

scripts/utils/parsing.py

@@ -0,0 +1,97 @@
+"""Parsing-related utility functions."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+from typing import List
+
+
+def get_ast_tree(file_path: Path) -> ast.AST:
+    """Get the parsed AST tree for a Python file.
+
+    Args:
+        file_path: Path to the Python file to parse.
+
+    Returns:
+        The parsed AST tree.
+    """
+    with open(file_path, "r", encoding="utf-8") as f:
+        source = f.read()
+    return ast.parse(source)
+
+
+def is_target_decorator(decorator: ast.AST, decorator_type: str) -> bool:
+    """Check if a decorator is a KFP component or pipeline decorator.
+
+    Supports the following decorator formats (using component as an example):
+    - @component (direct import: from kfp.dsl import component)
+    - @dsl.component (from kfp import dsl)
+    - @kfp.dsl.component (import kfp)
+    - All of the above with parentheses: @component(), @dsl.component(), etc.
+
+    Args:
+        decorator: AST node representing the decorator.
+        decorator_type: Type of decorator to find ('component' or 'pipeline').
+
+    Returns:
+        True if the decorator is the given decoration_type, False otherwise.
+    """
+    if isinstance(decorator, ast.Attribute):
+        # Handle attribute-based decorators
+        if decorator.attr == decorator_type:
+            # Check for @dsl.<function_type>
+            if isinstance(decorator.value, ast.Name) and decorator.value.id == "dsl":
+                return True
+            # Check for @kfp.dsl.<function_type>
+            if (
+                isinstance(decorator.value, ast.Attribute)
+                and decorator.value.attr == "dsl"
+                and isinstance(decorator.value.value, ast.Name)
+                and decorator.value.value.id == "kfp"
+            ):
+                return True
+        return False
+    elif isinstance(decorator, ast.Call):
+        # Handle decorators with arguments (e.g., @<function_type>(), @dsl.<function_type>())
+        return is_target_decorator(decorator.func, decorator_type)
+    elif isinstance(decorator, ast.Name):
+        # Handle @<function_type> (if imported directly)
+        return decorator.id == decorator_type
+    return False
+
+
+def find_pipeline_functions(file_path: Path) -> List[str]:
+    """Find all function names decorated with @dsl.pipeline.
+
+    Args:
+        file_path: Path to the Python file to parse.
+
+    Returns:
+        List of function names that are decorated with @dsl.pipeline.
+    """
+    return find_functions_with_decorator(file_path, "pipeline")
+
+
+def find_functions_with_decorator(file_path: Path, decorator_type: str) -> List[str]:
+    """Find all function names decorated with a specific KFP decorator
+
+    Args:
+        file_path: Path to the Python file to parse.
+        decorator_type: Type of decorator to find ('component' or 'pipeline').
+
+    Returns:
+        List of function names that are decorated with the specified decorator.
+    """
+    tree = get_ast_tree(file_path)
+
+    functions: List[str] = []
+
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            for decorator in node.decorator_list:
+                if is_target_decorator(decorator, decorator_type):
+                    functions.append(node.name)
+                    break  # Found the decorator, no need to check other decorators
+
+    return functions