#
# Copyright 2018-2025 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.
#
"""Tests for elyra-pipeline application"""
import json
from pathlib import Path
import shutil
from typing import List
from typing import Union

from click.testing import CliRunner
from conftest import KFP_COMPONENT_CACHE_INSTANCE
import pytest

from elyra.cli.pipeline_app import pipeline
from elyra.metadata.manager import MetadataManager
from elyra.metadata.metadata import Metadata
from elyra.metadata.schemaspaces import Runtimes
from elyra.pipeline.component_catalog import ComponentCache

# used to drive generic parameter handling tests
SUB_COMMANDS = ["run", "submit", "describe", "validate", "export"]


@pytest.fixture(autouse=True)
def destroy_component_cache():
    """
    This fixture clears any ComponentCache instances that
    may have been created during CLI processes so that
    those instance doesn't side-affect later tests.
    """
    yield
    ComponentCache.clear_instance()


@pytest.fixture
def kubeflow_pipelines_runtime_instance():
    """Creates a Kubeflow Pipelines RTC and removes it after test."""
    instance_name = "valid_kfp_test_config"
    instance_config_file = Path(__file__).parent / "resources" / "runtime_configs" / f"{instance_name}.json"
    with open(instance_config_file, "r") as fd:
        instance_config = json.load(fd)

    md_mgr = MetadataManager(schemaspace=Runtimes.RUNTIMES_SCHEMASPACE_ID)
    # clean possible orphaned instance...
    try:
        md_mgr.remove(instance_name)
    except Exception:
        pass
    runtime_instance = md_mgr.create(instance_name, Metadata(**instance_config))
    yield runtime_instance.name
    md_mgr.remove(runtime_instance.name)


@pytest.fixture
def airflow_runtime_instance():
    """Creates an airflow RTC and removes it after test."""
    instance_name = "valid_airflow_test_config"
    instance_config_file = Path(__file__).parent / "resources" / "runtime_configs" / f"{instance_name}.json"
    with open(instance_config_file, "r") as fd:
        instance_config = json.load(fd)

    md_mgr = MetadataManager(schemaspace=Runtimes.RUNTIMES_SCHEMASPACE_ID)
    # clean possible orphaned instance...
    try:
        md_mgr.remove(instance_name)
    except Exception:
        pass
    runtime_instance = md_mgr.create(instance_name, Metadata(**instance_config))
    yield runtime_instance.name
    md_mgr.remove(runtime_instance.name)


def test_no_opts():
    """Verify that all commands are displayed in help"""
    runner = CliRunner()
    result = runner.invoke(pipeline)
    assert "run       Run a pipeline in your local environment" in result.output
    assert "submit    Submit a pipeline to be executed on the server" in result.output
    assert "describe  Display pipeline summary" in result.output
    assert "export    Export a pipeline to a runtime-specific format" in result.output
    assert "validate  Validate pipeline" in result.output

    assert result.exit_code == 0


def test_bad_subcommand():
    runner = CliRunner()
    result = runner.invoke(pipeline, ["invalid_command"])
    assert "Error: No such command 'invalid_command'" in result.output
    assert result.exit_code != 0


@pytest.mark.parametrize("subcommand", SUB_COMMANDS)
def test_subcommand_no_opts(subcommand):
    runner = CliRunner()
    result = runner.invoke(pipeline, [subcommand])
    assert result.exit_code != 0
    assert "Error: Missing argument 'PIPELINE_PATH'" in result.output


@pytest.mark.parametrize("subcommand", SUB_COMMANDS)
def test_subcommand_invalid_pipeline_path(subcommand):
    """Verify that every command only accepts a valid pipeline_path file name"""
    runner = CliRunner()

    # test: file not found
    file_name = "no-such.pipeline"
    result = runner.invoke(pipeline, [subcommand, file_name])
    assert result.exit_code != 0
    assert f"Invalid value for 'PIPELINE_PATH': '{file_name}' is not a file." in result.output

    # test: file with wrong extension
    with runner.isolated_filesystem():
        file_name = "wrong.extension"
        with open(file_name, "w") as f:
            f.write("I am not a pipeline file.")
        result = runner.invoke(pipeline, [subcommand, file_name])
        assert result.exit_code != 0
        assert f"Invalid value for 'PIPELINE_PATH': '{file_name}' is not a .pipeline file." in result.output


