cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix: add missing builtins, fix type/lint issues

Browse Source
This commit is contained in:
Aditya Chhabra
2025-11-14 20:10:20 +05:30
parent 00c47416d0
commit 454f6c500f
5 changed files with 122 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cleveragents/reactive/route_bridge.py
+21 -17
View File
@@ -10,7 +10,6 @@ import logging
from typing import Any
from typing import Optional
from jinja2.exceptions import SecurityError
from rx.scheduler.eventloop import AsyncIOScheduler # type: ignore[attr-defined]
from cleveragents.agents.base import Agent
@@ -22,6 +21,7 @@ from cleveragents.langgraph.nodes import NodeType
from cleveragents.langgraph.state import GraphState
from cleveragents.reactive.route import RouteConfig
from cleveragents.reactive.route import RouteType
from cleveragents.core.exceptions import ConfigurationError
from cleveragents.reactive.stream_router import ReactiveStreamRouter
from cleveragents.reactive.stream_router import StreamConfig
from cleveragents.reactive.stream_router import StreamMessage
@@ -50,7 +50,7 @@ class RouteBridge:
self.logger = logging.getLogger(__name__)
self._active_conversions: dict[str, Any] = {}
async def check_upgrade_conditions(
async def check_upgrade_conditions( # pylint: disable=too-many-return-statements
self,
route_config: RouteConfig,
message: StreamMessage,
@@ -88,8 +88,9 @@ class RouteBridge:
# Only accept string predicate names (never callables)
if not isinstance(predicate, str):
self.logger.warning(
f"Route '{route_config.name}': custom_predicate must be a string. "
f"Got {type(predicate).__name__}. Ignoring."
"Route '%s': custom_predicate must be a string. Got %s. Ignoring.",
route_config.name,
type(predicate).__name__
)
return False
@@ -97,7 +98,7 @@ class RouteBridge:
predicate_params = conditions.get("custom_predicate_params", {})
# Whitelist of allowed predicates
ALLOWED_PREDICATES = {
allowed_predicates = {
'content_contains': lambda msg, params: (
str(params.get('text', '')) in str(msg.content or '')
),
@@ -115,23 +116,26 @@ class RouteBridge:
),
}
if predicate in ALLOWED_PREDICATES:
if predicate in allowed_predicates:
try:
predicate_func = ALLOWED_PREDICATES[predicate]
result = predicate_func(message, predicate_params)
predicate_func = allowed_predicates[predicate]
result = predicate_func( # type: ignore[no-untyped-call]
message, predicate_params
)
return bool(result)
except Exception as e:
except Exception as e: # pylint: disable=broad-exception-caught
self.logger.error(
f"Route '{route_config.name}': Error evaluating predicate '{predicate}': {e}"
"Route '%s': Error evaluating predicate '%s': %s",
route_config.name,
predicate,
e
)
return False
else:
# Use CleverAgents exception instead of jinja2
from cleveragents.core.exceptions import ConfigurationError
raise ConfigurationError(
f"Route '{route_config.name}': Predicate '{predicate}' not allowed. "
f"Allowed: {list(ALLOWED_PREDICATES.keys())}"
)
# Predicate not in whitelist
raise ConfigurationError(
f"Route '{route_config.name}': Predicate '{predicate}' not allowed. "
f"Allowed: {list(allowed_predicates.keys())}"
)
return False
src/cleveragents/reactive/stream_router.py
+89 -15
View File
@@ -227,8 +227,8 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
# Security: Validate function name type
if not isinstance(func_name, str):
self.logger.warning(
f"Function parameter must be a string, got {type(func_name).__name__}. "
"Ignoring function parameter."
"Function parameter must be a string, got %s. Ignoring function parameter.",
type(func_name).__name__
)
return ops.map(lambda x: x)
@@ -239,8 +239,9 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
"Function names must contain only alphanumeric characters and underscores."
)
# Security: Whitelist of allowed built-in functions
ALLOWED_STREAM_FUNCTIONS = {
# Security: Check for allowed built-in functions
# First check standard builtins
allowed_stream_functions = {
'uppercase': self._builtin_uppercase,
'lowercase': self._builtin_lowercase,
'extract_content': self._builtin_extract_content,
@@ -250,20 +251,27 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
'truncate': lambda: self._builtin_truncate_factory(func_params),
}
if func_name in ALLOWED_STREAM_FUNCTIONS:
func = ALLOWED_STREAM_FUNCTIONS[func_name]
if func_name in allowed_stream_functions:
func = allowed_stream_functions[func_name]
# Handle both direct functions and factories
if callable(func) and not isinstance(func, type(lambda: None)):
# Direct function
return ops.map(func)
else:
# Factory function
return ops.map(func())
else:
raise ConfigurationError(
f"Function '{func_name}' is not allowed. "
f"Allowed functions: {list(ALLOWED_STREAM_FUNCTIONS.keys())}"
)
# Factory function - call it to get the actual function
factory_result = func() # type: ignore[operator]
return ops.map(factory_result)
# Check for dynamically registered _builtin_ methods
builtin_method_name = f"_builtin_{func_name}"
if hasattr(self, builtin_method_name):
method = getattr(self, builtin_method_name)
if callable(method):
return ops.map(method)
# Function not found
raise ConfigurationError(
f"Function '{func_name}' is not allowed. "
f"Allowed functions: {list(allowed_stream_functions.keys())}"
)
if "transform" in params:
transform = params["transform"]
return ops.map(lambda x: self._apply_transform(x, transform))
@@ -517,6 +525,72 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
return acc
def _builtin_uppercase(self, msg: StreamMessage) -> StreamMessage:
"""Convert message content to uppercase."""
content = str(msg.content).upper() if msg.content is not None else ""
return msg.copy_with(content=content)
def _builtin_lowercase(self, msg: StreamMessage) -> StreamMessage:
"""Convert message content to lowercase."""
content = str(msg.content).lower() if msg.content is not None else ""
return msg.copy_with(content=content)
def _builtin_extract_content(self, msg: StreamMessage) -> Any:
"""Extract content from message."""
return msg.content
def _builtin_extract_metadata_factory(
self, params: dict[str, Any]
) -> Callable[[StreamMessage], Any]:
"""Create function to extract metadata field."""
key = params.get('key')
default = params.get('default')
def extract_metadata(msg: StreamMessage) -> Any:
return msg.metadata.get(key, default) if key else msg.metadata
return extract_metadata
def _builtin_add_prefix_factory(
self, params: dict[str, Any]
) -> Callable[[StreamMessage], StreamMessage]:
"""Create function to add prefix to content."""
prefix = params.get('prefix', '')
def add_prefix(msg: StreamMessage) -> StreamMessage:
content = prefix + str(msg.content) if msg.content is not None else prefix
return msg.copy_with(content=content)
return add_prefix
def _builtin_add_suffix_factory(
self, params: dict[str, Any]
) -> Callable[[StreamMessage], StreamMessage]:
"""Create function to add suffix to content."""
suffix = params.get('suffix', '')
def add_suffix(msg: StreamMessage) -> StreamMessage:
content = str(msg.content) + suffix if msg.content is not None else suffix
return msg.copy_with(content=content)
return add_suffix
def _builtin_truncate_factory(
self, params: dict[str, Any]
) -> Callable[[StreamMessage], StreamMessage]:
"""Create function to truncate content."""
max_length = params.get('max_length', 100)
def truncate(msg: StreamMessage) -> StreamMessage:
if msg.content is not None:
content = str(msg.content)
content = content[:max_length] if len(content) > max_length else content
else:
content = ""
return msg.copy_with(content=content)
return truncate
def _handle_stream_error(self, error: Exception, source: ObservableType) -> ObservableType:
"""Handle stream errors."""
error_msg = StreamMessage(
tests/unit/reactive/test_route_bridge.py
+9 -8
View File
@@ -149,13 +149,14 @@ class TestCheckUpgradeConditions:
"""Test upgrade with custom predicate returning True."""
bridge = RouteBridge(Mock(), {})
def custom_pred(msg, config):
return True
# Use string predicate name from whitelist
route_config = RouteConfig(
name="test",
type=RouteType.STREAM,
bridge=BridgeConfig(upgrade_conditions={"custom_predicate": custom_pred})
bridge=BridgeConfig(upgrade_conditions={
"custom_predicate": "content_contains",
"custom_predicate_params": {"text": "test"}
})
)
message = StreamMessage(content="test")
@@ -184,7 +185,8 @@ class TestCheckUpgradeConditions:
@pytest.mark.asyncio
async def test_check_upgrade_custom_predicate_non_callable(self):
"""Test upgrade with non-callable custom predicate."""
"""Test upgrade with invalid custom predicate raises error."""
from cleveragents.core.exceptions import ConfigurationError
bridge = RouteBridge(Mock(), {})
route_config = RouteConfig(
@@ -194,9 +196,8 @@ class TestCheckUpgradeConditions:
)
message = StreamMessage(content="test")
result = await bridge.check_upgrade_conditions(route_config, message)
assert result is False
with pytest.raises(ConfigurationError, match="Predicate 'not_callable' not allowed"):
await bridge.check_upgrade_conditions(route_config, message)
@pytest.mark.asyncio
async def test_check_upgrade_no_conditions_met(self):
tests/unit/reactive/test_stream_router.py
+1 -1
View File
@@ -262,7 +262,7 @@ class TestCreateStream:
config = StreamConfig(
name="test_stream",
operators=[
{"type": "map", "params": {"function": "lambda x: x"}}
{"type": "map", "params": {"function": "uppercase"}}
]
)
tests/unit/templates/test_renderer.py
+2 -1
View File
@@ -353,7 +353,8 @@ class TestTemplateRenderer:
result = simple_renderer.render("test", {"none_value": None})
assert result == "Value: "
# Jinja2 renders None as "None" by default
assert result == "Value: None"
def test_render_string_simple_with_format_and_jinja(self, simple_renderer):
"""Test rendering with both format and jinja-like placeholders."""