xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-10-22 16:35:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(pipeline): verify message and provider behavior

Browse Source
This commit is contained in:
huanghuoguoguo
2026-10-17 15:12:25 +08:00
parent bbf517edbc
commit bcfe78aa0e
11 changed files with 186 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit_tests/pipeline/test_aggregator.py
+59 -16
View File
@@ -465,8 +465,19 @@ class TestMessageAggregatorAddMessage:
pipeline_uuid=None, # None -> disabled
)
# Should have called query_pool.add_query
assert app.query_pool.add_query.called
app.query_pool.add_query.assert_awaited_once_with(
execution_context=execution_context(),
bot_uuid='test-bot',
launcher_type=provider_session.LauncherTypes.PERSON,
launcher_id=12345,
sender_id=12345,
message_event=event,
message_chain=chain,
adapter=adapter,
pipeline_uuid=None,
routed_by_rule=False,
variables=None,
)
@pytest.mark.asyncio
async def test_enabled_buffers_message(self):
@@ -505,8 +516,27 @@ class TestMessageAggregatorAddMessage:
pipeline_uuid='test-pipeline',
)
# Should have buffered the message
assert len(agg.buffers) == 1
context = execution_context(pipeline_uuid='test-pipeline')
key = aggregation_key(context, pipeline_uuid='test-pipeline')
assert list(agg.buffers) == [key]
assert agg.buffers[key].execution_context == context
assert len(agg.buffers[key].messages) == 1
pending = agg.buffers[key].messages[0]
assert {key: value for key, value in vars(pending).items() if key != 'timestamp'} == {
'execution_context': context,
'bot_uuid': 'test-bot',
'launcher_type': provider_session.LauncherTypes.PERSON,
'launcher_id': 12345,
'sender_id': 12345,
'message_event': event,
'message_chain': chain,
'adapter': adapter,
'pipeline_uuid': 'test-pipeline',
'routed_by_rule': False,
'variables': None,
}
app.query_pool.add_query.assert_not_awaited()
await agg.flush_all()
@pytest.mark.asyncio
async def test_control_variables_bypass_aggregation(self):
@@ -658,10 +688,7 @@ class TestMessageAggregatorMerge:
merged = agg._merge_messages([pending1, pending2])
# Should contain both messages with separator
merged_str = str(merged.message_chain)
assert 'hello' in merged_str
assert 'world' in merged_str
assert str(merged.message_chain) == 'hello\nworld'
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."""
@@ -760,7 +787,19 @@ class TestMessageAggregatorFlush:
await agg._flush_buffer(key, context)
assert app.query_pool.add_query.called
app.query_pool.add_query.assert_awaited_once_with(
execution_context=context,
bot_uuid='test-bot',
launcher_type=provider_session.LauncherTypes.PERSON,
launcher_id=12345,
sender_id=12345,
message_event=event,
message_chain=chain,
adapter=adapter,
pipeline_uuid=None,
routed_by_rule=False,
variables=None,
)
assert key not in agg.buffers
@pytest.mark.asyncio
@@ -825,8 +864,8 @@ class TestMessageAggregatorFlushAll:
app = make_aggregator_app()
agg = aggregator.MessageAggregator(app)
chain = text_chain('hello')
event = friend_message_event(chain)
first_chain = text_chain('first')
second_chain = text_chain('second')
adapter = mock_adapter()
# Create two buffers
@@ -836,8 +875,8 @@ class TestMessageAggregatorFlushAll:
launcher_type=provider_session.LauncherTypes.PERSON,
launcher_id=12345,
sender_id=12345,
message_event=event,
message_chain=chain,
message_event=friend_message_event(first_chain),
message_chain=first_chain,
adapter=adapter,
pipeline_uuid=None,
)
@@ -848,8 +887,8 @@ class TestMessageAggregatorFlushAll:
launcher_type=provider_session.LauncherTypes.PERSON,
launcher_id=67890,
sender_id=67890,
message_event=event,
message_chain=chain,
message_event=friend_message_event(second_chain),
message_chain=second_chain,
adapter=adapter,
pipeline_uuid=None,
)
@@ -875,7 +914,11 @@ class TestMessageAggregatorFlushAll:
# Both buffers should be flushed
assert len(agg.buffers) == 0
assert app.query_pool.add_query.call_count == 2
assert app.query_pool.add_query.await_count == 2
assert [
(sent.kwargs['launcher_id'], str(sent.kwargs['message_chain']), sent.kwargs['execution_context'])
for sent in app.query_pool.add_query.await_args_list
] == [(12345, 'first', context), (67890, 'second', context)]
class TestMessageAggregatorWorkspaceIsolation:
tests/unit_tests/pipeline/test_chat_handler.py
+38 -6
View File
@@ -89,6 +89,25 @@ def set_runner(fake_app):
return _set_runner
@pytest.fixture
def chat_session():
from langbot_plugin.api.entities.builtin.provider.session import Session, Conversation, LauncherTypes
from langbot_plugin.api.entities.builtin.provider.prompt import Prompt
return Session(
launcher_type=LauncherTypes.PERSON,
launcher_id=12345,
sender_id=12345,
bot_uuid='test-bot-uuid',
using_conversation=Conversation(
prompt=Prompt(name='default', messages=[]),
messages=[],
pipeline_uuid='test-pipeline-uuid',
bot_uuid='test-bot-uuid',
),
)
# ============== CACHED LAZY IMPORTS ==============
_chat_handler_module = None
@@ -223,7 +242,9 @@ class TestChatMessageHandlerReal:
assert query.user_message.content is not None
@pytest.mark.asyncio
async def test_adapter_without_stream_method_defaults_non_stream(self, fake_app, mock_event_ctx, set_runner):
async def test_adapter_without_stream_method_defaults_non_stream(
self, fake_app, mock_event_ctx, set_runner, chat_session
):
"""Adapter without is_stream_output_supported defaults to non-stream."""
from tests.factories import text_query
from langbot_plugin.api.entities.builtin.provider.message import Message, ContentElement
@@ -237,6 +258,7 @@ class TestChatMessageHandlerReal:
query = text_query('test')
query.adapter = Mock(spec=[])
query.user_message = Message(role='user', content=[ContentElement.from_text('test')])
query.session = chat_session
class SingleRunner:
name = 'local-agent'
@@ -256,7 +278,9 @@ class TestChatMessageHandlerReal:
async for result in handler.handle(query):
results.append(result)
assert len(results) >= 1
assert len(results) == 1
assert results[0].result_type == get_entities().ResultType.CONTINUE
assert [(message.role, message.content) for message in query.resp_messages] == [('assistant', 'response')]
@pytest.mark.usefixtures('mock_circular_import_chain')
@@ -315,7 +339,7 @@ class TestChatHandlerStreaming:
query.adapter.create_message_card.assert_awaited_once()
@pytest.mark.asyncio
async def test_streaming_chunks_collected(self, fake_app, mock_event_ctx, set_runner):
async def test_streaming_chunks_collected(self, fake_app, mock_event_ctx, set_runner, chat_session):
"""Streaming produces multiple results."""
from tests.factories import text_query
from langbot_plugin.api.entities.builtin.provider.message import Message, ContentElement, MessageChunk
@@ -330,6 +354,7 @@ class TestChatHandlerStreaming:
query.adapter.is_stream_output_supported = AsyncMock(return_value=True)
query.adapter.create_message_card = AsyncMock()
query.user_message = Message(role='user', content=[ContentElement.from_text('test')])
query.session = chat_session
class StreamRunner:
name = 'local-agent'
@@ -340,17 +365,24 @@ class TestChatHandlerStreaming:
async def run(self, query):
yield MessageChunk(role='assistant', content='Hello', is_final=False)
yield MessageChunk(role='assistant', content=' World', is_final=True)
yield MessageChunk(role='assistant', content='Hello World', is_final=True)
set_runner(StreamRunner)
handler = chat.ChatMessageHandler(fake_app)
results = []
delivered = []
async for result in handler.handle(query):
results.append(result)
delivered.append(query.resp_messages[-1])
assert len(results) >= 1
assert len(results) == 2
assert all(result.result_type == get_entities().ResultType.CONTINUE for result in results)
assert [chunk.content for chunk in delivered] == ['Hello', 'Hello World']
assert [chunk.is_final for chunk in delivered] == [False, True]
assert len({chunk.resp_message_id for chunk in delivered}) == 1
query.adapter.create_message_card.assert_awaited_once()
@pytest.mark.usefixtures('mock_circular_import_chain')
@@ -536,7 +568,7 @@ class TestChatHandlerHelper:
chat = get_chat_handler()
handler = chat.ChatMessageHandler(fake_app)
result = handler.cut_str('first line\nsecond line')
assert '...' in result
assert result == 'first line...'
def test_response_size_limit_uses_instance_config(self, fake_app):
from langbot_plugin.api.entities.builtin.provider.message import Message
tests/unit_tests/pipeline/test_command_handler.py
+3 -1
View File
@@ -367,6 +367,8 @@ class TestCommandHandlerReal:
assert len(msg.content) == 2
assert msg.content[0].type == 'text'
assert msg.content[1].type == 'image_url'
assert msg.content[0].text == 'Here is the image:'
assert msg.content[1].image_url.url == 'https://example.com/image.png'
@pytest.mark.asyncio
async def test_command_result_empty_interrupts(self, fake_app, mock_event_ctx, mock_execute_factory):
@@ -410,4 +412,4 @@ class TestCommandHandlerHelper:
command = get_command_handler()
handler = command.CommandHandler(fake_app)
result = handler.cut_str('first line\nsecond line')
assert '...' in result
assert result == 'first line...'
tests/unit_tests/pipeline/test_longtext.py
+8 -8
View File
@@ -13,6 +13,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,
@@ -73,7 +74,7 @@ class TestLongTextProcessStageInit:
async def test_initialize_forward_strategy(self):
"""Initialize with strategy='forward' should use ForwardComponentStrategy."""
longtext = get_longtext_module()
strategy = get_strategy_module()
forward = import_module('langbot.pkg.pipeline.longtext.strategies.forward')
app = FakeApp()
stage = longtext.LongTextProcessStage(app)
@@ -82,8 +83,7 @@ class TestLongTextProcessStageInit:
await stage.initialize(pipeline_config)
assert stage.strategy_impl is not None
assert isinstance(stage.strategy_impl, strategy.LongTextStrategy)
assert isinstance(stage.strategy_impl, forward.ForwardComponentStrategy)
@pytest.mark.asyncio
async def test_initialize_unknown_strategy_raises(self):
@@ -340,10 +340,11 @@ class TestLongTextProcessStageImageStrategy:
"""Tests for image strategy handling (requires PIL/font)."""
@pytest.mark.asyncio
async def test_image_strategy_missing_font_fallback(self):
async def test_image_strategy_missing_font_fallback(self, monkeypatch):
"""Missing font should fallback to forward strategy."""
longtext = get_longtext_module()
strategy = get_strategy_module()
forward = import_module('langbot.pkg.pipeline.longtext.strategies.forward')
monkeypatch.setattr(longtext, 'os', SimpleNamespace(name='posix', path=SimpleNamespace(exists=lambda _: False)))
app = FakeApp()
stage = longtext.LongTextProcessStage(app)
@@ -354,6 +355,5 @@ class TestLongTextProcessStageImageStrategy:
# On non-Windows without font, should fallback to forward
await stage.initialize(pipeline_config)
# Should have initialized (possibly with fallback strategy)
if stage.strategy_impl is not None:
assert isinstance(stage.strategy_impl, strategy.LongTextStrategy)
assert pipeline_config['output']['long-text-processing']['strategy'] == 'forward'
assert isinstance(stage.strategy_impl, forward.ForwardComponentStrategy)
tests/unit_tests/pipeline/test_preproc.py
+30 -9
View File
@@ -190,6 +190,7 @@ class TestPreProcessorNormalText:
assert result.new_query.user_message is not None
assert result.new_query.user_message.role == 'user'
assert [(part.type, part.text) for part in result.new_query.user_message.content] == [('text', 'test message')]
class TestPreProcessorEmptyMessage:
@@ -253,6 +254,7 @@ class TestPreProcessorImageSegment:
mock_model = Mock()
mock_model.model_entity = Mock(uuid='vision-model', abilities=['func_call', 'vision'])
app.model_mgr.get_model_by_uuid = AsyncMock(return_value=mock_model)
attach_runner_descriptor(app)
app.tool_mgr.get_all_tools = AsyncMock(return_value=[])
mock_event_ctx = Mock()
@@ -276,8 +278,13 @@ class TestPreProcessorImageSegment:
result = await stage.process(query, 'PreProcessor')
assert result.result_type == preproc.entities.ResultType.CONTINUE
# User message should have content
assert result.new_query.user_message.content is not None
assert result.new_query.use_llm_model_uuid == 'vision-model'
from langbot_plugin.api.entities.builtin.provider.message import ContentElement
assert result.new_query.user_message.content == [
ContentElement.from_text('look at this'),
ContentElement.from_image_base64('data:image/png;base64,abc123'),
]
@pytest.mark.asyncio
async def test_image_without_vision_model(self):
@@ -299,6 +306,7 @@ class TestPreProcessorImageSegment:
mock_model = Mock()
mock_model.model_entity = Mock(uuid='text-only-model', abilities=['func_call'])
app.model_mgr.get_model_by_uuid = AsyncMock(return_value=mock_model)
attach_runner_descriptor(app)
app.tool_mgr.get_all_tools = AsyncMock(return_value=[])
mock_event_ctx = Mock()
@@ -311,6 +319,12 @@ class TestPreProcessorImageSegment:
result = await stage.process(query, 'PreProcessor')
assert result.result_type == preproc.entities.ResultType.CONTINUE
assert result.new_query.use_llm_model_uuid == 'text-only-model'
assert [(part.type, part.text) for part in result.new_query.user_message.content] == [
('text', 'describe this'),
('text', '[Image]'),
]
assert result.new_query.variables['user_message_text'] == 'describe this[Image]'
class TestPreProcessorModelSelection:
@@ -455,11 +469,18 @@ class TestPreProcessorVariables:
result = await stage.process(query, 'PreProcessor')
variables = result.new_query.variables
assert 'launcher_type' in variables
assert 'launcher_id' in variables
assert 'sender_id' in variables
assert variables['sender_id'] == 67890
assert 'user_message_text' in variables
expected = {
'launcher_type': 'person',
'launcher_id': 12345,
'sender_id': 67890,
'session_id': 'person_12345',
'conversation_id': 'conv-123',
'msg_create_time': 1609459200,
'group_name': '',
'sender_name': 'TestUser',
'user_message_text': 'hello',
}
assert {key: variables[key] for key in expected} == expected
@pytest.mark.asyncio
async def test_group_variables_include_group_name(self):
@@ -493,8 +514,8 @@ class TestPreProcessorVariables:
result = await stage.process(query, 'PreProcessor')
variables = result.new_query.variables
assert 'group_name' in variables
assert 'sender_name' in variables
assert variables['group_name'] == 'TestGroup'
assert variables['sender_name'] == 'TestUser'
class TestPreProcessorToolSelection:
tests/unit_tests/pipeline/test_ratelimit.py
+6 -3
View File
@@ -272,11 +272,14 @@ class TestFixedWindowAlgo:
algo = fixedwin.FixedWindowAlgo(mock_app_for_algo)
await algo.initialize()
# release_access is empty in current implementation
await algo.require_access(sample_query_with_rate_limit, provider_session.LauncherTypes.PERSON, '12345')
containers = dict(algo.containers)
records = {key: dict(container.records) for key, container in containers.items()}
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
assert algo.containers == containers
assert {key: dict(container.records) for key, container in algo.containers.items()} == records
# Original mock-based tests for RateLimit stage integration
tests/unit_tests/pipeline/test_wrapper.py
+23 -12
View File
@@ -22,6 +22,7 @@ from tests.factories import (
import langbot_plugin.api.entities.builtin.platform.message as platform_message
import langbot_plugin.api.entities.builtin.provider.session as provider_session
from langbot_plugin.api.entities.builtin.provider.message import Message
def get_wrapper_module():
@@ -109,7 +110,7 @@ class TestResponseWrapperMessageChain:
assert len(results) == 1
assert results[0].result_type == entities.ResultType.CONTINUE
assert len(results[0].new_query.resp_message_chain) == 1
assert [str(chain) for chain in results[0].new_query.resp_message_chain] == ['response']
@pytest.mark.asyncio
async def test_message_chain_direct_append_consumes_pending_plugin_source(self):
@@ -171,12 +172,7 @@ class TestResponseWrapperCommand:
query.pipeline_config = pipeline_config
query.resp_message_chain = []
# Create a command response message
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')])
)
command_resp = Message(role='command', content='Help info')
query.resp_messages = [command_resp]
results = []
@@ -185,8 +181,7 @@ class TestResponseWrapperCommand:
assert len(results) == 1
assert results[0].result_type == entities.ResultType.CONTINUE
# Check that prefix was added (via get_content_platform_message_chain)
command_resp.get_content_platform_message_chain.assert_called_once()
assert [str(chain) for chain in query.resp_message_chain] == ['[bot] Help info']
class TestResponseWrapperPlugin:
@@ -223,6 +218,7 @@ class TestResponseWrapperPlugin:
assert len(results) == 1
assert results[0].result_type == entities.ResultType.CONTINUE
assert [str(chain) for chain in query.resp_message_chain] == ['Plugin response']
class TestResponseWrapperAssistant:
@@ -316,8 +312,11 @@ class TestResponseWrapperAssistant:
assert len(results) == 1
assert results[0].result_type == entities.ResultType.CONTINUE
# Event should have been emitted
app.plugin_connector.emit_event.assert_called()
assert [str(chain) for chain in query.resp_message_chain] == ['Hello back!']
app.plugin_connector.emit_event.assert_awaited_once()
event = app.plugin_connector.emit_event.await_args.args[0]
assert event.response_text == 'Hello back!'
assert event.funcs_called == []
@pytest.mark.asyncio
async def test_assistant_empty_content(self):
@@ -386,24 +385,36 @@ class TestResponseWrapperAssistant:
mock_tool_call = Mock()
mock_tool_call.function = Mock()
mock_tool_call.function.name = 'test_function'
other_tool_call = Mock()
other_tool_call.function.name = 'other_function'
assistant_resp = Mock()
assistant_resp.role = 'assistant'
assistant_resp.content = 'Processing...'
assistant_resp.tool_calls = [mock_tool_call]
assistant_resp.tool_calls = [mock_tool_call, other_tool_call]
assistant_resp.get_content_platform_message_chain = Mock(
return_value=platform_message.MessageChain([platform_message.Plain(text='Processing...')])
)
query.resp_messages = [assistant_resp]
results = []
delivered = []
async for result in stage.process(query, 'ResponseWrapper'):
results.append(result)
delivered.append(str(query.resp_message_chain[-1]))
assert len(results) == 2
for result in results:
assert result.result_type == entities.ResultType.CONTINUE
assert app.plugin_connector.emit_event.await_count == 2
expected = ['Processing...', 'Calling test_function, other_function...']
assert delivered == expected
events = [sent.args[0] for sent in app.plugin_connector.emit_event.await_args_list]
assert [event.response_text for event in events] == expected
assert [event.funcs_called for event in events] == [
['test_function', 'other_function'],
['test_function', 'other_function'],
]
class TestResponseWrapperInterrupt:
tests/unit_tests/provider/test_litellmchat.py
+4 -6
View File
@@ -1048,10 +1048,8 @@ class TestInvokeRerank:
documents=['Paris is the capital.', 'London is a city.', 'France is in Europe.'],
)
assert len(results) == 3
# Scores should be normalized
assert results[0]['index'] == 0
assert results[0]['relevance_score'] >= 0 and results[0]['relevance_score'] <= 1
assert [result['index'] for result in results] == [0, 1, 2]
assert [result['relevance_score'] for result in results] == pytest.approx([1.0, 0.0, 0.76923076923])
@pytest.mark.asyncio
async def test_invoke_rerank_normalization(self):
@@ -1297,8 +1295,8 @@ class TestConvertMessages:
result = requester._convert_messages(messages)
assert len(result) == 1
# LiteLLM handles multiple text parts, we pass them through
assert isinstance(result[0]['content'], list)
assert result[0]['role'] == 'user'
assert result[0]['content'] == [{'type': 'text', 'text': 'Hello'}, {'type': 'text', 'text': 'World'}]
class TestScanModels:
tests/unit_tests/provider/test_requester_base.py
+10 -14
View File
@@ -56,11 +56,8 @@ def test_requester_base_class_is_abstract():
mock_app = SimpleNamespace()
mock_app.logger = Mock()
# ProviderAPIRequester has abstract methods, but ABCMeta allows instantiation
# if you don't call the abstract methods. Test that it has abstract methods.
assert hasattr(requester.ProviderAPIRequester, 'invoke_llm')
# Check that invoke_llm is abstract
assert hasattr(requester.ProviderAPIRequester.invoke_llm, '__isabstractmethod__')
with pytest.raises(TypeError, match='abstract'):
requester.ProviderAPIRequester(mock_app, {})
def test_requester_default_config_merged():
@@ -251,13 +248,13 @@ def test_runtime_provider_initialization(runtime_provider, fake_persistence_data
def test_runtime_provider_has_invoke_methods(runtime_provider):
"""Test RuntimeProvider has invoke methods that delegate to requester."""
"""RuntimeProvider exposes callable invocation entrypoints."""
provider = runtime_provider
assert hasattr(provider, 'invoke_llm')
assert hasattr(provider, 'invoke_llm_stream')
assert hasattr(provider, 'invoke_embedding')
assert hasattr(provider, 'invoke_rerank')
assert callable(provider.invoke_llm)
assert callable(provider.invoke_llm_stream)
assert callable(provider.invoke_embedding)
assert callable(provider.invoke_rerank)
@pytest.mark.asyncio
@@ -404,6 +401,7 @@ async def test_runtime_provider_invoke_llm_stream_yields_chunks(runtime_provider
assert len(chunks) == 1
assert chunks[0].role == 'assistant'
assert [(part.type, part.text) for part in chunks[0].content] == [('text', 'Fake stream chunk')]
@pytest.mark.asyncio
@@ -513,13 +511,11 @@ def test_runtime_llm_model_initialization(runtime_llm_model, fake_persistence_da
assert model.reasoning_config_override is None
def test_runtime_llm_model_provider_ref(runtime_llm_model):
def test_runtime_llm_model_provider_ref(runtime_llm_model, runtime_provider):
"""Test RuntimeLLMModel has correct provider reference."""
model = runtime_llm_model
assert model.provider.provider_entity is not None
assert model.provider.token_mgr is not None
assert model.provider.requester is not None
assert model.provider is runtime_provider
# ============================================================================
tests/unit_tests/provider/test_session_manager.py
+4 -1
View File
@@ -365,7 +365,10 @@ class TestSessionManagerGetConversation:
)
assert conversation.prompt.name == 'default'
assert len(conversation.prompt.messages) == 2
assert [(message.role, message.content) for message in conversation.prompt.messages] == [
('system', 'System message'),
('user', 'User message'),
]
class TestSessionManagerWorkspaceIsolation:
tests/unit_tests/vector/test_filter_utils.py
+1 -3
View File
@@ -198,6 +198,4 @@ class TestSupportedOpsConstant:
def test_supported_ops_is_frozenset(self):
"""SUPPORTED_OPS is a frozenset for immutability."""
from collections.abc import Set
assert isinstance(SUPPORTED_OPS, Set)
assert isinstance(SUPPORTED_OPS, frozenset)