Jwiley84/assertnoreply

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

@@ -651,6 +651,46 @@ 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
+
+                elif 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."
+                    )
+
+                else:
+                    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

@@ -245,3 +245,32 @@ 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_recieved(
+        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")
+        await test_flow.assert_no_reply("should be no additional replies")