# Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. # pylint: disable=pointless-string-statement from enum import Enum from typing import Callable, List, Tuple import aiounittest from botbuilder.core import ( AutoSaveStateMiddleware, BotAdapter, ConversationState, MemoryStorage, MessageFactory, UserState, TurnContext, ) from botbuilder.core.adapters import TestAdapter from botbuilder.core.skills import SkillHandler, SkillConversationReference from botbuilder.dialogs import ( ComponentDialog, Dialog, DialogContext, DialogEvents, DialogInstance, DialogReason, TextPrompt, WaterfallDialog, DialogManager, DialogManagerResult, DialogTurnStatus, WaterfallStepContext, ) from botbuilder.dialogs.prompts import PromptOptions from botbuilder.schema import ( Activity, ActivityTypes, ChannelAccount, ConversationAccount, EndOfConversationCodes, InputHints, ) from botframework.connector.auth import AuthenticationConstants, ClaimsIdentity class SkillFlowTestCase(str, Enum): # DialogManager is executing on a root bot with no skills (typical standalone bot). root_bot_only = "RootBotOnly" # DialogManager is executing on a root bot handling replies from a skill. root_bot_consuming_skill = "RootBotConsumingSkill" # DialogManager is executing in a skill that is called from a root and calling another skill. middle_skill = "MiddleSkill" # DialogManager is executing in a skill that is called from a parent (a root or another skill) but doesn"t call # another skill. leaf_skill = "LeafSkill" class SimpleComponentDialog(ComponentDialog): # An App ID for a parent bot. parent_bot_id = "00000000-0000-0000-0000-0000000000PARENT" # An App ID for a skill bot. skill_bot_id = "00000000-0000-0000-0000-00000000000SKILL" # Captures an EndOfConversation if it was sent to help with assertions. eoc_sent: Activity = None # Property to capture the DialogManager turn results and do assertions. dm_turn_result: DialogManagerResult = None def __init__( self, id: str = None, prop: str = None ): # pylint: disable=unused-argument super().__init__(id or "SimpleComponentDialog") self.text_prompt = "TextPrompt" self.waterfall_dialog = "WaterfallDialog" self.add_dialog(TextPrompt(self.text_prompt)) self.add_dialog( WaterfallDialog( self.waterfall_dialog, [ self.prompt_for_name, self.final_step, ], ) ) self.initial_dialog_id = self.waterfall_dialog self.end_reason = None @staticmethod async def create_test_flow( dialog: Dialog, test_case: SkillFlowTestCase = SkillFlowTestCase.root_bot_only, enabled_trace=False, ) -> TestAdapter: conversation_id = "testFlowConversationId" storage = MemoryStorage() conversation_state = ConversationState(storage) user_state = UserState(storage) activity = Activity( channel_id="test", service_url="https://test.com", from_property=ChannelAccount(id="user1", name="User1"), recipient=ChannelAccount(id="bot", name="Bot"), conversation=ConversationAccount( is_group=False, conversation_type=conversation_id, id=conversation_id ), ) dialog_manager = DialogManager(dialog) dialog_manager.user_state = user_state dialog_manager.conversation_state = conversation_state async def logic(context: TurnContext): if test_case != SkillFlowTestCase.root_bot_only: # Create a skill ClaimsIdentity and put it in turn_state so isSkillClaim() returns True. claims_identity = ClaimsIdentity({}, False) claims_identity.claims["ver"] = ( "2.0" # AuthenticationConstants.VersionClaim ) claims_identity.claims["aud"] = ( SimpleComponentDialog.skill_bot_id ) # AuthenticationConstants.AudienceClaim claims_identity.claims["azp"] = ( SimpleComponentDialog.parent_bot_id ) # AuthenticationConstants.AuthorizedParty context.turn_state[BotAdapter.BOT_IDENTITY_KEY] = claims_identity if test_case == SkillFlowTestCase.root_bot_consuming_skill: # Simulate the SkillConversationReference with a channel OAuthScope stored in turn_state. # This emulates a response coming to a root bot through SkillHandler. context.turn_state[ SkillHandler.SKILL_CONVERSATION_REFERENCE_KEY ] = SkillConversationReference( None, AuthenticationConstants.TO_CHANNEL_FROM_BOT_OAUTH_SCOPE ) if test_case == SkillFlowTestCase.middle_skill: # Simulate the SkillConversationReference with a parent Bot ID stored in turn_state. # This emulates a response coming to a skill from another skill through SkillHandler. context.turn_state[ SkillHandler.SKILL_CONVERSATION_REFERENCE_KEY ] = SkillConversationReference( None, SimpleComponentDialog.parent_bot_id ) async def aux( turn_context: TurnContext, # pylint: disable=unused-argument activities: List[Activity], next: Callable, ): for activity in activities: if activity.type == ActivityTypes.end_of_conversation: SimpleComponentDialog.eoc_sent = activity break return await next() # Interceptor to capture the EoC activity if it was sent so we can assert it in the tests. context.on_send_activities(aux) SimpleComponentDialog.dm_turn_result = await dialog_manager.on_turn(context) adapter = TestAdapter(logic, activity, enabled_trace) adapter.use(AutoSaveStateMiddleware([user_state, conversation_state])) return adapter async def on_end_dialog( self, context: DialogContext, instance: DialogInstance, reason: DialogReason ): self.end_reason = reason return await super().on_end_dialog(context, instance, reason) async def prompt_for_name(self, step: WaterfallStepContext): return await step.prompt( self.text_prompt, PromptOptions( prompt=MessageFactory.text( "Hello, what is your name?", None, InputHints.expecting_input ), retry_prompt=MessageFactory.text( "Hello, what is your name again?", None, InputHints.expecting_input ), ), ) async def final_step(self, step: WaterfallStepContext): await step.context.send_activity(f"Hello { step.result }, nice to meet you!") return await step.end_dialog(step.result) class DialogManagerTests(aiounittest.AsyncTestCase): """ self.beforeEach(() => { _dmTurnResult = undefined }) """ async def test_handles_bot_and_skills(self): construction_data: List[Tuple[SkillFlowTestCase, bool]] = [ (SkillFlowTestCase.root_bot_only, False), (SkillFlowTestCase.root_bot_consuming_skill, False), (SkillFlowTestCase.middle_skill, True), (SkillFlowTestCase.leaf_skill, True), ] for test_case, should_send_eoc in construction_data: with self.subTest(test_case=test_case, should_send_eoc=should_send_eoc): SimpleComponentDialog.dm_turn_result = None SimpleComponentDialog.eoc_sent = None dialog = SimpleComponentDialog() test_flow = await SimpleComponentDialog.create_test_flow( dialog, test_case ) step1 = await test_flow.send("Hi") step2 = await step1.assert_reply("Hello, what is your name?") step3 = await step2.send("SomeName") await step3.assert_reply("Hello SomeName, nice to meet you!") self.assertEqual( SimpleComponentDialog.dm_turn_result.turn_result.status, DialogTurnStatus.Complete, ) self.assertEqual(dialog.end_reason, DialogReason.EndCalled) if should_send_eoc: self.assertTrue( bool(SimpleComponentDialog.eoc_sent), "Skills should send EndConversation to channel", ) self.assertEqual( SimpleComponentDialog.eoc_sent.type, ActivityTypes.end_of_conversation, ) self.assertEqual( SimpleComponentDialog.eoc_sent.code, EndOfConversationCodes.completed_successfully, ) self.assertEqual(SimpleComponentDialog.eoc_sent.value, "SomeName") else: self.assertIsNone( SimpleComponentDialog.eoc_sent, "Root bot should not send EndConversation to channel", ) async def test_skill_handles_eoc_from_parent(self): SimpleComponentDialog.dm_turn_result = None dialog = SimpleComponentDialog() test_flow = await SimpleComponentDialog.create_test_flow( dialog, SkillFlowTestCase.leaf_skill ) step1 = await test_flow.send("Hi") step2 = await step1.assert_reply("Hello, what is your name?") await step2.send(Activity(type=ActivityTypes.end_of_conversation)) self.assertEqual( SimpleComponentDialog.dm_turn_result.turn_result.status, DialogTurnStatus.Cancelled, ) async def test_skill_handles_reprompt_from_parent(self): SimpleComponentDialog.dm_turn_result = None dialog = SimpleComponentDialog() test_flow = await SimpleComponentDialog.create_test_flow( dialog, SkillFlowTestCase.leaf_skill ) step1 = await test_flow.send("Hi") step2 = await step1.assert_reply("Hello, what is your name?") step3 = await step2.send( Activity(type=ActivityTypes.event, name=DialogEvents.reprompt_dialog) ) await step3.assert_reply("Hello, what is your name?") self.assertEqual( SimpleComponentDialog.dm_turn_result.turn_result.status, DialogTurnStatus.Waiting, ) async def test_skill_should_return_empty_on_reprompt_with_no_dialog(self): SimpleComponentDialog.dm_turn_result = None dialog = SimpleComponentDialog() test_flow = await SimpleComponentDialog.create_test_flow( dialog, SkillFlowTestCase.leaf_skill ) await test_flow.send( Activity(type=ActivityTypes.event, name=DialogEvents.reprompt_dialog) ) self.assertEqual( SimpleComponentDialog.dm_turn_result.turn_result.status, DialogTurnStatus.Empty, ) async def test_trace_bot_state(self): SimpleComponentDialog.dm_turn_result = None dialog = SimpleComponentDialog() def assert_is_trace(activity, description): # pylint: disable=unused-argument assert activity.type == ActivityTypes.trace def assert_is_trace_and_label(activity, description): assert_is_trace(activity, description) assert activity.label == "Bot State" test_flow = await SimpleComponentDialog.create_test_flow( dialog, SkillFlowTestCase.root_bot_only, True ) step1 = await test_flow.send("Hi") step2 = await step1.assert_reply("Hello, what is your name?") step3 = await step2.assert_reply(assert_is_trace_and_label) step4 = await step3.send("SomeName") step5 = await step4.assert_reply("Hello SomeName, nice to meet you!") await step5.assert_reply(assert_is_trace_and_label) self.assertEqual( SimpleComponentDialog.dm_turn_result.turn_result.status, DialogTurnStatus.Complete, )