Spaces:
Build error
Build error
Validify-testbot-1
/
botbuilder-python
/libraries
/botbuilder-testing
/tests
/test_dialog_test_client.py
# Copyright (c) Microsoft Corporation. All rights reserved. | |
# Licensed under the MIT License. | |
import logging | |
from aiounittest import AsyncTestCase | |
from botbuilder.core import MessageFactory | |
from botbuilder.dialogs import ( | |
ComponentDialog, | |
DialogContext, | |
DialogTurnResult, | |
DialogTurnStatus, | |
PromptOptions, | |
TextPrompt, | |
WaterfallDialog, | |
WaterfallStepContext, | |
) | |
from botbuilder.schema import Activity | |
from botbuilder.testing import DialogTestClient, DialogTestLogger | |
class DialogTestClientTest(AsyncTestCase): | |
"""Tests for dialog test client.""" | |
def __init__(self, *args, **kwargs): | |
super(DialogTestClientTest, self).__init__(*args, **kwargs) | |
logging.basicConfig(format="", level=logging.INFO) | |
def test_init(self): | |
client = DialogTestClient(channel_or_adapter="test", target_dialog=None) | |
self.assertIsInstance(client, DialogTestClient) | |
def test_init_with_custom_channel_id(self): | |
client = DialogTestClient(channel_or_adapter="custom", target_dialog=None) | |
self.assertEqual("custom", client.test_adapter.template.channel_id) | |
async def test_single_turn_waterfall_dialog(self): | |
async def step1(step: DialogContext) -> DialogTurnResult: | |
await step.context.send_activity("hello") | |
return await step.end_dialog() | |
dialog = WaterfallDialog("waterfall", [step1]) | |
client = DialogTestClient("test", dialog) | |
reply = await client.send_activity("hello") | |
self.assertEqual("hello", reply.text) | |
self.assertEqual("test", reply.channel_id) | |
self.assertEqual(DialogTurnStatus.Complete, client.dialog_turn_result.status) | |
async def test_single_turn_waterfall_dialog_with_logger(self): | |
""" | |
Test for single turn waterfall dialog with logger with test client. | |
To view the console output: | |
* unittest | |
```bash | |
python -m unittest -v -k logger | |
``` | |
* pytest | |
```bash | |
pytest --log-cli-level=INFO --log-format="%(message)s" -k logger | |
``` | |
The results are similar to: | |
``` | |
User: Text = hello | |
-> ts: 13:39:59 | |
Bot: Text = hello | |
Speak = None | |
InputHint = acceptingInput | |
-> ts: 13:39:59 elapsed 8 ms | |
``` | |
:return: None | |
:rtype: None | |
""" | |
async def step1(step: DialogContext) -> DialogTurnResult: | |
await step.context.send_activity("hello") | |
return await step.end_dialog() | |
dialog = WaterfallDialog("waterfall", [step1]) | |
client = DialogTestClient( | |
"test", | |
dialog, | |
initial_dialog_options=None, | |
middlewares=[DialogTestLogger()], | |
) | |
reply = await client.send_activity("hello") | |
self.assertEqual("hello", reply.text) | |
self.assertEqual("test", reply.channel_id) | |
self.assertEqual(DialogTurnStatus.Complete, client.dialog_turn_result.status) | |
async def test_two_turn_waterfall_dialog(self): | |
async def step1(step: WaterfallStepContext) -> DialogTurnResult: | |
await step.context.send_activity("hello") | |
await step.context.send_activity(Activity(type="typing")) | |
return await step.next(result=None) | |
async def step2(step: WaterfallStepContext) -> DialogTurnResult: | |
await step.context.send_activity("hello 2") | |
return await step.end_dialog() | |
dialog = WaterfallDialog("waterfall", [step1, step2]) | |
client = DialogTestClient( | |
"test", | |
dialog, | |
initial_dialog_options=None, | |
middlewares=[DialogTestLogger()], | |
) | |
reply = await client.send_activity("hello") | |
self.assertEqual("hello", reply.text) | |
reply = client.get_next_reply() | |
self.assertEqual("typing", reply.type) | |
reply = client.get_next_reply() | |
self.assertEqual("hello 2", reply.text) | |
self.assertEqual(DialogTurnStatus.Complete, client.dialog_turn_result.status) | |
async def test_component_dialog(self): | |
component = MainDialog("component") | |
client = DialogTestClient( | |
"test", | |
component, | |
initial_dialog_options=None, | |
middlewares=[DialogTestLogger()], | |
) | |
reply = await client.send_activity("hello") | |
self.assertEqual("Tell me something", reply.text) | |
reply = await client.send_activity("foo") | |
self.assertEqual("you said: foo", reply.text) | |
self.assertEqual(DialogTurnStatus.Complete, client.dialog_turn_result.status) | |
class MainDialog(ComponentDialog): | |
def __init__(self, dialog_id: str): | |
super().__init__(dialog_id) | |
dialog = WaterfallDialog("waterfall", [self.step1, self.step2]) | |
self.add_dialog(TextPrompt(TextPrompt.__name__)) | |
self.add_dialog(dialog) | |
self.initial_dialog_id = dialog.id | |
async def step1(step: WaterfallStepContext) -> DialogTurnResult: | |
options = PromptOptions(prompt=MessageFactory.text("Tell me something")) | |
return await step.prompt(TextPrompt.__name__, options) | |
async def step2(step: WaterfallStepContext) -> DialogTurnResult: | |
await step.context.send_activity( | |
MessageFactory.text(f"you said: {step.result}") | |
) | |
return await step.end_dialog() | |