test cases

libraries/botbuilder-core/tests/teams/test_teams_info.py

@@ -1,7 +1,19 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
+import json
 import aiounittest
+from botbuilder.schema.teams._models_py3 import (
+    ContentType,
+    MeetingNotificationChannelData,
+    MeetingStageSurface,
+    MeetingTabIconSurface,
+    OnBehalfOf,
+    TargetedMeetingNotification,
+    TargetedMeetingNotificationValue,
+    TaskModuleContinueResponse,
+    TaskModuleTaskInfo,
+)
 from botframework.connector import Channels
 
 from botbuilder.core import TurnContext, MessageFactory
@@ -234,13 +246,62 @@ async def test_get_meeting_info(self):
         handler = TeamsActivityHandler()
         await handler.on_turn(turn_context)
 
+    async def test_send_meeting_notificationt(self):
+        test_cases = [
+            ("202", "accepted"),
+            (
+                "207",
+                "if the notifications are sent only to parital number of recipients\
+                      because the validation on some recipients' ids failed or some\
+                          recipients were not found in the roster. In this case, \
+                            SMBA will return the user MRIs of those failed recipients\
+                                  in a format that was given to a bot (ex: if a bot sent \
+                                    encrypted user MRIs, return encrypted one).",
+            ),
+            (
+                "400",
+                "when Meeting Notification request payload validation fails. For instance,\
+                      Recipients: # of recipients is greater than what the API allows ||\
+                        all of recipients' user ids were invalid, Surface: Surface list\
+                              is empty or null, Surface type is invalid, Duplicative \
+                                surface type exists in one payload",
+            ),
+            (
+                "403",
+                "if the bot is not allowed to send the notification. In this case,\
+                      the payload should contain more detail error message. \
+                        There can be many reasons: bot disabled by tenant admin,\
+                              blocked during live site mitigation, the bot does not\
+                                  have a correct RSC permission for a specific surface type, etc",
+            ),
+        ]
+        for status_code, expected_message in test_cases:
+            adapter = SimpleAdapterWithCreateConversation()
+
+            activity = Activity(
+                type="targetedMeetingNotification",
+                text="Test-send_meeting_notificationt",
+                channel_id=Channels.ms_teams,
+                from_property=ChannelAccount(
+                    aad_object_id="participantId-1", name=status_code
+                ),
+                service_url="https://test.coffee",
+                conversation=ConversationAccount(id="conversation-id"),
+            )
+
+            turn_context = TurnContext(adapter, activity)
+            handler = TeamsActivityHandler()
+            await handler.on_turn(turn_context)
+
 
 class TestTeamsActivityHandler(TeamsActivityHandler):
     async def on_turn(self, turn_context: TurnContext):
         await super().on_turn(turn_context)
 
         if turn_context.activity.text == "test_send_message_to_teams_channel":
             await self.call_send_message_to_teams(turn_context)
+        elif turn_context.activity.text == "test_send_meeting_notification":
+            await self.call_send_meeting_notification(turn_context)
 
     async def call_send_message_to_teams(self, turn_context: TurnContext):
         msg = MessageFactory.text("call_send_message_to_teams")
@@ -251,3 +312,71 @@ async def call_send_message_to_teams(self, turn_context: TurnContext):
 
         assert reference[0].activity_id == "new_conversation_id"
         assert reference[1] == "reference123"
+
+    async def call_send_meeting_notification(self, turn_context: TurnContext):
+        from_property = turn_context.activity.from_property
+        try:
+            # Send the meeting notification asynchronously
+            failed_participants = await TeamsInfo.send_meeting_notification(
+                turn_context,
+                self.get_targeted_meeting_notification(from_property),
+                "meeting-id",
+            )
+
+            # Handle based on the 'from_property.name'
+            if from_property.name == "207":
+                self.assertEqual(
+                    "failingid",
+                    failed_participants.recipients_failure_info[0].recipient_mri,
+                )
+            elif from_property.name == "202":
+                assert failed_participants is None
+            else:
+                raise TypeError(
+                    f"Expected HttpOperationException with response status code {from_property.name}."
+                )
+
+        except ValueError as ex:
+            # Assert that the response status code matches the from_property.name
+            assert from_property.name == str(int(ex.response.status_code))
+
+            # Deserialize the error response content to an ErrorResponse object
+            error_response = json.loads(ex.response.content)
+
+            # Handle based on error codes
+            if from_property.name == "400":
+                assert error_response["error"]["code"] == "BadSyntax"
+            elif from_property.name == "403":
+                assert error_response["error"]["code"] == "BotNotInConversationRoster"
+            else:
+                raise TypeError(
+                    f"Expected HttpOperationException with response status code {from_property.name}."
+                )
+
+    def get_targeted_meeting_notification(self, from_account: ChannelAccount):
+        recipients = [from_account.id]
+
+        if from_account.name == "207":
+            recipients.append("failingid")
+
+        meeting_stage_surface = MeetingStageSurface(
+            content=TaskModuleContinueResponse(
+                value=TaskModuleTaskInfo(title="title here", height=3, width=2)
+            ),
+            content_type=ContentType.Task,
+        )
+
+        meeting_tab_icon_surface = MeetingTabIconSurface(
+            tab_entity_id="test tab entity id"
+        )
+
+        value = TargetedMeetingNotificationValue(
+            recipients=recipients,
+            surfaces=[meeting_stage_surface, meeting_tab_icon_surface],
+        )
+
+        obo = OnBehalfOf(display_name=from_account.name, mri=from_account.id)
+
+        channel_data = MeetingNotificationChannelData(on_behalf_of_list=[obo])
+
+        return TargetedMeetingNotification(value=value, channel_data=channel_data)

libraries/botbuilder-schema/botbuilder/schema/teams/_models_py3.py

@@ -2799,7 +2799,9 @@ class MeetingNotification(MeetingNotificationBase):
         "value": {"key": "value", "type": "TargetedMeetingNotificationValue"},
     }
 
-    def __init__(self, *, value: "TargetedMeetingNotificationValue" =None, **kwargs) -> None:
+    def __init__(
+        self, *, value: "TargetedMeetingNotificationValue" = None, **kwargs
+    ) -> None:
         super(MeetingNotification, self).__init__(**kwargs)
         self.value = value
 
@@ -2953,20 +2955,14 @@ class TargetedMeetingNotificationValue(Model):
     }
 
     def __init__(
-        self,
-        *,
-        recipients: List[str] = None,
-        surfaces: List[Surface] = None,
-        **kwargs
+        self, *, recipients: List[str] = None, surfaces: List[Surface] = None, **kwargs
     ):
         super(TargetedMeetingNotificationValue, self).__init__(**kwargs)
         self.recipients = recipients
         self.surfaces = surfaces
 
 
-class TargetedMeetingNotification(
-    MeetingNotification
-):
+class TargetedMeetingNotification(MeetingNotification):
     """Specifies Teams targeted meeting notification.
 
     :param value: The value of the TargetedMeetingNotification.