#
# 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.
#
from datetime import datetime
import hashlib
import os
from pathlib import Path
import re
from typing import Any
from typing import Dict

import pytest
import yaml

from elyra.pipeline.catalog_connector import FilesystemComponentCatalogConnector
from elyra.pipeline.component import Component
from elyra.pipeline.kfp.kfp_processor import KfpPipelineProcessor
from elyra.pipeline.kfp.kfp_processor import RUN_ID_PLACEHOLDER
from elyra.pipeline.kfp.kfp_processor import WorkflowEngineType
from elyra.pipeline.kfp.kfp_properties import KfpPipelineParameter
from elyra.pipeline.pipeline import GenericOperation
from elyra.pipeline.pipeline import Operation
from elyra.pipeline.pipeline import Pipeline
from elyra.pipeline.pipeline_constants import KUBERNETES_SHARED_MEM_SIZE
from elyra.pipeline.pipeline_constants import MOUNTED_VOLUMES
from elyra.pipeline.processor import PipelineProcessor
from elyra.pipeline.properties import ComponentProperty
from elyra.pipeline.properties import CustomSharedMemorySize
from elyra.pipeline.properties import DisableNodeCaching
from elyra.pipeline.properties import ElyraPropertyList
from elyra.pipeline.properties import KubernetesAnnotation
from elyra.pipeline.properties import KubernetesLabel
from elyra.pipeline.properties import KubernetesSecret
from elyra.pipeline.properties import KubernetesToleration
from elyra.pipeline.properties import VolumeMount
from elyra.util.kubernetes import sanitize_label_value

PIPELINE_FILE_COMPLEX = str((Path("resources") / "sample_pipelines" / "pipeline_dependency_complex.json").as_posix())


@pytest.fixture
def processor() -> KfpPipelineProcessor:
    """
    Instantiate a Kubeflow Pipelines processor.
    """
    root_dir = str((Path(__file__).parent / "..").resolve())
    processor = KfpPipelineProcessor(root_dir=root_dir)
    return processor


@pytest.fixture
def processor_with_factory_data(setup_factory_data) -> KfpPipelineProcessor:
    """
    Instantiate a Kubeflow Pipelines processor and create
    system-owned runtime image configurations. This simulates the behavior
    of installing and running Elyra.
    """
    root_dir = str((Path(__file__).parent / "..").resolve())
    processor = KfpPipelineProcessor(root_dir=root_dir)
    return processor


# ---------------------------------------------------
# Tests for class WorkflowEngineType
# ---------------------------------------------------


def test_WorkflowEngineType_get_instance_by_value():
    """
    Validate that static method 'WorkflowEngineType.get_instance_by_value' yields the expected results for
    valid and invalid input.
    """
    # test valid inputs (the provided value is evalutaed in a case insensitive manner)
    assert WorkflowEngineType.get_instance_by_value("argo") == WorkflowEngineType.ARGO
    assert WorkflowEngineType.get_instance_by_value("ARGO") == WorkflowEngineType.ARGO
    assert WorkflowEngineType.get_instance_by_value("aRGo") == WorkflowEngineType.ARGO
    assert WorkflowEngineType.get_instance_by_value("Argo") == WorkflowEngineType.ARGO
    # test invalid inputs
    with pytest.raises(KeyError):
        WorkflowEngineType.get_instance_by_value(None)  # there is no default
    with pytest.raises(KeyError):
        WorkflowEngineType.get_instance_by_value("")  # there is no default
    with pytest.raises(KeyError):
        WorkflowEngineType.get_instance_by_value(" argo ")  # whitespaces are not trimmed
    with pytest.raises(KeyError):
        WorkflowEngineType.get_instance_by_value("bitcoin")
    with pytest.raises(KeyError):
        WorkflowEngineType.get_instance_by_value("ether")


# ---------------------------------------------------
# Test method KfpPipelineProcessor._compose_container_command_args
# ---------------------------------------------------


