Spaces:
Build error
Build error
| import json | |
| from os import path | |
| from unittest.mock import patch | |
| import aiounittest | |
| # from botbuilder.ai.qna import QnAMakerEndpoint, QnAMaker, QnAMakerOptions | |
| from botbuilder.ai.qna.dialogs import QnAMakerDialog | |
| from botbuilder.schema import Activity, ActivityTypes | |
| from botbuilder.core import ConversationState, MemoryStorage, TurnContext | |
| from botbuilder.core.adapters import TestAdapter, TestFlow | |
| from botbuilder.dialogs import DialogSet, DialogTurnStatus | |
| class QnaMakerDialogTest(aiounittest.AsyncTestCase): | |
| # Note this is NOT a real QnA Maker application ID nor a real QnA Maker subscription-key | |
| # theses are GUIDs edited to look right to the parsing and validation code. | |
| _knowledge_base_id: str = "f028d9k3-7g9z-11d3-d300-2b8x98227q8w" | |
| _endpoint_key: str = "1k997n7w-207z-36p3-j2u1-09tas20ci6011" | |
| _host: str = "https://dummyqnahost.azurewebsites.net/qnamaker" | |
| _tell_me_about_birds: str = "Tell me about birds" | |
| _choose_bird: str = "Choose one of the following birds to get more info" | |
| _bald_eagle: str = "Bald Eagle" | |
| _esper: str = "Esper" | |
| DEFAULT_ACTIVE_LEARNING_TITLE: str = "Did you mean:" | |
| DEFAULT_NO_MATCH_TEXT: str = "None of the above." | |
| DEFAULT_CARD_NO_MATCH_RESPONSE: str = "Thanks for the feedback." | |
| async def test_multiturn_dialog(self): | |
| # Set Up QnAMakerDialog | |
| convo_state = ConversationState(MemoryStorage()) | |
| dialog_state = convo_state.create_property("dialogState") | |
| dialogs = DialogSet(dialog_state) | |
| qna_dialog = QnAMakerDialog( | |
| self._knowledge_base_id, self._endpoint_key, self._host | |
| ) | |
| dialogs.add(qna_dialog) | |
| # Callback that runs the dialog | |
| async def execute_qna_dialog(turn_context: TurnContext) -> None: | |
| if turn_context.activity.type != ActivityTypes.message: | |
| raise TypeError( | |
| "Failed to execute QnA dialog. Should have received a message activity." | |
| ) | |
| response_json = self._get_json_res(turn_context.activity.text) | |
| dialog_context = await dialogs.create_context(turn_context) | |
| with patch( | |
| "aiohttp.ClientSession.post", | |
| return_value=aiounittest.futurized(response_json), | |
| ): | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| await dialog_context.begin_dialog("QnAMakerDialog") | |
| await convo_state.save_changes(turn_context) | |
| # Send and receive messages from QnA dialog | |
| test_adapter = TestAdapter(execute_qna_dialog) | |
| test_flow = TestFlow(None, test_adapter) | |
| tf2 = await test_flow.send(self._tell_me_about_birds) | |
| dialog_reply: Activity = tf2.adapter.activity_buffer[0] | |
| self._assert_has_valid_hero_card_buttons(dialog_reply, button_count=2) | |
| tf3 = await tf2.assert_reply(self._choose_bird) | |
| tf4 = await tf3.send(self._bald_eagle) | |
| await tf4.assert_reply("Apparently these guys aren't actually bald!") | |
| async def test_active_learning(self): | |
| # Set Up QnAMakerDialog | |
| convo_state = ConversationState(MemoryStorage()) | |
| dialog_state = convo_state.create_property("dialogState") | |
| dialogs = DialogSet(dialog_state) | |
| qna_dialog = QnAMakerDialog( | |
| self._knowledge_base_id, self._endpoint_key, self._host | |
| ) | |
| dialogs.add(qna_dialog) | |
| # Callback that runs the dialog | |
| async def execute_qna_dialog(turn_context: TurnContext) -> None: | |
| if turn_context.activity.type != ActivityTypes.message: | |
| raise TypeError( | |
| "Failed to execute QnA dialog. Should have received a message activity." | |
| ) | |
| response_json = self._get_json_res(turn_context.activity.text) | |
| dialog_context = await dialogs.create_context(turn_context) | |
| with patch( | |
| "aiohttp.ClientSession.post", | |
| return_value=aiounittest.futurized(response_json), | |
| ): | |
| results = await dialog_context.continue_dialog() | |
| if results.status == DialogTurnStatus.Empty: | |
| await dialog_context.begin_dialog("QnAMakerDialog") | |
| await convo_state.save_changes(turn_context) | |
| # Send and receive messages from QnA dialog | |
| test_adapter = TestAdapter(execute_qna_dialog) | |
| test_flow = TestFlow(None, test_adapter) | |
| tf2 = await test_flow.send(self._esper) | |
| dialog_reply: Activity = tf2.adapter.activity_buffer[0] | |
| self._assert_has_valid_hero_card_buttons(dialog_reply, button_count=3) | |
| tf3 = await tf2.assert_reply(self.DEFAULT_ACTIVE_LEARNING_TITLE) | |
| tf4 = await tf3.send(self.DEFAULT_NO_MATCH_TEXT) | |
| await tf4.assert_reply(self.DEFAULT_CARD_NO_MATCH_RESPONSE) | |
| print(tf2) | |
| def _assert_has_valid_hero_card_buttons( | |
| self, activity: Activity, button_count: int | |
| ): | |
| self.assertIsInstance(activity, Activity) | |
| attachments = activity.attachments | |
| self.assertTrue(attachments) | |
| self.assertEqual(len(attachments), 1) | |
| buttons = attachments[0].content.buttons | |
| button_count_err = ( | |
| f"Should have only received {button_count} buttons in multi-turn prompt" | |
| ) | |
| if activity.text == self._choose_bird: | |
| self.assertEqual(len(buttons), button_count, button_count_err) | |
| self.assertEqual(buttons[0].value, self._bald_eagle) | |
| self.assertEqual(buttons[1].value, "Hummingbird") | |
| if activity.text == self.DEFAULT_ACTIVE_LEARNING_TITLE: | |
| self.assertEqual(len(buttons), button_count, button_count_err) | |
| self.assertEqual(buttons[0].value, "Esper seeks") | |
| self.assertEqual(buttons[1].value, "Esper sups") | |
| self.assertEqual(buttons[2].value, self.DEFAULT_NO_MATCH_TEXT) | |
| def _get_json_res(self, text: str) -> object: | |
| if text == self._tell_me_about_birds: | |
| return QnaMakerDialogTest._get_json_for_file( | |
| "QnAMakerDialog_MultiTurn_Answer1.json" | |
| ) | |
| if text == self._bald_eagle: | |
| return QnaMakerDialogTest._get_json_for_file( | |
| "QnAMakerDialog_MultiTurn_Answer2.json" | |
| ) | |
| if text == self._esper: | |
| return QnaMakerDialogTest._get_json_for_file( | |
| "QnAMakerDialog_ActiveLearning.json" | |
| ) | |
| return None | |
| def _get_json_for_file(response_file: str) -> object: | |
| curr_dir = path.dirname(path.abspath(__file__)) | |
| response_path = path.join(curr_dir, "test_data", response_file) | |
| with open(response_path, "r", encoding="utf-8-sig") as file: | |
| response_str = file.read() | |
| response_json = json.loads(response_str) | |
| return response_json | |