stream: Only realm admins can post to an announcement_only streams.

If a non-admin tries to post to an announcement_only stream, error
message will be shown.

zerver/lib/actions.py

@@ -1937,6 +1937,9 @@ def check_message(sender: UserProfile, client: Client, addressee: Addressee,
             raise StreamDoesNotExistError(escape(stream_name))
         recipient = get_stream_recipient(stream.id)
 
+        if stream.is_announcement_only and not sender.is_realm_admin:
+            raise JsonableError(_("Only organization administrators can send to this stream."))
+
         if not stream.invite_only:
             # This is a public stream
             pass

zerver/tests/test_messages.py

@@ -1024,6 +1024,44 @@ class MessagePOSTTest(ZulipTestCase):
                                                            "subject": "Test subject"})
         self.assert_json_success(result)
 
+    def test_message_to_announce(self) -> None:
+        """
+        Sending a message to an announcement_only stream by a realm admin
+        successful.
+        """
+        user_profile = self.example_user("iago")
+        self.login(user_profile.email)
+
+        stream_name = "Verona"
+        stream = get_stream(stream_name, user_profile.realm)
+        stream.is_announcement_only = True
+        stream.save()
+        result = self.client_post("/json/messages", {"type": "stream",
+                                                     "to": stream_name,
+                                                     "client": "test suite",
+                                                     "content": "Test message",
+                                                     "subject": "Test subject"})
+        self.assert_json_success(result)
+
+    def test_message_fail_to_announce(self) -> None:
+        """
+        Sending a message to an announcement_only stream not by a realm
+        admin fails.
+        """
+        user_profile = self.example_user("hamlet")
+        self.login(user_profile.email)
+
+        stream_name = "Verona"
+        stream = get_stream(stream_name, user_profile.realm)
+        stream.is_announcement_only = True
+        stream.save()
+        result = self.client_post("/json/messages", {"type": "stream",
+                                                     "to": stream_name,
+                                                     "client": "test suite",
+                                                     "content": "Test message",
+                                                     "subject": "Test subject"})
+        self.assert_json_error(result, "Only organization administrators can send to this stream.")
+
     def test_api_message_with_default_to(self) -> None:
         """
         Sending messages without a to field should be sent to the default