notification.py

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

        self.failIf('joeuser' in tolist)
        
    def test_ignore_domains(self):
        """Non-SMTP domain exclusion"""
        self.env.config.set('notification', 'ignore_domains',
                            'example.com, example.org')
        self.env.known_users = \
            [('[e-mail]', 'No Email', ''), 
             ('[e-mail]', 'With Email', '[e-mail]')]
        ticket = Ticket(self.env)
        ticket['reporter'] = '[e-mail]'
        ticket['owner'] = '[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)
        # 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 not contain addresses with non-SMTP domains
        self.failIf('[e-mail]' in tolist)
        self.failIf('[e-mail]' in tolist)
        # 'To' list should have been resolved to the actual email address
        self.failIf('[e-mail]' not in tolist)
        self.failIf(len(tolist) != 1)
        
    def test_admit_domains(self):
        """SMTP domain inclusion"""
        self.env.config.set('notification', 'admit_domains',
                            'localdomain, server')
        ticket = Ticket(self.env)
        ticket['reporter'] = '[e-mail]'
        ticket['summary'] = 'This is a summary'
        ticket['cc'] = 'joe.user@localdomain, joe.user@unknown, ' \
                       'joe.user@server'
        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('Cc' not in headers)
        cclist = [addr.strip() for addr in headers['Cc'].split(',')]
        # 'Cc' list should contain addresses with SMTP included domains
        self.failIf('joe.user@localdomain' not in cclist)
        self.failIf('joe.user@server' not in cclist)
        # 'Cc' list should not contain non-FQDN domains
        self.failIf('joe.user@unknown' in cclist)
        self.failIf(len(cclist) != 2+2)

    def test_multiline_header(self):
        """Encoded headers split into multiple lines"""
        self.env.config.set('notification','mime_encoding', 'qp')
        ticket = Ticket(self.env)
        ticket['reporter'] = '[e-mail]'
        # Forces non-ascii characters
        ticket['summary'] = u'A_very %s s煤mm盲ry' % u' '.join(['long'] * 20)
        ticket.insert()
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        message = notifysuite.smtpd.get_message()
        (headers, body) = parse_smtp_message(message)
        # Discards the project name & ticket number
        subject = headers['Subject']
        summary = subject[subject.find(':')+2:]
        self.failIf(ticket['summary'] != summary)

    def test_mimebody_b64(self):
        """MIME Base64/utf-8 encoding"""
        self.env.config.set('notification','mime_encoding', 'base64')
        ticket = Ticket(self.env)
        ticket['reporter'] = '[e-mail]'
        ticket['summary'] = u'This is a long enough summary to cause Trac ' \
                            u'to generate a multi-line (2 lines) s煤mm盲ry'
        ticket.insert()
        self._validate_mimebody((base64, 'base64', 'utf-8'), \
                                ticket, True)

    def test_mimebody_qp(self):
        """MIME QP/utf-8 encoding"""
        self.env.config.set('notification','mime_encoding', 'qp')
        ticket = Ticket(self.env)
        ticket['reporter'] = '[e-mail]'
        ticket['summary'] = u'This is a long enough summary to cause Trac ' \
                            u'to generate a multi-line (2 lines) s煤mm盲ry'
        ticket.insert()
        self._validate_mimebody((quopri, 'quoted-printable', 'utf-8'),
                                ticket, True)

    def test_mimebody_none(self):
        """MIME None/ascii encoding"""
        self.env.config.set('notification','mime_encoding', 'none')
        ticket = Ticket(self.env)
        ticket['reporter'] = '[e-mail]'
        ticket['summary'] = u'This is a summary'
        ticket.insert()
        self._validate_mimebody((None, '7bit', 'ascii'), \
                                ticket, True)

    def test_md5_digest(self):
        """MD5 digest w/ non-ASCII recipient address (#3491)"""
        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', '')
        ticket = Ticket(self.env)
        ticket['reporter'] = u'"J枚e Us猫r" <[e-mail]>'
        ticket['summary'] = u'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)

    def test_updater(self):
        """No-self-notification option"""
        def _test_updater(disable):
            if disable:
                self.env.config.set('notification','always_notify_updater',
                                    'false')
            ticket = Ticket(self.env)
            ticket['reporter'] = '[e-mail]'
            ticket['summary'] = u'This is a s煤mm盲ry'
            ticket['cc'] = '[e-mail]'
            ticket.insert()
            ticket['component'] = 'dummy'
            now = datetime.now(utc)
            ticket.save_changes('[e-mail]', 'This is a change',
                                when=now)
            tn = TicketNotifyEmail(self.env)
            tn.notify(ticket, newticket=False, modtime=now)
            message = notifysuite.smtpd.get_message()
            (headers, body) = parse_smtp_message(message)
            # checks for header existence
            self.failIf(not headers)
            # checks for updater in the 'To' recipient list
            self.failIf('To' not in headers)
            tolist = [addr.strip() for addr in headers['To'].split(',')]
            if disable:
                self.failIf('[e-mail]' in tolist)
            else:
                self.failIf('[e-mail]' not in tolist)

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

    def test_updater_only(self):
        """Notification w/ updater, w/o any other recipient (#4188)"""
        self.env.config.set('notification', 'always_notify_owner', 'false')
        self.env.config.set('notification', 'always_notify_reporter', 'false')
        self.env.config.set('notification', 'always_notify_updater', 'true')
        self.env.config.set('notification', 'smtp_always_cc', '')
        self.env.config.set('notification', 'smtp_always_bcc', '')
        self.env.config.set('notification', 'use_public_cc', 'false')
        self.env.config.set('notification', 'use_short_addr', 'false')
        self.env.config.set('notification', 'smtp_replyto', 
                            '[e-mail]')
        ticket = Ticket(self.env)
        ticket['summary'] = 'Foo'
        ticket.insert()
        ticket['summary'] = 'Bar'
        ticket['component'] = 'New value'
        ticket.save_changes('[e-mail]', 'this is my comment')        
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        recipients = notifysuite.smtpd.get_recipients()
        self.failIf(recipients is None)
        self.failIf(len(recipients) != 1)
        self.failIf(recipients[0] != '[e-mail]')

    def test_updater_is_reporter(self):
        """Notification to reporter w/ updater option disabled (#3780)"""
        self.env.config.set('notification', 'always_notify_owner', 'false')
        self.env.config.set('notification', 'always_notify_reporter', 'true')
        self.env.config.set('notification', 'always_notify_updater', 'false')
        self.env.config.set('notification', 'smtp_always_cc', '')
        self.env.config.set('notification', 'smtp_always_bcc', '')
        self.env.config.set('notification', 'use_public_cc', 'false')
        self.env.config.set('notification', 'use_short_addr', 'false')
        self.env.config.set('notification', 'smtp_replyto', 
                            '[e-mail]')
        ticket = Ticket(self.env)
        ticket['summary'] = 'Foo'
        ticket['reporter'] = u'[e-mail]'
        ticket.insert()
        ticket['summary'] = 'Bar'
        ticket['component'] = 'New value'
        ticket.save_changes('[e-mail]', 'this is my comment')        
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=True)
        recipients = notifysuite.smtpd.get_recipients()
        self.failIf(recipients is None)
        self.failIf(len(recipients) != 1)
        self.failIf(recipients[0] != '[e-mail]')

    def _validate_mimebody(self, mime, ticket, newtk):
        """Body of a ticket notification message"""
        (mime_decoder, mime_name, mime_charset) = mime
        tn = TicketNotifyEmail(self.env)
        tn.notify(ticket, newticket=newtk)
        message = notifysuite.smtpd.get_message()
        (headers, body) = parse_smtp_message(message)
        self.failIf('MIME-Version' not in headers)
        self.failIf('Content-Type' not in headers)
        self.failIf('Content-Transfer-Encoding' not in headers)
        self.failIf(not re.compile(r"1.\d").match(headers['MIME-Version']))
        type_re = re.compile(r'^text/plain;\scharset="([\w\-\d]+)"$')
        charset = type_re.match(headers['Content-Type'])
        self.failIf(not charset)
        charset = charset.group(1)
        self.assertEqual(charset, mime_charset)
        self.assertEqual(headers['Content-Transfer-Encoding'], mime_name)
        # checks the width of each body line
        for line in body.splitlines():
            self.failIf(len(line) > MAXBODYWIDTH)
        # attempts to decode the body, following the specified MIME endoding 
        # and charset
        try:
            if mime_decoder:
                body = mime_decoder.decodestring(body)
            body = unicode(body, charset)
        except Exception, e:
            raise AssertionError, e
        # now processes each line of the body
        bodylines = body.splitlines()
        # body starts with one of more summary lines, first line is prefixed 
        # with the ticket number such as #<n>: summary        
        # finds the banner after the summary
        banner_delim_re = re.compile(r'^\-+\+\-+$')
        bodyheader = []
        while ( not banner_delim_re.match(bodylines[0]) ):
            bodyheader.append(bodylines.pop(0))
        # summary should be present
        self.failIf(not bodyheader)
        # banner should not be empty
        self.failIf(not bodylines)
        # extracts the ticket ID from the first line
        (tknum, bodyheader[0]) = bodyheader[0].split(' ', 1)
        self.assertEqual(tknum[0], '#')
        try:
            tkid = int(tknum[1:-1])
            self.assertEqual(tkid, 1)
        except ValueError:
            raise AssertionError, "invalid ticket number"
        self.assertEqual(tknum[-1], ':')
        summary = ' '.join(bodyheader)
        self.assertEqual(summary, ticket['summary'])
        # now checks the banner contents
        self.failIf(not banner_delim_re.match(bodylines[0]))
        banner = True
        footer = None
        props = {}
        for line in bodylines[1:]:
            # detect end of banner
            if banner_delim_re.match(line):
                banner = False
                continue
            if banner:
                # parse banner and fill in a property dict
                properties = line.split('|')
                self.assertEqual(len(properties), 2)
                for prop in properties:
                    if prop.strip() == '':
                        continue
                    (k, v) = prop.split(':')
                    props[k.strip().lower()] = v.strip()
            # detect footer marker (weak detection)
            if not footer:
                if line.strip() == '--':
                    footer = 0
                    continue
            # check footer
            if footer != None:
                footer += 1
                # invalid footer detection
                self.failIf(footer > 3)
                # check ticket link
                if line[:11] == 'Ticket URL:':
                    self.assertEqual(line[12:].strip(),
                                     "<%s>" % ticket['link'].strip())
                # note project title / URL are not validated yet

        # ticket properties which are not expected in the banner
        xlist = ['summary', 'description', 'link', 'comment', 'new']
        # check banner content (field exists, msg value matches ticket value)
        for p in [prop for prop in ticket.values.keys() if prop not in xlist]:
            self.failIf(not props.has_key(p))
            self.failIf(props[p] != ticket[p])


class NotificationTestSuite(unittest.TestSuite):
    """Thin test suite wrapper to start and stop the SMTP test server"""

    def __init__(self):
        """Start the local SMTP test server"""
        unittest.TestSuite.__init__(self)
        self.smtpd = SMTPThreadedServer(SMTP_TEST_PORT)
        self.smtpd.start()
        self.addTest(unittest.makeSuite(NotificationTestCase, 'test'))
        self.remaining = self.countTestCases()

    def tear_down(self):
        """Reset the local SMTP test server"""
        self.smtpd.cleanup()
        self.remaining = self.remaining-1
        if self.remaining > 0:
            return
        # stop the SMTP test server when all tests have been completed
        self.smtpd.stop()

def suite():
    global notifysuite
    if not notifysuite:
        notifysuite = NotificationTestSuite()
    return notifysuite

if __name__ == '__main__':
    unittest.TextTestRunner(verbosity=2).run(suite())