Pipeline Testing Introduced

[yoonspark]

The user may want to check if the generated pipeline files are valid and reliable before actually using them. As a first step to support this (more work in progress/planning), we implemented generation of simple scaffold/template that the user can customize to suit their unique testing needs. The feature is yet to be included in our next release, but you can give it a quick try now if you want:

pip install git+https://github.com/LineaLabs/lineapy.git --upgrade

Overview

Consider the following development code:

import pandas as pd
from sklearn.linear_model import LinearRegression

import lineapy


# Load train data
url1 = "https://raw.githubusercontent.com/LineaLabs/lineapy/main/examples/tutorials/data/iris.csv"
train_df = pd.read_csv(url1)
print(train_df)

# Initiate the model
mod = LinearRegression()

# Fit the model
mod.fit(
    X=train_df[["petal.width"]],
    y=train_df["petal.length"],
)

# Save the fitted model as an artifact
lineapy.save(mod, "iris_model")

# Load data to predict (assume it comes from a different source)
pred_df = pd.read_csv(url1)
print(pred_df)

# Make predictions
petal_length_pred = mod.predict(X=pred_df[["petal.width"]])
print(petal_length_pred)

# Save the predictions
lineapy.save(petal_length_pred, "iris_petal_length_pred")

Now, if we run

lineapy.to_pipeline(
    artifacts=["iris_model", "iris_petal_length_pred"],
    framework="SCRIPT",
    dependencies={"iris_petal_length_pred": {"iris_model"}},
    pipeline_name="iris",
    output_dir="./iris_pipeline/",
)

we would get a pipeline module file looking as follows:

# ./iris_pipeline/iris_module.py

import pandas as pd
from sklearn.linear_model import LinearRegression


def get_url1_for_artifact_iris_model_and_downstream():
    url1 = "https://raw.githubusercontent.com/LineaLabs/lineapy/main/examples/tutorials/data/iris.csv"
    return url1


def get_iris_model(url1):
    train_df = pd.read_csv(url1)
    mod = LinearRegression()
    mod.fit(
        X=train_df[["petal.width"]],
        y=train_df["petal.length"],
    )
    return mod


def get_iris_petal_length_pred(mod, url1):
    pred_df = pd.read_csv(url1)
    petal_length_pred = mod.predict(X=pred_df[["petal.width"]])
    return petal_length_pred

[...]

As shown, the module file includes a transformed version of the original development code where

• Extraneous operations (e.g., print() statements in this example) have been removed
• Duplicate code blocks have been factored out to reduce redundant computations (e.g., get_url1_for_artifact_iris_model_and_downstream() in this example)

With such changes, the user may want to ensure that the transformed code is valid and reliable before actually using it. To support this, the new feature generates simple testing scaffold/template looking as follows:

# ./iris_pipeline/test_iris.py

import unittest
import warnings
from typing import Any, Dict

from iris_module import (
    get_iris_model,
    get_iris_petal_length_pred,
    get_url1_for_artifact_iris_model_and_downstream,
)


class TestIris(unittest.TestCase):
    def setUp(self) -> None:
        # Add any processes to execute before each test in this class
        pass

    def tearDown(self) -> None:
        # Add any processes to execute after each test in this class
        pass

    @classmethod
    def setUpClass(cls) -> None:
        # Add any processes to execute once before all tests in this class run
        pass

    @classmethod
    def tearDownClass(cls) -> None:
        # Add any processes to execute once after all tests in this class run
        pass

    def test_get_url1_for_artifact_iris_model_and_downstream(self) -> None:
        # Adjust as needed
        sample_input: Dict[str, Any] = {}
        try:
            get_url1_for_artifact_iris_model_and_downstream(**sample_input)
        except Exception:
            warnings.warn(
                "Test failed, but this may be due to our limited templating. "
                "Please adapt the test as needed."
            )

    def test_get_iris_model(self) -> None:
        # Adjust as needed
        sample_input: Dict[str, Any] = {}
        sample_input["url1"] = None
        try:
            get_iris_model(**sample_input)
        except Exception:
            warnings.warn(
                "Test failed, but this may be due to our limited templating. "
                "Please adapt the test as needed."
            )

    def test_get_iris_petal_length_pred(self) -> None:
        # Adjust as needed
        sample_input: Dict[str, Any] = {}
        sample_input["mod"] = None
        sample_input["url1"] = None
        try:
            get_iris_petal_length_pred(**sample_input)
        except Exception:
            warnings.warn(
                "Test failed, but this may be due to our limited templating. "
                "Please adapt the test as needed."
            )

As shown, the scaffold contains placeholders for testing each function in the module file, which are meant to be fleshed out by the user to suit their needs. Though very simple, this bare-bones templating still delivers value to the user as it provides a concrete structure for thinking about pipeline testing; and the user can quickly focus on the testing logic itself.

References

• LIN-579 Provide user with pipeline test skeleton #772