6693

pending completion

Build # 6693

Build Type

push

travis-ci

Committed by

timabbott

Commit Message

Fix regression breaking websocket message sending.

In fe1ba6f3e, we change our auth
decorators etc. to use request.POST/request.GET rather than the (now
removed request.REQUEST).  This broke sending messages via our
websockets codepath, because we failed to update the artificial
requests we generate to use request._post as well.

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

22434 of 25311 relevant lines covered (88.63%)

1.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.92

/zerver/worker/queue_processors.py

from __future__ import absolute_import
from typing import Any, Callable, Mapping

from django.conf import settings
from django.core.handlers.wsgi import WSGIRequest
from django.core.handlers.base import BaseHandler
from zerver.models import get_user_profile_by_email, \
    get_user_profile_by_id, get_prereg_user_by_email, get_client
from zerver.lib.context_managers import lockfile
from zerver.lib.queue import SimpleQueueClient, queue_json_publish
from zerver.lib.timestamp import timestamp_to_datetime
from zerver.lib.notifications import handle_missedmessage_emails, enqueue_welcome_emails, \
    clear_followup_emails_queue, send_local_email_template_with_delay
from zerver.lib.actions import do_send_confirmation_email, \
    do_update_user_activity, do_update_user_activity_interval, do_update_user_presence, \
    internal_send_message, check_send_message, extract_recipients, \
    handle_push_notification
from zerver.lib.digest import handle_digest_email
from zerver.lib.email_mirror import process_message as mirror_email
from zerver.decorator import JsonableError
from zerver.lib.socket import req_redis_key
from confirmation.models import Confirmation
from zerver.lib.db import reset_queries
from django.core.mail import EmailMessage
from zerver.lib.redis_utils import get_redis_client

import os
import sys
import ujson
from collections import defaultdict
import email
import time
import datetime
import logging
import simplejson
from six.moves import cStringIO as StringIO

class WorkerDeclarationException(Exception):
    pass

def assign_queue(queue_name, enabled=True):
    # type: (str, bool) -> Callable[[QueueProcessingWorker], QueueProcessingWorker]
    def decorate(clazz):
        # type: (QueueProcessingWorker) -> QueueProcessingWorker
        clazz.queue_name = queue_name
        if enabled:
            register_worker(queue_name, clazz)
        return clazz
    return decorate

worker_classes = {} # type: Dict[str, Any] # Any here should be QueueProcessingWorker type
def register_worker(queue_name, clazz):
    # type: (str, QueueProcessingWorker) -> None
    worker_classes[queue_name] = clazz

def get_worker(queue_name):
    # type: (str) -> QueueProcessingWorker
    return worker_classes[queue_name]()

def get_active_worker_queues():
    # type: () -> List[str]
    return list(worker_classes.keys())

class QueueProcessingWorker(object):
    queue_name = None # type: str

    def __init__(self):
        # type: () -> None
        self.q = None # type: SimpleQueueClient
        if self.queue_name is None:
            raise WorkerDeclarationException("Queue worker declared without queue_name")

    def consume(self, data):
        # type: (Mapping[str, Any]) -> None
        raise WorkerDeclarationException("No consumer defined!")

    def consume_wrapper(self, data):
        # type: (Mapping[str, Any]) -> None
        try:
            self.consume(data)
        except Exception:
            self._log_problem()
            if not os.path.exists(settings.QUEUE_ERROR_DIR):
                os.mkdir(settings.QUEUE_ERROR_DIR)
            fname = '%s.errors' % (self.queue_name,)
            fn = os.path.join(settings.QUEUE_ERROR_DIR, fname)
            line = u'%s\t%s\n' % (time.asctime(), ujson.dumps(data))
            lock_fn = fn + '.lock'
            with lockfile(lock_fn):
                with open(fn, 'ab') as f:
                    f.write(line.encode('utf-8'))
        reset_queries()

    def _log_problem(self):
        # type: () -> None
        logging.exception("Problem handling data on queue %s" % (self.queue_name,))

    def setup(self):
        # type: () -> None
        self.q = SimpleQueueClient()

    def start(self):
        # type: () -> None
        self.q.register_json_consumer(self.queue_name, self.consume_wrapper)
        self.q.start_consuming()

    def stop(self):
        # type: () -> None
        self.q.stop_consuming()

