aihub-dag/elyra/tests/pipeline/test_pipeline_parser.py

#
# Copyright 2018-2022 Elyra 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.
#
import pytest

from elyra.pipeline.parser import PipelineParser
from elyra.pipeline.pipeline import GenericOperation
from elyra.tests.pipeline.util import _read_pipeline_resource


@pytest.fixture
def valid_operation():
    component_parameters = {
        "filename": "{{filename}}",
        "runtime_image": "{{runtime_image}}",
        "env_vars": ["var1=var1", "var2=var2"],
        "dependencies": ["a.txt", "b.txt", "c.txt"],
        "outputs": ["d.txt", "e.txt", "f.txt"],
    }
    return GenericOperation(
        id="{{uuid}}",
        type="execution_node",
        classifier="execute-notebook-node",
        name="{{label}}",
        component_params=component_parameters,
    )


def test_valid_pipeline(valid_operation):
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")

    pipeline = PipelineParser().parse(pipeline_json)

    assert pipeline.name == "{{name}}"
    assert pipeline.runtime == "{{runtime}}"
    assert pipeline.runtime_config == "{{runtime-config}}"
    assert len(pipeline.operations) == 1
    assert pipeline.operations["{{uuid}}"] == valid_operation


def test_pipeline_with_dirty_list_values(valid_operation):
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_with_invalid_list_values.json")

    pipeline = PipelineParser().parse(pipeline_json)

    assert pipeline.name == "{{name}}"
    assert pipeline.runtime == "{{runtime}}"
    assert pipeline.runtime_config == "{{runtime-config}}"
    assert len(pipeline.operations) == 1
    assert pipeline.operations["{{uuid}}"] == valid_operation


def test_multinode_pipeline():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_3_node_sample.json")

    pipeline = PipelineParser().parse(pipeline_json)

    assert len(pipeline.operations) == 3


def test_supernode_pipeline():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_with_supernode.json")

    pipeline = PipelineParser().parse(pipeline_json)

    assert len(pipeline.operations) == 4

    # Confirm structure of pipeline:
    # Two execution nodes feed their outputs to super-node with one execution_node.
    # Super-node's execution node, then sends its output to external execution node.
    # 4 nodes total.  Super-node execution node should have two parent-operations
    # pointing at first two nodes, and final node should have one parent pointing
    # at execution node WITHIN supernode.

    external_input_node_ids = ["db9f3f5b-b2e3-4824-aadd-c1c6bf652534", "f6584209-6f22-434f-9820-41327b6c749d"]
    supernode_excution_node_id = "079c0e12-eb5f-4fcc-983b-09e011869fee"
    external_node_id = "7628306d-2cc2-405c-94a1-fe42c95567a1"

    for node_id in pipeline.operations:
        # Validate operations list
        if node_id in external_input_node_ids:
            # These are input nodes, ensure parent_operation_ids are empty
            assert len(pipeline.operations[node_id].parent_operation_ids) == 0
            continue
        if node_id == supernode_excution_node_id:
            # Node within supernode, should have two parent_ops matching external_input_node_ids
            assert len(pipeline.operations[node_id].parent_operation_ids) == 2
            assert set(pipeline.operations[node_id].parent_operation_ids) == set(external_input_node_ids)
            continue
        if node_id == external_node_id:
            # Final external node, should have super_node embedded node as parent op.
            assert len(pipeline.operations[node_id].parent_operation_ids) == 1
            assert pipeline.operations[node_id].parent_operation_ids[0] == supernode_excution_node_id
            continue
        assert False, "Invalid node_id encountered in pipeline operations!"


def test_multiple_pipeline_definition():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/" "pipeline_multiple_pipeline_definitions.json")

    with pytest.raises(ValueError):
        PipelineParser().parse(pipeline_json)


def test_pipeline_operations_and_handle_artifact_file_details():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_3_node_sample.json")

    pipeline = PipelineParser().parse(pipeline_json)

    assert len(pipeline.operations) == 3

    for op in pipeline.operations.values():
        assert "." not in op.name


def test_pipeline_with_dependencies():
    pipeline_json = _read_pipeline_resource(
        "resources/sample_pipelines/" "pipeline_3_node_sample_with_dependencies.json"
    )

    pipeline = PipelineParser().parse(pipeline_json)

    assert len(pipeline.operations["acc4527d-7cc8-4c16-b520-5aa0f50a2e34"].parent_operation_ids) == 2


def test_pipeline_with_comments():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/" "pipeline_3_node_sample_with_comments.json")
    pipeline = PipelineParser().parse(pipeline_json)
    assert (
        pipeline.operations["d52ddfb4-dd0e-47ac-abc7-fa30bb95d45c"].doc
        == "Generate community stats and then aggregate them on an overview dashboard"
    )


def test_pipeline_global_attributes():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")

    pipeline = PipelineParser().parse(pipeline_json)

    assert pipeline.name == "{{name}}"
    assert pipeline.runtime == "{{runtime}}"
    assert pipeline.runtime_config == "{{runtime-config}}"


def test_missing_pipeline_name_should_default_to_untitled():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["app_data"]["properties"].pop("name")

    pipeline = PipelineParser().parse(pipeline_json)

    assert pipeline.name == "untitled"


def test_missing_pipeline_runtime():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["app_data"].pop("runtime")

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Invalid pipeline: Missing runtime." in str(e.value)


def test_missing_pipeline_runtime_configuration():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["app_data"].pop("runtime_config")

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Invalid pipeline: Missing runtime configuration" in str(e.value)


def test_missing_operation_id():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["nodes"][0].pop("id")

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Missing field 'operation id'" in str(e.value)


def test_missing_operation_type():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["nodes"][0].pop("type")

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Node type 'None' is invalid!" in str(e.value)


def test_invalid_node_type():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["nodes"][0]["type"] = "foo"

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Node type 'foo' is invalid!" in str(e.value)


def test_missing_operation_filename():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["nodes"][0]["app_data"]["component_parameters"].pop("filename")

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Missing field 'operation filename" in str(e.value)


def test_missing_operation_image():
    pipeline_json = _read_pipeline_resource("resources/sample_pipelines/pipeline_valid.json")
    pipeline_json["pipelines"][0]["nodes"][0]["app_data"]["component_parameters"].pop("runtime_image")

    with pytest.raises(ValueError) as e:
        PipelineParser().parse(pipeline_json)

    assert "Missing field 'operation runtime image'" in str(e.value)