xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-07-23 04:46:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into feat/workflow

Browse Source 
# Conflicts:
#	pyproject.toml
#	src/langbot/pkg/api/http/controller/main.py
#	src/langbot/pkg/core/app.py
#	src/langbot/pkg/core/stages/build_app.py
#	src/langbot/pkg/provider/modelmgr/requester.py
#	uv.lock
#	web/package-lock.json
#	web/package.json
#	web/pnpm-lock.yaml
#	web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx
#	web/src/app/home/components/dynamic-form/DynamicFormItemComponent.tsx
#	web/src/app/home/components/home-sidebar/HomeSidebar.tsx
#	web/src/app/infra/entities/api/index.ts
#	web/src/app/infra/http/BackendClient.ts
#	web/src/i18n/locales/en-US.ts
#	web/src/i18n/locales/es-ES.ts
#	web/src/i18n/locales/ja-JP.ts
#	web/src/i18n/locales/ru-RU.ts
#	web/src/i18n/locales/th-TH.ts
#	web/src/i18n/locales/vi-VN.ts
#	web/src/i18n/locales/zh-Hans.ts
#	web/src/i18n/locales/zh-Hant.ts
#	web/src/router.tsx
This commit is contained in:
RockChinQ
2026-07-01 20:33:05 +08:00
parent 417cca016c cc7a13158e
commit d770de7f7f
773 changed files with 84040 additions and 16377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit_tests/pipeline/conftest.py
+9
View File
@@ -12,6 +12,12 @@ from __future__ import annotations
import pytest
from unittest.mock import AsyncMock, Mock
# Preload pipelinemgr so the pipeline.stage module is fully initialised before
# any individual stage test (e.g. preproc, longtext) tries to import it. Without
# this, running a stage test in isolation triggers a circular-import error:
# stage.py → core.app → pipelinemgr → stage.stage_class (not yet bound).
import langbot.pkg.pipeline.pipelinemgr # noqa: F401
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
import langbot_plugin.api.entities.builtin.platform.message as platform_message
import langbot_plugin.api.entities.builtin.platform.events as platform_events
@@ -34,6 +40,9 @@ class MockApplication:
self.query_pool = self._create_mock_query_pool()
self.instance_config = self._create_mock_instance_config()
self.task_mgr = self._create_mock_task_manager()
# Skill manager is optional; PreProcessor only touches it for the
# local-agent runner. None keeps the skill-binding branch inert.
self.skill_mgr = None
def _create_mock_logger(self):
logger = Mock()
tests/unit_tests/pipeline/test_aggregator.py
+14 -14
View File
@@ -49,7 +49,7 @@ class TestPendingMessage:
"""PendingMessage should be created with correct fields."""
aggregator = get_aggregator_module()
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -88,7 +88,7 @@ class TestSessionBuffer:
"""SessionBuffer should accept initial messages."""
aggregator = get_aggregator_module()
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -309,7 +309,7 @@ class TestMessageAggregatorAddMessage:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -348,7 +348,7 @@ class TestMessageAggregatorAddMessage:
agg = aggregator.MessageAggregator(app)
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -387,7 +387,7 @@ class TestMessageAggregatorAddMessage:
agg = aggregator.MessageAggregator(app)
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -419,7 +419,7 @@ class TestMessageAggregatorMerge:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -445,8 +445,8 @@ class TestMessageAggregatorMerge:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain1 = text_chain("hello")
chain2 = text_chain("world")
chain1 = text_chain('hello')
chain2 = text_chain('world')
event = friend_message_event(chain1)
adapter = mock_adapter()
@@ -476,8 +476,8 @@ class TestMessageAggregatorMerge:
# Should contain both messages with separator
merged_str = str(merged.message_chain)
assert "hello" in merged_str
assert "world" in merged_str
assert 'hello' in merged_str
assert 'world' in merged_str
def test_merge_messages_preserves_routed_by_rule_if_any_input_matches(self):
"""Merged PendingMessage should keep routed_by_rule when any input was rule-routed."""
@@ -486,8 +486,8 @@ class TestMessageAggregatorMerge:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain1 = text_chain("first")
chain2 = text_chain("second")
chain1 = text_chain('first')
chain2 = text_chain('second')
event = friend_message_event(chain1)
adapter = mock_adapter()
@@ -545,7 +545,7 @@ class TestMessageAggregatorFlush:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
@@ -597,7 +597,7 @@ class TestMessageAggregatorFlushAll:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain = text_chain("hello")
chain = text_chain('hello')
event = friend_message_event(chain)
adapter = mock_adapter()
tests/unit_tests/pipeline/test_chat_handler.py
+34 -4
View File
@@ -15,6 +15,7 @@ from tests.factories import FakeApp
# ============== FIXTURE USING IMPORT ISOLATION UTILITY ==============
@pytest.fixture(scope='module')
def mock_circular_import_chain():
"""
@@ -36,9 +37,11 @@ def mock_circular_import_chain():
# Create a default runner that yields a simple response
class DefaultRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
yield Message(role='assistant', content='fake response')
@@ -70,9 +73,12 @@ def mock_event_ctx():
@pytest.fixture
def set_runner():
"""Factory fixture to set a custom runner for tests."""
def _set_runner(runner_class):
import sys
sys.modules['langbot.pkg.provider.runner'].preregistered_runners = [runner_class]
return _set_runner
@@ -87,6 +93,7 @@ def get_chat_handler():
global _chat_handler_module
if _chat_handler_module is None:
from importlib import import_module
_chat_handler_module = import_module('langbot.pkg.pipeline.process.handlers.chat')
return _chat_handler_module
@@ -96,12 +103,14 @@ def get_entities():
global _entities_module
if _entities_module is None:
from importlib import import_module
_entities_module = import_module('langbot.pkg.pipeline.entities')
return _entities_module
# ============== REAL ChatMessageHandler Tests ==============
@pytest.mark.usefixtures('mock_circular_import_chain')
class TestChatMessageHandlerReal:
"""Tests for real ChatMessageHandler class."""
@@ -188,9 +197,11 @@ class TestChatMessageHandlerReal:
class QuickRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
yield Message(role='assistant', content='ok')
@@ -222,9 +233,11 @@ class TestChatMessageHandlerReal:
class SingleRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
yield Message(role='assistant', content='response')
@@ -262,9 +275,11 @@ class TestChatHandlerStreaming:
class StreamRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
yield MessageChunk(role='assistant', content='Hello', is_final=False)
yield MessageChunk(role='assistant', content=' World', is_final=True)
@@ -303,14 +318,19 @@ class TestChatHandlerExceptions:
query.pipeline_config = {
'output': {'misc': {'exception-handling': 'show-hint', 'failure-hint': 'Request failed.'}},
'ai': {'runner': {'runner': 'local-agent'}, 'local-agent': {'prompt': 'default', 'model': {'primary': 'test'}}},
'ai': {
'runner': {'runner': 'local-agent'},
'local-agent': {'prompt': 'default', 'model': {'primary': 'test'}},
},
}
class FailingRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
raise ValueError('API error')
yield
@@ -346,14 +366,19 @@ class TestChatHandlerExceptions:
query.pipeline_config = {
'output': {'misc': {'exception-handling': 'show-error'}},
'ai': {'runner': {'runner': 'local-agent'}, 'local-agent': {'prompt': 'default', 'model': {'primary': 'test'}}},
'ai': {
'runner': {'runner': 'local-agent'},
'local-agent': {'prompt': 'default', 'model': {'primary': 'test'}},
},
}
class ErrorRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
raise ValueError('Custom error')
yield
@@ -386,14 +411,19 @@ class TestChatHandlerExceptions:
query.pipeline_config = {
'output': {'misc': {'exception-handling': 'hide'}},
'ai': {'runner': {'runner': 'local-agent'}, 'local-agent': {'prompt': 'default', 'model': {'primary': 'test'}}},
'ai': {
'runner': {'runner': 'local-agent'},
'local-agent': {'prompt': 'default', 'model': {'primary': 'test'}},
},
}
class HideErrorRunner:
name = 'local-agent'
def __init__(self, app, config):
self.app = app
self.config = config
async def run(self, query):
raise RuntimeError('hidden')
yield
@@ -433,4 +463,4 @@ class TestChatHandlerHelper:
chat = get_chat_handler()
handler = chat.ChatMessageHandler(fake_app)
result = handler.cut_str('first line\nsecond line')
assert '...' in result
assert '...' in result
tests/unit_tests/pipeline/test_chat_handler_logging.py
+78
View File
@@ -0,0 +1,78 @@
from __future__ import annotations
from unittest.mock import Mock
import pytest
import langbot_plugin.api.entities.builtin.provider.message as provider_message
# TODO: unskip once the handler ↔ app circular import is resolved
pytest.skip(
'circular import in handler ↔ app; will be unblocked once resolved',
allow_module_level=True,
)
from langbot.pkg.pipeline.process.handler import MessageHandler # noqa: E402
class _StubHandler(MessageHandler):
async def handle(self, query):
raise NotImplementedError
handler = _StubHandler(ap=Mock())
def test_chat_handler_formats_tool_call_request_log():
result = provider_message.Message(
role='assistant',
content='',
tool_calls=[
provider_message.ToolCall(
id='call-1',
type='function',
function=provider_message.FunctionCall(name='exec', arguments='{}'),
)
],
)
summary = handler.format_result_log(result)
assert summary == 'assistant: requested tools: exec'
def test_chat_handler_formats_tool_result_log():
result = provider_message.Message(
role='tool',
content='{"status":"completed","exit_code":0,"backend":"podman","stdout":"42\\n"}',
tool_call_id='call-1',
)
summary = handler.format_result_log(result)
# Tool results use generic cut_str truncation
assert summary is not None
assert summary.startswith('tool: {"status":"com')
assert summary.endswith('...')
def test_chat_handler_formats_tool_error_log():
result = provider_message.MessageChunk(
role='tool',
content='err: host_path must point to an existing directory on the host',
tool_call_id='call-1',
is_final=True,
)
summary = handler.format_result_log(result)
assert summary is not None
assert summary.startswith('tool error: err: host_path must')
assert summary.endswith('...')
def test_chat_handler_skips_empty_assistant_log():
result = provider_message.Message(role='assistant', content='')
summary = handler.format_result_log(result)
assert summary is None
tests/unit_tests/pipeline/test_cntfilter.py
+22 -24
View File
@@ -67,7 +67,11 @@ def make_pipeline_config(**overrides):
for key, value in overrides.items():
if key in base_config and isinstance(base_config[key], dict) and isinstance(value, dict):
for sub_key, sub_value in value.items():
if sub_key in base_config[key] and isinstance(base_config[key][sub_key], dict) and isinstance(sub_value, dict):
if (
sub_key in base_config[key]
and isinstance(base_config[key][sub_key], dict)
and isinstance(sub_value, dict)
):
base_config[key][sub_key].update(sub_value)
else:
base_config[key][sub_key] = sub_value
@@ -141,7 +145,7 @@ class TestPreContentFilter:
await stage.initialize(pipeline_config)
query = text_query("hello world")
query = text_query('hello world')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -163,7 +167,7 @@ class TestPreContentFilter:
await stage.initialize(pipeline_config)
# Empty message chain
query = text_query("")
query = text_query('')
query.message_chain = platform_message.MessageChain([])
query.pipeline_config = pipeline_config
@@ -185,7 +189,7 @@ class TestPreContentFilter:
await stage.initialize(pipeline_config)
query = text_query(" ") # Only whitespace
query = text_query(' ') # Only whitespace
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -234,7 +238,7 @@ class TestPreContentFilter:
await stage.initialize(pipeline_config)
query = text_query("hello world")
query = text_query('hello world')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -266,7 +270,7 @@ class TestContentIgnoreFilter:
await stage.initialize(pipeline_config)
query = text_query("/help me")
query = text_query('/help me')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -294,7 +298,7 @@ class TestContentIgnoreFilter:
await stage.initialize(pipeline_config)
query = text_query("http://example.com")
query = text_query('http://example.com')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -322,7 +326,7 @@ class TestContentIgnoreFilter:
await stage.initialize(pipeline_config)
query = text_query("normal message")
query = text_query('normal message')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -343,7 +347,7 @@ class TestContentIgnoreFilter:
await stage.initialize(pipeline_config)
query = text_query("/help me")
query = text_query('/help me')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
@@ -368,12 +372,10 @@ class TestPostContentFilter:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
# Add a response message
query.resp_messages = [
provider_message.Message(role='assistant', content='Hello back!')
]
query.resp_messages = [provider_message.Message(role='assistant', content='Hello back!')]
result = await stage.process(query, 'PostContentFilterStage')
@@ -398,11 +400,9 @@ class TestPostContentFilter:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_messages = [
provider_message.Message(role='assistant', content='Response')
]
query.resp_messages = [provider_message.Message(role='assistant', content='Response')]
result = await stage.process(query, 'PostContentFilterStage')
@@ -422,7 +422,7 @@ class TestPostContentFilter:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
# Non-string content - use model_construct to bypass validation
# The actual content type could be a list of ContentElement objects
@@ -450,11 +450,9 @@ class TestPostContentFilter:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_messages = [
provider_message.Message(role='assistant', content='')
]
query.resp_messages = [provider_message.Message(role='assistant', content='')]
result = await stage.process(query, 'PostContentFilterStage')
@@ -476,7 +474,7 @@ class TestContentFilterStageInvalidName:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
with pytest.raises(ValueError, match='未知的 stage_inst_name'):
@@ -506,7 +504,7 @@ class TestContentIgnoreFilterDirect:
await stage.initialize(pipeline_config)
query = text_query("normal message without prefix")
query = text_query('normal message without prefix')
query.pipeline_config = pipeline_config
result = await stage.process(query, 'PreContentFilterStage')
tests/unit_tests/pipeline/test_command_handler.py
+10 -3
View File
@@ -15,6 +15,7 @@ from tests.factories import FakeApp, command_query
# ============== FIXTURE USING IMPORT ISOLATION UTILITY ==============
@pytest.fixture(scope='module')
def mock_circular_import_chain():
"""
@@ -56,6 +57,7 @@ def mock_event_ctx():
@pytest.fixture
def mock_execute_factory():
"""Factory fixture to create mock cmd_mgr.execute generators."""
def _create_execute(
text: str | None = 'ok',
error: str | None = None,
@@ -71,7 +73,9 @@ def mock_execute_factory():
ret.image_base64 = image_base64
ret.file_url = file_url
yield ret
return mock_execute
return _create_execute
@@ -86,6 +90,7 @@ def get_command_handler():
global _command_handler_module
if _command_handler_module is None:
from importlib import import_module
_command_handler_module = import_module('langbot.pkg.pipeline.process.handlers.command')
return _command_handler_module
@@ -95,12 +100,14 @@ def get_entities():
global _entities_module
if _entities_module is None:
from importlib import import_module
_entities_module = import_module('langbot.pkg.pipeline.entities')
return _entities_module
# ============== REAL CommandHandler Tests ==============
@pytest.mark.usefixtures('mock_circular_import_chain')
class TestCommandHandlerReal:
"""Tests for real CommandHandler class."""
@@ -127,6 +134,7 @@ class TestCommandHandlerReal:
fake_app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
executed_commands = []
async def track_execute(command_text, full_command_text, query, session):
executed_commands.append(command_text)
ret = Mock()
@@ -334,8 +342,7 @@ class TestCommandHandlerReal:
command = get_command_handler()
fake_app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
fake_app.cmd_mgr.execute = mock_execute_factory(
text='Here is the image:',
image_url='https://example.com/image.png'
text='Here is the image:', image_url='https://example.com/image.png'
)
handler = command.CommandHandler(fake_app)
@@ -393,4 +400,4 @@ class TestCommandHandlerHelper:
command = get_command_handler()
handler = command.CommandHandler(fake_app)
result = handler.cut_str('first line\nsecond line')
assert '...' in result
assert '...' in result
tests/unit_tests/pipeline/test_longtext.py
+19 -27
View File
@@ -126,11 +126,9 @@ class TestLongTextProcessStageProcess:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = [
platform_message.MessageChain([platform_message.Plain(text="very long response")])
]
query.resp_message_chain = [platform_message.MessageChain([platform_message.Plain(text='very long response')])]
result = await stage.process(query, 'LongTextProcessStage')
@@ -151,11 +149,9 @@ class TestLongTextProcessStageProcess:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = [
platform_message.MessageChain([platform_message.Plain(text="short response")])
]
query.resp_message_chain = [platform_message.MessageChain([platform_message.Plain(text='short response')])]
result = await stage.process(query, 'LongTextProcessStage')
@@ -179,14 +175,13 @@ class TestLongTextProcessStageProcess:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
# Non-Plain component (Image)
query.resp_message_chain = [
platform_message.MessageChain([
platform_message.Plain(text="short"),
platform_message.Image(url="https://example.com/img.png")
])
platform_message.MessageChain(
[platform_message.Plain(text='short'), platform_message.Image(url='https://example.com/img.png')]
)
]
result = await stage.process(query, 'LongTextProcessStage')
@@ -213,7 +208,7 @@ class TestLongTextProcessStageProcess:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
@@ -232,7 +227,7 @@ class TestLongTextProcessStageProcess:
stage = longtext.LongTextProcessStage(app)
stage.strategy_impl = AsyncMock()
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = make_longtext_config(strategy='forward', threshold=1)
query.resp_message_chain = []
@@ -242,6 +237,7 @@ class TestLongTextProcessStageProcess:
assert result.new_query is query
stage.strategy_impl.process.assert_not_called()
class TestForwardStrategy:
"""Tests for ForwardComponentStrategy."""
@@ -260,7 +256,7 @@ class TestForwardStrategy:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
# Create a mock adapter with bot_account_id
mock_adapter = Mock()
@@ -268,10 +264,8 @@ class TestForwardStrategy:
query.adapter = mock_adapter
# Long text exceeding threshold
long_text = "This is a very long response that exceeds the threshold"
query.resp_message_chain = [
platform_message.MessageChain([platform_message.Plain(text=long_text)])
]
long_text = 'This is a very long response that exceeds the threshold'
query.resp_message_chain = [platform_message.MessageChain([platform_message.Plain(text=long_text)])]
result = await stage.process(query, 'LongTextProcessStage')
@@ -297,13 +291,13 @@ class TestForwardStrategy:
await strat.initialize()
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = make_longtext_config()
mock_adapter = Mock()
mock_adapter.bot_account_id = '12345'
query.adapter = mock_adapter
components = await strat.process("test message", query)
components = await strat.process('test message', query)
assert len(components) == 1
assert isinstance(components[0], platform_message.Forward)
@@ -326,14 +320,12 @@ class TestLongTextThreshold:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
# Text below threshold
short_text = "x" * (threshold - 1)
query.resp_message_chain = [
platform_message.MessageChain([platform_message.Plain(text=short_text)])
]
short_text = 'x' * (threshold - 1)
query.resp_message_chain = [platform_message.MessageChain([platform_message.Plain(text=short_text)])]
result = await stage.process(query, 'LongTextProcessStage')
tests/unit_tests/pipeline/test_msgtrun.py
+7 -7
View File
@@ -115,7 +115,7 @@ class TestRoundTruncatorProcess:
await stage.initialize(pipeline_config)
# Create query with 3 messages (within limit)
query = text_query("current message")
query = text_query('current message')
query.pipeline_config = pipeline_config
query.messages = [
provider_message.Message(role='user', content='message 1'),
@@ -154,7 +154,7 @@ class TestRoundTruncatorProcess:
# Create query with many messages exceeding limit
# 7 messages = 3 full rounds + 1 current user
query = text_query("current message")
query = text_query('current message')
query.pipeline_config = pipeline_config
query.messages = [
provider_message.Message(role='user', content='message 1'),
@@ -194,7 +194,7 @@ class TestRoundTruncatorProcess:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.messages = []
@@ -216,7 +216,7 @@ class TestRoundTruncatorProcess:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.messages = [
provider_message.Message(role='user', content='hello'),
@@ -240,7 +240,7 @@ class TestRoundTruncatorProcess:
await stage.initialize(pipeline_config)
query = text_query("current")
query = text_query('current')
query.pipeline_config = pipeline_config
query.messages = [
provider_message.Message(role='user', content='user1'),
@@ -274,7 +274,7 @@ class TestRoundTruncatorProcess:
await stage.initialize(pipeline_config)
query = text_query("current")
query = text_query('current')
query.pipeline_config = pipeline_config
query.messages = [
provider_message.Message(role='user', content='old1'),
@@ -305,7 +305,7 @@ class TestRoundTruncatorDirect:
trun = trun_cls(app)
break
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = make_truncate_config(max_round=3)
query.messages = [
provider_message.Message(role='user', content='m1'),
tests/unit_tests/pipeline/test_n8nsvapi.py
+31 -18
View File
@@ -14,33 +14,43 @@ import json
import sys
from unittest.mock import AsyncMock, MagicMock, Mock, patch
# Break the circular import chain before importing n8nsvapi:
import pytest
import langbot_plugin.api.entities.builtin.provider.message as provider_message
# Break the circular import chain while importing n8nsvapi:
# n8nsvapi → runner → app → pipelinemgr → all runners → runner (partially init)
_mock_runner = MagicMock()
_mock_runner.runner_class = lambda name: (lambda cls: cls) # no-op decorator
_mock_runner.RequestRunner = object
_mocked_imports = {
'langbot.pkg.provider.runner': _mock_runner,
# The stubs are restored in a ``finally`` block so this module does NOT pollute
# sys.modules for other test modules (e.g. ones importing the real
# LocalAgentRunner, which would otherwise inherit ``object`` and break).
# Mirrors master's intent but uses try/finally so a raised import doesn't
# leave the global namespace in a stubbed state, and includes
# ``langbot.pkg.utils.httpclient`` which master didn't stub.
_runner_stub = MagicMock()
_runner_stub.runner_class = lambda name: (lambda cls: cls) # no-op decorator
_runner_stub.RequestRunner = object
_import_stubs = {
'langbot.pkg.provider.runner': _runner_stub,
'langbot.pkg.core.app': MagicMock(),
'langbot.pkg.utils.httpclient': MagicMock(),
}
_original_imports = {name: sys.modules.get(name) for name in _mocked_imports}
sys.modules.update(_mocked_imports)
import pytest # noqa: E402
import langbot_plugin.api.entities.builtin.provider.message as provider_message # noqa: E402
from langbot.pkg.provider.runners.n8nsvapi import N8nServiceAPIRunner # noqa: E402
for _name, _original in _original_imports.items():
if _original is None:
sys.modules.pop(_name, None)
else:
sys.modules[_name] = _original
_saved_modules = {name: sys.modules.get(name) for name in _import_stubs}
for _name, _stub in _import_stubs.items():
sys.modules[_name] = _stub
try:
from langbot.pkg.provider.runners.n8nsvapi import N8nServiceAPIRunner
finally:
for _name, _original in _saved_modules.items():
if _original is None:
sys.modules.pop(_name, None)
else:
sys.modules[_name] = _original
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def make_runner(output_key: str = 'response') -> N8nServiceAPIRunner:
ap = Mock()
ap.logger = Mock()
@@ -83,6 +93,7 @@ async def collect_chunks(runner: N8nServiceAPIRunner, chunks: list[bytes | str])
# _process_response: stream format (type:item/end)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_stream_format_single_item():
"""Single item + end in one chunk yields final chunk with full content."""
@@ -165,6 +176,7 @@ async def test_stream_format_no_spurious_empty_yield():
# _process_response: plain JSON fallback
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_plain_json_with_output_key():
"""Plain JSON with matching output_key extracts value via output_key."""
@@ -235,6 +247,7 @@ async def test_invalid_json_returns_raw_text():
# _call_webhook: output type depends on is_stream
# ---------------------------------------------------------------------------
def make_query(is_stream: bool):
"""Build a minimal Query mock."""
query = Mock()
tests/unit_tests/pipeline/test_pipelinemgr.py
+43
View File
@@ -162,3 +162,46 @@ async def test_runtime_pipeline_execute(mock_app, sample_query):
# Verify stage was called
mock_stage.process.assert_called_once()
def test_runtime_pipeline_prefers_local_agent_mcp_resources(mock_app):
"""Local Agent resource selection should override legacy extension prefs."""
pipelinemgr = get_pipelinemgr_module()
persistence_pipeline = get_persistence_pipeline_module()
pipeline_entity = Mock(spec=persistence_pipeline.LegacyPipeline)
pipeline_entity.config = {
'ai': {
'local-agent': {
'mcp-resources': [{'server_uuid': 'srv-new', 'uri': 'file:///new.md'}],
'mcp-resource-agent-read-enabled': False,
}
}
}
pipeline_entity.extensions_preferences = {
'mcp_resources': [{'server_uuid': 'srv-old', 'uri': 'file:///old.md'}],
'mcp_resource_agent_read_enabled': True,
}
runtime_pipeline = pipelinemgr.RuntimePipeline(mock_app, pipeline_entity, [])
assert runtime_pipeline.mcp_resource_attachments == [{'server_uuid': 'srv-new', 'uri': 'file:///new.md'}]
assert runtime_pipeline.mcp_resource_agent_read_enabled is False
def test_runtime_pipeline_falls_back_to_extension_mcp_resources(mock_app):
"""Existing extension prefs remain compatible until a Local Agent value exists."""
pipelinemgr = get_pipelinemgr_module()
persistence_pipeline = get_persistence_pipeline_module()
pipeline_entity = Mock(spec=persistence_pipeline.LegacyPipeline)
pipeline_entity.config = {'ai': {'local-agent': {}}}
pipeline_entity.extensions_preferences = {
'mcp_resources': [{'server_uuid': 'srv-old', 'uri': 'file:///old.md'}],
'mcp_resource_agent_read_enabled': False,
}
runtime_pipeline = pipelinemgr.RuntimePipeline(mock_app, pipeline_entity, [])
assert runtime_pipeline.mcp_resource_attachments == [{'server_uuid': 'srv-old', 'uri': 'file:///old.md'}]
assert runtime_pipeline.mcp_resource_agent_read_enabled is False
tests/unit_tests/pipeline/test_preproc.py
+73 -12
View File
@@ -14,6 +14,7 @@ from __future__ import annotations
import pytest
from unittest.mock import AsyncMock, Mock
from importlib import import_module
from types import SimpleNamespace
from tests.factories import (
FakeApp,
@@ -78,7 +79,7 @@ class TestPreProcessorNormalText:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = text_query("hello world")
query = text_query('hello world')
result = await stage.process(query, 'PreProcessor')
@@ -113,7 +114,7 @@ class TestPreProcessorNormalText:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = text_query("test message")
query = text_query('test message')
result = await stage.process(query, 'PreProcessor')
@@ -194,13 +195,16 @@ class TestPreProcessorImageSegment:
stage = preproc.PreProcessor(app)
# Image query with base64
query = image_query(text="look at this", url=None)
query = image_query(text='look at this', url=None)
# Set base64 on the image component
import langbot_plugin.api.entities.builtin.platform.message as platform_message
chain = platform_message.MessageChain([
platform_message.Plain(text="look at this"),
platform_message.Image(base64="data:image/png;base64,abc123"),
])
chain = platform_message.MessageChain(
[
platform_message.Plain(text='look at this'),
platform_message.Image(base64='data:image/png;base64,abc123'),
]
)
query.message_chain = chain
result = await stage.process(query, 'PreProcessor')
@@ -238,7 +242,7 @@ class TestPreProcessorImageSegment:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = image_query(text="describe this")
query = image_query(text='describe this')
result = await stage.process(query, 'PreProcessor')
@@ -276,7 +280,7 @@ class TestPreProcessorModelSelection:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = text_query("hello")
query = text_query('hello')
# Set pipeline config with primary model
query.pipeline_config = {
@@ -335,7 +339,7 @@ class TestPreProcessorModelSelection:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = {
'ai': {
@@ -384,7 +388,7 @@ class TestPreProcessorVariables:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = text_query("hello", sender_id=67890)
query = text_query('hello', sender_id=67890)
result = await stage.process(query, 'PreProcessor')
@@ -421,10 +425,67 @@ class TestPreProcessorVariables:
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = group_text_query("hello", group_id=99999)
query = group_text_query('hello', group_id=99999)
result = await stage.process(query, 'PreProcessor')
variables = result.new_query.variables
assert 'group_name' in variables
assert 'sender_name' in variables
class TestPreProcessorToolSelection:
"""Tests for Local Agent tool selection."""
@pytest.mark.asyncio
async def test_local_agent_filters_selected_tools(self):
"""Only selected tools should be exposed when all-tools mode is off."""
preproc = get_preproc_module()
app = FakeApp()
mock_session = Mock()
mock_session.launcher_type = Mock(value='person')
mock_session.launcher_id = 12345
app.sess_mgr.get_session = AsyncMock(return_value=mock_session)
mock_conversation = Mock()
mock_conversation.prompt = Mock(messages=[])
mock_conversation.prompt.copy = Mock(return_value=Mock(messages=[]))
mock_conversation.messages = []
mock_conversation.uuid = None
app.sess_mgr.get_conversation = AsyncMock(return_value=mock_conversation)
mock_model = Mock()
mock_model.model_entity = Mock(uuid='primary-model-uuid', abilities=['func_call'])
app.model_mgr.get_model_by_uuid = AsyncMock(return_value=mock_model)
app.tool_mgr.get_all_tools = AsyncMock(
return_value=[
SimpleNamespace(name='exec'),
SimpleNamespace(name='plugin_tool'),
SimpleNamespace(name='mcp_tool'),
]
)
mock_event_ctx = Mock()
mock_event_ctx.event = Mock(default_prompt=[], prompt=[])
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = preproc.PreProcessor(app)
query = text_query('hello')
query.pipeline_config = {
'ai': {
'runner': {'runner': 'local-agent'},
'local-agent': {
'model': {'primary': 'primary-model-uuid', 'fallbacks': []},
'prompt': 'default',
'enable-all-tools': False,
'tools': ['plugin_tool'],
},
},
'output': {'misc': {'at-sender': False}},
'trigger': {'misc': {}},
}
result = await stage.process(query, 'PreProcessor')
assert [tool.name for tool in result.new_query.use_funcs] == ['plugin_tool']
tests/unit_tests/pipeline/test_ratelimit.py
+18 -58
View File
@@ -46,7 +46,7 @@ class TestFixedWindowAlgo:
'safety': {
'rate-limit': {
'window-length': 60, # 60 seconds window
'limitation': 10, # 10 requests per window
'limitation': 10, # 10 requests per window
'strategy': 'drop',
}
}
@@ -75,11 +75,9 @@ class TestFixedWindowAlgo:
# Make requests within limit
for i in range(10):
result = await algo.require_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'12345'
sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, '12345'
)
assert result is True, f"Request {i+1} should be allowed"
assert result is True, f'Request {i + 1} should be allowed'
@pytest.mark.asyncio
async def test_fixedwin_exceeds_limit_drop_strategy(self, mock_app_for_algo, sample_query_with_rate_limit):
@@ -91,20 +89,12 @@ class TestFixedWindowAlgo:
# Exhaust the limit
for i in range(10):
await algo.require_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'12345'
)
await algo.require_access(sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, '12345')
# Next request should be denied
result = await algo.require_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'12345'
)
result = await algo.require_access(sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, '12345')
assert result is False, "Request exceeding limit should be denied"
assert result is False, 'Request exceeding limit should be denied'
@pytest.mark.asyncio
async def test_fixedwin_different_sessions_isolated(self, mock_app_for_algo, sample_query_with_rate_limit):
@@ -116,20 +106,14 @@ class TestFixedWindowAlgo:
# Exhaust limit for session 1
for i in range(10):
await algo.require_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'session1'
)
await algo.require_access(sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, 'session1')
# Session 2 should still have its own limit
result = await algo.require_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'session2'
sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, 'session2'
)
assert result is True, "Different session should have independent limit"
assert result is True, 'Different session should have independent limit'
@pytest.mark.asyncio
async def test_fixedwin_limit_one_request(self, mock_app_for_algo, sample_query):
@@ -150,19 +134,11 @@ class TestFixedWindowAlgo:
await algo.initialize()
# First request allowed
result1 = await algo.require_access(
sample_query,
provider_session.LauncherTypes.PERSON,
'12345'
)
result1 = await algo.require_access(sample_query, provider_session.LauncherTypes.PERSON, '12345')
assert result1 is True
# Second request denied
result2 = await algo.require_access(
sample_query,
provider_session.LauncherTypes.PERSON,
'12345'
)
result2 = await algo.require_access(sample_query, provider_session.LauncherTypes.PERSON, '12345')
assert result2 is False
@pytest.mark.asyncio
@@ -174,11 +150,7 @@ class TestFixedWindowAlgo:
await algo.initialize()
# First request creates container
await algo.require_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'12345'
)
await algo.require_access(sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, '12345')
# Key format: 'LauncherTypes.PERSON_12345' (enum string representation)
expected_key = 'LauncherTypes.PERSON_12345'
@@ -230,7 +202,7 @@ class TestFixedWindowAlgo:
# New request should be allowed (new window)
result = await algo.require_access(sample_query, provider_session.LauncherTypes.PERSON, 'test')
assert result is True, "New window should allow new requests"
assert result is True, 'New window should allow new requests'
@pytest.mark.asyncio
async def test_fixedwin_wait_strategy_blocks_until_next_window(self, mock_app_for_algo, sample_query):
@@ -256,29 +228,21 @@ class TestFixedWindowAlgo:
# First request allowed
start_time = time.time()
result1 = await algo.require_access(
sample_query,
provider_session.LauncherTypes.PERSON,
'wait_test'
)
result1 = await algo.require_access(sample_query, provider_session.LauncherTypes.PERSON, 'wait_test')
assert result1 is True
# Exhaust limit
await algo.require_access(sample_query, provider_session.LauncherTypes.PERSON, 'wait_test')
# Third request should wait and then succeed
result3 = await algo.require_access(
sample_query,
provider_session.LauncherTypes.PERSON,
'wait_test'
)
result3 = await algo.require_access(sample_query, provider_session.LauncherTypes.PERSON, 'wait_test')
elapsed = time.time() - start_time
assert result3 is True, "After wait, request should succeed"
assert result3 is True, 'After wait, request should succeed'
# Should have waited approximately until next window
# With 1-second window, elapsed should be > 0.5 second (allowing for timing variance)
# Note: This is a timing-sensitive test, so we use a generous tolerance
assert elapsed >= 0.5, f"Should have waited for next window, elapsed={elapsed:.2f}s"
assert elapsed >= 0.5, f'Should have waited for next window, elapsed={elapsed:.2f}s'
@pytest.mark.asyncio
async def test_fixedwin_release_access(self, mock_app_for_algo, sample_query_with_rate_limit):
@@ -289,11 +253,7 @@ class TestFixedWindowAlgo:
await algo.initialize()
# release_access is empty in current implementation
await algo.release_access(
sample_query_with_rate_limit,
provider_session.LauncherTypes.PERSON,
'12345'
)
await algo.release_access(sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, '12345')
# Should not raise or change state
assert 'person_12345' not in algo.containers
tests/unit_tests/pipeline/test_wrapper.py
+167 -27
View File
@@ -36,6 +36,11 @@ def get_entities_module():
return import_module('langbot.pkg.pipeline.entities')
def get_plugin_diagnostics_module():
"""Lazy import for plugin diagnostic attribution helpers."""
return import_module('langbot.pkg.pipeline.plugin_diagnostics')
def make_wrapper_config():
"""Create a pipeline config for wrapper tests."""
return {
@@ -55,7 +60,7 @@ def make_session():
launcher_type=provider_session.LauncherTypes.PERSON,
launcher_id=12345,
sender_id=12345,
use_prompt_name="default",
use_prompt_name='default',
using_conversation=None,
conversations=[],
)
@@ -93,11 +98,9 @@ class TestResponseWrapperMessageChain:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_messages = [
platform_message.MessageChain([platform_message.Plain(text="response")])
]
query.resp_messages = [platform_message.MessageChain([platform_message.Plain(text='response')])]
query.resp_message_chain = []
results = []
@@ -108,6 +111,45 @@ class TestResponseWrapperMessageChain:
assert results[0].result_type == entities.ResultType.CONTINUE
assert len(results[0].new_query.resp_message_chain) == 1
@pytest.mark.asyncio
async def test_message_chain_direct_append_consumes_pending_plugin_source(self):
"""MessageChain replies from earlier plugin events keep attribution."""
wrapper = get_wrapper_module()
app = FakeApp()
stage = wrapper.ResponseWrapper(app)
await stage.initialize(make_wrapper_config())
reply_chain = platform_message.MessageChain([platform_message.Plain(text='response')])
query = text_query('hello')
query.pipeline_config = make_wrapper_config()
query.resp_messages = [reply_chain]
query.resp_message_chain = []
plugin_diagnostics = get_plugin_diagnostics_module()
plugin_diagnostics.record_pending_plugin_response_source(
query,
reply_chain,
[
{
'kind': 'reply_message_chain',
'plugin': {'author': 'tester', 'name': 'demo'},
}
],
[{'manifest': {'metadata': {'author': 'observer', 'name': 'not-reply-source'}}}],
'PersonNormalMessageReceived',
)
results = []
async for result in stage.process(query, 'ResponseWrapper'):
results.append(result)
sources = plugin_diagnostics._get_response_sources(results[0].new_query, 0)
assert sources[0].plugin == {'author': 'tester', 'name': 'demo'}
assert sources[0].event_name == 'PersonNormalMessageReceived'
assert sources[0].is_approximate is False
assert '_plugin_response_sources' not in query.variables
assert '_plugin_pending_response_sources' not in query.variables
class TestResponseWrapperCommand:
"""Tests for command response wrapping."""
@@ -125,7 +167,7 @@ class TestResponseWrapperCommand:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
@@ -133,7 +175,7 @@ class TestResponseWrapperCommand:
command_resp = Mock()
command_resp.role = 'command'
command_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Help info")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Help info')])
)
query.resp_messages = [command_resp]
@@ -163,7 +205,7 @@ class TestResponseWrapperPlugin:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
@@ -171,7 +213,7 @@ class TestResponseWrapperPlugin:
plugin_resp = Mock()
plugin_resp.role = 'plugin'
plugin_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Plugin response")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Plugin response')])
)
query.resp_messages = [plugin_resp]
@@ -211,17 +253,17 @@ class TestResponseWrapperAssistant:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
# Create assistant response with content
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = "Hello back!"
assistant_resp.content = 'Hello back!'
assistant_resp.tool_calls = None
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Hello back!")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Hello back!')])
)
query.resp_messages = [assistant_resp]
@@ -247,7 +289,7 @@ class TestResponseWrapperAssistant:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
@@ -292,7 +334,7 @@ class TestResponseWrapperAssistant:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
@@ -303,10 +345,10 @@ class TestResponseWrapperAssistant:
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = "Processing..."
assistant_resp.content = 'Processing...'
assistant_resp.tool_calls = [mock_tool_call]
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Processing...")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Processing...')])
)
query.resp_messages = [assistant_resp]
@@ -346,17 +388,17 @@ class TestResponseWrapperInterrupt:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
# Create assistant response with content
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = "Hello!"
assistant_resp.content = 'Hello!'
assistant_resp.tool_calls = None
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Hello!")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Hello!')])
)
query.resp_messages = [assistant_resp]
@@ -384,7 +426,7 @@ class TestResponseWrapperCustomReply:
app.sess_mgr.get_session = AsyncMock(return_value=session)
# Mock plugin connector with custom reply
custom_chain = platform_message.MessageChain([platform_message.Plain(text="Custom reply")])
custom_chain = platform_message.MessageChain([platform_message.Plain(text='Custom reply')])
mock_event_ctx = Mock()
mock_event_ctx.is_prevented_default = Mock(return_value=False)
mock_event_ctx.event = Mock()
@@ -397,17 +439,17 @@ class TestResponseWrapperCustomReply:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
# Create assistant response
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = "Default reply"
assistant_resp.content = 'Default reply'
assistant_resp.tool_calls = None
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Default reply")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Default reply')])
)
query.resp_messages = [assistant_resp]
@@ -421,7 +463,105 @@ class TestResponseWrapperCustomReply:
assert len(results[0].new_query.resp_message_chain) == 1
# Should be the custom chain
chain = results[0].new_query.resp_message_chain[0]
assert "Custom reply" in str(chain)
assert 'Custom reply' in str(chain)
@pytest.mark.asyncio
async def test_custom_reply_records_plugin_source(self):
"""Plugin reply_message_chain should keep emitted plugin attribution."""
wrapper = get_wrapper_module()
app = FakeApp()
app.sess_mgr.get_session = AsyncMock(return_value=make_session())
custom_chain = platform_message.MessageChain([platform_message.Plain(text='Custom reply')])
mock_event_ctx = Mock()
mock_event_ctx.is_prevented_default = Mock(return_value=False)
mock_event_ctx.event = Mock()
mock_event_ctx.event.reply_message_chain = custom_chain
mock_event_ctx._emitted_plugins = [
{
'manifest': {'metadata': {'author': 'observer', 'name': 'not-reply-source'}},
'plugin_config': {'token': 'secret-token'},
},
]
mock_event_ctx._response_sources = [
{
'kind': 'reply_message_chain',
'plugin': {'author': 'tester', 'name': 'demo'},
}
]
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = wrapper.ResponseWrapper(app)
pipeline_config = make_wrapper_config()
await stage.initialize(pipeline_config)
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = 'Default reply'
assistant_resp.tool_calls = None
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text='Default reply')])
)
query.resp_messages = [assistant_resp]
results = []
async for result in stage.process(query, 'ResponseWrapper'):
results.append(result)
plugin_diagnostics = get_plugin_diagnostics_module()
sources = plugin_diagnostics._get_response_sources(results[0].new_query, 0)
assert sources[0].plugin == {'author': 'tester', 'name': 'demo'}
assert sources[0].event_name == 'NormalMessageResponded'
assert sources[0].is_approximate is False
assert 'secret-token' not in str(sources)
assert '_plugin_response_sources' not in query.variables
@pytest.mark.asyncio
async def test_custom_reply_falls_back_to_emitted_plugins_for_old_runtime(self):
"""Older plugin runtimes without response_sources keep approximate attribution."""
wrapper = get_wrapper_module()
app = FakeApp()
app.sess_mgr.get_session = AsyncMock(return_value=make_session())
custom_chain = platform_message.MessageChain([platform_message.Plain(text='Custom reply')])
mock_event_ctx = Mock()
mock_event_ctx.is_prevented_default = Mock(return_value=False)
mock_event_ctx.event = Mock()
mock_event_ctx.event.reply_message_chain = custom_chain
mock_event_ctx._emitted_plugins = [
{'manifest': {'metadata': {'author': 'tester', 'name': 'demo'}}},
]
app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
stage = wrapper.ResponseWrapper(app)
pipeline_config = make_wrapper_config()
await stage.initialize(pipeline_config)
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = 'Default reply'
assistant_resp.tool_calls = None
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text='Default reply')])
)
query.resp_messages = [assistant_resp]
results = []
async for result in stage.process(query, 'ResponseWrapper'):
results.append(result)
plugin_diagnostics = get_plugin_diagnostics_module()
sources = plugin_diagnostics._get_response_sources(results[0].new_query, 0)
assert sources[0].plugin == {'author': 'tester', 'name': 'demo'}
assert sources[0].is_approximate is True
class TestResponseWrapperVariables:
@@ -452,7 +592,7 @@ class TestResponseWrapperVariables:
await stage.initialize(pipeline_config)
query = text_query("hello")
query = text_query('hello')
query.pipeline_config = pipeline_config
query.resp_message_chain = []
query.variables['_pipeline_bound_plugins'] = ['plugin1', 'plugin2']
@@ -460,10 +600,10 @@ class TestResponseWrapperVariables:
# Create assistant response
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = "Hello"
assistant_resp.content = 'Hello'
assistant_resp.tool_calls = None
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text="Hello")])
return_value=platform_message.MessageChain([platform_message.Plain(text='Hello')])
)
query.resp_messages = [assistant_resp]
tests/unit_tests/pipeline/test_wrapper_outbound_attachments.py
+146
View File
@@ -0,0 +1,146 @@
"""Unit tests for ResponseWrapper outbound-attachment helpers.
Covers the sandbox -> user attachment path added for the Box attachment
round-trip:
* ``_is_final_assistant_message`` — only the terminal, tool-call-free assistant
message (or a final MessageChunk) should trigger collection.
* ``_append_outbound_attachments`` — collects sandbox outbox files exactly once
per query and maps each descriptor to the right platform component, swallowing
collection errors.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
import pytest
import langbot_plugin.api.entities.builtin.platform.message as platform_message
import langbot_plugin.api.entities.builtin.provider.message as provider_message
from langbot.pkg.pipeline.wrapper.wrapper import ResponseWrapper
def _make_wrapper(box_service) -> ResponseWrapper:
app = SimpleNamespace(logger=Mock())
wrapper = ResponseWrapper.__new__(ResponseWrapper)
wrapper.ap = app
return wrapper
def _make_query():
return SimpleNamespace(variables={})
def test_is_final_assistant_message_plain_assistant():
wrapper = _make_wrapper(box_service=None)
msg = provider_message.Message(role='assistant', content='done')
assert wrapper._is_final_assistant_message(msg) is True
def test_is_final_assistant_message_rejects_non_assistant():
wrapper = _make_wrapper(box_service=None)
msg = provider_message.Message(role='tool', content='{}')
assert wrapper._is_final_assistant_message(msg) is False
def test_is_final_assistant_message_rejects_tool_call_round():
wrapper = _make_wrapper(box_service=None)
msg = provider_message.Message(
role='assistant',
content='calling',
tool_calls=[
provider_message.ToolCall(
id='c1',
type='function',
function=provider_message.FunctionCall(name='exec', arguments='{}'),
)
],
)
assert wrapper._is_final_assistant_message(msg) is False
def test_is_final_assistant_message_non_final_chunk():
wrapper = _make_wrapper(box_service=None)
chunk = provider_message.MessageChunk(role='assistant', content='partial', is_final=False)
assert wrapper._is_final_assistant_message(chunk) is False
final_chunk = provider_message.MessageChunk(role='assistant', content='partial', is_final=True)
assert wrapper._is_final_assistant_message(final_chunk) is True
@pytest.mark.asyncio
async def test_append_outbound_attachments_maps_each_type():
box_service = SimpleNamespace(
available=True,
collect_outbound_attachments=AsyncMock(
return_value=[
{'type': 'Image', 'base64': 'data:image/png;base64,iVBORw0K'},
{'type': 'Voice', 'base64': 'data:audio/wav;base64,UklGRg=='},
{'type': 'File', 'name': 'report.xlsx', 'base64': 'data:app;base64,UEsDBA=='},
]
),
)
wrapper = _make_wrapper(box_service)
wrapper.ap.box_service = box_service
query = _make_query()
chain = platform_message.MessageChain([])
await wrapper._append_outbound_attachments(query, chain)
kinds = [type(c).__name__ for c in chain]
assert kinds == ['Image', 'Voice', 'File']
assert query.variables['_sandbox_outbound_collected'] is True
# File keeps its name
file_comp = chain[2]
assert getattr(file_comp, 'name', None) == 'report.xlsx'
@pytest.mark.asyncio
async def test_append_outbound_attachments_runs_once_per_query():
box_service = SimpleNamespace(
available=True,
collect_outbound_attachments=AsyncMock(return_value=[]),
)
wrapper = _make_wrapper(box_service)
wrapper.ap.box_service = box_service
query = _make_query()
query.variables['_sandbox_outbound_collected'] = True
chain = platform_message.MessageChain([])
await wrapper._append_outbound_attachments(query, chain)
box_service.collect_outbound_attachments.assert_not_awaited()
assert len(chain) == 0
@pytest.mark.asyncio
async def test_append_outbound_attachments_noop_without_box_service():
wrapper = _make_wrapper(box_service=None)
wrapper.ap.box_service = None
query = _make_query()
chain = platform_message.MessageChain([])
await wrapper._append_outbound_attachments(query, chain)
assert len(chain) == 0
# not marked collected, since service is unavailable
assert '_sandbox_outbound_collected' not in query.variables
@pytest.mark.asyncio
async def test_append_outbound_attachments_swallows_collection_error():
box_service = SimpleNamespace(
available=True,
collect_outbound_attachments=AsyncMock(side_effect=RuntimeError('boom')),
)
wrapper = _make_wrapper(box_service)
wrapper.ap.box_service = box_service
query = _make_query()
chain = platform_message.MessageChain([])
# must not raise
await wrapper._append_outbound_attachments(query, chain)
assert len(chain) == 0
wrapper.ap.logger.warning.assert_called_once()