<filename>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)

<filename>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)

# # 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. # # 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. # Validate operations list # These are input nodes, ensure parent_operation_ids are empty # Node within supernode, should have two parent_ops matching external_input_node_ids # Final external node, should have super_node embedded node as parent op.