if settings.MAILCHIMP_API_KEY:
    from postmonkey import PostMonkey, MailChimpException

@assign_queue('signups')
class SignupWorker(QueueProcessingWorker):
    def __init__(self):
        # type: () -> None
        super(SignupWorker, self).__init__()
        if settings.MAILCHIMP_API_KEY:
            self.pm = PostMonkey(settings.MAILCHIMP_API_KEY, timeout=10)

    def consume(self, data):
        # type: (Mapping[str, Any]) -> None
        merge_vars=data['merge_vars']
        # This should clear out any invitation reminder emails
        clear_followup_emails_queue(data["EMAIL"])
        if settings.MAILCHIMP_API_KEY and settings.PRODUCTION:
            try:
                self.pm.listSubscribe(
                        id=settings.ZULIP_FRIENDS_LIST_ID,
                        email_address=data['EMAIL'],
                        merge_vars=merge_vars,
                        double_optin=False,
                        send_welcome=False)
            except MailChimpException as e:
                if e.code == 214:
                    logging.warning("Attempted to sign up already existing email to list: %s" % (data['EMAIL'],))
                else:
                    raise e

        email = data.get("EMAIL")
        name = merge_vars.get("NAME")
        enqueue_welcome_emails(email, name)

@assign_queue('invites')
class ConfirmationEmailWorker(QueueProcessingWorker):
    def consume(self, data):
        # type: (Mapping[str, Any]) -> None
        invitee = get_prereg_user_by_email(data["email"])
        referrer = get_user_profile_by_email(data["referrer_email"])
        do_send_confirmation_email(invitee, referrer)

        # queue invitation reminder for two days from now.
        link = Confirmation.objects.get_link_for_object(invitee, host=referrer.realm.host)
        send_local_email_template_with_delay([{'email': data["email"], 'name': ""}],
                                             "zerver/emails/invitation/invitation_reminder_email",
                                             {'activate_url': link,
                                              'referrer': referrer,
                                              'verbose_support_offers': settings.VERBOSE_SUPPORT_OFFERS,
                                              'external_host': settings.EXTERNAL_HOST,
                                              'external_uri_scheme': settings.EXTERNAL_URI_SCHEME,
                                              'server_uri': settings.SERVER_URI,
                                              'realm_uri': referrer.realm.uri,
                                              'support_email': settings.ZULIP_ADMINISTRATOR},
                                             datetime.timedelta(days=2),
                                             tags=["invitation-reminders"],
                                             sender={'email': settings.ZULIP_ADMINISTRATOR, 'name': 'Zulip'})

@assign_queue('user_activity')
class UserActivityWorker(QueueProcessingWorker):
    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        client = get_client(event["client"])
        log_time = timestamp_to_datetime(event["time"])
        query = event["query"]
        do_update_user_activity(user_profile, client, query, log_time)

@assign_queue('user_activity_interval')
class UserActivityIntervalWorker(QueueProcessingWorker):
    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        log_time = timestamp_to_datetime(event["time"])
        do_update_user_activity_interval(user_profile, log_time)

@assign_queue('user_presence')
class UserPresenceWorker(QueueProcessingWorker):
    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        logging.info("Received event: %s" % (event),)
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        client = get_client(event["client"])
        log_time = timestamp_to_datetime(event["time"])
        status = event["status"]
        do_update_user_presence(user_profile, client, log_time, status)

@assign_queue('missedmessage_emails')
class MissedMessageWorker(QueueProcessingWorker):
    def start(self):
        # type: () -> None
        while True:
            missed_events = self.q.drain_queue("missedmessage_emails", json=True)
            by_recipient = defaultdict(list) # type: Dict[int, List[Dict[str, Any]]]

            for event in missed_events:
                logging.info("Received event: %s" % (event,))
                by_recipient[event['user_profile_id']].append(event)

            for user_profile_id, events in by_recipient.items():
                handle_missedmessage_emails(user_profile_id, events)

            reset_queries()
            # Aggregate all messages received every 2 minutes to let someone finish sending a batch
            # of messages
            time.sleep(2 * 60)

