Jwiley84/assertnoreply

libraries/botbuilder-core/botbuilder/core/adapters/test_adapter.py

@@ -13,6 +13,12 @@
 from typing import Awaitable, Coroutine, Dict, List, Callable, Union
 from copy import copy
 from threading import Lock
+from botframework.connector.auth import AppCredentials, ClaimsIdentity
+from botframework.connector.token_api.models import (
+    SignInUrlResponse,
+    TokenExchangeResource,
+    TokenExchangeRequest,
+)
 from botbuilder.schema import (
     ActivityTypes,
     Activity,
@@ -22,12 +28,6 @@
     ResourceResponse,
     TokenResponse,
 )
-from botframework.connector.auth import AppCredentials, ClaimsIdentity
-from botframework.connector.token_api.models import (
-    SignInUrlResponse,
-    TokenExchangeResource,
-    TokenExchangeRequest,
-)
 from ..bot_adapter import BotAdapter
 from ..turn_context import TurnContext
 from ..oauth.extended_user_token_provider import ExtendedUserTokenProvider
@@ -595,6 +595,7 @@ async def assert_reply(
         :param is_substring:
         :return:
         """
+
         # TODO: refactor method so expected can take a Callable[[Activity], None]
         def default_inspector(reply, description=None):
             if isinstance(expected, Activity):
@@ -651,6 +652,45 @@ async def wait_for_activity():
 
         return TestFlow(await test_flow_previous(), self.adapter)
 
+    async def assert_no_reply(
+        self, description=None, timeout=None,  # pylint: disable=unused-argument
+    ) -> "TestFlow":
+        """
+        Generates an assertion if the bot responds when no response is expected.
+        :param description:
+        :param timeout:
+        """
+        if description is None:
+            description = ""
+
+        async def test_flow_previous():
+            nonlocal timeout
+            if not timeout:
+                timeout = 3000
+            start = datetime.now()
+            adapter = self.adapter
+
+            async def wait_for_activity():
+                nonlocal timeout
+                current = datetime.now()
+
+                if (current - start).total_seconds() * 1000 > timeout:
+                    # operation timed out and recieved no reply
+                    return
+
+                if adapter.activity_buffer:
+                    reply = adapter.activity_buffer.pop(0)
+                    raise RuntimeError(
+                        f"TestAdapter.assert_no_reply(): '{reply.text}' is responded when waiting for no reply."
+                    )
+
+                await asyncio.sleep(0.05)
+                await wait_for_activity()
+
+            await wait_for_activity()
+
+        return TestFlow(await test_flow_previous(), self.adapter)
+
 
 def validate_activity(activity, expected) -> None:
     """

libraries/botbuilder-core/tests/test_test_adapter.py

@@ -3,10 +3,10 @@
 
 import aiounittest
 
+from botframework.connector.auth import MicrosoftAppCredentials
 from botbuilder.core import TurnContext
 from botbuilder.core.adapters import TestAdapter
 from botbuilder.schema import Activity, ConversationReference, ChannelAccount
-from botframework.connector.auth import MicrosoftAppCredentials
 
 RECEIVED_MESSAGE = Activity(type="message", text="received")
 UPDATED_ACTIVITY = Activity(type="message", text="update")
@@ -245,3 +245,34 @@ async def test_get_user_token_returns_token_with_magice_code(self):
         assert token_response
         assert token == token_response.token
         assert connection_name == token_response.connection_name
+
+    async def test_should_validate_no_reply_when_no_reply_expected(self):
+        async def logic(context: TurnContext):
+            await context.send_activity(RECEIVED_MESSAGE)
+
+        adapter = TestAdapter(logic)
+        test_flow = await adapter.test("test", "received")
+        await test_flow.assert_no_reply("should be no additional replies")
+
+    async def test_should_timeout_waiting_for_assert_no_reply_when_no_reply_expected(
+        self,
+    ):
+        async def logic(context: TurnContext):
+            await context.send_activity(RECEIVED_MESSAGE)
+
+        adapter = TestAdapter(logic)
+        test_flow = await adapter.test("test", "received")
+        await test_flow.assert_no_reply("no reply received", 500)
+
+    async def test_should_throw_error_with_assert_no_reply_when_no_reply_expected_but_was_received(
+        self,
+    ):
+        async def logic(context: TurnContext):
+            activities = [RECEIVED_MESSAGE, RECEIVED_MESSAGE]
+            await context.send_activities(activities)
+
+        adapter = TestAdapter(logic)
+        test_flow = await adapter.test("test", "received")
+
+        with self.assertRaises(Exception):
+            await test_flow.assert_no_reply("should be no additional replies")