mirror of
https://github.com/langgenius/dify-plugin-sdks.git
synced 2026-07-22 02:15:22 -04:00
Harry
350 lines
12 KiB
Python
350 lines
12 KiB
Python
from collections.abc import Mapping
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from werkzeug import Request, Response
|
|
|
|
from dify_plugin.core.runtime import Session
|
|
from dify_plugin.core.trigger_factory import TriggerFactory
|
|
from dify_plugin.entities import I18nObject, ParameterOption
|
|
from dify_plugin.entities.provider_config import CredentialType
|
|
from dify_plugin.entities.trigger import (
|
|
EventConfiguration,
|
|
EventConfigurationExtra,
|
|
EventDispatch,
|
|
EventIdentity,
|
|
EventParameter,
|
|
Subscription,
|
|
TriggerProviderConfiguration,
|
|
TriggerProviderConfigurationExtra,
|
|
TriggerProviderIdentity,
|
|
TriggerSubscriptionConstructorConfiguration,
|
|
TriggerSubscriptionConstructorConfigurationExtra,
|
|
TriggerSubscriptionConstructorRuntime,
|
|
UnsubscribeResult,
|
|
Variables,
|
|
)
|
|
from dify_plugin.interfaces.trigger import Event, EventRuntime, Trigger, TriggerSubscriptionConstructor
|
|
|
|
|
|
class MockTriggerProvider(Trigger):
|
|
"""
|
|
Mock Trigger Provider
|
|
"""
|
|
|
|
def _dispatch_event(self, subscription: Subscription, request: Request) -> EventDispatch:
|
|
"""
|
|
Dispatch event from webhook
|
|
"""
|
|
return EventDispatch(events=["test_event"], response=Response("OK", status=200))
|
|
|
|
|
|
class MockTriggerSubscriptionConstructor(TriggerSubscriptionConstructor):
|
|
"""
|
|
Mock Trigger Subscription Constructor
|
|
"""
|
|
|
|
def _validate_api_key(self, credentials: dict):
|
|
"""
|
|
Validate API key
|
|
"""
|
|
pass
|
|
|
|
def _create_subscription(
|
|
self, endpoint: str, credentials: Mapping[str, Any], selected_events: list[str], parameters: Mapping[str, Any]
|
|
) -> Subscription:
|
|
"""
|
|
Create subscription
|
|
"""
|
|
return Subscription(
|
|
expires_at=1234567890,
|
|
endpoint=endpoint,
|
|
properties={
|
|
"external_id": "test_external_id",
|
|
"webhook_secret": "test_secret",
|
|
},
|
|
subscribed_events=selected_events,
|
|
)
|
|
|
|
def _delete_subscription(self, subscription: Subscription, credentials: Mapping[str, Any]) -> UnsubscribeResult:
|
|
"""
|
|
Delete subscription
|
|
"""
|
|
return UnsubscribeResult(success=True, message="Successfully unsubscribed")
|
|
|
|
def _refresh_subscription(self, subscription: Subscription, credentials: Mapping[str, Any]) -> Subscription:
|
|
"""
|
|
Refresh subscription
|
|
"""
|
|
return Subscription(
|
|
expires_at=9999999999,
|
|
endpoint=subscription.endpoint,
|
|
properties=subscription.properties,
|
|
)
|
|
|
|
def _fetch_parameter_options(self, credentials: Mapping[str, Any], parameter: str) -> list[ParameterOption]:
|
|
"""
|
|
Fetch parameter options
|
|
"""
|
|
return [
|
|
ParameterOption(value="option1", label=I18nObject(en_US="Option 1")),
|
|
ParameterOption(value="option2", label=I18nObject(en_US="Option 2")),
|
|
]
|
|
|
|
|
|
class MockEventHandler(Event):
|
|
"""
|
|
Mock Event
|
|
"""
|
|
|
|
def _on_event(self, request: Request, parameters: Mapping[str, Any], payload: Mapping[str, Any]) -> Variables:
|
|
"""
|
|
Transform the webhook request into Variables
|
|
"""
|
|
return Variables(variables={"test_variable": "test_value", "event_data": request.get_data(as_text=True)})
|
|
|
|
|
|
def test_trigger_factory_register_and_get_provider():
|
|
"""
|
|
Test trigger factory registration and retrieval of provider
|
|
"""
|
|
factory = TriggerFactory()
|
|
session = MagicMock(spec=Session)
|
|
|
|
# Create provider configuration
|
|
provider_config = TriggerProviderConfiguration(
|
|
identity=TriggerProviderIdentity(
|
|
author="test",
|
|
name="test_provider",
|
|
label=I18nObject(en_US="Test Provider"),
|
|
description=I18nObject(en_US="Test Provider Description"),
|
|
),
|
|
subscription_constructor=TriggerSubscriptionConstructorConfiguration(
|
|
parameters=[],
|
|
credentials_schema=[],
|
|
extra=TriggerSubscriptionConstructorConfigurationExtra(
|
|
python=TriggerSubscriptionConstructorConfigurationExtra.Python(source="test_constructor.py")
|
|
),
|
|
),
|
|
extra=TriggerProviderConfigurationExtra(
|
|
python=TriggerProviderConfigurationExtra.Python(source="test_provider.py")
|
|
),
|
|
)
|
|
|
|
# Create trigger configuration
|
|
trigger_config = EventConfiguration(
|
|
identity=EventIdentity(
|
|
author="test",
|
|
name="test_event",
|
|
label=I18nObject(en_US="Test Event"),
|
|
),
|
|
parameters=[
|
|
EventParameter(
|
|
name="test_param",
|
|
label=I18nObject(en_US="Test Parameter"),
|
|
type=EventParameter.EventParameterType.STRING,
|
|
)
|
|
],
|
|
description=I18nObject(en_US="Human description"),
|
|
extra=EventConfigurationExtra(python=EventConfigurationExtra.Python(source="test_event.py")),
|
|
output_schema={"test_variable": {"type": "string"}},
|
|
)
|
|
|
|
# Register trigger provider with events
|
|
factory.register_trigger_provider(
|
|
configuration=provider_config,
|
|
provider_cls=MockTriggerProvider,
|
|
subscription_constructor_cls=MockTriggerSubscriptionConstructor,
|
|
events={"test_event": (trigger_config, MockEventHandler)},
|
|
)
|
|
|
|
# Test getting provider instance
|
|
provider = factory.get_trigger_provider("test_provider", session, None, None)
|
|
assert isinstance(provider, MockTriggerProvider)
|
|
assert provider.runtime.session == session
|
|
|
|
# Test getting provider class
|
|
provider_cls = factory.get_provider_cls("test_provider")
|
|
assert provider_cls == MockTriggerProvider
|
|
|
|
# Test getting configuration
|
|
config = factory.get_configuration("test_provider")
|
|
assert config == provider_config
|
|
|
|
|
|
def test_trigger_factory_subscription_constructor():
|
|
"""
|
|
Test trigger factory subscription constructor
|
|
"""
|
|
factory = TriggerFactory()
|
|
runtime = TriggerSubscriptionConstructorRuntime(
|
|
credentials={"api_key": "test_key"}, session=MagicMock(spec=Session), credential_type=CredentialType.API_KEY
|
|
)
|
|
|
|
# Create provider configuration with subscription constructor
|
|
provider_config = TriggerProviderConfiguration(
|
|
identity=TriggerProviderIdentity(
|
|
author="test",
|
|
name="test_provider",
|
|
label=I18nObject(en_US="Test Provider"),
|
|
description=I18nObject(en_US="Test Provider Description"),
|
|
),
|
|
subscription_constructor=TriggerSubscriptionConstructorConfiguration(
|
|
parameters=[],
|
|
credentials_schema=[],
|
|
extra=TriggerSubscriptionConstructorConfigurationExtra(
|
|
python=TriggerSubscriptionConstructorConfigurationExtra.Python(source="test_constructor.py")
|
|
),
|
|
),
|
|
extra=TriggerProviderConfigurationExtra(
|
|
python=TriggerProviderConfigurationExtra.Python(source="test_provider.py")
|
|
),
|
|
)
|
|
|
|
# Register provider
|
|
factory.register_trigger_provider(
|
|
configuration=provider_config,
|
|
provider_cls=MockTriggerProvider,
|
|
subscription_constructor_cls=MockTriggerSubscriptionConstructor,
|
|
events={},
|
|
)
|
|
|
|
# Test has_subscription_constructor
|
|
assert factory.has_subscription_constructor("test_provider") is True
|
|
|
|
# Test get subscription constructor instance
|
|
constructor = factory.get_subscription_constructor("test_provider", runtime)
|
|
assert isinstance(constructor, MockTriggerSubscriptionConstructor)
|
|
assert constructor.runtime.session == runtime.session
|
|
|
|
# Test get subscription constructor class
|
|
constructor_cls = factory.get_subscription_constructor_cls("test_provider")
|
|
assert constructor_cls == MockTriggerSubscriptionConstructor
|
|
|
|
|
|
def test_trigger_factory_trigger_events():
|
|
"""
|
|
Test trigger factory trigger event handling
|
|
"""
|
|
factory = TriggerFactory()
|
|
session = MagicMock(spec=Session)
|
|
|
|
# Create configurations
|
|
provider_config = TriggerProviderConfiguration(
|
|
identity=TriggerProviderIdentity(
|
|
author="test",
|
|
name="test_provider",
|
|
label=I18nObject(en_US="Test Provider"),
|
|
description=I18nObject(en_US="Test Provider Description"),
|
|
),
|
|
extra=TriggerProviderConfigurationExtra(
|
|
python=TriggerProviderConfigurationExtra.Python(source="test_provider.py")
|
|
),
|
|
)
|
|
|
|
trigger_config = EventConfiguration(
|
|
identity=EventIdentity(
|
|
author="test",
|
|
name="test_event",
|
|
label=I18nObject(en_US="Test Event"),
|
|
),
|
|
parameters=[],
|
|
description=I18nObject(en_US="Human description"),
|
|
extra=EventConfigurationExtra(python=EventConfigurationExtra.Python(source="test_event.py")),
|
|
)
|
|
|
|
# Register provider with events
|
|
registration = factory.register_trigger_provider(
|
|
configuration=provider_config,
|
|
provider_cls=MockTriggerProvider,
|
|
subscription_constructor_cls=None,
|
|
events={},
|
|
)
|
|
|
|
# Register trigger after provider registration
|
|
registration.register_trigger(
|
|
name="test_event",
|
|
configuration=trigger_config,
|
|
trigger_cls=MockEventHandler,
|
|
)
|
|
|
|
# Test get Event
|
|
event = factory.get_trigger_event_handler(
|
|
"test_provider",
|
|
"test_event",
|
|
EventRuntime(
|
|
session=session,
|
|
credential_type=CredentialType.UNAUTHORIZED,
|
|
subscription=Subscription(
|
|
expires_at=1234567890,
|
|
endpoint="test_endpoint",
|
|
properties={"external_id": "test_external_id", "webhook_secret": "test_secret"},
|
|
),
|
|
),
|
|
)
|
|
assert isinstance(event, MockEventHandler)
|
|
assert event.runtime.session == session
|
|
|
|
# Test get trigger configuration
|
|
config = factory.get_trigger_configuration("test_provider", "test_event")
|
|
assert config == trigger_config
|
|
|
|
# Test iterate events
|
|
events = factory.iter_events("test_provider")
|
|
assert "test_event" in events
|
|
assert events["test_event"][0] == trigger_config
|
|
assert events["test_event"][1] == MockEventHandler
|
|
|
|
|
|
def test_trigger_factory_error_handling():
|
|
"""
|
|
Test trigger factory error handling
|
|
"""
|
|
factory = TriggerFactory()
|
|
session = MagicMock(spec=Session)
|
|
|
|
# Test getting non-existent provider
|
|
with pytest.raises(ValueError, match="Trigger provider `non_existent` not found"):
|
|
factory.get_trigger_provider("non_existent", session, None, None)
|
|
|
|
# Create and register a provider
|
|
provider_config = TriggerProviderConfiguration(
|
|
identity=TriggerProviderIdentity(
|
|
author="test",
|
|
name="test_provider",
|
|
label=I18nObject(en_US="Test Provider"),
|
|
description=I18nObject(en_US="Test Provider Description"),
|
|
),
|
|
extra=TriggerProviderConfigurationExtra(
|
|
python=TriggerProviderConfigurationExtra.Python(source="test_provider.py")
|
|
),
|
|
)
|
|
|
|
factory.register_trigger_provider(
|
|
configuration=provider_config,
|
|
provider_cls=MockTriggerProvider,
|
|
subscription_constructor_cls=None,
|
|
events={},
|
|
)
|
|
|
|
# Test duplicate registration
|
|
with pytest.raises(ValueError, match="Trigger provider `test_provider` is already registered"):
|
|
factory.register_trigger_provider(
|
|
configuration=provider_config,
|
|
provider_cls=MockTriggerProvider,
|
|
subscription_constructor_cls=None,
|
|
events={},
|
|
)
|
|
|
|
# Test getting non-existent event
|
|
with pytest.raises(ValueError, match="Event `non_existent` not found in provider `test_provider`"):
|
|
factory.get_trigger_event_handler("test_provider", "non_existent", session)
|
|
|
|
# Test getting subscription constructor when none exists
|
|
runtime = TriggerSubscriptionConstructorRuntime(
|
|
credentials={"api_key": "test_key"}, session=MagicMock(spec=Session), credential_type=CredentialType.API_KEY
|
|
)
|
|
with pytest.raises(ValueError, match="Trigger provider `test_provider` does not define a subscription constructor"):
|
|
factory.get_subscription_constructor("test_provider", runtime)
|