@assign_queue('missedmessage_mobile_notifications')
class PushNotificationsWorker(QueueProcessingWorker):
    def consume(self, data):
        # type: (Mapping[str, Any]) -> None
        handle_push_notification(data['user_profile_id'], data)

def make_feedback_client():
    # type: () -> Any # Should be zulip.Client, but not necessarily importable
    sys.path.append(os.path.join(os.path.dirname(__file__), '../../api'))
    import zulip
    return zulip.Client(
        client="ZulipFeedback/0.1",
        email=settings.DEPLOYMENT_ROLE_NAME,
        api_key=settings.DEPLOYMENT_ROLE_KEY,
        verbose=True,
        site=settings.FEEDBACK_TARGET)

# We probably could stop running this queue worker at all if ENABLE_FEEDBACK is False
@assign_queue('feedback_messages')
class FeedbackBot(QueueProcessingWorker):
    def start(self):
        # type: () -> None
        if settings.ENABLE_FEEDBACK and settings.FEEDBACK_EMAIL is None:
            self.staging_client = make_feedback_client()
            self.staging_client._register(
                'forward_feedback',
                method='POST',
                url='deployments/feedback',
                make_request=(lambda request: {'message': simplejson.dumps(request)}),
            )
        QueueProcessingWorker.start(self)

    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        if not settings.ENABLE_FEEDBACK:
            return
        if settings.FEEDBACK_EMAIL is not None:
            to_email = settings.FEEDBACK_EMAIL
            subject = "Zulip feedback from %s" % (event["sender_email"],)
            content = event["content"]
            from_email = '"%s" <%s>' % (event["sender_full_name"], settings.ZULIP_ADMINISTRATOR)
            headers = {'Reply-To' : '"%s" <%s>' % (event["sender_full_name"], event["sender_email"])}
            msg = EmailMessage(subject, content, from_email, [to_email], headers=headers)
            msg.send()
        else:
            self.staging_client.forward_feedback(event)

@assign_queue('error_reports')
class ErrorReporter(QueueProcessingWorker):
    def start(self):
        # type: () -> None
        if settings.DEPLOYMENT_ROLE_KEY:
            self.staging_client = make_feedback_client()
            self.staging_client._register(
                'forward_error',
                method='POST',
                url='deployments/report_error',
                make_request=(lambda type, report: {'type': type, 'report': simplejson.dumps(report)}),
                )
        QueueProcessingWorker.start(self)

    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        if settings.DEPLOYMENT_ROLE_KEY:
            self.staging_client.forward_error(event['type'], event['report'])
        elif settings.ZILENCER_ENABLED:
            from zilencer.views import do_report_error
            do_report_error(settings.DEPLOYMENT_ROLE_NAME, event['type'], event['report'])

@assign_queue('slow_queries')
class SlowQueryWorker(QueueProcessingWorker):
    def start(self):
        # type: () -> None
        while True:
            self.process_one_batch()
            # Aggregate all slow query messages in 1-minute chunks to avoid message spam
            time.sleep(1 * 60)

    def process_one_batch(self):
        # type: () -> None
        slow_queries = self.q.drain_queue("slow_queries", json=True)

        if settings.ERROR_BOT is None:
            return

        if len(slow_queries) > 0:
            topic = "%s: slow queries" % (settings.EXTERNAL_HOST,)

            content = ""
            for query in slow_queries:
                content += "    %s\n" % (query,)

            internal_send_message(settings.ERROR_BOT, "stream", "logs", topic, content)

        reset_queries()

