test_subs: Add test for all notification bot messages.

In #31206, we changed `add_subscriptions_backend` to exclude the
notification bot DMs when subscribing users to a channel when the
number of subscriptions exceeds MAX_BULK_NEW_SUBSCRIPTION_MESSAGES.

This caused a bug where new channel announcement and creation
notification messages were not being sent. Here we add a test that
confirmes the current behavior, which will be corrected in a
subsequent commit.

Co-authored-by: Lauryn Menard <lauryn@zulip.com>
(cherry picked from commit 06d2ecb482)

zerver/tests/test_subs.py

@@ -4985,8 +4985,12 @@ class SubscriptionAPITest(ZulipTestCase):
 
     def test_notification_bot_dm_on_subscription(self) -> None:
         desdemona = self.example_user("desdemona")
+        realm = desdemona.realm
         self.login_user(desdemona)
-
+        bot = self.notification_bot(realm)
+        test_channel = self.make_stream("test A")
+        announce = realm.new_stream_announcements_stream
+        assert announce is not None
         user_ids = [
             desdemona.id,
             self.example_user("cordelia").id,
@@ -4995,20 +4999,168 @@ class SubscriptionAPITest(ZulipTestCase):
             self.example_user("iago").id,
             self.example_user("prospero").id,
         ]
-
+        principals_dict = dict(
+            principals=orjson.dumps(user_ids).decode(), announce=orjson.dumps(True).decode()
+        )
+        # When subscribing to an already existing channel with announce=True,
+        # the bot should send DMs to all the newly subscribed users, but
+        # no announcement message.
+        now = timezone_now()
         response = self.subscribe_via_post(
-            desdemona, ["Test stream 1"], dict(principals=orjson.dumps(user_ids).decode())
+            desdemona,
+            [test_channel.name],
+            principals_dict,
         )
         data = self.assert_json_success(response)
         self.assertEqual(data["new_subscription_messages_sent"], True)
 
+        notification_bot_dms = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.PERSONAL,
+            date_sent__gt=now,
+        )
+        self.assert_length(notification_bot_dms, 5)
+        notif_bot_dm_recipients = [
+            dm["recipient__type_id"] for dm in notification_bot_dms.values("recipient__type_id")
+        ]
+        self.assertSetEqual(
+            {id for id in user_ids if id != desdemona.id}, set(notif_bot_dm_recipients)
+        )
+
+        announcement_channel_message = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.STREAM,
+            recipient__type_id=announce.id,
+            date_sent__gt=now,
+        )
+        self.assertEqual(announcement_channel_message.count(), 0)
+
+        # When subscribing to a newly created channel with announce=True,
+        # we expect an announcement message and new channel message,
+        # but no DM notifications.
+        now = timezone_now()
+        response = self.subscribe_via_post(
+            desdemona,
+            ["test B"],
+            principals_dict,
+        )
+        data = self.assert_json_success(response)
+        self.assertEqual(data["new_subscription_messages_sent"], True)
+
+        notification_bot_dms = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.PERSONAL,
+            date_sent__gt=now,
+        )
+        self.assertEqual(notification_bot_dms.count(), 0)
+
+        announcement_channel_message = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.STREAM,
+            recipient__type_id=announce.id,
+            date_sent__gt=now,
+        )
+        self.assert_length(announcement_channel_message.values(), 1)
+        self.assertEqual(
+            announcement_channel_message[0].content,
+            f"@_**{desdemona.full_name}|{desdemona.id}** created a new channel #**test B**.",
+        )
+
+        new_channel = get_stream("test B", realm)
+        new_channel_message = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.STREAM,
+            recipient__type_id=new_channel.id,
+            date_sent__gt=now,
+        )
+        self.assert_length(new_channel_message.values(), 1)
+        self.assertEqual(
+            new_channel_message[0].topic_name(), Realm.STREAM_EVENTS_NOTIFICATION_TOPIC_NAME
+        )
+        self.assertEqual(
+            new_channel_message[0].content,
+            f"**Public** channel created by @_**{desdemona.full_name}|{desdemona.id}**. **Description:**\n"
+            "```` quote\n*No description.*\n````",
+        )
+
+        # When subscribing to an already existing channel, if the number of new
+        # subscriptions exceeds the limit, no DMs or announcement message should
+        # be sent.
+        now = timezone_now()
+        test_channel = self.make_stream("test C")
         with self.settings(MAX_BULK_NEW_SUBSCRIPTION_MESSAGES=5):
             response = self.subscribe_via_post(
-                desdemona, ["Test stream 2"], dict(principals=orjson.dumps(user_ids).decode())
+                desdemona,
+                [test_channel.name],
+                principals_dict,
             )
         data = self.assert_json_success(response)
         self.assertEqual(data["new_subscription_messages_sent"], False)
 
+        notification_bot_dms = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.PERSONAL,
+            date_sent__gt=now,
+        )
+        self.assertEqual(notification_bot_dms.count(), 0)
+
+        announcement_channel_message = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.STREAM,
+            recipient__type_id=announce.id,
+            date_sent__gt=now,
+        )
+        self.assertEqual(announcement_channel_message.count(), 0)
+
+        # When subscribing to an already existing channel, if the number of new
+        # subscriptions exceeds the limit, no DMs are sent, but an announcement
+        # message and new channel message should be sent.
+        now = timezone_now()
+        with self.settings(MAX_BULK_NEW_SUBSCRIPTION_MESSAGES=5):
+            response = self.subscribe_via_post(
+                desdemona,
+                ["test D"],
+                principals_dict,
+            )
+        data = self.assert_json_success(response)
+        self.assertEqual(data["new_subscription_messages_sent"], False)
+
+        notification_bot_dms = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.PERSONAL,
+            date_sent__gt=now,
+        )
+        self.assertEqual(notification_bot_dms.count(), 0)
+
+        announcement_channel_message = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.STREAM,
+            recipient__type_id=announce.id,
+            date_sent__gt=now,
+        )
+        # Currently, this notification is not sent even though it should be.
+        self.assertEqual(announcement_channel_message.count(), 0)
+
+        new_channel = get_stream("test D", realm)
+        new_channel_message = Message.objects.filter(
+            realm_id=realm.id,
+            sender=bot.id,
+            recipient__type=Recipient.STREAM,
+            recipient__type_id=new_channel.id,
+            date_sent__gt=now,
+        )
+        # Currently, this notification is not sent even though it should be.
+        self.assert_length(new_channel_message.values(), 0)
+
 
 class InviteOnlyStreamTest(ZulipTestCase):
     def test_must_be_subbed_to_send(self) -> None: