zcommands: Add zcommand module and separate test module.

Move the zcommands from '/views/messages.py' to
'/lib/zcommand'.

Also, move the zcommand tests from '/tests/test_messages.py'
to '/tests/test_zcommand'.

zerver/lib/zcommand.py

@@ -0,0 +1,59 @@
+from typing import Any, Dict
+from django.utils.translation import ugettext as _
+from django.core.exceptions import ObjectDoesNotExist
+
+from zerver.models import UserProfile
+from zerver.lib.topic_mutes import topic_is_muted
+from zerver.lib.actions import do_set_user_display_setting, do_mute_topic
+from zerver.models import get_realm_stream, topic_exists, get_stream_recipient
+from zerver.lib.exceptions import JsonableError
+
+def process_zcommands(command: str, user_profile: UserProfile) -> Dict[str, Any]:
+    tokens = command.split(' ')
+
+    if command == 'ping':
+        ret = dict()  # type: Dict[str, Any]
+        return ret
+
+    if command == 'night':
+        if user_profile.night_mode:
+            msg = 'You are still in night mode.'
+        else:
+            msg = 'Changed to night mode! To revert night mode, type `/day`.'
+            do_set_user_display_setting(user_profile, 'night_mode', True)
+        ret = dict(msg=msg)
+        return ret
+
+    if command == 'day':
+        if user_profile.night_mode:
+            msg = 'Changed to day mode! To revert day mode, type `/night`.'
+            do_set_user_display_setting(user_profile, 'night_mode', False)
+        else:
+            msg = 'You are still in day mode.'
+        ret = dict(msg=msg)
+        return ret
+
+    if tokens[0] == 'mute_topic':
+        if len(tokens) < 3 or tokens[1] == 'help':
+            return dict(msg="Usage: /mute_topic <stream_name> <topic_name>.")
+        else:
+            topic = ' '.join(tokens[2:])
+            stream_name = tokens[1]
+            try:
+                stream = get_realm_stream(stream_name=stream_name,
+                                          realm_id=user_profile.realm.id)
+                recipient = get_stream_recipient(stream.id)
+                if not topic_exists(topic, recipient):
+                    msg = ("A valid topic name is required.")
+                else:
+                    if topic_is_muted(user_profile, stream.id, topic):
+                        msg = ("The topic '%s' is already muted." % (topic))
+                    else:
+                        do_mute_topic(user_profile, stream, recipient, topic)
+                        msg = ("The topic '%s' has been muted." % (topic))
+            except ObjectDoesNotExist:
+                msg = ('A valid stream name is required.')
+            ret = dict(msg=msg)
+            return ret
+
+    raise JsonableError(_('No such command: %s') % (command,))

zerver/tests/test_messages.py

@@ -1009,69 +1009,6 @@ def test_sew_messages_and_reaction(self) -> None:
 
 class MessagePOSTTest(ZulipTestCase):
 
