mlflow--mlflow/tests/pyfunc/docker/test_docker_flavors.py

import contextlib
import os
import shutil
import sys
import threading
import time

import pandas as pd
import pytest
import requests

import mlflow
from mlflow.environment_variables import _MLFLOW_RUN_SLOW_TESTS
from mlflow.models.flavor_backend_registry import get_flavor_backend
from mlflow.models.utils import load_serving_example

# Only import model fixtures if when MLFLOW_RUN_SLOW_TESTS environment variable is set to true
if _MLFLOW_RUN_SLOW_TESTS.get():
    from tests.catboost.test_catboost_model_export import reg_model  # noqa: F401
    from tests.h2o.test_h2o_model_export import h2o_iris_model  # noqa: F401
    from tests.helper_functions import get_safe_port
    from tests.langchain.test_langchain_model_export import fake_chat_model  # noqa: F401
    from tests.lightgbm.test_lightgbm_model_export import lgb_model  # noqa: F401
    from tests.models.test_model import iris_data, sklearn_knn_model  # noqa: F401
    from tests.pmdarima.test_pmdarima_model_export import (  # noqa: F401
        auto_arima_object_model,
        test_data,
    )
    from tests.prophet.test_prophet_model_export import (
        prophet_model as prophet_raw_model,  # noqa: F401
    )
    from tests.pyfunc.docker.conftest import (
        MLFLOW_ROOT,
        TEST_IMAGE_NAME,
        docker_client,
        save_model_with_latest_mlflow_version,
    )
    from tests.spacy.test_spacy_model_export import spacy_model_with_data  # noqa: F401
    from tests.spark.test_spark_model_export import (  # noqa: F401
        iris_df,
        spark,
        spark_model_iris,
    )
    from tests.statsmodels.model_fixtures import ols_model
    from tests.tensorflow.test_tensorflow2_core_model_export import tf2_toy_model  # noqa: F401
    from tests.transformers.helper import load_text_classification_pipeline


pytestmark = pytest.mark.skipif(
    not _MLFLOW_RUN_SLOW_TESTS.get(),
    reason="Skip slow tests. Set MLFLOW_RUN_SLOW_TESTS environment variable to run them.",
)


@pytest.fixture
def model_path(tmp_path):
    model_path = tmp_path.joinpath("model")

    yield model_path

    # Pytest keeps the temporary directory created by `tmp_path` fixture for 3 recent test sessions
    # by default. This is useful for debugging during local testing, but in CI it just wastes the
    # disk space.
    if os.environ.get("GITHUB_ACTIONS") == "true":
        shutil.rmtree(model_path, ignore_errors=True)


@contextlib.contextmanager
def start_container(port: int):
    container = docker_client.containers.run(
        image=TEST_IMAGE_NAME,
        ports={8080: port},
        detach=True,
    )

    def stream_logs():
        for line in container.logs(stream=True):
            sys.stdout.write(line.decode("utf-8"))

    # Start a thread to stream logs from the container
    t = threading.Thread(name="docker-log-stream", target=stream_logs, daemon=True)
    t.start()

    try:
        # Wait for the server to start
        for _ in range(30):
            try:
                response = requests.get(url=f"http://localhost:{port}/ping")
                if response.ok:
                    break
            except requests.exceptions.ConnectionError as e:
                sys.stdout.write(f"An exception occurred when calling the server: {e}\n")

            container.reload()  # update container status
            if container.status == "exited":
                raise Exception("Container exited unexpectedly.")

            sys.stdout.write(f"Container status: {container.status}\n")
            time.sleep(5)

        else:
            raise TimeoutError("Failed to start server.")

        yield container
    finally:
        container.stop()
        container.remove()
        t.join(timeout=5)