@pytest.mark.parametrize("subcommand", SUB_COMMANDS)
def test_subcommand_with_no_pipelines_field(subcommand, kubeflow_pipelines_runtime_instance):
    """Verify that every command properly detects pipeline issues"""
    runner = CliRunner()
    with runner.isolated_filesystem():
        pipeline_file = "pipeline_without_pipelines_field.pipeline"
        pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / pipeline_file
        assert pipeline_file_path.is_file()

        # every CLI command invocation requires these parameters
        invoke_parameters = [subcommand, str(pipeline_file_path)]
        if subcommand in ["submit", "export"]:
            # these commands also require a runtime configuration
            invoke_parameters.extend(["--runtime-config", kubeflow_pipelines_runtime_instance])

        result = runner.invoke(pipeline, invoke_parameters)
        assert result.exit_code != 0
        assert "Pipeline is missing 'pipelines' field." in result.output


@pytest.mark.parametrize("subcommand", SUB_COMMANDS)
def test_subcommand_with_zero_length_pipelines_field(subcommand, kubeflow_pipelines_runtime_instance):
    """Verify that every command properly detects pipeline issues"""
    runner = CliRunner()
    with runner.isolated_filesystem():
        pipeline_file = "pipeline_with_zero_length_pipelines_field.pipeline"
        pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / pipeline_file
        assert pipeline_file_path.is_file()

        # every CLI command invocation requires these parameters
        invoke_parameters = [subcommand, str(pipeline_file_path)]
        if subcommand in ["submit", "export"]:
            # these commands also require a runtime configuration
            invoke_parameters.extend(["--runtime-config", kubeflow_pipelines_runtime_instance])

        result = runner.invoke(pipeline, invoke_parameters)
        assert result.exit_code != 0
        assert "Pipeline has zero length 'pipelines' field." in result.output


@pytest.mark.parametrize("subcommand", SUB_COMMANDS)
def test_subcommand_with_no_nodes(subcommand, kubeflow_pipelines_runtime_instance):
    """Verify that every command properly detects pipeline issues"""

    # don't run this test for the `describe` command
    # (see test_describe_with_no_nodes)
    if subcommand == "describe":
        return

    runner = CliRunner()
    with runner.isolated_filesystem():
        pipeline_file = "pipeline_with_zero_nodes.pipeline"
        pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / pipeline_file
        assert pipeline_file_path.is_file()

        # every CLI command invocation requires these parameters
        invoke_parameters = [subcommand, str(pipeline_file_path)]
        if subcommand in ["submit", "export"]:
            # these commands also require a runtime configuration
            invoke_parameters.extend(["--runtime-config", kubeflow_pipelines_runtime_instance])

        result = runner.invoke(pipeline, invoke_parameters)
        assert result.exit_code != 0


# ------------------------------------------------------------------
# tests for 'describe' command
# ------------------------------------------------------------------


def test_describe_with_no_nodes():
    """
    Verify that describe yields the expected results if a pipeline without any
    nodes is is provided as input.
    """
    runner = CliRunner()
    with runner.isolated_filesystem():
        pipeline_file = "pipeline_with_zero_nodes.pipeline"
        pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / pipeline_file
        assert pipeline_file_path.is_file()

        # verify human-readable output
        result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
        assert result.exit_code == 0, result.output
        assert "Pipeline name: pipeline_with_zero_nodes" in result.output
        assert "Description: None specified" in result.output
        assert "Pipeline type: None specified" in result.output
        assert "Pipeline runtime: Generic" in result.output
        assert "Pipeline format version: 7" in result.output
        assert "Number of generic nodes: 0" in result.output
        assert "Number of generic nodes: 0" in result.output
        assert "Script dependencies: None specified" in result.output
        assert "Notebook dependencies: None specified" in result.output
        assert "Local file dependencies: None specified" in result.output
        assert "Component dependencies: None specified" in result.output
        assert "Volume dependencies: None specified" in result.output
        assert "Container image dependencies: None specified" in result.output
        assert "Kubernetes secret dependencies: None specified" in result.output

        # verify machine-readable output
        result = runner.invoke(pipeline, ["describe", str(pipeline_file_path), "--json"])
        assert result.exit_code == 0, result.output
        result_json = json.loads(result.output)
        assert result_json["name"] == "pipeline_with_zero_nodes"
        assert result_json["description"] is None
        assert result_json["pipeline_type"] is None
        assert result_json["pipeline_format_version"] == 7
        assert result_json["pipeline_runtime"] == "Generic"
        assert result_json["generic_node_count"] == 0
        assert result_json["custom_node_count"] == 0
        for property in [
            "scripts",
            "notebooks",
            "files",
            "custom_components",
            "container_images",
            "volumes",
            "kubernetes_secrets",
        ]:
            assert isinstance(result_json["dependencies"][property], list)
            assert len(result_json["dependencies"][property]) == 0


