code: def test_put_trigger(
      mock_put_rule,
      mock_put_target,
      mock_get_tags,
      mock_add_tags,
      mock_describe_rule,
      mock_create_trigger_lineage,
  ):
      session = Mock(
          Session,
          sagemaker_client=Mock(
              describe_pipeline=Mock(
                  return_value={"PipelineArn": "test-pipeline-arn"}
              )
          ),
          boto_session=Mock(),
      )
      source_pipeline_events = [
          FeatureProcessorPipelineEvents(
              pipeline_name="test-pipeline",
              pipeline_execution_status=[
                  FeatureProcessorPipelineExecutionStatus.SUCCEEDED
              ],
          )
      ]
      put_trigger(
          source_pipeline_events=source_pipeline_events,
          target_pipeline="test-target-pipeline",
          state="Enabled",
          event_pattern="test-pattern",
          role_arn=SCHEDULE_ROLE_ARN,
          sagemaker_session=session,
      )
  
      mock_put_rule.assert_called_once_with(
          source_pipeline_events=source_pipeline_events,
          target_pipeline="test-target-pipeline",
          state="Enabled",
          event_pattern="test-pattern",
      )
      mock_put_target.assert_called_once_with(
          rule_name="test-rule",
          target_pipeline="test-target-pipeline",
          target_pipeline_parameters=None,
          role_arn=SCHEDULE_ROLE_ARN,
      )
      mock_add_tags.assert_called_once_with(
          rule_arn=EVENT_BRIDGE_RULE_ARN, tags=TAGS
      )
      mock_create_trigger_lineage.assert_called_once_with(
          pipeline_name="test-target-pipeline",
          trigger_arn=EVENT_BRIDGE_RULE_ARN,
          state="Enabled",
          tags=TAGS,
          event_pattern="test-pattern",
      )
  
imports: from __future__ import absolute_import
  from datetime import datetime
  from typing import Callable
  import pytest
  import json
  from botocore.exceptions import ClientError
  from mock import Mock, patch, call
  from sagemaker.feature_store.feature_processor.feature_scheduler import FeatureProcessorLineageHandler
  from sagemaker.feature_store.feature_processor import FeatureProcessorPipelineEvents, FeatureProcessorPipelineExecutionStatus
  from sagemaker.lineage.context import Context
  from sagemaker.remote_function.spark_config import SparkConfig
  from sagemaker import Session
  from sagemaker.feature_store.feature_processor._enums import FeatureProcessorMode
  from sagemaker.feature_store.feature_processor._constants import FEATURE_PROCESSOR_TAG_KEY, FEATURE_PROCESSOR_TAG_VALUE, EXECUTION_TIME_PIPELINE_PARAMETER_FORMAT, PIPELINE_NAME_MAXIMUM_LENGTH
  from sagemaker.feature_store.feature_processor.feature_scheduler import schedule, to_pipeline, execute, delete_schedule, describe, list_pipelines, put_trigger, enable_trigger, disable_trigger, delete_trigger, _validate_fg_lineage_resources, _validate_pipeline_lineage_resources
  from sagemaker.remote_function.job import _JobSettings, SPARK_APP_SCRIPT_PATH, RUNTIME_SCRIPTS_CHANNEL_NAME, REMOTE_FUNCTION_WORKSPACE, ENTRYPOINT_SCRIPT_NAME, SPARK_CONF_CHANNEL_NAME
  from sagemaker.workflow.parameters import Parameter, ParameterTypeEnum
  from sagemaker.workflow.retry import StepRetryPolicy, StepExceptionTypeEnum, SageMakerJobStepRetryPolicy, SageMakerJobExceptionTypeEnum
  import test_data_helpers
label_desc: Use image node_id 26 for calling a global function with example usage: test_put_trigger(mock_put_rule, mock_put_target, mock_get_tags, mock_add_tags, mock_describe_rule, mock_create_trigger_lineage) without return types