-    def test_zcommand(self) -> None:
-        self.login(self.example_email("hamlet"))
-
-        payload = dict(command="boil-ocean")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_error(result, "No such command: boil-ocean")
-
-        payload = dict(command="ping")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-
-        user = self.example_user('hamlet')
-        user.night_mode = False
-        user.save()
-
-        payload = dict(command="night")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('Changed to night', result.json()['msg'])
-
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('still in night mode', result.json()['msg'])
-
-        payload = dict(command="day")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('Changed to day', result.json()['msg'])
-
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('still in day mode', result.json()['msg'])
-
-        payload = dict(command="mute_topic help")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('Usage: /mute_topic <stream_name> <topic_name>.', result.json()['msg'])
-
-        payload = dict(command="mute_topic invalid")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('Usage: /mute_topic <stream_name> <topic_name>.', result.json()['msg'])
-
-        payload = dict(command="mute_topic invalid_stream topic")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('A valid stream name is required.', result.json()['msg'])
-
-        payload = dict(command="mute_topic Denmark invalid_topic")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn('A valid topic name is required.', result.json()['msg'])
-
-        payload = dict(command="mute_topic Verona Verona1")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn("The topic 'Verona1' has been muted.", result.json()['msg'])
-
-        payload = dict(command="mute_topic Verona Verona1")
-        result = self.client_post("/json/zcommand", payload)
-        self.assert_json_success(result)
-        self.assertIn("The topic 'Verona1' is already muted.", result.json()['msg'])
-
     def test_message_to_self(self) -> None:
         """
         Sending a message to a stream to which you are subscribed is

zerver/tests/test_zcommand.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+
+from zerver.lib.test_classes import (
+    ZulipTestCase,
+)
+
+class ZcommandTest(ZulipTestCase):
+
+    def test_invalid_zcommand(self) -> None:
+        self.login(self.example_email("hamlet"))
+
+        payload = dict(command="boil-ocean")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_error(result, "No such command: boil-ocean")
+
+    def test_ping_zcommand(self) -> None:
+        self.login(self.example_email("hamlet"))
+
+        payload = dict(command="ping")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+
+    def test_night_zcommand(self) -> None:
+        self.login(self.example_email("hamlet"))
+        user = self.example_user('hamlet')
+        user.night_mode = False
+        user.save()
+
+        payload = dict(command="night")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('Changed to night', result.json()['msg'])
+
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('still in night mode', result.json()['msg'])
+
+    def test_day_zcommand(self) -> None:
+        self.login(self.example_email("hamlet"))
+        user = self.example_user('hamlet')
+        user.night_mode = True
+        user.save()
+
+        payload = dict(command="day")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('Changed to day', result.json()['msg'])
+
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('still in day mode', result.json()['msg'])
+
+    def test_mute_topic_zcommand(self) -> None:
+        self.login(self.example_email("hamlet"))
+
+        payload = dict(command="mute_topic help")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('Usage: /mute_topic <stream_name> <topic_name>.', result.json()['msg'])
+
+        payload = dict(command="mute_topic invalid")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('Usage: /mute_topic <stream_name> <topic_name>.', result.json()['msg'])
+
+        payload = dict(command="mute_topic invalid_stream topic")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('A valid stream name is required.', result.json()['msg'])
+
+        payload = dict(command="mute_topic Denmark invalid_topic")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn('A valid topic name is required.', result.json()['msg'])
+
+        payload = dict(command="mute_topic Verona Verona1")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn("The topic 'Verona1' has been muted.", result.json()['msg'])
+
+        payload = dict(command="mute_topic Verona Verona1")
+        result = self.client_post("/json/zcommand", payload)
+        self.assert_json_success(result)
+        self.assertIn("The topic 'Verona1' is already muted.", result.json()['msg'])

zerver/views/messages.py

@@ -2,7 +2,7 @@
 from django.utils.timezone import now as timezone_now
 from django.conf import settings
 from django.core import validators
-from django.core.exceptions import ValidationError, ObjectDoesNotExist
+from django.core.exceptions import ValidationError
 from django.db import connection
 from django.http import HttpRequest, HttpResponse
 from typing import Dict, List, Set, Any, Callable, Iterable, \
@@ -13,12 +13,12 @@
     REQ, to_non_negative_int
 from django.utils.html import escape as escape_html
 from zerver.lib import bugdown
+from zerver.lib.zcommand import process_zcommands
 from zerver.lib.actions import recipient_for_emails, do_update_message_flags, \
     compute_mit_user_fullname, compute_irc_user_fullname, compute_jabber_user_fullname, \
     create_mirror_user_if_needed, check_send_message, do_update_message, \
     extract_recipients, truncate_body, render_incoming_message, do_delete_message, \
     do_mark_all_as_read, do_mark_stream_messages_as_read, \
-    do_set_user_display_setting, do_mute_topic, \
     get_user_info_for_message_updates, check_schedule_message
 from zerver.lib.queue import queue_json_publish
 from zerver.lib.message import (
@@ -32,14 +32,14 @@
 from zerver.lib.streams import access_stream_by_id, can_access_stream_history_by_name
 from zerver.lib.timestamp import datetime_to_timestamp, convert_to_UTC
 from zerver.lib.timezone import get_timezone
-from zerver.lib.topic_mutes import exclude_topic_mutes, topic_is_muted
+from zerver.lib.topic_mutes import exclude_topic_mutes
 from zerver.lib.utils import statsd
 from zerver.lib.validator import \
     check_list, check_int, check_dict, check_string, check_bool
 from zerver.models import Message, UserProfile, Stream, Subscription, Client,\
     Realm, RealmDomain, Recipient, UserMessage, bulk_get_recipients, get_personal_recipient, \
-    get_stream, email_to_domain, get_realm, get_active_streams, get_realm_stream, \
-    get_user_including_cross_realm, get_stream_recipient, topic_exists
+    get_stream, email_to_domain, get_realm, get_active_streams, \
+    get_user_including_cross_realm, get_stream_recipient
 
 from sqlalchemy import func
 from sqlalchemy.sql import select, join, column, literal_column, literal, and_, \
@@ -684,56 +684,7 @@ def find_first_unread_anchor(sa_conn: Any,
 @has_request_variables
 def zcommand_backend(request: HttpRequest, user_profile: UserProfile,
                      command: str=REQ('command')) -> HttpResponse:
-
-    tokens = command.split(' ')
-
-    if command == 'ping':
-        ret = dict()  # type: Dict[str, Any]
-        return json_success(ret)
-
-    if command == 'night':
-        if user_profile.night_mode:
-            msg = 'You are still in night mode.'
-        else:
-            msg = 'Changed to night mode! To revert night mode, type `/day`.'
-            do_set_user_display_setting(user_profile, 'night_mode', True)
-        ret = dict(msg=msg)
-        return json_success(ret)
-
-    if command == 'day':
-        if user_profile.night_mode:
-            msg = 'Changed to day mode! To revert day mode, type `/night`.'
-            do_set_user_display_setting(user_profile, 'night_mode', False)
-        else:
-            msg = 'You are still in day mode.'
-        ret = dict(msg=msg)
-        return json_success(ret)
-
-    if tokens[0] == 'mute_topic':
-        if len(tokens) < 3 or tokens[1] == 'help':
-            ret = dict(msg="Usage: /mute_topic <stream_name> <topic_name>.")
-            return json_success(ret)
-
-        topic = ' '.join(tokens[2:])
-        stream_name = tokens[1]
-        try:
-            stream = get_realm_stream(stream_name=stream_name,
-                                      realm_id=user_profile.realm.id)
-            recipient = get_stream_recipient(stream.id)
-            if not topic_exists(topic, recipient):
-                msg = ("A valid topic name is required.")
-            else:
-                if topic_is_muted(user_profile, stream.id, topic):
-                    msg = ("The topic '%s' is already muted." % (topic))
-                else:
-                    do_mute_topic(user_profile, stream, recipient, topic)
-                    msg = ("The topic '%s' has been muted." % (topic))
-        except ObjectDoesNotExist:
-            msg = ('A valid stream name is required.')
-        ret = dict(msg=msg)
-        return json_success(ret)
-
-    raise JsonableError(_('No such command: %s') % (command,))
+    return json_success(process_zcommands(command, user_profile))
 
 @has_request_variables
 def get_messages_backend(request: HttpRequest, user_profile: UserProfile,