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

test: format test suite

Browse Source
This commit is contained in:
huanghuoguoguo
2026-06-16 11:13:05 +08:00
parent 1ae5aacc00
commit ff0c5a6f0a
92 changed files with 1658 additions and 1713 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/smoke/__init__.py
+1 -1
View File
@@ -3,4 +3,4 @@ Smoke tests package.
Smoke tests verify basic functionality works without testing edge cases.
Run with: uv run pytest tests/smoke/ -q
"""
"""
tests/smoke/test_fake_message_flow.py
+67 -64
View File
@@ -39,19 +39,19 @@ class TestFakeMessageFlow:
assert app.instance_config is not None
# Verify default config
assert app.instance_config.data["command"]["prefix"] == ["/", "!"]
assert app.instance_config.data["command"]["enable"] is True
assert app.instance_config.data['command']['prefix'] == ['/', '!']
assert app.instance_config.data['command']['enable'] is True
@pytest.mark.asyncio
async def test_fake_provider_returns_text(self):
"""Test FakeProvider returns configured response."""
provider = FakeProvider(default_response="test response")
provider = FakeProvider(default_response='test response')
# Create mock model with provider
model = fake_model(provider=provider)
# Create a simple query
query = text_query("hello")
query = text_query('hello')
# Simulate invoke
result = await provider.invoke_llm(
@@ -63,15 +63,15 @@ class TestFakeMessageFlow:
)
assert result is not None
assert result.role == "assistant"
assert result.content == "test response"
assert result.role == 'assistant'
assert result.content == 'test response'
@pytest.mark.asyncio
async def test_fake_provider_pong(self):
"""Test FakeProvider returns LANGBOT_FAKE_PONG marker."""
provider = fake_provider_pong()
model = fake_model(provider=provider)
query = text_query("ping")
query = text_query('ping')
result = await provider.invoke_llm(
query=query,
@@ -86,9 +86,9 @@ class TestFakeMessageFlow:
@pytest.mark.asyncio
async def test_fake_provider_streaming(self):
"""Test FakeProvider streaming response."""
provider = FakeProvider().returns_streaming(["Hello", " World"])
provider = FakeProvider().returns_streaming(['Hello', ' World'])
model = fake_model(provider=provider)
query = text_query("hello")
query = text_query('hello')
chunks = []
# invoke_llm_stream returns an async generator, don't await it
@@ -102,8 +102,8 @@ class TestFakeMessageFlow:
chunks.append(chunk)
assert len(chunks) == 2
assert chunks[0].content == "Hello"
assert chunks[1].content == " World"
assert chunks[0].content == 'Hello'
assert chunks[1].content == ' World'
assert chunks[1].is_final is True
@pytest.mark.asyncio
@@ -111,9 +111,9 @@ class TestFakeMessageFlow:
"""Test FakeProvider simulates timeout error."""
provider = FakeProvider().timeout()
model = fake_model(provider=provider)
query = text_query("hello")
query = text_query('hello')
with pytest.raises(TimeoutError, match="Provider timeout"):
with pytest.raises(TimeoutError, match='Provider timeout'):
await provider.invoke_llm(
query=query,
model=model,
@@ -127,9 +127,9 @@ class TestFakeMessageFlow:
"""Test FakeProvider simulates rate limit error."""
provider = FakeProvider().rate_limit()
model = fake_model(provider=provider)
query = text_query("hello")
query = text_query('hello')
with pytest.raises(Exception, match="Rate limit exceeded"):
with pytest.raises(Exception, match='Rate limit exceeded'):
await provider.invoke_llm(
query=query,
model=model,
@@ -142,34 +142,34 @@ class TestFakeMessageFlow:
async def test_fake_provider_captures_requests(self):
"""Test FakeProvider captures request arguments."""
provider = FakeProvider()
model = fake_model(name="gpt-4", provider=provider)
query = text_query("hello")
model = fake_model(name='gpt-4', provider=provider)
query = text_query('hello')
await provider.invoke_llm(
query=query,
model=model,
messages=[{"role": "user", "content": "hello"}],
funcs=[{"name": "test_func"}],
extra_args={"temperature": 0.7},
messages=[{'role': 'user', 'content': 'hello'}],
funcs=[{'name': 'test_func'}],
extra_args={'temperature': 0.7},
)
captured = provider.get_captured_requests()
assert len(captured) == 1
assert captured[0]["model"] == "gpt-4"
assert captured[0]["messages"] == [{"role": "user", "content": "hello"}]
assert captured[0]["funcs"] == [{"name": "test_func"}]
assert captured[0]["extra_args"] == {"temperature": 0.7}
assert captured[0]['model'] == 'gpt-4'
assert captured[0]['messages'] == [{'role': 'user', 'content': 'hello'}]
assert captured[0]['funcs'] == [{'name': 'test_func'}]
assert captured[0]['extra_args'] == {'temperature': 0.7}
@pytest.mark.asyncio
async def test_fake_platform_capture_outbound(self):
"""Test FakePlatform captures outbound messages."""
platform = FakePlatform(bot_account_id="test-bot")
query = text_query("hello")
platform = FakePlatform(bot_account_id='test-bot')
query = text_query('hello')
# Simulate sending reply
from tests.factories.message import text_chain
reply_chain = text_chain("response text")
reply_chain = text_chain('response text')
event = query.message_event
await platform.reply_message(event, reply_chain, quote_origin=False)
@@ -177,38 +177,38 @@ class TestFakeMessageFlow:
# Verify captured
outbound = platform.get_outbound_messages()
assert len(outbound) == 1
assert outbound[0]["type"] == "reply"
assert outbound[0]["message"] == reply_chain
assert outbound[0]['type'] == 'reply'
assert outbound[0]['message'] == reply_chain
@pytest.mark.asyncio
async def test_fake_platform_friend_message(self):
"""Test FakePlatform creates friend message events."""
platform = FakePlatform(bot_account_id="test-bot")
platform = FakePlatform(bot_account_id='test-bot')
event = platform.create_friend_message(
text="hello bot",
text='hello bot',
sender_id=12345,
nickname="TestUser",
nickname='TestUser',
)
assert event.type == "FriendMessage"
assert event.type == 'FriendMessage'
assert event.sender.id == 12345
assert event.sender.nickname == "TestUser"
assert str(event.message_chain) == "hello bot"
assert event.sender.nickname == 'TestUser'
assert str(event.message_chain) == 'hello bot'
@pytest.mark.asyncio
async def test_fake_platform_group_message_with_mention(self):
"""Test FakePlatform creates group message with @mention."""
platform = FakePlatform(bot_account_id="test-bot")
platform = FakePlatform(bot_account_id='test-bot')
event = platform.create_group_message(
text="hello everyone",
text='hello everyone',
sender_id=12345,
group_id=99999,
mention_bot=True,
)
assert event.type == "GroupMessage"
assert event.type == 'GroupMessage'
assert event.sender.id == 12345
assert event.group.id == 99999
@@ -220,54 +220,57 @@ class TestFakeMessageFlow:
async def test_query_factories_basic(self):
"""Test basic query factory functions."""
# Text query
q1 = text_query("hello world")
assert q1.launcher_type.value == "person"
assert str(q1.message_chain) == "hello world"
q1 = text_query('hello world')
assert q1.launcher_type.value == 'person'
assert str(q1.message_chain) == 'hello world'
# Group query
from tests.factories import group_text_query
q2 = group_text_query("hello group", group_id=88888)
assert q2.launcher_type.value == "group"
q2 = group_text_query('hello group', group_id=88888)
assert q2.launcher_type.value == 'group'
assert q2.launcher_id == 88888
# Command query
from tests.factories import command_query
q3 = command_query("help", prefix="/")
assert str(q3.message_chain) == "/help"
q3 = command_query('help', prefix='/')
assert str(q3.message_chain) == '/help'
# Mention query
from tests.factories import mention_query
q4 = mention_query("hi", target="test-bot", group_id=77777)
assert q4.launcher_type.value == "group"
q4 = mention_query('hi', target='test-bot', group_id=77777)
assert q4.launcher_type.value == 'group'
@pytest.mark.asyncio
async def test_fake_platform_send_failure(self):
"""Test FakePlatform simulates send failure."""
platform = FakePlatform().send_failure()
query = text_query("hello")
query = text_query('hello')
from tests.factories.message import text_chain
with pytest.raises(Exception, match="Platform send failure"):
with pytest.raises(Exception, match='Platform send failure'):
await platform.reply_message(
query.message_event,
text_chain("response"),
text_chain('response'),
)
@pytest.mark.asyncio
async def test_mock_platform_adapter(self):
"""Test mock_platform_adapter helper."""
platform = FakePlatform(bot_account_id="bot-123")
platform = FakePlatform(bot_account_id='bot-123')
adapter = mock_platform_adapter(platform)
assert adapter.bot_account_id == "bot-123"
assert adapter.bot_account_id == 'bot-123'
assert adapter._fake_platform is platform
# Test reply_message is wired
from tests.factories.message import text_chain
query = text_query("test")
await adapter.reply_message(query.message_event, text_chain("response"))
query = text_query('test')
await adapter.reply_message(query.message_event, text_chain('response'))
# Verify platform captured it
assert len(platform.get_outbound_messages()) == 1
@@ -293,18 +296,18 @@ class TestMessageFlowIntegration:
Note: This does NOT run actual LangBot pipeline stages.
"""
# Setup
platform = FakePlatform(bot_account_id="test-bot")
platform = FakePlatform(bot_account_id='test-bot')
provider = fake_provider_pong()
model = fake_model(provider=provider)
# Create inbound message
query = text_query("ping")
query = text_query('ping')
# Simulate provider processing
response = await provider.invoke_llm(
query=query,
model=model,
messages=[{"role": "user", "content": "ping"}],
messages=[{'role': 'user', 'content': 'ping'}],
funcs=[],
extra_args={},
)
@@ -321,16 +324,16 @@ class TestMessageFlowIntegration:
# Verify platform captured outbound
outbound = platform.get_outbound_messages()
assert len(outbound) == 1
assert outbound[0]["type"] == "reply"
assert str(outbound[0]["message"]) == FakeProvider.PONG_RESPONSE
assert outbound[0]['type'] == 'reply'
assert str(outbound[0]['message']) == FakeProvider.PONG_RESPONSE
@pytest.mark.asyncio
async def test_streaming_message_flow(self):
"""Smoke test: streaming message flow."""
platform = FakePlatform().supports_streaming()
provider = FakeProvider().returns_streaming(["Hello", " there"])
provider = FakeProvider().returns_streaming(['Hello', ' there'])
model = fake_model(provider=provider)
query = text_query("hi")
query = text_query('hi')
chunks = []
async for chunk in provider.invoke_llm_stream(
@@ -344,8 +347,8 @@ class TestMessageFlowIntegration:
# Verify streaming worked
assert len(chunks) == 2
full_content = "".join(c.content for c in chunks)
assert full_content == "Hello there"
full_content = ''.join(c.content for c in chunks)
assert full_content == 'Hello there'
# Verify platform supports streaming
assert await platform.is_stream_output_supported() is True
assert await platform.is_stream_output_supported() is True