@pytest.mark.parametrize(
    ("flavor"),
    [
        "catboost",
        "h2o",
        # "johnsnowlabs", # Couldn't test JohnSnowLab locally due to license issue
        "keras",
        "langchain",
        "lightgbm",
        "onnx",
        # "openai", # OPENAI API KEY is not necessarily available for everyone
        # "paddle",  # Disabled: https://github.com/PaddlePaddle/PaddleOCR/issues/16402
        "pmdarima",
        "prophet",
        "pyfunc",
        "pytorch",
        "sklearn",
        "spacy",
        "spark",
        "statsmodels",
        "tensorflow",
        "transformers_pt",  # Test with Pytorch-based model
    ],
)
def test_build_image_and_serve(flavor, request):
    model_path = str(request.getfixturevalue(f"{flavor}_model"))
    flavor = flavor.split("_")[0]  # Remove _pt or _tf from the flavor name

    # Build an image
    backend = get_flavor_backend(model_uri=model_path, docker_build=True, env_manager=None)
    backend.build_image(
        model_uri=model_path,
        image_name=TEST_IMAGE_NAME,
        mlflow_home=MLFLOW_ROOT,  # Required to prevent installing dev version of MLflow from PyPI
    )

    # Run a container
    port = get_safe_port()
    with start_container(port):
        # Make a scoring request with a saved serving input example
        inference_payload = load_serving_example(model_path)

        response = requests.post(
            url=f"http://localhost:{port}/invocations",
            data=inference_payload,
            headers={"Content-Type": "application/json"},
        )

        assert response.status_code == 200, f"Response: {response.text}"
        if flavor == "langchain":
            # "messages" key is unified llm input, output is not wrapped into predictions
            assert response.json() == ["Hi"]
        else:
            assert "predictions" in response.json(), f"Response: {response.text}"


@pytest.fixture
def catboost_model(model_path, reg_model):
    save_model_with_latest_mlflow_version(
        flavor="catboost",
        cb_model=reg_model.model,
        path=model_path,
        input_example=reg_model.inference_dataframe[:1],
    )
    return model_path


@pytest.fixture
def h2o_model(model_path, h2o_iris_model):
    save_model_with_latest_mlflow_version(
        flavor="h2o",
        h2o_model=h2o_iris_model.model,
        path=model_path,
        input_example=h2o_iris_model.inference_data.as_data_frame()[:1],
    )
    return model_path


@pytest.fixture
def keras_model(model_path, iris_data):
    from sklearn import datasets
    from tensorflow.keras.layers import Dense
    from tensorflow.keras.models import Sequential

    model = Sequential()
    model.add(Dense(3, input_dim=4))
    model.add(Dense(1))

    X, y = datasets.load_iris(return_X_y=True)
    save_model_with_latest_mlflow_version(
        flavor="tensorflow",
        model=model,
        path=model_path,
        input_example=X[:3, :],
    )
    return model_path


@pytest.fixture
def langchain_model(model_path, tmp_path):
    # LangChain v1+ requires models-from-code
    model_code = """
from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough
import mlflow

mlflow.models.set_model(RunnablePassthrough() | itemgetter("messages"))
"""
    code_path = tmp_path / "langchain_model.py"
    code_path.write_text(model_code)

    save_model_with_latest_mlflow_version(
        flavor="langchain",
        lc_model=str(code_path),
        path=model_path,
        input_example={"messages": "Hi"},
    )
    return model_path


@pytest.fixture
def lightgbm_model(model_path, lgb_model):
    save_model_with_latest_mlflow_version(
        flavor="lightgbm",
        lgb_model=lgb_model.model,
        path=model_path,
        input_example=lgb_model.inference_dataframe.to_numpy()[:1],
    )
    return model_path


@pytest.fixture
def onnx_model(tmp_path, model_path):
    import numpy as np
    import onnx
    import torch
    from torch import nn

    model = torch.nn.Sequential(nn.Linear(4, 3), nn.ReLU(), nn.Linear(3, 1))
    onnx_model_path = os.path.join(tmp_path, "torch_onnx")
    torch.onnx.export(
        model,
        torch.randn(1, 4),
        onnx_model_path,
        dynamic_axes={"input": {0: "batch"}},
        input_names=["input"],
    )
    onnx_model = onnx.load(onnx_model_path)

    model_path = str(tmp_path / "onnx_model")
    save_model_with_latest_mlflow_version(
        flavor="onnx",
        onnx_model=onnx_model,
        path=model_path,
        input_example=np.random.rand(1, 4).astype(np.float32),
    )
    return model_path


# Paddle fixture disabled: https://github.com/PaddlePaddle/PaddleOCR/issues/16402
# @pytest.fixture
# def paddle_model(model_path, pd_model):
#     save_model_with_latest_mlflow_version(
#         flavor="paddle",
#         pd_model=pd_model.model,
#         path=model_path,
#         input_example=pd_model.inference_dataframe[:1],
#     )
#     return model_path


@pytest.fixture
def pmdarima_model(model_path, auto_arima_object_model):
    save_model_with_latest_mlflow_version(
        flavor="pmdarima",
        pmdarima_model=auto_arima_object_model,
        path=model_path,
        input_example=pd.DataFrame({"n_periods": [30]}),
    )
    return model_path


@pytest.fixture
def prophet_model(model_path, prophet_raw_model):
    save_model_with_latest_mlflow_version(
        flavor="prophet",
        pr_model=prophet_raw_model.model,
        path=model_path,
        input_example=prophet_raw_model.data[:1],
        # Prophet does not handle numpy 2 yet. https://github.com/facebook/prophet/issues/2595
        extra_pip_requirements=["numpy<2"],
    )
    return model_path


@pytest.fixture
def pyfunc_model(model_path):
    class CustomModel(mlflow.pyfunc.PythonModel):
        def __init__(self):
            pass

        def predict(self, context, model_input):
            return model_input

    save_model_with_latest_mlflow_version(
        flavor="pyfunc",
        python_model=CustomModel(),
        path=model_path,
        input_example=[1, 2, 3],
    )
    return model_path


@pytest.fixture
def pytorch_model(model_path):
    from torch import nn, randn

    model = nn.Sequential(nn.Linear(4, 3), nn.ReLU(), nn.Linear(3, 1))
    save_model_with_latest_mlflow_version(
        flavor="pytorch",
        pytorch_model=model,
        path=model_path,
        input_example=randn(1, 4).numpy(),
    )
    return model_path


@pytest.fixture
def sklearn_model(model_path, sklearn_knn_model, iris_data):
    save_model_with_latest_mlflow_version(
        flavor="sklearn",
        sk_model=sklearn_knn_model,
        path=model_path,
        input_example=iris_data[0][:1],
    )
    return model_path


@pytest.fixture
def spacy_model(model_path, spacy_model_with_data):
    save_model_with_latest_mlflow_version(
        flavor="spacy",
        spacy_model=spacy_model_with_data.model,
        path=model_path,
        input_example=spacy_model_with_data.inference_data[:1],
    )
    return model_path


@pytest.fixture
def spark_model(model_path, spark_model_iris):
    save_model_with_latest_mlflow_version(
        flavor="spark",
        spark_model=spark_model_iris.model,
        path=model_path,
        input_example=spark_model_iris.spark_df.toPandas()[:1],
    )
    return model_path


@pytest.fixture
def statsmodels_model(model_path):
    model = ols_model()
    save_model_with_latest_mlflow_version(
        flavor="statsmodels",
        statsmodels_model=model.model,
        path=model_path,
        input_example=model.inference_dataframe[:1],
    )
    return model_path


@pytest.fixture
def tensorflow_model(model_path, tf2_toy_model):
    save_model_with_latest_mlflow_version(
        flavor="tensorflow",
        model=tf2_toy_model.model,
        path=model_path,
        input_example=tf2_toy_model.inference_data[:1],
    )
    return model_path


@pytest.fixture
def transformers_pt_model(model_path):
    pipeline = load_text_classification_pipeline()
    save_model_with_latest_mlflow_version(
        flavor="transformers",
        transformers_model=pipeline,
        path=model_path,
        input_example="hi",
    )
    return model_path