Compare commits

..

1 Commits

Author SHA1 Message Date
huanghuoguoguo
aa4b5d6732 fix(plugin): validate plugin id format 2026-05-16 10:36:58 +08:00
4 changed files with 29 additions and 18 deletions

View File

@@ -633,11 +633,12 @@ class PluginRuntimeConnector:
Raises: Raises:
ValueError: If plugin_id is not in the expected 'author/name' format. ValueError: If plugin_id is not in the expected 'author/name' format.
""" """
if '/' not in plugin_id: segments = plugin_id.split('/')
if len(segments) != 2 or not all(segments):
raise ValueError( raise ValueError(
f"Invalid plugin_id format: '{plugin_id}'. Expected 'author/name' format (e.g. 'langbot/rag-engine')." f"Invalid plugin_id format: '{plugin_id}'. Expected 'author/name' format (e.g. 'langbot/rag-engine')."
) )
return plugin_id.split('/', 1) return segments[0], segments[1]
async def call_rag_ingest(self, plugin_id: str, context_data: dict[str, Any]) -> dict[str, Any]: async def call_rag_ingest(self, plugin_id: str, context_data: dict[str, Any]) -> dict[str, Any]:
"""Call plugin to ingest document. """Call plugin to ingest document.

View File

@@ -83,7 +83,7 @@ def get_func_schema(function: typing.Callable) -> dict:
parameters['properties'][param.name] = { parameters['properties'][param.name] = {
'type': param_type, 'type': param_type,
'description': args_doc.get(param.name, ''), 'description': args_doc[param.name],
} }
# add schema for array # add schema for array

View File

@@ -0,0 +1,25 @@
"""Test plugin ID parsing validation."""
import pytest
from src.langbot.pkg.plugin.connector import PluginRuntimeConnector
def test_parse_plugin_id_accepts_author_name():
assert PluginRuntimeConnector._parse_plugin_id('langbot/rag-engine') == ('langbot', 'rag-engine')
@pytest.mark.parametrize(
'plugin_id',
[
'',
'author',
'author/',
'/name',
'author/name/extra',
'/',
],
)
def test_parse_plugin_id_rejects_malformed_ids(plugin_id):
with pytest.raises(ValueError, match='Expected'):
PluginRuntimeConnector._parse_plugin_id(plugin_id)

View File

@@ -1,15 +0,0 @@
from langbot.pkg.utils.funcschema import get_func_schema
def test_get_func_schema_uses_empty_description_for_undocumented_parameter():
def sample_function(documented: str, undocumented: int):
"""Sample function.
Args:
documented(str): documented parameter description
"""
schema = get_func_schema(sample_function)
assert schema['parameters']['properties']['documented']['description'] == 'documented parameter description'
assert schema['parameters']['properties']['undocumented']['description'] == ''