Spaces:
Build error
Build error
| # Copyright (c) Microsoft Corporation. All rights reserved. | |
| # Licensed under the MIT License. | |
| import aiounittest | |
| from botbuilder.schema import ( | |
| Activity, | |
| ConversationReference, | |
| ConversationAccount, | |
| ChannelAccount, | |
| Entity, | |
| ResourceResponse, | |
| Attachment, | |
| ) | |
| from botbuilder.schema._connector_client_enums import ActivityTypes | |
| class TestActivity(aiounittest.AsyncTestCase): | |
| def test_constructor(self): | |
| # Arrange | |
| activity = Activity() | |
| # Assert | |
| self.assertIsNotNone(activity) | |
| self.assertIsNone(activity.type) | |
| self.assertIsNone(activity.id) | |
| self.assertIsNone(activity.timestamp) | |
| self.assertIsNone(activity.local_timestamp) | |
| self.assertIsNone(activity.local_timezone) | |
| self.assertIsNone(activity.service_url) | |
| self.assertIsNone(activity.channel_id) | |
| self.assertIsNone(activity.from_property) | |
| self.assertIsNone(activity.conversation) | |
| self.assertIsNone(activity.recipient) | |
| self.assertIsNone(activity.text_format) | |
| self.assertIsNone(activity.attachment_layout) | |
| self.assertIsNone(activity.members_added) | |
| self.assertIsNone(activity.members_removed) | |
| self.assertIsNone(activity.reactions_added) | |
| self.assertIsNone(activity.reactions_removed) | |
| self.assertIsNone(activity.topic_name) | |
| self.assertIsNone(activity.history_disclosed) | |
| self.assertIsNone(activity.locale) | |
| self.assertIsNone(activity.text) | |
| self.assertIsNone(activity.speak) | |
| self.assertIsNone(activity.input_hint) | |
| self.assertIsNone(activity.summary) | |
| self.assertIsNone(activity.suggested_actions) | |
| self.assertIsNone(activity.attachments) | |
| self.assertIsNone(activity.entities) | |
| self.assertIsNone(activity.channel_data) | |
| self.assertIsNone(activity.action) | |
| self.assertIsNone(activity.reply_to_id) | |
| self.assertIsNone(activity.label) | |
| self.assertIsNone(activity.value_type) | |
| self.assertIsNone(activity.value) | |
| self.assertIsNone(activity.name) | |
| self.assertIsNone(activity.relates_to) | |
| self.assertIsNone(activity.code) | |
| self.assertIsNone(activity.expiration) | |
| self.assertIsNone(activity.importance) | |
| self.assertIsNone(activity.delivery_mode) | |
| self.assertIsNone(activity.listen_for) | |
| self.assertIsNone(activity.text_highlights) | |
| self.assertIsNone(activity.semantic_action) | |
| self.assertIsNone(activity.caller_id) | |
| def test_apply_conversation_reference(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| conversation_reference = ConversationReference( | |
| channel_id="123", | |
| service_url="serviceUrl", | |
| conversation=ConversationAccount(id="456"), | |
| user=ChannelAccount(id="abc"), | |
| bot=ChannelAccount(id="def"), | |
| activity_id="12345", | |
| locale="en-uS", | |
| ) | |
| # Act | |
| activity.apply_conversation_reference(reference=conversation_reference) | |
| # Assert | |
| self.assertEqual(conversation_reference.channel_id, activity.channel_id) | |
| self.assertEqual(conversation_reference.locale, activity.locale) | |
| self.assertEqual(conversation_reference.service_url, activity.service_url) | |
| self.assertEqual( | |
| conversation_reference.conversation.id, activity.conversation.id | |
| ) | |
| self.assertEqual(conversation_reference.bot.id, activity.from_property.id) | |
| self.assertEqual(conversation_reference.user.id, activity.recipient.id) | |
| self.assertEqual(conversation_reference.activity_id, activity.reply_to_id) | |
| def test_apply_conversation_reference_with_is_incoming_true(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| conversation_reference = ConversationReference( | |
| channel_id="cr_123", | |
| service_url="cr_serviceUrl", | |
| conversation=ConversationAccount(id="cr_456"), | |
| user=ChannelAccount(id="cr_abc"), | |
| bot=ChannelAccount(id="cr_def"), | |
| activity_id="cr_12345", | |
| locale="en-uS", | |
| ) | |
| # Act | |
| activity.apply_conversation_reference( | |
| reference=conversation_reference, is_incoming=True | |
| ) | |
| # Assert | |
| self.assertEqual(conversation_reference.channel_id, activity.channel_id) | |
| self.assertEqual(conversation_reference.locale, activity.locale) | |
| self.assertEqual(conversation_reference.service_url, activity.service_url) | |
| self.assertEqual( | |
| conversation_reference.conversation.id, activity.conversation.id | |
| ) | |
| self.assertEqual(conversation_reference.user.id, activity.from_property.id) | |
| self.assertEqual(conversation_reference.bot.id, activity.recipient.id) | |
| self.assertEqual(conversation_reference.activity_id, activity.id) | |
| def test_as_contact_relation_update_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.contact_relation_update | |
| # Act | |
| result = activity.as_contact_relation_update_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.contact_relation_update) | |
| def test_as_contact_relation_update_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_contact_relation_update_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_conversation_update_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.conversation_update | |
| # Act | |
| result = activity.as_conversation_update_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.conversation_update) | |
| def test_as_conversation_update_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_conversation_update_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_end_of_conversation_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.end_of_conversation | |
| # Act | |
| result = activity.as_end_of_conversation_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.end_of_conversation) | |
| def test_as_end_of_conversation_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_end_of_conversation_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_event_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.event | |
| # Act | |
| result = activity.as_event_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.event) | |
| def test_as_event_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_event_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_handoff_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.handoff | |
| # Act | |
| result = activity.as_handoff_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.handoff) | |
| def test_as_handoff_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_handoff_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_installation_update_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.installation_update | |
| # Act | |
| result = activity.as_installation_update_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.installation_update) | |
| def test_as_installation_update_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_installation_update_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_invoke_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.invoke | |
| # Act | |
| result = activity.as_invoke_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.invoke) | |
| def test_as_invoke_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_invoke_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_message_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_message_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.message) | |
| def test_as_message_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.invoke | |
| # Act | |
| result = activity.as_message_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_message_activity_type_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = None | |
| # Act | |
| result = activity.as_message_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_message_delete_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message_delete | |
| # Act | |
| result = activity.as_message_delete_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.message_delete) | |
| def test_as_message_delete_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_message_delete_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_message_reaction_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message_reaction | |
| # Act | |
| result = activity.as_message_reaction_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.message_reaction) | |
| def test_as_message_reaction_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_message_reaction_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_message_update_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message_update | |
| # Act | |
| result = activity.as_message_update_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.message_update) | |
| def test_as_message_update_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_message_update_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_suggestion_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.suggestion | |
| # Act | |
| result = activity.as_suggestion_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.suggestion) | |
| def test_as_suggestion_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_suggestion_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_trace_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.trace | |
| # Act | |
| result = activity.as_trace_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.trace) | |
| def test_as_trace_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_trace_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_as_typing_activity_return_activity(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.typing | |
| # Act | |
| result = activity.as_typing_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.typing) | |
| def test_as_typing_activity_return_none(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.type = ActivityTypes.message | |
| # Act | |
| result = activity.as_typing_activity() | |
| # Assert | |
| self.assertIsNone(result) | |
| def test_create_contact_relation_update_activity(self): | |
| # Act | |
| result = Activity.create_contact_relation_update_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.contact_relation_update) | |
| def test_create_conversation_update_activity(self): | |
| # Act | |
| result = Activity.create_conversation_update_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.conversation_update) | |
| def test_create_end_of_conversation_activity(self): | |
| # Act | |
| result = Activity.create_end_of_conversation_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.end_of_conversation) | |
| def test_create_event_activity(self): | |
| # Act | |
| result = Activity.create_event_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.event) | |
| def test_create_handoff_activity(self): | |
| # Act | |
| result = Activity.create_handoff_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.handoff) | |
| def test_create_invoke_activity(self): | |
| # Act | |
| result = Activity.create_invoke_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.invoke) | |
| def test_create_message_activity(self): | |
| # Act | |
| result = Activity.create_message_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.message) | |
| def test_create_reply(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| text = "test reply" | |
| locale = "en-us" | |
| # Act | |
| result = activity.create_reply(text=text, locale=locale) | |
| # Assert | |
| self.assertEqual(result.text, text) | |
| self.assertEqual(result.locale, locale) | |
| self.assertEqual(result.type, ActivityTypes.message) | |
| def test_create_reply_without_arguments(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| # Act | |
| result = activity.create_reply() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.message) | |
| self.assertEqual(result.text, "") | |
| self.assertEqual(result.locale, activity.locale) | |
| def test_create_trace(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| name = "test-activity" | |
| value_type = "string" | |
| value = "test-value" | |
| label = "test-label" | |
| # Act | |
| result = activity.create_trace( | |
| name=name, value_type=value_type, value=value, label=label | |
| ) | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.trace) | |
| self.assertEqual(result.name, name) | |
| self.assertEqual(result.value_type, value_type) | |
| self.assertEqual(result.value, value) | |
| self.assertEqual(result.label, label) | |
| def test_create_trace_activity_no_recipient(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| activity.recipient = None | |
| # Act | |
| result = activity.create_trace("test") | |
| # Assert | |
| self.assertIsNone(result.from_property.id) | |
| self.assertIsNone(result.from_property.name) | |
| def test_crete_trace_activity_no_value_type(self): | |
| # Arrange | |
| name = "test-activity" | |
| value = "test-value" | |
| label = "test-label" | |
| # Act | |
| result = Activity.create_trace_activity(name=name, value=value, label=label) | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.trace) | |
| self.assertEqual(result.value_type, type(value)) | |
| self.assertEqual(result.label, label) | |
| def test_create_trace_activity(self): | |
| # Arrange | |
| name = "test-activity" | |
| value_type = "string" | |
| value = "test-value" | |
| label = "test-label" | |
| # Act | |
| result = Activity.create_trace_activity( | |
| name=name, value_type=value_type, value=value, label=label | |
| ) | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.trace) | |
| self.assertEqual(result.name, name) | |
| self.assertEqual(result.value_type, value_type) | |
| self.assertEqual(result.label, label) | |
| def test_create_typing_activity(self): | |
| # Act | |
| result = Activity.create_typing_activity() | |
| # Assert | |
| self.assertEqual(result.type, ActivityTypes.typing) | |
| def test_get_conversation_reference(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| # Act | |
| result = activity.get_conversation_reference() | |
| # Assert | |
| self.assertEqual(activity.id, result.activity_id) | |
| self.assertEqual(activity.from_property.id, result.user.id) | |
| self.assertEqual(activity.recipient.id, result.bot.id) | |
| self.assertEqual(activity.conversation.id, result.conversation.id) | |
| self.assertEqual(activity.channel_id, result.channel_id) | |
| self.assertEqual(activity.locale, result.locale) | |
| self.assertEqual(activity.service_url, result.service_url) | |
| def test_get_mentions(self): | |
| # Arrange | |
| mentions = [Entity(type="mention"), Entity(type="reaction")] | |
| activity = Activity(entities=mentions) | |
| # Act | |
| result = Activity.get_mentions(activity) | |
| # Assert | |
| self.assertEqual(len(result), 1) | |
| self.assertEqual(result[0].type, "mention") | |
| def test_get_reply_conversation_reference(self): | |
| # Arrange | |
| activity = self.__create_activity() | |
| reply = ResourceResponse(id="1234") | |
| # Act | |
| result = activity.get_reply_conversation_reference(reply=reply) | |
| # Assert | |
| self.assertEqual(reply.id, result.activity_id) | |
| self.assertEqual(activity.from_property.id, result.user.id) | |
| self.assertEqual(activity.recipient.id, result.bot.id) | |
| self.assertEqual(activity.conversation.id, result.conversation.id) | |
| self.assertEqual(activity.channel_id, result.channel_id) | |
| self.assertEqual(activity.locale, result.locale) | |
| self.assertEqual(activity.service_url, result.service_url) | |
| def test_has_content_empty(self): | |
| # Arrange | |
| activity_empty = Activity() | |
| # Act | |
| result_empty = activity_empty.has_content() | |
| # Assert | |
| self.assertEqual(result_empty, False) | |
| def test_has_content_with_text(self): | |
| # Arrange | |
| activity_with_text = Activity(text="test-text") | |
| # Act | |
| result_with_text = activity_with_text.has_content() | |
| # Assert | |
| self.assertEqual(result_with_text, True) | |
| def test_has_content_with_summary(self): | |
| # Arrange | |
| activity_with_summary = Activity(summary="test-summary") | |
| # Act | |
| result_with_summary = activity_with_summary.has_content() | |
| # Assert | |
| self.assertEqual(result_with_summary, True) | |
| def test_has_content_with_attachment(self): | |
| # Arrange | |
| activity_with_attachment = Activity(attachments=[Attachment()]) | |
| # Act | |
| result_with_attachment = activity_with_attachment.has_content() | |
| # Assert | |
| self.assertEqual(result_with_attachment, True) | |
| def test_has_content_with_channel_data(self): | |
| # Arrange | |
| activity_with_channel_data = Activity(channel_data="test-channel-data") | |
| # Act | |
| result_with_channel_data = activity_with_channel_data.has_content() | |
| # Assert | |
| self.assertEqual(result_with_channel_data, True) | |
| def test_is_from_streaming_connection(self): | |
| # Arrange | |
| non_streaming = [ | |
| "http://yayay.com", | |
| "https://yayay.com", | |
| "HTTP://yayay.com", | |
| "HTTPS://yayay.com", | |
| ] | |
| streaming = [ | |
| "urn:botframework:WebSocket:wss://beep.com", | |
| "urn:botframework:WebSocket:http://beep.com", | |
| "URN:botframework:WebSocket:wss://beep.com", | |
| "URN:botframework:WebSocket:http://beep.com", | |
| ] | |
| activity = self.__create_activity() | |
| activity.service_url = None | |
| # Assert | |
| self.assertEqual(activity.is_from_streaming_connection(), False) | |
| for s in non_streaming: | |
| activity.service_url = s | |
| self.assertEqual(activity.is_from_streaming_connection(), False) | |
| for s in streaming: | |
| activity.service_url = s | |
| self.assertEqual(activity.is_from_streaming_connection(), True) | |
| def __create_activity() -> Activity: | |
| account1 = ChannelAccount( | |
| id="ChannelAccount_Id_1", | |
| name="ChannelAccount_Name_1", | |
| aad_object_id="ChannelAccount_aadObjectId_1", | |
| role="ChannelAccount_Role_1", | |
| ) | |
| account2 = ChannelAccount( | |
| id="ChannelAccount_Id_2", | |
| name="ChannelAccount_Name_2", | |
| aad_object_id="ChannelAccount_aadObjectId_2", | |
| role="ChannelAccount_Role_2", | |
| ) | |
| conversation_account = ConversationAccount( | |
| conversation_type="a", | |
| id="123", | |
| is_group=True, | |
| name="Name", | |
| role="ConversationAccount_Role", | |
| ) | |
| activity = Activity( | |
| id="123", | |
| from_property=account1, | |
| recipient=account2, | |
| conversation=conversation_account, | |
| channel_id="ChannelId123", | |
| locale="en-uS", | |
| service_url="ServiceUrl123", | |
| ) | |
| return activity | |