Spaces:
Build error
Build error
Validify-testbot-1
/
botbuilder-python
/libraries
/botbuilder-dialogs
/tests
/test_activity_prompt.py
| # Copyright (c) Microsoft Corporation. All rights reserved. | |
| # Licensed under the MIT License. | |
| import unittest | |
| import aiounittest | |
| from botbuilder.dialogs.prompts import ( | |
| ActivityPrompt, | |
| PromptOptions, | |
| PromptValidatorContext, | |
| ) | |
| from botbuilder.schema import Activity, ActivityTypes | |
| from botbuilder.core import ( | |
| ConversationState, | |
| MemoryStorage, | |
| TurnContext, | |
| MessageFactory, | |
| ) | |
| from botbuilder.core.adapters import TestAdapter | |
| from botbuilder.dialogs import DialogSet, DialogTurnStatus, DialogReason | |
| async def validator(prompt_context: PromptValidatorContext): | |
| tester = unittest.TestCase() | |
| tester.assertTrue(prompt_context.attempt_count > 0) | |
| activity = prompt_context.recognized.value | |
| if activity.type == ActivityTypes.event: | |
| if int(activity.value) == 2: | |
| prompt_context.recognized.value = MessageFactory.text(str(activity.value)) | |
| return True | |
| else: | |
| await prompt_context.context.send_activity( | |
| "Please send an 'event'-type Activity with a value of 2." | |
| ) | |
| return False | |
| class SimpleActivityPrompt(ActivityPrompt): | |
| pass | |
| class ActivityPromptTests(aiounittest.AsyncTestCase): | |
| def test_activity_prompt_with_empty_id_should_fail(self): | |
| empty_id = "" | |
| with self.assertRaises(TypeError): | |
| SimpleActivityPrompt(empty_id, validator) | |
| def test_activity_prompt_with_none_id_should_fail(self): | |
| none_id = None | |
| with self.assertRaises(TypeError): | |
| SimpleActivityPrompt(none_id, validator) | |
| def test_activity_prompt_with_none_validator_should_fail(self): | |
| none_validator = None | |
| with self.assertRaises(TypeError): | |
| SimpleActivityPrompt("EventActivityPrompt", none_validator) | |
| async def test_basic_activity_prompt(self): | |
| async def exec_test(turn_context: TurnContext): | |
| dialog_context = await dialogs.create_context(turn_context) | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| options = PromptOptions( | |
| prompt=Activity( | |
| type=ActivityTypes.message, text="please send an event." | |
| ) | |
| ) | |
| await dialog_context.prompt("EventActivityPrompt", options) | |
| elif results.status == DialogTurnStatus.Complete: | |
| await turn_context.send_activity(results.result) | |
| await convo_state.save_changes(turn_context) | |
| # Initialize TestAdapter. | |
| adapter = TestAdapter(exec_test) | |
| # Create ConversationState with MemoryStorage and register the state as middleware. | |
| convo_state = ConversationState(MemoryStorage()) | |
| # Create a DialogState property, DialogSet and AttachmentPrompt. | |
| dialog_state = convo_state.create_property("dialog_state") | |
| dialogs = DialogSet(dialog_state) | |
| dialogs.add(SimpleActivityPrompt("EventActivityPrompt", validator)) | |
| event_activity = Activity(type=ActivityTypes.event, value=2) | |
| step1 = await adapter.send("hello") | |
| step2 = await step1.assert_reply("please send an event.") | |
| step3 = await step2.send(event_activity) | |
| await step3.assert_reply("2") | |
| async def test_retry_activity_prompt(self): | |
| async def exec_test(turn_context: TurnContext): | |
| dialog_context = await dialogs.create_context(turn_context) | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| options = PromptOptions( | |
| prompt=Activity( | |
| type=ActivityTypes.message, text="please send an event." | |
| ) | |
| ) | |
| await dialog_context.prompt("EventActivityPrompt", options) | |
| elif results.status == DialogTurnStatus.Complete: | |
| await turn_context.send_activity(results.result) | |
| await convo_state.save_changes(turn_context) | |
| # Initialize TestAdapter. | |
| adapter = TestAdapter(exec_test) | |
| # Create ConversationState with MemoryStorage and register the state as middleware. | |
| convo_state = ConversationState(MemoryStorage()) | |
| # Create a DialogState property, DialogSet and AttachmentPrompt. | |
| dialog_state = convo_state.create_property("dialog_state") | |
| dialogs = DialogSet(dialog_state) | |
| dialogs.add(SimpleActivityPrompt("EventActivityPrompt", validator)) | |
| event_activity = Activity(type=ActivityTypes.event, value=2) | |
| step1 = await adapter.send("hello") | |
| step2 = await step1.assert_reply("please send an event.") | |
| step3 = await step2.send("hello again") | |
| step4 = await step3.assert_reply( | |
| "Please send an 'event'-type Activity with a value of 2." | |
| ) | |
| step5 = await step4.send(event_activity) | |
| await step5.assert_reply("2") | |
| async def test_activity_prompt_should_return_dialog_end_if_validation_failed(self): | |
| async def exec_test(turn_context: TurnContext): | |
| dialog_context = await dialogs.create_context(turn_context) | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| options = PromptOptions( | |
| prompt=Activity( | |
| type=ActivityTypes.message, text="please send an event." | |
| ), | |
| retry_prompt=Activity( | |
| type=ActivityTypes.message, text="event not received." | |
| ), | |
| ) | |
| await dialog_context.prompt("EventActivityPrompt", options) | |
| elif results.status == DialogTurnStatus.Complete: | |
| await turn_context.send_activity(results.result) | |
| await convo_state.save_changes(turn_context) | |
| async def aux_validator(prompt_context: PromptValidatorContext): | |
| assert prompt_context, "Validator missing prompt_context" | |
| return False | |
| # Initialize TestAdapter. | |
| adapter = TestAdapter(exec_test) | |
| # Create ConversationState with MemoryStorage and register the state as middleware. | |
| convo_state = ConversationState(MemoryStorage()) | |
| # Create a DialogState property, DialogSet and AttachmentPrompt. | |
| dialog_state = convo_state.create_property("dialog_state") | |
| dialogs = DialogSet(dialog_state) | |
| dialogs.add(SimpleActivityPrompt("EventActivityPrompt", aux_validator)) | |
| step1 = await adapter.send("hello") | |
| step2 = await step1.assert_reply("please send an event.") | |
| step3 = await step2.send("test") | |
| await step3.assert_reply("event not received.") | |
| async def test_activity_prompt_resume_dialog_should_return_dialog_end(self): | |
| async def exec_test(turn_context: TurnContext): | |
| dialog_context = await dialogs.create_context(turn_context) | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| options = PromptOptions( | |
| prompt=Activity( | |
| type=ActivityTypes.message, text="please send an event." | |
| ) | |
| ) | |
| await dialog_context.prompt("EventActivityPrompt", options) | |
| second_results = await event_prompt.resume_dialog( | |
| dialog_context, DialogReason.NextCalled | |
| ) | |
| assert ( | |
| second_results.status == DialogTurnStatus.Waiting | |
| ), "resume_dialog did not returned Dialog.EndOfTurn" | |
| await convo_state.save_changes(turn_context) | |
| async def aux_validator(prompt_context: PromptValidatorContext): | |
| assert prompt_context, "Validator missing prompt_context" | |
| return False | |
| # Initialize TestAdapter. | |
| adapter = TestAdapter(exec_test) | |
| # Create ConversationState with MemoryStorage and register the state as middleware. | |
| convo_state = ConversationState(MemoryStorage()) | |
| # Create a DialogState property, DialogSet and AttachmentPrompt. | |
| dialog_state = convo_state.create_property("dialog_state") | |
| dialogs = DialogSet(dialog_state) | |
| event_prompt = SimpleActivityPrompt("EventActivityPrompt", aux_validator) | |
| dialogs.add(event_prompt) | |
| step1 = await adapter.send("hello") | |
| step2 = await step1.assert_reply("please send an event.") | |
| await step2.assert_reply("please send an event.") | |
| async def test_activity_prompt_onerror_should_return_dialogcontext(self): | |
| # Create ConversationState with MemoryStorage and register the state as middleware. | |
| convo_state = ConversationState(MemoryStorage()) | |
| # Create a DialogState property, DialogSet and AttachmentPrompt. | |
| dialog_state = convo_state.create_property("dialog_state") | |
| dialogs = DialogSet(dialog_state) | |
| dialogs.add(SimpleActivityPrompt("EventActivityPrompt", validator)) | |
| async def exec_test(turn_context: TurnContext): | |
| dialog_context = await dialogs.create_context(turn_context) | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| options = PromptOptions( | |
| prompt=Activity( | |
| type=ActivityTypes.message, text="please send an event." | |
| ) | |
| ) | |
| try: | |
| await dialog_context.prompt("EventActivityPrompt", options) | |
| await dialog_context.prompt("Non existent id", options) | |
| except Exception as err: | |
| self.assertIsNotNone( | |
| err.data["DialogContext"] # pylint: disable=no-member | |
| ) | |
| self.assertEqual( | |
| err.data["DialogContext"][ # pylint: disable=no-member | |
| "active_dialog" | |
| ], | |
| "EventActivityPrompt", | |
| ) | |
| else: | |
| raise Exception("Should have thrown an error.") | |
| elif results.status == DialogTurnStatus.Complete: | |
| await turn_context.send_activity(results.result) | |
| await convo_state.save_changes(turn_context) | |
| # Initialize TestAdapter. | |
| adapter = TestAdapter(exec_test) | |
| await adapter.send("hello") | |
| async def test_activity_replace_dialog_onerror_should_return_dialogcontext(self): | |
| # Create ConversationState with MemoryStorage and register the state as middleware. | |
| convo_state = ConversationState(MemoryStorage()) | |
| # Create a DialogState property, DialogSet and AttachmentPrompt. | |
| dialog_state = convo_state.create_property("dialog_state") | |
| dialogs = DialogSet(dialog_state) | |
| dialogs.add(SimpleActivityPrompt("EventActivityPrompt", validator)) | |
| async def exec_test(turn_context: TurnContext): | |
| dialog_context = await dialogs.create_context(turn_context) | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| options = PromptOptions( | |
| prompt=Activity( | |
| type=ActivityTypes.message, text="please send an event." | |
| ) | |
| ) | |
| try: | |
| await dialog_context.prompt("EventActivityPrompt", options) | |
| await dialog_context.replace_dialog("Non existent id", options) | |
| except Exception as err: | |
| self.assertIsNotNone( | |
| err.data["DialogContext"] # pylint: disable=no-member | |
| ) | |
| else: | |
| raise Exception("Should have thrown an error.") | |
| elif results.status == DialogTurnStatus.Complete: | |
| await turn_context.send_activity(results.result) | |
| await convo_state.save_changes(turn_context) | |
| # Initialize TestAdapter. | |
| adapter = TestAdapter(exec_test) | |
| await adapter.send("hello") | |