notification.py

一款基于web的项目管理、bug跟踪系统。提供了与svn集成的操作界面、问题跟踪

# -*- coding: utf-8 -*-
#
# Copyright (C) 2005-2008 Edgewall Software
# Copyright (C) 2005-2006 Emmanuel Blot <[e-mail]>
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution. The terms
# are also available at http://trac.edgewall.org/wiki/TracLicense.
#
# This software consists of voluntary contributions made by many
# individuals. For the exact contribution history, see the revision
# history and logs, available at http://trac.edgewall.org/log/.
#
# Include a basic SMTP server, based on L. Smithson 
# ([e-mail]) extensible Python SMTP Server
#

from trac.core import TracError
from trac.util.datefmt import utc
from trac.ticket.model import Ticket
from trac.ticket.notification import TicketNotifyEmail
from trac.test import EnvironmentStub, Mock, MockPerm
from trac.tests.notification import SMTPThreadedServer, parse_smtp_message, \
                                    smtp_address
                                    
import base64
from datetime import datetime
import os
import quopri
import re
import time
import unittest

SMTP_TEST_PORT = 7000 + os.getpid() % 1000
MAXBODYWIDTH = 76
notifysuite = None


class NotificationTestCase(unittest.TestCase):
    """Notification test cases that send email over SMTP"""
    
    def setUp(self):
        self.env = EnvironmentStub(default_data=True)
        self.env.config.set('project','name', 'TracTest')
        self.env.config.set('notification', 'smtp_enabled', 'true')
        self.env.config.set('notification', 'always_notify_owner', 'true')
        self.env.config.set('notification', 'always_notify_reporter', 'true')
        self.env.config.set('notification', 'smtp_always_cc', 
                            '[e-mail], [e-mail]')
        self.env.config.set('notification', 'use_public_cc', 'true')
        self.env.config.set('notification', 'smtp_port', str(SMTP_TEST_PORT))
        self.env.config.set('notification', 'smtp_server','localhost')
        self.req = Mock(href=self.env.href, abs_href=self.env.abs_href, tz=utc,
                        perm=MockPerm())

    def tearDown(self):
        """Signal the notification test suite that a test is over"""
        notifysuite.tear_down()

    def test_recipients(self):
        """To/Cc recipients"""
        ticket = Ticket(self.env)
        ticket['reporter'] = '"Joe User" <[e-mail]>'
        ticket['owner']    = '[e-mail]'
        ticket['cc']       = '[e-mail], [e-mail], ' \
                             '[e-mail]'
        ticket['summary'] = 'Foo'
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        recipients = notifysuite.smtpd.get_recipients()
        # checks there is no duplicate in the recipient list
        rcpts = []
        for r in recipients:
            self.failIf(r in rcpts)
            rcpts.append(r)
        # checks that all cc recipients have been notified
        cc_list = self.env.config.get('notification', 'smtp_always_cc')
        cc_list = "%s, %s" % (cc_list, ticket['cc'])
        for r in cc_list.replace(',', ' ').split():
            self.failIf(r not in recipients)
        # checks that owner has been notified
        self.failIf(smtp_address(ticket['owner']) not in recipients)
        # checks that reporter has been notified
        self.failIf(smtp_address(ticket['reporter']) not in recipients)

    def test_no_recipient(self):
        """No recipient case"""
        self.env.config.set('notification', 'smtp_always_cc', '')
        ticket = Ticket(self.env)
        ticket['reporter'] = 'anonymous'
        ticket['summary'] = 'Foo'
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        sender = notifysuite.smtpd.get_sender()
        recipients = notifysuite.smtpd.get_recipients()
        message = notifysuite.smtpd.get_message()
        # checks that no message has been sent
        self.failIf(recipients)
        self.failIf(sender)
        self.failIf(message)

    def test_cc_only(self):
        """Notification w/o explicit recipients but Cc: (#3101)"""
        ticket = Ticket(self.env)
        ticket['summary'] = 'Foo'
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        recipients = notifysuite.smtpd.get_recipients()
        # checks that all cc recipients have been notified
        cc_list = self.env.config.get('notification', 'smtp_always_cc')
        for r in cc_list.replace(',', ' ').split():
            self.failIf(r not in recipients)

    def test_structure(self):
        """Basic SMTP message structure (headers, body)"""
        ticket = Ticket(self.env)
        ticket['reporter'] = '"Joe User" <[e-mail]>'
        ticket['owner']    = '[e-mail]'
        ticket['cc']       = '[e-mail], [e-mail], ' \
                             '[e-mail]'
        ticket['summary'] = 'This is a summary'
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        message = notifysuite.smtpd.get_message()
        (headers, body) = parse_smtp_message(message)
        # checks for header existence
        self.failIf(not headers)
        # checks for body existance
        self.failIf(not body)
        # checks for expected headers
        self.failIf('Date' not in headers)
        self.failIf('Subject' not in headers)
        self.failIf('Message-ID' not in headers)
        self.failIf('From' not in headers)

    def test_date(self):
        """Date format compliance (RFC822) 
           we do not support 'military' format""" 
        date_str = r"^((?P<day>\w{3}),\s*)*(?P<dm>\d{2})\s+" \
                   r"(?P<month>\w{3})\s+(?P<year>200\d)\s+" \
                   r"(?P<hour>\d{2}):(?P<min>[0-5][0-9])" \
                   r"(:(?P<sec>[0-5][0-9]))*\s" \
                   r"((?P<tz>\w{2,3})|(?P<offset>[+\-]\d{4}))$"
        date_re = re.compile(date_str)
        # python time module does not detect incorrect time values
        days = ['Mon','Tue','Wed','Thu','Fri','Sat','Sun']
        months = ['Jan','Feb','Mar','Apr','May','Jun', \
                  'Jul','Aug','Sep','Oct','Nov','Dec']
        tz = ['UT','GMT','EST','EDT','CST','CDT','MST','MDT''PST','PDT']
        ticket = Ticket(self.env)
        ticket['reporter'] = '"Joe User" <[e-mail]>'
        ticket['summary'] = 'This is a summary'
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        message = notifysuite.smtpd.get_message()
        (headers, body) = parse_smtp_message(message)
        self.failIf('Date' not in headers)
        mo = date_re.match(headers['Date'])
        self.failIf(not mo)
        if mo.group('day'):
            self.failIf(mo.group('day') not in days)
        self.failIf(int(mo.group('dm')) not in range(1,32))
        self.failIf(mo.group('month') not in months)
        self.failIf(int(mo.group('hour')) not in range(0,24))
        if mo.group('tz'):
            self.failIf(mo.group('tz') not in tz)

    def test_bcc_privacy(self):
        """Visibility of recipients"""
        def run_bcc_feature(public):
            # CC list should be private
            self.env.config.set('notification', 'use_public_cc',
                                public and 'true' or 'false')
            self.env.config.set('notification', 'smtp_always_bcc', 
                                '[e-mail]')
            ticket = Ticket(self.env)
            ticket['reporter'] = '"Joe User" <[e-mail]>'
            ticket['summary'] = 'This is a summary'
            ticket.insert()
            tn = TicketNotifyEmail(self.env)
            tn.notify(ticket, newticket=True)
            message = notifysuite.smtpd.get_message()
            (headers, body) = parse_smtp_message(message)
            if public:
                # Msg should have a To list
                self.failIf('To' not in headers)
                # Extract the list of 'To' recipients from the message
                to = [rcpt.strip() for rcpt in headers['To'].split(',')]
            else:
                # Msg should not have a To list
                self.failIf('To' in headers)
                # Extract the list of 'To' recipients from the message
                to = []            
            # Extract the list of 'Cc' recipients from the message
            cc = [rcpt.strip() for rcpt in headers['Cc'].split(',')]
            # Extract the list of the actual SMTP recipients
            rcptlist = notifysuite.smtpd.get_recipients()
            # Build the list of the expected 'Cc' recipients 
            ccrcpt = self.env.config.get('notification', 'smtp_always_cc')
            cclist = [ccr.strip() for ccr in ccrcpt.split(',')]
            for rcpt in cclist:
                # Each recipient of the 'Cc' list should appear 
                # in the 'Cc' header
                self.failIf(rcpt not in cc)
                # Check the message has actually been sent to the recipients
                self.failIf(rcpt not in rcptlist)
            # Build the list of the expected 'Bcc' recipients 
            bccrcpt = self.env.config.get('notification', 'smtp_always_bcc')
            bcclist = [bccr.strip() for bccr in bccrcpt.split(',')]
            for rcpt in bcclist:
                # Check none of the 'Bcc' recipients appears 
                # in the 'To' header
                self.failIf(rcpt in to)
                # Check the message has actually been sent to the recipients
                self.failIf(rcpt not in rcptlist)
        run_bcc_feature(True)
        run_bcc_feature(False)

    def test_short_login(self):
        """Email addresses without a FQDN"""
        def _test_short_login(enabled):
            ticket = Ticket(self.env)
            ticket['reporter'] = 'joeuser'
            ticket['summary'] = 'This is a summary'
            ticket.insert()
            # Be sure that at least one email address is valid, so that we 
            # send a notification even if other addresses are not valid
            self.env.config.set('notification', 'smtp_always_cc',
                                '[e-mail]')
            if enabled:
                self.env.config.set('notification', 'use_short_addr', 'true')
            tn = TicketNotifyEmail(self.env)
            tn.notify(ticket, newticket=True)
            message = notifysuite.smtpd.get_message()
            (headers, body) = parse_smtp_message(message)
            # Msg should not have a 'To' header
            if not enabled:
                self.failIf('To' in headers)
            else:
                tolist = [addr.strip() for addr in headers['To'].split(',')]
            # Msg should have a 'Cc' field
            self.failIf('Cc' not in headers)
            cclist = [addr.strip() for addr in headers['Cc'].split(',')]
            if enabled:
                # Msg should be delivered to the reporter
                self.failIf(ticket['reporter'] not in tolist)
            else:
                # Msg should not be delivered to joeuser
                self.failIf(ticket['reporter'] in cclist)
            # Msg should still be delivered to the always_cc list
            self.failIf(self.env.config.get('notification',
                        'smtp_always_cc') not in cclist)
        # Validate with and without the short addr option enabled
        for enable in [False, True]:
            _test_short_login(enable)

    def test_default_domain(self):
        """Default domain name"""
        def _test_default_domain(enabled):
            self.env.config.set('notification', 'always_notify_owner',
                                'false')
            self.env.config.set('notification', 'always_notify_reporter',
                                'false')
            self.env.config.set('notification', 'smtp_always_cc', '')
            ticket = Ticket(self.env)
            ticket['cc'] = 'joenodom, [e-mail]'
            ticket['summary'] = 'This is a summary'
            ticket.insert()
            # Be sure that at least one email address is valid, so that we 
            # send a notification even if other addresses are not valid
            self.env.config.set('notification', 'smtp_always_cc',
                                '[e-mail]')
            if enabled:
                self.env.config.set('notification', 'smtp_default_domain',
                                    'example.org')
            tn = TicketNotifyEmail(self.env)
            tn.notify(ticket, newticket=True)
            message = notifysuite.smtpd.get_message()
            (headers, body) = parse_smtp_message(message)
            # Msg should always have a 'Cc' field
            self.failIf('Cc' not in headers)
            cclist = [addr.strip() for addr in headers['Cc'].split(',')]
            self.failIf('[e-mail]' not in cclist)
            self.failIf('[e-mail]' not in cclist)
            if not enabled:
                self.failIf(len(cclist) != 2)
                self.failIf('joenodom' in cclist)
            else:
                self.failIf(len(cclist) != 3)
                self.failIf('[e-mail]' not in cclist)

        # Validate with and without a default domain
        for enable in [False, True]:
            _test_default_domain(enable)

    def test_email_map(self):
        """Login-to-email mapping"""
        self.env.config.set('notification', 'always_notify_owner', 'false')
        self.env.config.set('notification', 'always_notify_reporter', 'true')
        self.env.config.set('notification', 'smtp_always_cc',
                            '[e-mail]')
        self.env.known_users = [('joeuser', 'Joe User',
                                '[e-mail]')]
        ticket = Ticket(self.env)
        ticket['reporter'] = 'joeuser'
        ticket['summary'] = 'This is a summary'
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        message = notifysuite.smtpd.get_message()
        (headers, body) = parse_smtp_message(message)
        # Msg should always have a 'To' field
        self.failIf('To' not in headers)
        tolist = [addr.strip() for addr in headers['To'].split(',')]
        # 'To' list should have been resolved to the real email address
        self.failIf('[e-mail]' not in tolist)