mirror of
https://github.com/langgenius/dify-plugin-sdks.git
synced 2026-07-24 13:06:56 -04:00
Harry
8cdd62509d
- Updated the PluginExecutor and related classes to improve type annotations and clarity, particularly in methods handling OAuth credentials and trigger subscriptions. - Refactored the PluginRegistration and TriggerFactory classes to streamline the retrieval of trigger providers and event handlers, ensuring better type safety and consistency. - Adjusted the YAML loader function to specify return types, enhancing code readability and maintainability. These changes improve the overall structure and usability of the trigger subscription system, making it more robust and easier to understand.
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from collections.abc import Mapping
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Any
|
|
|
|
from werkzeug import Request
|
|
|
|
from dify_plugin.core.runtime import Session
|
|
from dify_plugin.core.server.stdio.request_reader import StdioRequestReader
|
|
from dify_plugin.core.server.stdio.response_writer import StdioResponseWriter
|
|
from dify_plugin.entities.trigger import Variables
|
|
from dify_plugin.interfaces.trigger import Event
|
|
|
|
|
|
def test_construct_trigger():
|
|
"""
|
|
Test the constructor of Event
|
|
|
|
NOTE:
|
|
- This test is to ensure that the constructor of Event is not overridden.
|
|
- And ensure a breaking change will be detected by CI.
|
|
"""
|
|
|
|
class TriggerEventImpl(Event):
|
|
def _on_event(self, request: Request, parameters: Mapping[str, Any]) -> Variables:
|
|
return Variables(variables={})
|
|
|
|
session = Session(
|
|
session_id="test",
|
|
executor=ThreadPoolExecutor(max_workers=1),
|
|
reader=StdioRequestReader(),
|
|
writer=StdioResponseWriter(),
|
|
)
|
|
|
|
trigger = TriggerEventImpl(session=session)
|
|
assert trigger is not None
|