def test_describe_with_kfp_components():
    runner = CliRunner()
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"

    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert "Description: 3-node custom component pipeline" in result.output
    assert "Pipeline type: KUBEFLOW_PIPELINES" in result.output
    assert "Number of custom nodes: 3" in result.output
    assert "Number of generic nodes: 0" in result.output
    assert "Local file dependencies: None specified" in result.output
    assert (
        '- {"catalog_type": "elyra-kfp-examples-catalog", "component_ref": {"component-id": "download_data.yaml"}}'
        in result.output
    )
    assert (
        '- {"catalog_type": "elyra-kfp-examples-catalog", "component_ref": '
        '{"component-id": "filter_text_using_shell_and_grep.yaml"}}' in result.output
    )
    assert (
        '- {"catalog_type": "elyra-kfp-examples-catalog", "component_ref": {"component-id": "calculate_hash.yaml"}}'
        in result.output
    )
    assert result.exit_code == 0


def test_describe_with_missing_kfp_component():
    runner = CliRunner()
    with runner.isolated_filesystem():
        valid_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
        pipeline_file_path = Path.cwd() / "foo.pipeline"
        with open(pipeline_file_path, "w") as pipeline_file:
            with open(valid_file_path) as valid_file:
                valid_data = json.load(valid_file)
                # Update known component name to trigger a missing component
                valid_data["pipelines"][0]["nodes"][0]["op"] = valid_data["pipelines"][0]["nodes"][0]["op"] + "Missing"
                pipeline_file.write(json.dumps(valid_data))

        result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
        assert "Description: 3-node custom component pipeline" in result.output
        assert "Pipeline type: KUBEFLOW_PIPELINES" in result.output
        assert "Number of custom nodes: 3" in result.output
        assert "Number of generic nodes: 0" in result.output

        assert result.exit_code == 0


def test_describe_notebooks_scripts_report():
    """
    Test human-readable output for notebooks/scripts property when none, one or many instances are present
    """
    runner = CliRunner()

    #
    # Pipeline references no notebooks/no scripts:
    # - Pipeline does not contain nodes -> test_describe_with_no_nodes
    # - Pipeline contains only script nodes
    # - Pipeline contains only notebook nodes
    # - Pipeline contains only custom components

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Notebook dependencies: None specified" in result.output

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Script dependencies: None specified" in result.output

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Notebook dependencies: None specified" in result.output
    assert "Script dependencies: None specified" in result.output

    #
    # Pipeline references multiple notebooks:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Notebook dependencies:\n" in result.output
    assert "notebooks/notebook_1.ipynb" in result.output
    assert "notebooks/notebook_2.ipynb" in result.output
    # Ensure no entries for scripts
    assert "Script dependencies: None specified" in result.output
    assert "Number of generic nodes: 2" in result.output
    assert "Number of custom nodes: 0" in result.output

    #
    # Pipeline references multiple scripts:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Script dependencies:\n" in result.output
    assert "scripts/script_1.py" in result.output
    assert "scripts/script_2.py" in result.output
    assert "scripts/script_3.py" in result.output
    # Ensure no entries for notebooks
    assert "Notebook dependencies: None specified" in result.output
    assert "Number of generic nodes: 3" in result.output
    assert "Number of custom nodes: 0" in result.output

    #
    # Pipeline references multiple notebooks and scripts:
    #
    pipeline_file_path = (
        Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks_and_scripts.pipeline"
    )
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Notebook dependencies:\n" in result.output
    assert "notebooks/notebook_1.ipynb" in result.output
    assert "notebooks/notebook_2.ipynb" in result.output
    assert "Script dependencies:\n" in result.output
    assert "scripts/script_1.py" in result.output
    assert "scripts/script_2.py" in result.output
    assert "scripts/script_3.py" in result.output
    assert "Number of generic nodes: 5" in result.output
    assert "Number of custom nodes: 0" in result.output


def test_describe_notebooks_scripts_json():
    """
    Test machine-readable output for notebooks/scripts property when none, one or many instances are present
    """
    runner = CliRunner()

    #
    # Pipeline references no notebooks/no scripts:
    # - Pipeline does not contain nodes -> test_describe_with_no_nodes
    # - Pipeline contains only script nodes
    # - Pipeline contains only notebook nodes
    # - Pipeline contains only custom components

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 3
    assert result_json["custom_node_count"] == 0
    dependencies = result_json.get("dependencies")
    assert isinstance(dependencies.get("notebooks"), list)
    assert len(dependencies.get("notebooks")) == 0

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 2
    assert result_json["custom_node_count"] == 0
    dependencies = result_json.get("dependencies")
    assert isinstance(dependencies.get("scripts"), list)
    assert len(dependencies.get("scripts")) == 0

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 0
    assert result_json["custom_node_count"] == 3
    dependencies = result_json.get("dependencies")
    assert isinstance(dependencies.get("notebooks"), list)
    assert len(dependencies.get("notebooks")) == 0
    assert isinstance(dependencies.get("scripts"), list)
    assert len(dependencies.get("scripts")) == 0

    #
    # Pipeline references multiple notebooks:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 2
    assert result_json["custom_node_count"] == 0
    dependencies = result_json.get("dependencies")
    assert isinstance(dependencies.get("notebooks"), list)
    assert any(x.endswith("notebooks/notebook_1.ipynb") for x in dependencies["notebooks"]), dependencies["notebooks"]
    assert any(x.endswith("notebooks/notebook_2.ipynb") for x in dependencies["notebooks"]), dependencies["notebooks"]
    # Ensure no entries for scripts
    assert isinstance(dependencies.get("scripts"), list)
    assert len(dependencies.get("scripts")) == 0

    #
    # Pipeline references multiple scripts:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 3
    assert result_json["custom_node_count"] == 0
    dependencies = result_json.get("dependencies")
    assert isinstance(dependencies.get("scripts"), list)
    assert len(dependencies.get("scripts")) == 3
    assert any(x.endswith("scripts/script_1.py") for x in dependencies["scripts"]), dependencies["scripts"]
    assert any(x.endswith("scripts/script_2.py") for x in dependencies["scripts"]), dependencies["scripts"]
    assert any(x.endswith("scripts/script_3.py") for x in dependencies["scripts"]), dependencies["scripts"]

    # Ensure no entries for notebooks
    assert isinstance(dependencies.get("notebooks"), list)
    assert len(dependencies.get("notebooks")) == 0

    #
    # Pipeline references multiple notebooks and scripts:
    #
    pipeline_file_path = (
        Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks_and_scripts.pipeline"
    )
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 5
    assert result_json["custom_node_count"] == 0
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json.get("dependencies")
    assert isinstance(dependencies.get("scripts"), list)
    assert len(dependencies.get("scripts")) == 3
    assert any(x.endswith("scripts/script_1.py") for x in dependencies["scripts"]), dependencies["scripts"]
    assert any(x.endswith("scripts/script_2.py") for x in dependencies["scripts"]), dependencies["scripts"]
    assert any(x.endswith("scripts/script_3.py") for x in dependencies["scripts"]), dependencies["scripts"]
    assert isinstance(dependencies.get("notebooks"), list)
    assert len(dependencies.get("notebooks")) == 2
    assert any(x.endswith("notebooks/notebook_1.ipynb") for x in dependencies["notebooks"]), dependencies["notebooks"]
    assert any(x.endswith("notebooks/notebook_2.ipynb") for x in dependencies["notebooks"]), dependencies["notebooks"]


def test_describe_container_images_report():
    """
    Test report output for container_images property when none, one or many instances are present
    """
    runner = CliRunner()

    #
    # Pipeline references no container images
    # - Pipeline does not contain nodes -> test_describe_with_no_nodes
    # - Pipeline contains only custom components

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Container image dependencies: None specified" in result.output

    #
    # Pipeline references multiple container images through notebook nodes:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Container image dependencies:\n" in result.output
    assert "- tensorflow/tensorflow:2.8.0" in result.output, result.output

    #
    # Pipeline references multiple container images through script nodes
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Container image dependencies:\n" in result.output, result.output
    assert "- tensorflow/tensorflow:2.8.0-gpu" in result.output, result.output
    assert "- tensorflow/tensorflow:2.8.0" in result.output, result.output

    #
    # Pipeline references multiple notebooks and scripts:
    #
    pipeline_file_path = (
        Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks_and_scripts.pipeline"
    )
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Container image dependencies:\n" in result.output
    assert "- tensorflow/tensorflow:2.8.0-gpu" in result.output, result.output
    assert "- tensorflow/tensorflow:2.8.0" in result.output, result.output
    assert "- amancevice/pandas:1.4.1" in result.output, result.output


def test_describe_container_images_json():
    """
    Test JSON output for runtime_images property when none, one or many instances are present
    """
    runner = CliRunner()

    #
    # Pipeline references no container images
    # - Pipeline does not contain nodes -> test_describe_with_no_nodes
    # - Pipeline contains only custom components

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 0
    assert result_json["custom_node_count"] == 3
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert isinstance(dependencies["container_images"], list)
    assert len(dependencies["container_images"]) == 0

    #
    # Pipeline references multiple container images through notebook nodes:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 2
    assert result_json["custom_node_count"] == 0
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert isinstance(dependencies["container_images"], list)
    assert len(dependencies["container_images"]) == 1
    assert "tensorflow/tensorflow:2.8.0" in dependencies["container_images"], dependencies["container_images"]

    #
    # Pipeline references multiple container images through script nodes
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 3
    assert result_json["custom_node_count"] == 0
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert isinstance(dependencies["container_images"], list)
    assert len(dependencies["container_images"]) == 2
    assert "tensorflow/tensorflow:2.8.0" in dependencies["container_images"], dependencies["container_images"]
    assert "tensorflow/tensorflow:2.8.0-gpu" in dependencies["container_images"], dependencies["container_images"]

    #
    # Pipeline references multiple notebooks and scripts:
    #
    pipeline_file_path = (
        Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks_and_scripts.pipeline"
    )
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 5
    assert result_json["custom_node_count"] == 0
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert isinstance(dependencies["container_images"], list)
    assert len(dependencies["container_images"]) == 3
    assert "tensorflow/tensorflow:2.8.0" in dependencies["container_images"], dependencies["container_images"]
    assert "tensorflow/tensorflow:2.8.0-gpu" in dependencies["container_images"], dependencies["container_images"]
    assert "amancevice/pandas:1.4.1" in dependencies["container_images"], dependencies["container_images"]


def test_describe_volumes_report():
    """
    Test report format output for volumes property when none, one or many volume mounts are present
    """
    runner = CliRunner()

    #
    # Pipeline references no volumes
    # - Pipeline does not contain nodes -> test_describe_with_no_nodes
    # - Pipeline contains only custom components
    # - No generic nodes mount a volume

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Volume dependencies: None specified" in result.output

    #
    # Pipeline references multiple volumes through notebook nodes:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Volume dependencies:\n" in result.output
    assert "- pvc-claim-1" in result.output, result.output

    #
    # Pipeline references multiple volumes through script nodes
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Volume dependencies:\n" in result.output, result.output
    assert "- pvc-claim-2" in result.output, result.output
    assert "- pvc-claim-3" in result.output, result.output

    #
    # Pipeline references multiple volumes through notebook and script nodes:
    #
    pipeline_file_path = (
        Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks_and_scripts.pipeline"
    )
    result = runner.invoke(pipeline, ["describe", str(pipeline_file_path)])
    assert result.exit_code == 0
    assert "Volume dependencies:\n" in result.output, result.output
    assert "- pvc-claim-1" in result.output, result.output
    assert "- pvc-claim-2" in result.output, result.output
    assert "- pvc-claim-3" in result.output, result.output


def test_describe_volumes_json():
    """
    Test JSON output for volumes property when none, one or many volume mounts are present
    """
    runner = CliRunner()

    #
    # Pipeline references no volumes
    # - Pipeline does not contain nodes -> test_describe_with_no_nodes
    # - Pipeline contains only custom components
    # - No generic nodes mount a volume

    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "kfp_3_node_custom.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 0
    assert result_json["custom_node_count"] == 3
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert isinstance(dependencies["volumes"], list)
    assert len(dependencies["volumes"]) == 0

    #
    # Pipeline references multiple volumes through notebook nodes:
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 2
    assert result_json["custom_node_count"] == 0
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert len(dependencies["volumes"]) == 1
    assert "pvc-claim-1" in dependencies["volumes"], dependencies["volumes"]

    #
    # Pipeline references multiple volumes through script nodes
    #
    pipeline_file_path = Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_scripts.pipeline"
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
    assert result_json["generic_node_count"] == 3
    assert result_json["custom_node_count"] == 0
    assert isinstance(result_json.get("dependencies"), dict)
    dependencies = result_json["dependencies"]
    assert len(dependencies["volumes"]) == 2
    assert "pvc-claim-2" in dependencies["volumes"], dependencies["volumes"]
    assert "pvc-claim-3" in dependencies["volumes"], dependencies["volumes"]

    #
    # Pipeline references multiple volumes through notebook and script nodes:
    #
    pipeline_file_path = (
        Path(__file__).parent / "resources" / "pipelines" / "pipeline_with_notebooks_and_scripts.pipeline"
    )
    result = runner.invoke(pipeline, ["describe", "--json", str(pipeline_file_path)])
    assert result.exit_code == 0
    result_json = json.loads(result.output)