def test_compose_container_command_args(processor: KfpPipelineProcessor):
    """
    Verify that _compose_container_command_args yields the expected output for valid input
    """

    pipeline_name = "test pipeline"
    cos_endpoint = "https://minio:9000"
    cos_bucket = "test_bucket"
    cos_directory = "a_dir"
    cos_dependencies_archive = "dummy-notebook-0815.tar.gz"
    filename = "dummy-notebook.ipynb"

    command_args = processor._compose_container_command_args(
        pipeline_name=pipeline_name,
        cos_endpoint=cos_endpoint,
        cos_bucket=cos_bucket,
        cos_directory=cos_directory,
        cos_dependencies_archive=cos_dependencies_archive,
        filename=filename,
    )
    command_args = "".join(command_args)
    assert f"--pipeline-name '{pipeline_name}'" in command_args
    assert f"--cos-endpoint '{cos_endpoint}'" in command_args
    assert f"--cos-bucket '{cos_bucket}'" in command_args
    assert f"--cos-directory '{cos_directory}'" in command_args
    assert f"--cos-dependencies-archive '{cos_dependencies_archive}'" in command_args
    assert f"--file '{filename}'" in command_args

    assert "--inputs" not in command_args
    assert "--outputs" not in command_args

    # verify correct handling of file dependencies and file outputs
    for file_dependency in [[], ["input_file.txt"], ["input_file.txt", "input_file_2.txt"]]:
        for file_output in [[], ["output.csv"], ["output_1.csv", "output_2.pdf"]]:
            command_args = processor._compose_container_command_args(
                pipeline_name=pipeline_name,
                cos_endpoint=cos_endpoint,
                cos_bucket=cos_bucket,
                cos_directory=cos_directory,
                cos_dependencies_archive=cos_dependencies_archive,
                filename=filename,
                cos_inputs=file_dependency,
                cos_outputs=file_output,
            )
            command_args = "".join(command_args)

            if len(file_dependency) < 1:
                assert "--inputs" not in command_args
            else:
                assert f"--inputs '{';'.join(file_dependency)}'" in command_args

            if len(file_output) < 1:
                assert "--outputs" not in command_args
            else:
                assert f"--outputs '{';'.join(file_output)}'" in command_args


def test_compose_container_command_args_invalid_dependency_filename(processor: KfpPipelineProcessor):
    """
    Verify that _compose_container_command_args fails if one or more of the
    specified input file dependencies contains the reserved separator character
    """

    pipeline_name = "test pipeline"
    cos_endpoint = "https://minio:9000"
    cos_bucket = "test_bucket"
    cos_directory = "a_dir"
    cos_dependencies_archive = "dummy-notebook-0815.tar.gz"
    filename = "dummy-notebook.ipynb"

    reserved_separator_char = ";"

    for file_dependency in [
        [f"input_file{reserved_separator_char}txt"],
        ["input_file.txt", f"input{reserved_separator_char}_file_2.txt"],
    ]:
        # identify invalid file dependency name
        invalid_file_name = [file for file in file_dependency if reserved_separator_char in file][0]
        for file_output in [[], ["output.csv"], ["output_1.csv", "output_2.pdf"]]:
            with pytest.raises(
                ValueError,
                match=re.escape(
                    f"Illegal character ({reserved_separator_char}) found in list item '{invalid_file_name}'."
                ),
            ):
                command_args = processor._compose_container_command_args(
                    pipeline_name=pipeline_name,
                    cos_endpoint=cos_endpoint,
                    cos_bucket=cos_bucket,
                    cos_directory=cos_directory,
                    cos_dependencies_archive=cos_dependencies_archive,
                    filename=filename,
                    cos_inputs=file_dependency,
                    cos_outputs=file_output,
                )
                assert command_args is None


# ---------------------------------------------------
# Tests for methods
#  - KfpPipelineProcessor._add_disable_node_caching
#  - KfpPipelineProcessor._add_custom_shared_memory_size
#  - KfpPipelineProcessor._add_kubernetes_secret
#  - KfpPipelineProcessor._add_mounted_volume
#  - KfpPipelineProcessor._add_kubernetes_pod_annotation
#  - KfpPipelineProcessor._add_kubernetes_pod_label
#  - KfpPipelineProcessor._add_kubernetes_toleration
# ---------------------------------------------------


def test_add_disable_node_caching(processor: KfpPipelineProcessor):
    """
    Verify that add_disable_node_caching updates the execution object as expected
    """
    execution_object = {}
    for instance in [
        DisableNodeCaching("True"),
        DisableNodeCaching("False"),
    ]:
        processor.add_disable_node_caching(instance=instance, execution_object=execution_object)
        assert execution_object.get("disable_node_caching") is instance.selection
    assert len(execution_object.keys()) == 1


