|
4 | 4 | from uuid import uuid4
|
5 | 5 | import aiounittest
|
6 | 6 |
|
7 |
| -from botbuilder.core import ShowTypingMiddleware, TurnContext |
8 |
| -from botbuilder.core.adapters import TestAdapter |
| 7 | +from botbuilder.core import Middleware, ShowTypingMiddleware, TurnContext |
| 8 | +from botbuilder.core.adapters import TestAdapter, TestFlow |
9 | 9 | from botbuilder.schema import Activity, ActivityTypes
|
10 | 10 | from botframework.connector.auth import AuthenticationConstants, ClaimsIdentity
|
11 | 11 |
|
@@ -47,6 +47,28 @@ def assert_is_typing(activity, description): # pylint: disable=unused-argument
|
47 | 47 | step5 = await step4.assert_reply(assert_is_typing)
|
48 | 48 | await step5.assert_reply("echo:bar")
|
49 | 49 |
|
| 50 | + async def test_should_stop_sendind_typing_indicators_when_exception_raised(self): |
| 51 | + async def aux(context): |
| 52 | + await context.send_activity(f"echo:{context.activity.text}") |
| 53 | + raise Exception("test:error") |
| 54 | + |
| 55 | + class ErrorHandlingMiddleware(Middleware): |
| 56 | + async def on_turn(self, context, logic): |
| 57 | + try: |
| 58 | + await logic() |
| 59 | + except Exception as e: |
| 60 | + await context.send_activity(f"error:{e.args[0]}") |
| 61 | + |
| 62 | + adapter = TestAdapter(aux) |
| 63 | + |
| 64 | + adapter.use(ErrorHandlingMiddleware()) |
| 65 | + adapter.use(ShowTypingMiddleware(0.2, 0.4)) |
| 66 | + |
| 67 | + step1: TestFlow = await adapter.send("foo") |
| 68 | + step2 = await step1.assert_reply("echo:foo") |
| 69 | + step2 = await step1.assert_reply("error:test:error") |
| 70 | + await step2.assert_no_reply("no typing activity should be shown", timeout=450) |
| 71 | + |
50 | 72 | async def test_should_not_automatically_send_a_typing_indicator_if_no_middleware(
|
51 | 73 | self,
|
52 | 74 | ):
|
|
0 commit comments