@assign_queue("message_sender")
class MessageSenderWorker(QueueProcessingWorker):
    def __init__(self):
        # type: () -> None
        super(MessageSenderWorker, self).__init__()
        self.redis_client = get_redis_client()
        self.handler = BaseHandler()
        self.handler.load_middleware()

    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        server_meta = event['server_meta']

        environ = {'REQUEST_METHOD': 'SOCKET',
                   'SCRIPT_NAME': '',
                   'PATH_INFO': '/json/messages',
                   'SERVER_NAME': '127.0.0.1',
                   'SERVER_PORT': 9993,
                   'SERVER_PROTOCOL': 'ZULIP_SOCKET/1.0',
                   'wsgi.version': (1, 0),
                   'wsgi.input': StringIO(),
                   'wsgi.errors': sys.stderr,
                   'wsgi.multithread': False,
                   'wsgi.multiprocess': True,
                   'wsgi.run_once': False,
                   'zulip.emulated_method': 'POST'}
        # We're mostly using a WSGIRequest for convenience
        environ.update(server_meta['request_environ'])
        request = WSGIRequest(environ)
        # Note: If we ever support non-POST methods, we'll need to change this.
        request._post = event['request']
        request.csrf_processing_done = True

        user_profile = get_user_profile_by_id(server_meta['user_id'])
        request._cached_user = user_profile

        resp = self.handler.get_response(request)
        server_meta['time_request_finished'] = time.time()
        server_meta['worker_log_data'] = request._log_data

        resp_content = resp.content.decode('utf-8')
        result = {'response': ujson.loads(resp_content), 'req_id': event['req_id'],
                  'server_meta': server_meta}

        redis_key = req_redis_key(event['req_id'])
        self.redis_client.hmset(redis_key, {'status': 'complete',
                                            'response': resp_content});

        queue_json_publish(server_meta['return_queue'], result, lambda e: None)

@assign_queue('digest_emails')
class DigestWorker(QueueProcessingWorker):
    # Who gets a digest is entirely determined by the enqueue_digest_emails
    # management command, not here.
    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        logging.info("Received digest event: %s" % (event,))
        handle_digest_email(event["user_profile_id"], event["cutoff"])

@assign_queue('email_mirror')
class MirrorWorker(QueueProcessingWorker):
    # who gets a digest is entirely determined by the enqueue_digest_emails
    # management command, not here.
    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        mirror_email(email.message_from_string(event["message"]),
                     rcpt_to=event["rcpt_to"], pre_checked=True)

@assign_queue('test')
class TestWorker(QueueProcessingWorker):
    # This worker allows you to test the queue worker infrastructure without
    # creating significant side effects.  It can be useful in development or
    # for troubleshooting prod/staging.  It pulls a message off the test queue
    # and appends it to a file in /tmp.
    def consume(self, event):
        # type: (Mapping[str, Any]) -> None
        fn = settings.ZULIP_WORKER_TEST_FILE
        message = ujson.dumps(event)
        logging.info("TestWorker should append this message to %s: %s" % (fn, message))
        with open(fn, 'a') as f:
            f.write(message + '\n')