def test_add_custom_shared_memory_size(processor):
    """
    Verify that add_custom_shared_memory_size updates the execution object as expected
    """
    execution_object = {}
    for instance in [CustomSharedMemorySize(None, None), CustomSharedMemorySize("", None)]:
        processor.add_custom_shared_memory_size(instance=instance, execution_object=execution_object)
        assert execution_object.get("kubernetes_shared_mem_size") is None

    for instance in [
        CustomSharedMemorySize("0.5", None),
        CustomSharedMemorySize("3.14", "G"),
        CustomSharedMemorySize("256", "M"),
    ]:
        processor.add_custom_shared_memory_size(instance=instance, execution_object=execution_object)
        assert execution_object["kubernetes_shared_mem_size"]["size"] == instance.size
        assert execution_object["kubernetes_shared_mem_size"]["units"] == instance.units


def test_add_kubernetes_secret(processor: KfpPipelineProcessor):
    """
    Verify that add_kubernetes_secret updates the execution object as expected
    """
    execution_object = {}
    for instance in [
        KubernetesSecret("var", "secret_name", "secret_key"),
        KubernetesSecret("var2", "secret_name", "secret_key"),
        KubernetesSecret("var", "secret_name_2", "secret_key_2"),
    ]:
        processor.add_kubernetes_secret(instance=instance, execution_object=execution_object)
        assert execution_object["kubernetes_secrets"][instance.env_var]["name"] == instance.name
        assert execution_object["kubernetes_secrets"][instance.env_var]["key"] == instance.key

    # given above instances, there should be two entries in the modified execution_object
    assert len(execution_object["kubernetes_secrets"].keys()) == 2


def test_add_mounted_volume(processor: KfpPipelineProcessor):
    """
    Verify that add_mounted_volume updates the execution object as expected
    """
    execution_object = {}
    for instance in [
        VolumeMount("/mount/path", "test-pvc", None, None),
        VolumeMount("/mount/path2", "test-pvc-2", None, True),
        VolumeMount("/mount/path3", "test-pvc-3", None, False),
        VolumeMount("/mount/path4", "test-pvc-4", "sub/path", True),
        VolumeMount("/mount/path", "test-pvc", None, True),
    ]:
        processor.add_mounted_volume(instance=instance, execution_object=execution_object)
        assert execution_object["kubernetes_volumes"][instance.path]["pvc_name"] == instance.pvc_name
        assert execution_object["kubernetes_volumes"][instance.path]["sub_path"] == instance.sub_path
        assert execution_object["kubernetes_volumes"][instance.path]["read_only"] == instance.read_only

    # given above instances, there should be four entries in the modified execution_object
    assert len(execution_object["kubernetes_volumes"].keys()) == 4


def test_add_kubernetes_pod_annotation(processor: KfpPipelineProcessor):
    """
    Verify that add_kubernetes_pod_annotation updates the execution object as expected
    """
    execution_object = {}
    for instance in [
        KubernetesAnnotation("annotation-key", None),
        KubernetesAnnotation("prefix/annotation-key-2", ""),
        KubernetesAnnotation("annotation-key-3", "annotation value"),
        KubernetesAnnotation("annotation-key-3", "another annotation value"),
    ]:
        processor.add_kubernetes_pod_annotation(instance=instance, execution_object=execution_object)
        if instance.value is not None:
            assert execution_object["pod_annotations"][instance.key] == instance.value
        else:
            assert execution_object["pod_annotations"][instance.key] == ""

    # given above instances, there should be three entries in the modified execution_object
    assert len(execution_object["pod_annotations"].keys()) == 3


def test_add_kubernetes_pod_label(processor: KfpPipelineProcessor):
    """
    Verify that add_kubernetes_pod_label updates the execution object as expected
    """
    execution_object = {}
    for instance in [
        KubernetesLabel("label-key", None),
        KubernetesLabel("label-key-2", ""),
        KubernetesLabel("label-key-3", "label-value"),
        KubernetesLabel("label-key-2", "a-different-label-value"),
    ]:
        processor.add_kubernetes_pod_label(instance=instance, execution_object=execution_object)
        if instance.value is not None:
            assert execution_object["pod_labels"][instance.key] == instance.value
        else:
            assert execution_object["pod_labels"][instance.key] == ""

    # given above instances, there should be three entries in the modified execution_object
    assert len(execution_object["pod_labels"].keys()) == 3


def test_add_kubernetes_toleration(processor: KfpPipelineProcessor):
    """
    Verify that add_kubernetes_toleration updates the execution object as expected
    """
    execution_object = {}
    expected_unique_execution_object_entries = []
    for instance in [
        KubernetesToleration("toleration-key", "Exists", None, "NoExecute"),
        KubernetesToleration("toleration-key", "Equals", 42, ""),
    ]:
        processor.add_kubernetes_toleration(instance=instance, execution_object=execution_object)
        toleration_hash = hashlib.sha256(
            f"{instance.key}::{instance.operator}::{instance.value}::{instance.effect}".encode()
        ).hexdigest()
        if toleration_hash not in expected_unique_execution_object_entries:
            expected_unique_execution_object_entries.append(toleration_hash)
        assert execution_object["kubernetes_tolerations"][toleration_hash]["key"] == instance.key
        assert execution_object["kubernetes_tolerations"][toleration_hash]["value"] == instance.value
        assert execution_object["kubernetes_tolerations"][toleration_hash]["operator"] == instance.operator
        assert execution_object["kubernetes_tolerations"][toleration_hash]["effect"] == instance.effect
    assert len(expected_unique_execution_object_entries) == len(execution_object["kubernetes_tolerations"].keys())


# ---------------------------------------------------
# Tests for methods
#  - KfpPipelineProcessor._generate_pipeline_dsl
#  - KfpPipelineProcessor._compile_pipeline_dsl
# ---------------------------------------------------


def test_generate_pipeline_dsl_compile_pipeline_dsl_custom_component_pipeline(
    processor: KfpPipelineProcessor, component_cache, tmpdir
):
    """
    Verify that _generate_pipeline_dsl and _compile_pipeline_dsl yield
    the expected output for pipeline the includes a custom component
    """

    # load test component definition
    component_def_path = Path(__file__).parent / ".." / "resources" / "components" / "download_data.yaml"

    # Read contents of given path -- read_component_definition() returns a
    # a dictionary of component definition content indexed by path
    reader = FilesystemComponentCatalogConnector([".yaml"])
    entry_data = reader.get_entry_data({"path": str(component_def_path.absolute())}, {})
    component_definition = entry_data.definition

    properties = [
        ComponentProperty(
            id="url",
            name="Url",
            json_data_type="string",
            value="",
            description="",
            allowed_input_types=["file", "inputpath", "inputvalue"],
        ),
        ComponentProperty(
            id="curl_options",
            name="Curl Options",
            json_data_type="string",
            value="--location",
            description="Additional options given to the curl program",
            allowed_input_types=["file", "inputpath", "inputvalue"],
        ),
    ]

    # Instantiate a file-based component
    component_id = "test-component"
    component = Component(
        id=component_id,
        name="Download data",
        description="download data from web",
        op="download-data",
        catalog_type="elyra-kfp-examples-catalog",
        component_reference={"path": component_def_path.as_posix()},
        definition=component_definition,
        properties=properties,
        categories=[],
    )

    # Fabricate the component cache to include single filename-based component for testing
    component_cache._component_cache[processor._type.name] = {
        "spoofed_catalog": {"components": {component_id: component}}
    }

    # Construct operation for component
    operation_name = "Download data test"
    operation_params = {
        "url": {
            "widget": "string",
            "value": "https://raw.githubusercontent.com/elyra-ai/examples/"
            "main/pipelines/run-pipelines-on-kubeflow-pipelines/data/data.csv",
        },
        "curl_options": {"widget": "string", "value": "--location"},
    }
    operation = Operation(
        id="download-data-id",
        type="execution_node",
        classifier=component_id,
        name=operation_name,
        parent_operation_ids=[],
        component_props=operation_params,
    )

    # Construct single-operation pipeline
    pipeline = Pipeline(
        id="pipeline-id",
        name="code-gen-test-custom-components",
        description="Test code generation for custom components",
        runtime="kfp",
        runtime_config="test",
        source="download_data.pipeline",
    )
    pipeline.operations[operation.id] = operation

    # generate Python DSL for the Argo workflow engine
    generated_argo_dsl = processor._generate_pipeline_dsl(
        pipeline=pipeline, pipeline_name=pipeline.name, workflow_engine=WorkflowEngineType.ARGO
    )

    assert generated_argo_dsl is not None
    # Generated DSL includes workflow engine specific code in the _main_ function
    assert "kfp.compiler.Compiler().compile(" in generated_argo_dsl

    compiled_argo_output_file = Path(tmpdir) / "compiled_kfp_test_argo.yaml"

    # make sure the output file does not exist (3.8+ use unlink("missing_ok=True"))
    if compiled_argo_output_file.is_file():
        compiled_argo_output_file.unlink()

    # if the compiler discovers an issue with the generated DSL this call fails
    processor._compile_pipeline_dsl(
        dsl=generated_argo_dsl,
        workflow_engine=WorkflowEngineType.ARGO,
        output_file=compiled_argo_output_file.as_posix(),
        pipeline_conf=None,
    )

    # verify that the output file exists
    assert compiled_argo_output_file.is_file()

    # verify the file content
    with open(compiled_argo_output_file) as fh:
        argo_spec = yaml.safe_load(fh.read())

    assert argo_spec["pipelineInfo"]["name"] == pipeline.name
    assert argo_spec["pipelineInfo"]["description"] == pipeline.description


@pytest.mark.parametrize(
    "metadata_dependencies",
    [
        {
            "pipeline_file": Path(__file__).parent
            / ".."
            / "resources"
            / "test_pipelines"
            / "kfp"
            / "kfp-one-node-generic.pipeline",
            "workflow_engine": WorkflowEngineType.ARGO,
        },
    ],
    indirect=True,
)
def test_generate_pipeline_dsl_compile_pipeline_dsl_workflow_engine_test(
    monkeypatch, processor: KfpPipelineProcessor, metadata_dependencies: Dict[str, Any], tmpdir
):
    """
    This test validates the following:
     - _generate_pipeline_dsl generates Python code for the supported workflow engines
     - _compile_pipeline_dsl compiles the generated code using the workflow engine's compiler

    This test does not validate that the output artifacts correctly reflect the test pipeline.
    Other tests do that.
    """

    # Obtain artifacts from metadata_dependencies fixture
    test_pipeline_file = metadata_dependencies["pipeline_file"]
    pipeline = metadata_dependencies["pipeline_object"]
    assert pipeline is not None
    runtime_config = metadata_dependencies["runtime_config"]
    assert runtime_config is not None
    assert runtime_config.name == pipeline.runtime_config

    workflow_engine = WorkflowEngineType.get_instance_by_value(runtime_config.metadata["engine"])

    # Mock calls that require access to object storage, because their side effects
    # have no bearing on the outcome of this test.
    monkeypatch.setattr(processor, "_upload_dependencies_to_object_store", lambda w, x, y, prefix: True)
    monkeypatch.setattr(processor, "_verify_cos_connectivity", lambda x: True)

    compiled_output_file = Path(tmpdir) / test_pipeline_file.with_suffix(".yaml").name
    compiled_output_file_name = str(compiled_output_file.absolute())

    # generate Python DSL for the specified workflow engine
    pipeline_version = f"{pipeline.name}-test-0"
    pipeline_instance_id = f"{pipeline.name}-{datetime.now().strftime('%m%d%H%M%S')}"
    experiment_name = f"{pipeline.name}-test-0"
    generated_dsl = processor._generate_pipeline_dsl(
        pipeline=pipeline,
        pipeline_name=pipeline.name,
        workflow_engine=workflow_engine,
        pipeline_version=pipeline_version,
        pipeline_instance_id=pipeline_instance_id,
        experiment_name=experiment_name,
    )

    # Compile the generated Python DSL
    processor._compile_pipeline_dsl(
        dsl=generated_dsl,
        workflow_engine=workflow_engine,
        output_file=compiled_output_file_name,
        pipeline_conf=None,
    )

    # Load compiled workflow
    with open(compiled_output_file_name) as f:
        workflow_spec_docs = list(yaml.safe_load_all(f.read()))

    assert len(workflow_spec_docs) == 2
    assert "components" in workflow_spec_docs[0]
    assert "platforms" in workflow_spec_docs[1]


@pytest.mark.parametrize(
    "metadata_dependencies",
    [
        {
            "pipeline_file": Path(__file__).parent
            / ".."
            / "resources"
            / "test_pipelines"
            / "kfp"
            / "kfp-one-node-generic.pipeline",
            "workflow_engine": WorkflowEngineType.ARGO,
            "with_disable_node_caching": True,
        },
        {
            "pipeline_file": Path(__file__).parent
            / ".."
            / "resources"
            / "test_pipelines"
            / "kfp"
            / "kfp-one-node-generic.pipeline",
            "workflow_engine": WorkflowEngineType.ARGO,
            "with_disable_node_caching": False,
        },
    ],
    indirect=True,
)
def test_disable_node_caching_on_compiled_pipeline(
    monkeypatch, processor: KfpPipelineProcessor, metadata_dependencies: Dict[str, Any], tmpdir
):
    """
    This test validates the disable_node_caching option all the way to the compiled pipeline.
    """

    # Obtain artifacts from metadata_dependencies fixture
    test_pipeline_file = metadata_dependencies["pipeline_file"]
    pipeline = metadata_dependencies["pipeline_object"]
    assert pipeline is not None
    runtime_config = metadata_dependencies["runtime_config"]
    assert runtime_config is not None
    assert runtime_config.name == pipeline.runtime_config

    disable_node_caching = metadata_dependencies["fixture_parameters"]["with_disable_node_caching"]

    workflow_engine = WorkflowEngineType.get_instance_by_value(runtime_config.metadata["engine"])

    # Mock calls that require access to object storage, because their side effects
    # have no bearing on the outcome of this test.
    monkeypatch.setattr(processor, "_upload_dependencies_to_object_store", lambda w, x, y, prefix: True)
    monkeypatch.setattr(processor, "_verify_cos_connectivity", lambda x: True)

    compiled_output_file = Path(tmpdir) / test_pipeline_file.with_suffix(".yaml").name
    compiled_output_file_name = str(compiled_output_file.absolute())

    # generate Python DSL for the specified workflow engine
    pipeline_version = f"{pipeline.name}-test-0"
    pipeline_instance_id = f"{pipeline.name}-{datetime.now().strftime('%m%d%H%M%S')}"
    experiment_name = f"{pipeline.name}-test-0"
    generated_dsl = processor._generate_pipeline_dsl(
        pipeline=pipeline,
        pipeline_name=pipeline.name,
        workflow_engine=workflow_engine,
        pipeline_version=pipeline_version,
        pipeline_instance_id=pipeline_instance_id,
        experiment_name=experiment_name,
    )

    if disable_node_caching:
        assert "set_caching_options(enable_caching=False)" in generated_dsl
    else:
        # Default behavior is to enable node caching
        assert "set_caching_options" not in generated_dsl

    # Compile the generated Python DSL
    processor._compile_pipeline_dsl(
        dsl=generated_dsl,
        workflow_engine=workflow_engine,
        output_file=compiled_output_file_name,
        pipeline_conf=None,
    )

    # Load compiled workflow
    with open(compiled_output_file_name) as f:
        workflow_spec_docs = list(yaml.safe_load_all(f.read()))

    assert len(workflow_spec_docs) == 2
    assert "components" in workflow_spec_docs[0]
    assert "platforms" in workflow_spec_docs[1]

    caching_options = workflow_spec_docs[0]["root"]["dag"]["tasks"]["run-a-file"]["cachingOptions"]
    if disable_node_caching:
        assert caching_options == {}
    else:
        assert caching_options["enableCache"] is True


@pytest.mark.parametrize(
    "metadata_dependencies",
    [
        {
            "pipeline_file": Path(__file__).parent
            / ".."
            / "resources"
            / "test_pipelines"
            / "kfp"
            / "kfp-one-node-generic.pipeline",
            "workflow_engine": WorkflowEngineType.ARGO,
        },
    ],
    indirect=True,
)
def test_generate_pipeline_dsl_compile_pipeline_dsl_one_generic_node_pipeline_test_1(
    monkeypatch, processor: KfpPipelineProcessor, metadata_dependencies: Dict[str, Any], tmpdir
):
    """
    This test validates that the output of _generate_pipeline_dsl and _compile_pipeline_dsl
    yields the expected results for a generic node that has only the required inputs defined.

    This test covers:
     - the Argo workflow engine
     - runtime configurations that use cloud storage authentication types KUBERNETES_SECRET
       and USER_CREDENTIALS (the generated code varies depending on the selected type)

    Other tests cover the scenarios where the user defined optional properties,
    such as environment variables, Kubernetes labels, or data volumes.
    """

    # Obtain artifacts from metadata_dependencies fixture
    test_pipeline_file = metadata_dependencies["pipeline_file"]
    pipeline = metadata_dependencies["pipeline_object"]
    assert pipeline is not None
    runtime_config = metadata_dependencies["runtime_config"]
    assert runtime_config is not None
    assert runtime_config.name == pipeline.runtime_config

    workflow_engine = WorkflowEngineType.get_instance_by_value(runtime_config.metadata["engine"])

    # Make sure this is a one generic node pipeline
    assert len(pipeline.operations.keys()) == 1
    assert isinstance(list(pipeline.operations.values())[0], GenericOperation)
    # Use 'op' variable to access the operation
    op = list(pipeline.operations.values())[0]

    # Mock calls that require access to object storage, because their side effects
    # have no bearing on the outcome of this test.
    monkeypatch.setattr(processor, "_upload_dependencies_to_object_store", lambda w, x, y, prefix: True)
    monkeypatch.setattr(processor, "_verify_cos_connectivity", lambda x: True)

    # Mock pipeline to not include any parameters
    monkeypatch.setattr(pipeline, "_pipeline_parameters", ElyraPropertyList([]))

    compiled_argo_output_file = Path(tmpdir) / test_pipeline_file.with_suffix(".yaml").name
    compiled_argo_output_file_name = str(compiled_argo_output_file.absolute())

    # generate Python DSL for the Argo workflow engine
    pipeline_version = f"{pipeline.name}-0815"
    pipeline_instance_id = f"{pipeline.name}-{datetime.now().strftime('%m%d%H%M%S')}"
    experiment_name = f"{pipeline.name}-0815"
    generated_dsl = processor._generate_pipeline_dsl(
        pipeline=pipeline,
        pipeline_name=pipeline.name,
        workflow_engine=workflow_engine,
        pipeline_version=pipeline_version,
        pipeline_instance_id=pipeline_instance_id,
        experiment_name=experiment_name,
    )

    # if the compiler discovers an issue with the generated DSL this call fails
    processor._compile_pipeline_dsl(
        dsl=generated_dsl,
        workflow_engine=workflow_engine,
        output_file=compiled_argo_output_file_name,
        pipeline_conf=None,
    )

    # Load generated workflow
    with open(compiled_argo_output_file_name) as f:
        spec_docs = list(yaml.safe_load_all(f.read()))

    assert len(spec_docs) == 2
    components, platforms = spec_docs[0], spec_docs[1]["platforms"]

    assert components["pipelineInfo"]["name"] == pipeline.name
    assert components["pipelineInfo"]["description"] == pipeline.description

    # There should be two templates, one for the DAG and one for the generic node.
    # Locate the one for the generic node and inspect its properties.
    assert components["root"]["dag"]
    assert len(components["components"]) == 1
    executors = components["deploymentSpec"]["executors"]
    assert len(executors) == 1

    node_template = components["deploymentSpec"]["executors"]["exec-run-a-file"]

    # Verify component definition information (see generic_component_definition_template.jinja2)
    #  - property 'implementation.container.command'
    assert node_template["container"]["command"] == ["sh", "-c"]
    #  - property 'implementation.container.args'
    #    This is a CLOB, which we need to spot check.
    assert isinstance(node_template["container"]["args"], list) and len(node_template["container"]["args"]) == 1
    #    Check for things that must be in this CLOB:
    #    - the pipeline name
    assert f"--pipeline-name '{pipeline.name}'" in node_template["container"]["args"][0]
    #    - the object storage endpoint that this node uses for file I/O
    assert f"--cos-endpoint '{runtime_config.metadata['cos_endpoint']}'" in node_template["container"]["args"][0]
    #    - the object storage bucket name that this node uses for file I/O
    assert f"--cos-bucket '{runtime_config.metadata['cos_bucket']}'" in node_template["container"]["args"][0]
    #    - the directory within that object storage bucket
    assert f"--cos-directory '{pipeline_instance_id}" in node_template["container"]["args"][0]
    #  - the name of the archive in that directory
    expected_archive_name = processor._get_dependency_archive_name(op)
    assert f"--cos-dependencies-archive '{expected_archive_name}' " in node_template["container"]["args"][0]
    #  - the name of the file that this node processes, which is included in that archive
    assert f"--file '{op.filename}'" in node_template["container"]["args"][0]

    # Check for things that should not be in this CLOB:
    #  - Since it's a one-node pipeline, the component cannot have any "--inputs",
    #    which are declared object storage output files from upstream components.
    assert "--inputs" not in node_template["container"]["args"]
    #  - The component does not declare "--outputs",
    #    which are output files that need to be stored on object storage.
    assert "--outputs" not in node_template["container"]["args"]

    #  - property 'implementation.container.image'
    assert node_template["container"]["image"] == op.runtime_image

