mirror of
https://github.com/langgenius/dify.git
synced 2026-01-19 11:45:05 +08:00
chore: use from __future__ import annotations (#30254)
Co-authored-by: Dev <dev@Devs-MacBook-Pro-4.local> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Asuka Minato <i@asukaminato.eu.org> Co-authored-by: crazywoola <100913391+crazywoola@users.noreply.github.com>
This commit is contained in:
@ -5,6 +5,8 @@ This module provides a flexible configuration system for customizing
|
||||
the behavior of mock nodes during testing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
@ -95,67 +97,67 @@ class MockConfigBuilder:
|
||||
def __init__(self) -> None:
|
||||
self._config = MockConfig()
|
||||
|
||||
def with_auto_mock(self, enabled: bool = True) -> "MockConfigBuilder":
|
||||
def with_auto_mock(self, enabled: bool = True) -> MockConfigBuilder:
|
||||
"""Enable or disable auto-mocking."""
|
||||
self._config.enable_auto_mock = enabled
|
||||
return self
|
||||
|
||||
def with_delays(self, enabled: bool = True) -> "MockConfigBuilder":
|
||||
def with_delays(self, enabled: bool = True) -> MockConfigBuilder:
|
||||
"""Enable or disable simulated execution delays."""
|
||||
self._config.simulate_delays = enabled
|
||||
return self
|
||||
|
||||
def with_llm_response(self, response: str) -> "MockConfigBuilder":
|
||||
def with_llm_response(self, response: str) -> MockConfigBuilder:
|
||||
"""Set default LLM response."""
|
||||
self._config.default_llm_response = response
|
||||
return self
|
||||
|
||||
def with_agent_response(self, response: str) -> "MockConfigBuilder":
|
||||
def with_agent_response(self, response: str) -> MockConfigBuilder:
|
||||
"""Set default agent response."""
|
||||
self._config.default_agent_response = response
|
||||
return self
|
||||
|
||||
def with_tool_response(self, response: dict[str, Any]) -> "MockConfigBuilder":
|
||||
def with_tool_response(self, response: dict[str, Any]) -> MockConfigBuilder:
|
||||
"""Set default tool response."""
|
||||
self._config.default_tool_response = response
|
||||
return self
|
||||
|
||||
def with_retrieval_response(self, response: str) -> "MockConfigBuilder":
|
||||
def with_retrieval_response(self, response: str) -> MockConfigBuilder:
|
||||
"""Set default retrieval response."""
|
||||
self._config.default_retrieval_response = response
|
||||
return self
|
||||
|
||||
def with_http_response(self, response: dict[str, Any]) -> "MockConfigBuilder":
|
||||
def with_http_response(self, response: dict[str, Any]) -> MockConfigBuilder:
|
||||
"""Set default HTTP response."""
|
||||
self._config.default_http_response = response
|
||||
return self
|
||||
|
||||
def with_template_transform_response(self, response: str) -> "MockConfigBuilder":
|
||||
def with_template_transform_response(self, response: str) -> MockConfigBuilder:
|
||||
"""Set default template transform response."""
|
||||
self._config.default_template_transform_response = response
|
||||
return self
|
||||
|
||||
def with_code_response(self, response: dict[str, Any]) -> "MockConfigBuilder":
|
||||
def with_code_response(self, response: dict[str, Any]) -> MockConfigBuilder:
|
||||
"""Set default code execution response."""
|
||||
self._config.default_code_response = response
|
||||
return self
|
||||
|
||||
def with_node_output(self, node_id: str, outputs: dict[str, Any]) -> "MockConfigBuilder":
|
||||
def with_node_output(self, node_id: str, outputs: dict[str, Any]) -> MockConfigBuilder:
|
||||
"""Set outputs for a specific node."""
|
||||
self._config.set_node_outputs(node_id, outputs)
|
||||
return self
|
||||
|
||||
def with_node_error(self, node_id: str, error: str) -> "MockConfigBuilder":
|
||||
def with_node_error(self, node_id: str, error: str) -> MockConfigBuilder:
|
||||
"""Set error for a specific node."""
|
||||
self._config.set_node_error(node_id, error)
|
||||
return self
|
||||
|
||||
def with_node_config(self, config: NodeMockConfig) -> "MockConfigBuilder":
|
||||
def with_node_config(self, config: NodeMockConfig) -> MockConfigBuilder:
|
||||
"""Add a node-specific configuration."""
|
||||
self._config.set_node_config(config.node_id, config)
|
||||
return self
|
||||
|
||||
def with_default_config(self, node_type: NodeType, config: dict[str, Any]) -> "MockConfigBuilder":
|
||||
def with_default_config(self, node_type: NodeType, config: dict[str, Any]) -> MockConfigBuilder:
|
||||
"""Set default configuration for a node type."""
|
||||
self._config.set_default_config(node_type, config)
|
||||
return self
|
||||
|
||||
@ -1,3 +1,5 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import types
|
||||
from collections.abc import Generator
|
||||
@ -21,7 +23,7 @@ if TYPE_CHECKING: # pragma: no cover - imported for type checking only
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tool_node(monkeypatch) -> "ToolNode":
|
||||
def tool_node(monkeypatch) -> ToolNode:
|
||||
module_name = "core.ops.ops_trace_manager"
|
||||
if module_name not in sys.modules:
|
||||
ops_stub = types.ModuleType(module_name)
|
||||
@ -85,7 +87,7 @@ def _collect_events(generator: Generator) -> tuple[list[Any], LLMUsage]:
|
||||
return events, stop.value
|
||||
|
||||
|
||||
def _run_transform(tool_node: "ToolNode", message: ToolInvokeMessage) -> tuple[list[Any], LLMUsage]:
|
||||
def _run_transform(tool_node: ToolNode, message: ToolInvokeMessage) -> tuple[list[Any], LLMUsage]:
|
||||
def _identity_transform(messages, *_args, **_kwargs):
|
||||
return messages
|
||||
|
||||
@ -103,7 +105,7 @@ def _run_transform(tool_node: "ToolNode", message: ToolInvokeMessage) -> tuple[l
|
||||
return _collect_events(generator)
|
||||
|
||||
|
||||
def test_link_messages_with_file_populate_files_output(tool_node: "ToolNode"):
|
||||
def test_link_messages_with_file_populate_files_output(tool_node: ToolNode):
|
||||
file_obj = File(
|
||||
tenant_id="tenant-id",
|
||||
type=FileType.DOCUMENT,
|
||||
@ -139,7 +141,7 @@ def test_link_messages_with_file_populate_files_output(tool_node: "ToolNode"):
|
||||
assert files_segment.value == [file_obj]
|
||||
|
||||
|
||||
def test_plain_link_messages_remain_links(tool_node: "ToolNode"):
|
||||
def test_plain_link_messages_remain_links(tool_node: ToolNode):
|
||||
message = ToolInvokeMessage(
|
||||
type=ToolInvokeMessage.MessageType.LINK,
|
||||
message=ToolInvokeMessage.TextMessage(text="https://dify.ai"),
|
||||
|
||||
Reference in New Issue
Block a user