1	from __future__ import absolute_import	2✔
2	from typing import Any, Callable, Mapping	2✔
3
4	from django.conf import settings	2✔
5	from django.core.handlers.wsgi import WSGIRequest	2✔
6	from django.core.handlers.base import BaseHandler	2✔
7	from zerver.models import get_user_profile_by_email, \	2✔
8	get_user_profile_by_id, get_prereg_user_by_email, get_client
9	from zerver.lib.context_managers import lockfile	2✔
10	from zerver.lib.queue import SimpleQueueClient, queue_json_publish	2✔
11	from zerver.lib.timestamp import timestamp_to_datetime	2✔
12	from zerver.lib.notifications import handle_missedmessage_emails, enqueue_welcome_emails, \	2✔
13	clear_followup_emails_queue, send_local_email_template_with_delay
14	from zerver.lib.actions import do_send_confirmation_email, \	2✔
15	do_update_user_activity, do_update_user_activity_interval, do_update_user_presence, \
16	internal_send_message, check_send_message, extract_recipients, \
17	handle_push_notification
18	from zerver.lib.digest import handle_digest_email	2✔
19	from zerver.lib.email_mirror import process_message as mirror_email	2✔
20	from zerver.decorator import JsonableError	2✔
21	from zerver.lib.socket import req_redis_key	2✔
22	from confirmation.models import Confirmation	2✔
23	from zerver.lib.db import reset_queries	2✔
24	from django.core.mail import EmailMessage	2✔
25	from zerver.lib.redis_utils import get_redis_client	2✔
26
27	import os	2✔
28	import sys	2✔
29	import ujson	2✔
30	from collections import defaultdict	2✔
31	import email	2✔
32	import time	2✔
33	import datetime	2✔
34	import logging	2✔
35	import simplejson	2✔
36	from six.moves import cStringIO as StringIO	2✔
37
38	class WorkerDeclarationException(Exception):	2✔
39	pass	2✔
40
41	def assign_queue(queue_name, enabled=True):	2✔
42	# type: (str, bool) -> Callable[[QueueProcessingWorker], QueueProcessingWorker]
43	def decorate(clazz):	2✔
44	# type: (QueueProcessingWorker) -> QueueProcessingWorker
45	clazz.queue_name = queue_name	2✔
46	if enabled:	2✔
47	register_worker(queue_name, clazz)	2✔
48	return clazz	2✔
49	return decorate	2✔
50
51	worker_classes = {} # type: Dict[str, Any] # Any here should be QueueProcessingWorker type	2✔
52	def register_worker(queue_name, clazz):	2✔
53	# type: (str, QueueProcessingWorker) -> None
54	worker_classes[queue_name] = clazz	2✔
55
56	def get_worker(queue_name):	2✔
57	# type: (str) -> QueueProcessingWorker
58	return worker_classes[queue_name]()	×
59
60	def get_active_worker_queues():	2✔
61	# type: () -> List[str]
62	return list(worker_classes.keys())	×
63
64	class QueueProcessingWorker(object):	2✔
65	queue_name = None # type: str	2✔
66
67	def __init__(self):	2✔
68	# type: () -> None
69	self.q = None # type: SimpleQueueClient	2✔
70	if self.queue_name is None:	2✔
71	raise WorkerDeclarationException("Queue worker declared without queue_name")	2✔
72
73	def consume(self, data):	2✔
74	# type: (Mapping[str, Any]) -> None
75	raise WorkerDeclarationException("No consumer defined!")	2✔
76
77	def consume_wrapper(self, data):	2✔
78	# type: (Mapping[str, Any]) -> None
79	try:	2✔
80	self.consume(data)	2✔
81	except Exception:	2✔
82	self._log_problem()	2✔
83	if not os.path.exists(settings.QUEUE_ERROR_DIR):	2✔
84	os.mkdir(settings.QUEUE_ERROR_DIR)	2✔
85	fname = '%s.errors' % (self.queue_name,)	2✔
86	fn = os.path.join(settings.QUEUE_ERROR_DIR, fname)	2✔
87	line = u'%s\t%s\n' % (time.asctime(), ujson.dumps(data))	2✔
88	lock_fn = fn + '.lock'	2✔
89	with lockfile(lock_fn):	2✔
90	with open(fn, 'ab') as f:	2✔
91	f.write(line.encode('utf-8'))	2✔
92	reset_queries()	2✔
93
94	def _log_problem(self):	2✔
95	# type: () -> None
96	logging.exception("Problem handling data on queue %s" % (self.queue_name,))	×
97
98	def setup(self):	2✔
99	# type: () -> None
100	self.q = SimpleQueueClient()	2✔
101
102	def start(self):	2✔
103	# type: () -> None
104	self.q.register_json_consumer(self.queue_name, self.consume_wrapper)	2✔
105	self.q.start_consuming()	2✔
106
107	def stop(self):	2✔
108	# type: () -> None
109	self.q.stop_consuming()	×
110
111	if settings.MAILCHIMP_API_KEY:	2✔
112	from postmonkey import PostMonkey, MailChimpException	×
113
114	@assign_queue('signups')	2✔
115	class SignupWorker(QueueProcessingWorker):	2✔
116	def __init__(self):	2✔
117	# type: () -> None
118	super(SignupWorker, self).__init__()	×
119	if settings.MAILCHIMP_API_KEY:	×
120	self.pm = PostMonkey(settings.MAILCHIMP_API_KEY, timeout=10)	×
121
122	def consume(self, data):	2✔
123	# type: (Mapping[str, Any]) -> None
124	merge_vars=data['merge_vars']	×
125	# This should clear out any invitation reminder emails
126	clear_followup_emails_queue(data["EMAIL"])	×
127	if settings.MAILCHIMP_API_KEY and settings.PRODUCTION:	×
128	try:	×
129	self.pm.listSubscribe(	×
130	id=settings.ZULIP_FRIENDS_LIST_ID,
131	email_address=data['EMAIL'],
132	merge_vars=merge_vars,
133	double_optin=False,
134	send_welcome=False)
135	except MailChimpException as e:	×
136	if e.code == 214:	×
137	logging.warning("Attempted to sign up already existing email to list: %s" % (data['EMAIL'],))	×
138	else:
139	raise e	×
140
141	email = data.get("EMAIL")	×
142	name = merge_vars.get("NAME")	×
143	enqueue_welcome_emails(email, name)	×
144
145	@assign_queue('invites')	2✔
146	class ConfirmationEmailWorker(QueueProcessingWorker):	2✔
147	def consume(self, data):	2✔
148	# type: (Mapping[str, Any]) -> None
149	invitee = get_prereg_user_by_email(data["email"])	×
150	referrer = get_user_profile_by_email(data["referrer_email"])	×
151	do_send_confirmation_email(invitee, referrer)	×
152
153	# queue invitation reminder for two days from now.
154	link = Confirmation.objects.get_link_for_object(invitee, host=referrer.realm.host)	×
155	send_local_email_template_with_delay([{'email': data["email"], 'name': ""}],	×
156	"zerver/emails/invitation/invitation_reminder_email",
157	{'activate_url': link,
158	'referrer': referrer,
159	'verbose_support_offers': settings.VERBOSE_SUPPORT_OFFERS,
160	'external_host': settings.EXTERNAL_HOST,
161	'external_uri_scheme': settings.EXTERNAL_URI_SCHEME,
162	'server_uri': settings.SERVER_URI,
163	'realm_uri': referrer.realm.uri,
164	'support_email': settings.ZULIP_ADMINISTRATOR},
165	datetime.timedelta(days=2),
166	tags=["invitation-reminders"],
167	sender={'email': settings.ZULIP_ADMINISTRATOR, 'name': 'Zulip'})
168
169	@assign_queue('user_activity')	2✔
170	class UserActivityWorker(QueueProcessingWorker):	2✔
171	def consume(self, event):	2✔
172	# type: (Mapping[str, Any]) -> None
173	user_profile = get_user_profile_by_id(event["user_profile_id"])	2✔
174	client = get_client(event["client"])	2✔
175	log_time = timestamp_to_datetime(event["time"])	2✔
176	query = event["query"]	2✔
177	do_update_user_activity(user_profile, client, query, log_time)	2✔
178
179	@assign_queue('user_activity_interval')	2✔
180	class UserActivityIntervalWorker(QueueProcessingWorker):	2✔
181	def consume(self, event):	2✔
182	# type: (Mapping[str, Any]) -> None
183	user_profile = get_user_profile_by_id(event["user_profile_id"])	×
184	log_time = timestamp_to_datetime(event["time"])	×
185	do_update_user_activity_interval(user_profile, log_time)	×
186
187	@assign_queue('user_presence')	2✔
188	class UserPresenceWorker(QueueProcessingWorker):	2✔
189	def consume(self, event):	2✔
190	# type: (Mapping[str, Any]) -> None
191	logging.info("Received event: %s" % (event),)	×
192	user_profile = get_user_profile_by_id(event["user_profile_id"])	×
193	client = get_client(event["client"])	×
194	log_time = timestamp_to_datetime(event["time"])	×
195	status = event["status"]	×
196	do_update_user_presence(user_profile, client, log_time, status)	×
197
198	@assign_queue('missedmessage_emails')	2✔
199	class MissedMessageWorker(QueueProcessingWorker):	2✔
200	def start(self):	2✔
201	# type: () -> None
202	while True:	×
203	missed_events = self.q.drain_queue("missedmessage_emails", json=True)	×
204	by_recipient = defaultdict(list) # type: Dict[int, List[Dict[str, Any]]]	×
205
206	for event in missed_events:	×
207	logging.info("Received event: %s" % (event,))	×
208	by_recipient[event['user_profile_id']].append(event)	×
209
210	for user_profile_id, events in by_recipient.items():	×
211	handle_missedmessage_emails(user_profile_id, events)	×
212
213	reset_queries()	×
214	# Aggregate all messages received every 2 minutes to let someone finish sending a batch
215	# of messages
216	time.sleep(2 * 60)	×
217
218	@assign_queue('missedmessage_mobile_notifications')	2✔
219	class PushNotificationsWorker(QueueProcessingWorker):	2✔
220	def consume(self, data):	2✔
221	# type: (Mapping[str, Any]) -> None
222	handle_push_notification(data['user_profile_id'], data)	×
223
224	def make_feedback_client():	2✔
225	# type: () -> Any # Should be zulip.Client, but not necessarily importable
226	sys.path.append(os.path.join(os.path.dirname(__file__), '../../api'))	×
227	import zulip	×
228	return zulip.Client(	×
229	client="ZulipFeedback/0.1",
230	email=settings.DEPLOYMENT_ROLE_NAME,
231	api_key=settings.DEPLOYMENT_ROLE_KEY,
232	verbose=True,
233	site=settings.FEEDBACK_TARGET)
234
235	# We probably could stop running this queue worker at all if ENABLE_FEEDBACK is False
236	@assign_queue('feedback_messages')	2✔
237	class FeedbackBot(QueueProcessingWorker):	2✔
238	def start(self):	2✔
239	# type: () -> None
240	if settings.ENABLE_FEEDBACK and settings.FEEDBACK_EMAIL is None:	×
241	self.staging_client = make_feedback_client()	×
242	self.staging_client._register(	×
243	'forward_feedback',
244	method='POST',
245	url='deployments/feedback',
246	make_request=(lambda request: {'message': simplejson.dumps(request)}),
247	)
248	QueueProcessingWorker.start(self)	×
249
250	def consume(self, event):	2✔
251	# type: (Mapping[str, Any]) -> None
252	if not settings.ENABLE_FEEDBACK:	×
253	return	×
254	if settings.FEEDBACK_EMAIL is not None:	×
255	to_email = settings.FEEDBACK_EMAIL	×
256	subject = "Zulip feedback from %s" % (event["sender_email"],)	×
257	content = event["content"]	×
258	from_email = '"%s" <%s>' % (event["sender_full_name"], settings.ZULIP_ADMINISTRATOR)	×
259	headers = {'Reply-To' : '"%s" <%s>' % (event["sender_full_name"], event["sender_email"])}	×
260	msg = EmailMessage(subject, content, from_email, [to_email], headers=headers)	×
261	msg.send()	×
262	else:
263	self.staging_client.forward_feedback(event)	×
264
265	@assign_queue('error_reports')	2✔
266	class ErrorReporter(QueueProcessingWorker):	2✔
267	def start(self):	2✔
268	# type: () -> None
269	if settings.DEPLOYMENT_ROLE_KEY:	×
270	self.staging_client = make_feedback_client()	×
271	self.staging_client._register(	×
272	'forward_error',
273	method='POST',
274	url='deployments/report_error',
275	make_request=(lambda type, report: {'type': type, 'report': simplejson.dumps(report)}),
276	)
277	QueueProcessingWorker.start(self)	×
278
279	def consume(self, event):	2✔
280	# type: (Mapping[str, Any]) -> None
281	if settings.DEPLOYMENT_ROLE_KEY:	×
282	self.staging_client.forward_error(event['type'], event['report'])	×
283	elif settings.ZILENCER_ENABLED:	×
284	from zilencer.views import do_report_error	×
285	do_report_error(settings.DEPLOYMENT_ROLE_NAME, event['type'], event['report'])	×
286
287	@assign_queue('slow_queries')	2✔
288	class SlowQueryWorker(QueueProcessingWorker):	2✔
289	def start(self):	2✔
290	# type: () -> None
291	while True:	×
292	self.process_one_batch()	×
293	# Aggregate all slow query messages in 1-minute chunks to avoid message spam
294	time.sleep(1 * 60)	×
295
296	def process_one_batch(self):	2✔
297	# type: () -> None
298	slow_queries = self.q.drain_queue("slow_queries", json=True)	×
299
300	if settings.ERROR_BOT is None:	×
301	return	×
302
303	if len(slow_queries) > 0:	×
304	topic = "%s: slow queries" % (settings.EXTERNAL_HOST,)	×
305
306	content = ""	×
307	for query in slow_queries:	×
308	content += " %s\n" % (query,)	×
309
310	internal_send_message(settings.ERROR_BOT, "stream", "logs", topic, content)	×
311
312	reset_queries()	×
313
314	@assign_queue("message_sender")	2✔
315	class MessageSenderWorker(QueueProcessingWorker):	2✔
316	def __init__(self):	2✔
317	# type: () -> None
318	super(MessageSenderWorker, self).__init__()	×
319	self.redis_client = get_redis_client()	×
320	self.handler = BaseHandler()	×
321	self.handler.load_middleware()	×
322
323	def consume(self, event):	2✔
324	# type: (Mapping[str, Any]) -> None
325	server_meta = event['server_meta']	×
326
327	environ = {'REQUEST_METHOD': 'SOCKET',	×
328	'SCRIPT_NAME': '',
329	'PATH_INFO': '/json/messages',
330	'SERVER_NAME': '127.0.0.1',
331	'SERVER_PORT': 9993,
332	'SERVER_PROTOCOL': 'ZULIP_SOCKET/1.0',
333	'wsgi.version': (1, 0),
334	'wsgi.input': StringIO(),
335	'wsgi.errors': sys.stderr,
336	'wsgi.multithread': False,
337	'wsgi.multiprocess': True,
338	'wsgi.run_once': False,
339	'zulip.emulated_method': 'POST'}
340	# We're mostly using a WSGIRequest for convenience
341	environ.update(server_meta['request_environ'])	×
342	request = WSGIRequest(environ)	×
343	# Note: If we ever support non-POST methods, we'll need to change this.
344	request._post = event['request']	×
345	request.csrf_processing_done = True	×
346
347	user_profile = get_user_profile_by_id(server_meta['user_id'])	×
348	request._cached_user = user_profile	×
349
350	resp = self.handler.get_response(request)	×
351	server_meta['time_request_finished'] = time.time()	×
352	server_meta['worker_log_data'] = request._log_data	×
353
354	resp_content = resp.content.decode('utf-8')	×
355	result = {'response': ujson.loads(resp_content), 'req_id': event['req_id'],	×
356	'server_meta': server_meta}
357
358	redis_key = req_redis_key(event['req_id'])	×
359	self.redis_client.hmset(redis_key, {'status': 'complete',	×
360	'response': resp_content});
361
362	queue_json_publish(server_meta['return_queue'], result, lambda e: None)	×
363
364	@assign_queue('digest_emails')	2✔
365	class DigestWorker(QueueProcessingWorker):	2✔
366	# Who gets a digest is entirely determined by the enqueue_digest_emails
367	# management command, not here.
368	def consume(self, event):	2✔
369	# type: (Mapping[str, Any]) -> None
370	logging.info("Received digest event: %s" % (event,))	×
371	handle_digest_email(event["user_profile_id"], event["cutoff"])	×
372
373	@assign_queue('email_mirror')	2✔
374	class MirrorWorker(QueueProcessingWorker):	2✔
375	# who gets a digest is entirely determined by the enqueue_digest_emails
376	# management command, not here.
377	def consume(self, event):	2✔
378	# type: (Mapping[str, Any]) -> None
379	mirror_email(email.message_from_string(event["message"]),	×
380	rcpt_to=event["rcpt_to"], pre_checked=True)
381
382	@assign_queue('test')	2✔
383	class TestWorker(QueueProcessingWorker):	2✔
384	# This worker allows you to test the queue worker infrastructure without
385	# creating significant side effects. It can be useful in development or
386	# for troubleshooting prod/staging. It pulls a message off the test queue
387	# and appends it to a file in /tmp.
388	def consume(self, event):	2✔
389	# type: (Mapping[str, Any]) -> None
390	fn = settings.ZULIP_WORKER_TEST_FILE	×
391	message = ujson.dumps(event)	×
392	logging.info("TestWorker should append this message to %s: %s" % (fn, message))	×
393	with open(fn, 'a') as f:	×
394	f.write(message + '\n')	×

zulip / zulip / 6693

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous