test_flow.py

# License: BSD 3-Clause
from __future__ import annotations

import collections
import copy
import hashlib
import re
import os
import time
from packaging.version import Version
from unittest import mock

import pytest
import scipy.stats
import sklearn
import sklearn.datasets
import sklearn.decomposition
import sklearn.dummy
import sklearn.ensemble
import sklearn.feature_selection
import sklearn.model_selection
import sklearn.naive_bayes
import sklearn.pipeline
import sklearn.preprocessing
import sklearn.tree
import xmltodict

from openml_sklearn import SklearnExtension

import openml
import openml.exceptions
import openml.utils
from openml._api_calls import _perform_api_call
from openml.testing import SimpleImputer, TestBase


class TestFlow(TestBase):
    _multiprocess_can_split_ = True

    def setUp(self):
        super().setUp()
        self.extension = SklearnExtension()

    def tearDown(self):
        super().tearDown()

    @pytest.mark.production_server()
    def test_get_flow(self):
        # We need to use the production server here because 4024 is not the
        # test server
        self.use_production_server()

        flow = openml.flows.get_flow(4024)
        assert isinstance(flow, openml.OpenMLFlow)
        assert flow.flow_id == 4024
        assert len(flow.parameters) == 24
        assert len(flow.components) == 1

        subflow_1 = next(iter(flow.components.values()))
        assert isinstance(subflow_1, openml.OpenMLFlow)
        assert subflow_1.flow_id == 4025
        assert len(subflow_1.parameters) == 14
        assert subflow_1.parameters["E"] == "CC"
        assert len(subflow_1.components) == 1

        subflow_2 = next(iter(subflow_1.components.values()))
        assert isinstance(subflow_2, openml.OpenMLFlow)
        assert subflow_2.flow_id == 4026
        assert len(subflow_2.parameters) == 13
        assert subflow_2.parameters["I"] == "10"
        assert len(subflow_2.components) == 1

        subflow_3 = next(iter(subflow_2.components.values()))
        assert isinstance(subflow_3, openml.OpenMLFlow)
        assert subflow_3.flow_id == 1724
        assert len(subflow_3.parameters) == 11
        assert subflow_3.parameters["L"] == "-1"
        assert len(subflow_3.components) == 0

    @pytest.mark.production_server()
    @pytest.mark.xfail(reason="failures_issue_1544", strict=False)
    def test_get_structure(self):
        # also responsible for testing: flow.get_subflow
        # We need to use the production server here because 4024 is not the
        # test server
        self.use_production_server()

        flow = openml.flows.get_flow(4024)
        flow_structure_name = flow.get_structure("name")
        flow_structure_id = flow.get_structure("flow_id")
        # components: root (filteredclassifier), multisearch, loginboost,
        # reptree
        assert len(flow_structure_name) == 4
        assert len(flow_structure_id) == 4

        for sub_flow_name, structure in flow_structure_name.items():
            if len(structure) > 0:  # skip root element
                subflow = flow.get_subflow(structure)
                assert subflow.name == sub_flow_name

        for sub_flow_id, structure in flow_structure_id.items():
            if len(structure) > 0:  # skip root element
                subflow = flow.get_subflow(structure)
                assert subflow.flow_id == sub_flow_id

    @pytest.mark.test_server()
    def test_tagging(self):
        flows = openml.flows.list_flows(size=1)
        flow_id = flows["id"].iloc[0]
        flow = openml.flows.get_flow(flow_id)
        # tags can be at most 64 alphanumeric (+ underscore) chars
        unique_indicator = str(time.time()).replace(".", "")
        tag = f"test_tag_TestFlow_{unique_indicator}"
        flows = openml.flows.list_flows(tag=tag)
        assert len(flows) == 0
        flow.push_tag(tag)
        flows = openml.flows.list_flows(tag=tag)
        assert len(flows) == 1
        assert flow_id in flows["id"]
        flow.remove_tag(tag)
        flows = openml.flows.list_flows(tag=tag)
        assert len(flows) == 0

    @pytest.mark.test_server()
    def test_from_xml_to_xml(self):
        # Get the raw xml thing
        # TODO maybe get this via get_flow(), which would have to be refactored
        # to allow getting only the xml dictionary
        # TODO: no sklearn flows.
        for flow_id in [
            3,
            5,
            7,
            9,
        ]:
            flow_xml = _perform_api_call("flow/%d" % flow_id, request_method="get")
            flow_dict = xmltodict.parse(flow_xml)

            flow = openml.OpenMLFlow._from_dict(flow_dict)
            new_xml = flow._to_xml()

            flow_xml = (
                flow_xml.replace("  ", "")
                .replace("\t", "")
                .strip()
                .replace("\n\n", "\n")
                .replace(""", '"')
            )
            flow_xml = re.sub(r"^$", "", flow_xml)
            new_xml = (
                new_xml.replace("  ", "")
                .replace("\t", "")
                .strip()
                .replace("\n\n", "\n")
                .replace(""", '"')
            )
            new_xml = re.sub(r"^$", "", new_xml)

            assert new_xml == flow_xml

    @pytest.mark.sklearn()
    def test_to_xml_from_xml(self):
        scaler = sklearn.preprocessing.StandardScaler(with_mean=False)
        estimator_name = (
            "base_estimator" if Version(sklearn.__version__) < Version("1.4") else "estimator"
        )
        boosting = sklearn.ensemble.AdaBoostClassifier(
            **{estimator_name: sklearn.tree.DecisionTreeClassifier()},
        )
        model = sklearn.pipeline.Pipeline(steps=(("scaler", scaler), ("boosting", boosting)))
        flow = self.extension.model_to_flow(model)
        flow.flow_id = -234
        # end of setup

        xml = flow._to_xml()
        xml_dict = xmltodict.parse(xml)
        new_flow = openml.flows.OpenMLFlow._from_dict(xml_dict)

        # Would raise exception if they are not legal
        openml.flows.functions.assert_flows_equal(new_flow, flow)
        assert new_flow is not flow

    @pytest.mark.sklearn()
    @pytest.mark.test_server()
    def test_publish_flow(self):
        flow = openml.OpenMLFlow(
            name="sklearn.dummy.DummyClassifier",
            class_name="sklearn.dummy.DummyClassifier",
            description="test description",
            model=sklearn.dummy.DummyClassifier(),
            components=collections.OrderedDict(),
            parameters=collections.OrderedDict(),
            parameters_meta_info=collections.OrderedDict(),
            external_version=self.extension._format_external_version(
                "sklearn",
                sklearn.__version__,
            ),
            tags=[],
            language="English",
            dependencies=None,
        )

        flow, _ = self._add_sentinel_to_flow_name(flow, None)

        flow.publish()
        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow.flow_id}")
        assert isinstance(flow.flow_id, int)

    @pytest.mark.sklearn()
    @mock.patch("openml.flows.functions.flow_exists")
    def test_publish_existing_flow(self, flow_exists_mock):
        clf = sklearn.tree.DecisionTreeClassifier(max_depth=2)
        flow = self.extension.model_to_flow(clf)
        flow_exists_mock.return_value = 1

        with pytest.raises(openml.exceptions.PyOpenMLError, match="OpenMLFlow already exists"):
            flow.publish(raise_error_if_exists=True)

        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(
            f"collected from {__file__.split('/')[-1]}: {flow.flow_id}",
        )

    @pytest.mark.sklearn()
    @pytest.mark.test_server()
    def test_publish_flow_with_similar_components(self):
        clf = sklearn.ensemble.VotingClassifier(
            [("lr", sklearn.linear_model.LogisticRegression(solver="lbfgs"))],
        )
        flow = self.extension.model_to_flow(clf)
        flow, _ = self._add_sentinel_to_flow_name(flow, None)
        flow.publish()
        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow.flow_id}")
        # For a flow where both components are published together, the upload
        # date should be equal
        assert flow.upload_date == flow.components["lr"].upload_date, (
            flow.name,
            flow.flow_id,
            flow.components["lr"].name,
            flow.components["lr"].flow_id,
        )

        clf1 = sklearn.tree.DecisionTreeClassifier(max_depth=2)
        flow1 = self.extension.model_to_flow(clf1)
        flow1, sentinel = self._add_sentinel_to_flow_name(flow1, None)
        flow1.publish()
        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow1.flow_id}")

        # In order to assign different upload times to the flows!
        time.sleep(1)

        clf2 = sklearn.ensemble.VotingClassifier(
            [("dt", sklearn.tree.DecisionTreeClassifier(max_depth=2))],
        )
        flow2 = self.extension.model_to_flow(clf2)
        flow2, _ = self._add_sentinel_to_flow_name(flow2, sentinel)
        flow2.publish()
        TestBase._mark_entity_for_removal("flow", flow2.flow_id, flow2.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow2.flow_id}")
        # If one component was published before the other, the components in
        # the flow should have different upload dates
        assert flow2.upload_date != flow2.components["dt"].upload_date

        clf3 = sklearn.ensemble.AdaBoostClassifier(sklearn.tree.DecisionTreeClassifier(max_depth=3))
        flow3 = self.extension.model_to_flow(clf3)
        flow3, _ = self._add_sentinel_to_flow_name(flow3, sentinel)
        # Child flow has different parameter. Check for storing the flow
        # correctly on the server should thus not check the child's parameters!
        flow3.publish()
        TestBase._mark_entity_for_removal("flow", flow3.flow_id, flow3.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow3.flow_id}")

    @pytest.mark.sklearn()
    @pytest.mark.test_server()
    def test_semi_legal_flow(self):
        # TODO: Test if parameters are set correctly!
        # should not throw error as it contains two differentiable forms of
        # Bagging i.e., Bagging(Bagging(J48)) and Bagging(J48)
        estimator_name = (
            "base_estimator" if Version(sklearn.__version__) < Version("1.4") else "estimator"
        )
        semi_legal = sklearn.ensemble.BaggingClassifier(
            **{
                estimator_name: sklearn.ensemble.BaggingClassifier(
                    **{
                        estimator_name: sklearn.tree.DecisionTreeClassifier(),
                    }
                )
            }
        )
        flow = self.extension.model_to_flow(semi_legal)
        flow, _ = self._add_sentinel_to_flow_name(flow, None)

        flow.publish()
        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow.flow_id}")

    @pytest.mark.sklearn()
    @mock.patch("openml.flows.functions.get_flow")
    @mock.patch("openml.flows.functions.flow_exists")
    @mock.patch("openml._api_calls._perform_api_call")
    def test_publish_error(self, api_call_mock, flow_exists_mock, get_flow_mock):
        model = sklearn.ensemble.RandomForestClassifier()
        flow = self.extension.model_to_flow(model)
        api_call_mock.return_value = (
            "<oml:upload_flow>\n" "    <oml:id>1</oml:id>\n" "</oml:upload_flow>"
        )
        flow_exists_mock.return_value = False
        get_flow_mock.return_value = flow

        flow.publish()
        # Not collecting flow_id for deletion since this is a test for failed upload

        assert api_call_mock.call_count == 1
        assert get_flow_mock.call_count == 1
        assert flow_exists_mock.call_count == 1

        flow_copy = copy.deepcopy(flow)
        flow_copy.name = flow_copy.name[:-1]
        get_flow_mock.return_value = flow_copy
        flow_exists_mock.return_value = 1

        if Version(sklearn.__version__) < Version("0.22"):
            fixture = (
                "The flow on the server is inconsistent with the local flow. "
                "The server flow ID is 1. Please check manually and remove "
                "the flow if necessary! Error is:\n"
                "'Flow sklearn.ensemble.forest.RandomForestClassifier: "
                "values for attribute 'name' differ: "
                "'sklearn.ensemble.forest.RandomForestClassifier'"
                "\nvs\n'sklearn.ensemble.forest.RandomForestClassifie'.'"
            )
        else:
            # sklearn.ensemble.forest -> sklearn.ensemble._forest
            fixture = (
                "The flow on the server is inconsistent with the local flow. "
                "The server flow ID is 1. Please check manually and remove "
                "the flow if necessary! Error is:\n"
                "'Flow sklearn.ensemble._forest.RandomForestClassifier: "
                "values for attribute 'name' differ: "
                "'sklearn.ensemble._forest.RandomForestClassifier'"
                "\nvs\n'sklearn.ensemble._forest.RandomForestClassifie'.'"
            )
        with pytest.raises(ValueError, match=fixture):
            flow.publish()

        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(
            f"collected from {__file__.split('/')[-1]}: {flow.flow_id}",
        )

        assert get_flow_mock.call_count == 2

    @pytest.mark.sklearn()
    def test_illegal_flow(self):
        # should throw error as it contains two imputers
        illegal = sklearn.pipeline.Pipeline(
            steps=[
                ("imputer1", SimpleImputer()),
                ("imputer2", SimpleImputer()),
                ("classif", sklearn.tree.DecisionTreeClassifier()),
            ],
        )
        self.assertRaises(ValueError, self.extension.model_to_flow, illegal)

    @pytest.mark.test_server()
    def test_nonexisting_flow_exists(self):
        def get_sentinel():
            # Create a unique prefix for the flow. Necessary because the flow
            # is identified by its name and external version online. Having a
            # unique name allows us to publish the same flow in each test run
            md5 = hashlib.md5()
            md5.update(str(time.time()).encode("utf-8"))
            sentinel = md5.hexdigest()[:10]
            return f"TEST{sentinel}"

        name = get_sentinel() + get_sentinel()
        version = get_sentinel()

        flow_id = openml.flows.flow_exists(name, version)
        assert not flow_id

    @pytest.mark.sklearn()
    @pytest.mark.test_server()
    def test_existing_flow_exists(self):
        # create a flow
        nb = sklearn.naive_bayes.GaussianNB()

        sparse = "sparse" if Version(sklearn.__version__) < Version("1.4") else "sparse_output"
        ohe_params = {sparse: False, "handle_unknown": "ignore"}
        if Version(sklearn.__version__) >= Version("0.20"):
            ohe_params["categories"] = "auto"
        steps = [
            ("imputation", SimpleImputer(strategy="median")),
            ("hotencoding", sklearn.preprocessing.OneHotEncoder(**ohe_params)),
            (
                "variencethreshold",
                sklearn.feature_selection.VarianceThreshold(),
            ),
            ("classifier", sklearn.tree.DecisionTreeClassifier()),
        ]
        complicated = sklearn.pipeline.Pipeline(steps=steps)

        for classifier in [nb, complicated]:
            flow = self.extension.model_to_flow(classifier)
            flow, _ = self._add_sentinel_to_flow_name(flow, None)
            # publish the flow
            flow = flow.publish()
            TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
            TestBase.logger.info(
                f"collected from {__file__.split('/')[-1]}: {flow.flow_id}",
            )
            # redownload the flow
            flow = openml.flows.get_flow(flow.flow_id)

            # check if flow exists can find it
            flow = openml.flows.get_flow(flow.flow_id)
            downloaded_flow_id = openml.flows.flow_exists(
                flow.name,
                flow.external_version,
            )
            assert downloaded_flow_id == flow.flow_id

    @pytest.mark.sklearn()
    @pytest.mark.test_server()
    def test_sklearn_to_upload_to_flow(self):
        iris = sklearn.datasets.load_iris()
        X = iris.data
        y = iris.target

        # Test a more complicated flow
        ohe_params = {"handle_unknown": "ignore"}
        if Version(sklearn.__version__) >= Version("0.20"):
            ohe_params["categories"] = "auto"
        ohe = sklearn.preprocessing.OneHotEncoder(**ohe_params)
        scaler = sklearn.preprocessing.StandardScaler(with_mean=False)
        pca = sklearn.decomposition.TruncatedSVD()
        fs = sklearn.feature_selection.SelectPercentile(
            score_func=sklearn.feature_selection.f_classif,
            percentile=30,
        )
        fu = sklearn.pipeline.FeatureUnion(transformer_list=[("pca", pca), ("fs", fs)])
        estimator_name = (
            "base_estimator" if Version(sklearn.__version__) < Version("1.4") else "estimator"
        )
        boosting = sklearn.ensemble.AdaBoostClassifier(
            **{estimator_name: sklearn.tree.DecisionTreeClassifier()},
        )
        model = sklearn.pipeline.Pipeline(
            steps=[("ohe", ohe), ("scaler", scaler), ("fu", fu), ("boosting", boosting)],
        )
        parameter_grid = {
            "boosting__n_estimators": [1, 5, 10, 100],
            "boosting__learning_rate": scipy.stats.uniform(0.01, 0.99),
            f"boosting__{estimator_name}__max_depth": scipy.stats.randint(1, 10),
        }
        cv = sklearn.model_selection.StratifiedKFold(n_splits=5, shuffle=True)
        rs = sklearn.model_selection.RandomizedSearchCV(
            estimator=model,
            param_distributions=parameter_grid,
            cv=cv,
        )
        rs.fit(X, y)
        flow = self.extension.model_to_flow(rs)
        # Tags may be sorted in any order (by the server). Just using one tag
        # makes sure that the xml comparison does not fail because of that.
        subflows = [flow]
        while len(subflows) > 0:
            f = subflows.pop()
            f.tags = []
            subflows.extend(list(f.components.values()))

        flow, sentinel = self._add_sentinel_to_flow_name(flow, None)

        flow.publish()
        TestBase._mark_entity_for_removal("flow", flow.flow_id, flow.name)
        TestBase.logger.info(f"collected from {__file__.split('/')[-1]}: {flow.flow_id}")
        assert isinstance(flow.flow_id, int)

        # Check whether we can load the flow again
        # Remove the sentinel from the name again so that we can reinstantiate
        # the object again
        new_flow = openml.flows.get_flow(flow_id=flow.flow_id, reinstantiate=True)

        local_xml = flow._to_xml()
        server_xml = new_flow._to_xml()

        for _i in range(10):
            # Make sure that we replace all occurences of two newlines
            local_xml = local_xml.replace(sentinel, "")
            local_xml = (
                local_xml.replace("  ", "")
                .replace("\t", "")
                .strip()
                .replace("\n\n", "\n")
                .replace("&quot;", '"')
            )
            local_xml = re.sub(r"(^$)", "", local_xml)
            server_xml = server_xml.replace(sentinel, "")
            server_xml = (
                server_xml.replace("  ", "")
                .replace("\t", "")
                .strip()
                .replace("\n\n", "\n")
                .replace("&quot;", '"')
            )
            server_xml = re.sub(r"^$", "", server_xml)

        assert server_xml == local_xml

        # Would raise exception if they are not equal!
        openml.flows.functions.assert_flows_equal(new_flow, flow)
        assert new_flow is not flow

        # OneHotEncoder was moved to _encoders module in 0.20
        module_name_encoder = (
            "_encoders" if Version(sklearn.__version__) >= Version("0.20") else "data"
        )
        if Version(sklearn.__version__) < Version("0.22"):
            fixture_name = (
                f"{sentinel}sklearn.model_selection._search.RandomizedSearchCV("
                "estimator=sklearn.pipeline.Pipeline("
                f"ohe=sklearn.preprocessing.{module_name_encoder}.OneHotEncoder,"
                "scaler=sklearn.preprocessing.data.StandardScaler,"
                "fu=sklearn.pipeline.FeatureUnion("
                "pca=sklearn.decomposition.truncated_svd.TruncatedSVD,"
                "fs="
                "sklearn.feature_selection.univariate_selection.SelectPercentile),"
                "boosting=sklearn.ensemble.weight_boosting.AdaBoostClassifier("
                "base_estimator=sklearn.tree.tree.DecisionTreeClassifier)))"
            )
        else:
            # sklearn.sklearn.preprocessing.data -> sklearn.sklearn.preprocessing._data
            # sklearn.sklearn.decomposition.truncated_svd -> sklearn.decomposition._truncated_svd
            # sklearn.feature_selection.univariate_selection ->
            #     sklearn.feature_selection._univariate_selection
            # sklearn.ensemble.weight_boosting -> sklearn.ensemble._weight_boosting
            # sklearn.tree.tree.DecisionTree... -> sklearn.tree._classes.DecisionTree...
            fixture_name = (
                f"{sentinel}sklearn.model_selection._search.RandomizedSearchCV("
                "estimator=sklearn.pipeline.Pipeline("
                f"ohe=sklearn.preprocessing.{module_name_encoder}.OneHotEncoder,"
                "scaler=sklearn.preprocessing._data.StandardScaler,"
                "fu=sklearn.pipeline.FeatureUnion("
                "pca=sklearn.decomposition._truncated_svd.TruncatedSVD,"
                "fs="
                "sklearn.feature_selection._univariate_selection.SelectPercentile),"
                "boosting=sklearn.ensemble._weight_boosting.AdaBoostClassifier("
                f"{estimator_name}=sklearn.tree._classes.DecisionTreeClassifier)))"
            )
        assert new_flow.name == fixture_name
        new_flow.model.fit(X, y)

    def test_extract_tags(self):
        flow_xml = "<oml:tag>study_14</oml:tag>"
        flow_dict = xmltodict.parse(flow_xml)
        tags = openml.utils.extract_xml_tags("oml:tag", flow_dict)
        assert tags == ["study_14"]

        flow_xml = "<oml:flow><oml:tag>OpenmlWeka</oml:tag>\n" "<oml:tag>weka</oml:tag></oml:flow>"
        flow_dict = xmltodict.parse(flow_xml)
        tags = openml.utils.extract_xml_tags("oml:tag", flow_dict["oml:flow"])
        assert tags == ["OpenmlWeka", "weka"]

    @pytest.mark.production_server()
    def test_download_non_scikit_learn_flows(self):
        self.use_production_server()

        flow = openml.flows.get_flow(6742)
        assert isinstance(flow, openml.OpenMLFlow)
        assert flow.flow_id == 6742
        assert len(flow.parameters) == 19
        assert len(flow.components) == 1
        assert flow.model is None

        subflow_1 = next(iter(flow.components.values()))
        assert isinstance(subflow_1, openml.OpenMLFlow)
        assert subflow_1.flow_id == 6743
        assert len(subflow_1.parameters) == 8
        assert subflow_1.parameters["U"] == "0"
        assert len(subflow_1.components) == 1
        assert subflow_1.model is None

        subflow_2 = next(iter(subflow_1.components.values()))
        assert isinstance(subflow_2, openml.OpenMLFlow)
        assert subflow_2.flow_id == 5888
        assert len(subflow_2.parameters) == 4
        assert subflow_2.parameters["batch-size"] is None
        assert len(subflow_2.components) == 0
        assert subflow_2.model is None