CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-botbuilder-core

Microsoft Bot Framework Bot Builder core functionality for building conversational AI bots and chatbots in Python.

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

testing-utilities.mddocs/

Testing Utilities

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.

Capabilities

TestAdapter

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
        """

TestFlow

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
        """

Usage Examples

Basic Bot Testing

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()

Complex Bot Testing

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()

Testing with State

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()

Testing with Middleware

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()

Custom Assertions

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()

Testing Exception Handling

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()

Performance Testing

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"

Types

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_actions

Install with Tessl CLI

npx tessl i tessl/pypi-botbuilder-core

docs

activity-handling.md

bot-adapters.md

index.md

message-factories.md

middleware.md

oauth-authentication.md

state-management.md

storage.md

telemetry-logging.md

testing-utilities.md

turn-context.md

tile.json