Microsoft Bot Framework Bot Builder core functionality for building conversational AI bots and chatbots in Python.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Testing utilities for unit testing bots including test adapters, test flows, and assertion helpers. Enables comprehensive testing of bot logic without requiring actual Bot Framework channels.
Test adapter for unit testing bots that simulates Bot Framework functionality without requiring actual HTTP endpoints or channel connections.
class TestAdapter(BotAdapter):
def __init__(self, conversation_reference=None):
"""
Initialize test adapter.
Args:
conversation_reference (ConversationReference, optional): Default conversation reference
"""
async def send_activities(self, context: TurnContext, activities):
"""Send activities in test environment."""
async def update_activity(self, context: TurnContext, activity):
"""Update activity in test environment."""
async def delete_activity(self, context: TurnContext, reference):
"""Delete activity in test environment."""
async def process_activity(self, activity, logic):
"""
Process activity with bot logic.
Args:
activity (Activity): Activity to process
logic: Bot logic function to execute
Returns:
list: List of response activities
"""
async def test(self, user_says: str, expected_replies, description: str = None, timeout: int = 3):
"""
Test a single conversation turn.
Args:
user_says (str): User input text
expected_replies (str or list): Expected bot responses
description (str, optional): Test description
timeout (int): Timeout in seconds
Returns:
TestFlow: Test flow for chaining
"""
def make_activity(self, text: str = None):
"""
Create test activity.
Args:
text (str, optional): Activity text
Returns:
Activity: Test activity
"""
def get_next_reply(self):
"""
Get next reply from bot.
Returns:
Activity: Next bot reply or None
"""
def activity_buffer(self):
"""
Get all activities in buffer.
Returns:
list: List of activities
"""Fluent interface for testing bot conversations that allows chaining multiple conversation turns and assertions for comprehensive bot testing.
class TestFlow:
def __init__(self, test_task, adapter):
"""Initialize test flow."""
def test(self, user_says: str, expected_replies, description: str = None, timeout: int = 3):
"""
Test a single turn of conversation.
Args:
user_says (str): User input text
expected_replies (str or list): Expected bot responses
description (str, optional): Test description
timeout (int): Timeout in seconds
Returns:
TestFlow: Self for method chaining
"""
def send(self, user_says: str):
"""
Send user message to bot.
Args:
user_says (str): User input text
Returns:
TestFlow: Self for method chaining
"""
def assert_reply(self, expected_reply, description: str = None, timeout: int = 3):
"""
Assert bot reply matches expected response.
Args:
expected_reply (str or callable): Expected reply or validator function
description (str, optional): Assertion description
timeout (int): Timeout in seconds
Returns:
TestFlow: Self for method chaining
"""
def assert_reply_one_of(self, expected_replies, description: str = None, timeout: int = 3):
"""
Assert bot reply matches one of expected responses.
Args:
expected_replies (list): List of possible expected replies
description (str, optional): Assertion description
timeout (int): Timeout in seconds
Returns:
TestFlow: Self for method chaining
"""
def assert_no_reply(self, description: str = None, timeout: int = 3):
"""
Assert bot sends no reply.
Args:
description (str, optional): Assertion description
timeout (int): Timeout in seconds
Returns:
TestFlow: Self for method chaining
"""
async def start_test(self):
"""
Start the test conversation flow.
Returns:
TestFlow: Self for method chaining
"""import pytest
from botbuilder.core import TestAdapter, ActivityHandler, TurnContext, MessageFactory
class EchoBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
reply_text = f"You said: {turn_context.activity.text}"
await turn_context.send_activity(MessageFactory.text(reply_text))
class TestEchoBot:
@pytest.mark.asyncio
async def test_echo_response(self):
# Create test adapter and bot
adapter = TestAdapter()
bot = EchoBot()
# Test single interaction
await adapter.test("hello", "You said: hello") \
.start_test()
@pytest.mark.asyncio
async def test_multiple_turns(self):
adapter = TestAdapter()
bot = EchoBot()
# Test conversation flow
await adapter.test("hello", "You said: hello") \
.test("how are you?", "You said: how are you?") \
.test("goodbye", "You said: goodbye") \
.start_test()class WeatherBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
text = turn_context.activity.text.lower()
if "weather" in text:
await turn_context.send_activity(MessageFactory.text("It's sunny today!"))
elif "hello" in text:
await turn_context.send_activity(MessageFactory.text("Hello! Ask me about the weather."))
else:
await turn_context.send_activity(MessageFactory.text("I can help with weather information."))
class TestWeatherBot:
@pytest.mark.asyncio
async def test_weather_query(self):
adapter = TestAdapter()
bot = WeatherBot()
await adapter.test("what's the weather?", "It's sunny today!") \
.start_test()
@pytest.mark.asyncio
async def test_greeting(self):
adapter = TestAdapter()
bot = WeatherBot()
await adapter.test("hello", "Hello! Ask me about the weather.") \
.start_test()
@pytest.mark.asyncio
async def test_unknown_input(self):
adapter = TestAdapter()
bot = WeatherBot()
await adapter.test("random text", "I can help with weather information.") \
.start_test()
@pytest.mark.asyncio
async def test_conversation_flow(self):
adapter = TestAdapter()
bot = WeatherBot()
await adapter.test("hi", "Hello! Ask me about the weather.") \
.test("weather", "It's sunny today!") \
.test("thanks", "I can help with weather information.") \
.start_test()from botbuilder.core import ConversationState, UserState, MemoryStorage
class CounterBot(ActivityHandler):
def __init__(self, conversation_state: ConversationState):
self.conversation_state = conversation_state
self.count_accessor = conversation_state.create_property("CountProperty")
async def on_message_activity(self, turn_context: TurnContext):
count = await self.count_accessor.get(turn_context, lambda: 0)
count += 1
await self.count_accessor.set(turn_context, count)
await turn_context.send_activity(MessageFactory.text(f"Turn {count}"))
# Save state
await self.conversation_state.save_changes(turn_context)
class TestCounterBot:
@pytest.mark.asyncio
async def test_counter_increments(self):
# Create storage and state
storage = MemoryStorage()
conversation_state = ConversationState(storage)
# Create bot with state
bot = CounterBot(conversation_state)
# Create adapter
adapter = TestAdapter()
# Test counter increments
await adapter.test("anything", "Turn 1") \
.test("something", "Turn 2") \
.test("else", "Turn 3") \
.start_test()from botbuilder.core import AutoSaveStateMiddleware
class TestBotWithMiddleware:
@pytest.mark.asyncio
async def test_with_auto_save_middleware(self):
# Create storage and states
storage = MemoryStorage()
conversation_state = ConversationState(storage)
user_state = UserState(storage)
# Create adapter and add middleware
adapter = TestAdapter()
adapter.use(AutoSaveStateMiddleware([conversation_state, user_state]))
# Create bot
bot = CounterBot(conversation_state)
# Test - state should be auto-saved by middleware
await adapter.test("test", "Turn 1") \
.test("test2", "Turn 2") \
.start_test()class TestAdvancedAssertions:
@pytest.mark.asyncio
async def test_custom_assertion(self):
adapter = TestAdapter()
bot = EchoBot()
# Custom assertion function
def validate_echo_response(activity):
assert activity.type == "message"
assert "You said:" in activity.text
assert len(activity.text) > 10
await adapter.send("hello world") \
.assert_reply(validate_echo_response) \
.start_test()
@pytest.mark.asyncio
async def test_multiple_possible_responses(self):
class RandomBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
import random
responses = ["Hello!", "Hi there!", "Greetings!"]
reply = random.choice(responses)
await turn_context.send_activity(MessageFactory.text(reply))
adapter = TestAdapter()
bot = RandomBot()
await adapter.send("hi") \
.assert_reply_one_of(["Hello!", "Hi there!", "Greetings!"]) \
.start_test()
@pytest.mark.asyncio
async def test_no_reply_scenario(self):
class SilentBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
# Bot doesn't respond to certain inputs
if turn_context.activity.text == "ignore":
return # No response
await turn_context.send_activity(MessageFactory.text("I heard you"))
adapter = TestAdapter()
bot = SilentBot()
await adapter.send("ignore") \
.assert_no_reply() \
.send("hello") \
.assert_reply("I heard you") \
.start_test()class TestErrorHandling:
@pytest.mark.asyncio
async def test_bot_exception_handling(self):
class ErrorBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
if turn_context.activity.text == "error":
raise ValueError("Test error")
await turn_context.send_activity(MessageFactory.text("OK"))
adapter = TestAdapter()
bot = ErrorBot()
# Test that exception is properly handled
with pytest.raises(ValueError, match="Test error"):
await adapter.send("error").start_test()
# Test normal operation still works
await adapter.send("normal").assert_reply("OK").start_test()import time
class TestPerformance:
@pytest.mark.asyncio
async def test_response_time(self):
class SlowBot(ActivityHandler):
async def on_message_activity(self, turn_context: TurnContext):
# Simulate processing time
await asyncio.sleep(0.1)
await turn_context.send_activity(MessageFactory.text("Processed"))
adapter = TestAdapter()
bot = SlowBot()
start_time = time.time()
await adapter.test("test", "Processed").start_test()
duration = time.time() - start_time
# Assert response time is reasonable
assert duration < 1.0, f"Bot took too long to respond: {duration}s"class TestActivityInspector:
"""Helper for inspecting test activities."""
@staticmethod
def assert_message_activity(activity, text: str = None):
"""Assert activity is a message with optional text check."""
assert activity.type == "message"
if text:
assert activity.text == text
@staticmethod
def assert_suggested_actions(activity, expected_actions):
"""Assert activity has expected suggested actions."""
assert activity.suggested_actions is not None
actual_actions = [action.title for action in activity.suggested_actions.actions]
assert actual_actions == expected_actionsInstall with Tessl CLI
npx tessl i tessl/pypi-botbuilder-core