~launchpad-pqm/launchpad/devel

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

"""Test the sendbranchmail script"""

import transaction

from lp.code.enums import (
    BranchSubscriptionDiffSize,
    BranchSubscriptionNotificationLevel,
    CodeReviewNotificationLevel,
    )
from lp.code.model.branchjob import (
    RevisionMailJob,
    RevisionsAddedJob,
    )
from lp.services.osutils import override_environ
from lp.services.scripts.tests import run_script
from lp.testing import TestCaseWithFactory
from lp.testing.layers import ZopelessAppServerLayer


class TestSendbranchmail(TestCaseWithFactory):

    layer = ZopelessAppServerLayer

    def createBranch(self):
        branch, tree = self.create_branch_and_tree()
        branch.subscribe(
            branch.registrant,
            BranchSubscriptionNotificationLevel.FULL,
            BranchSubscriptionDiffSize.WHOLEDIFF,
            CodeReviewNotificationLevel.FULL,
            branch.registrant)
        transport = tree.bzrdir.root_transport
        transport.put_bytes('foo', 'bar')
        tree.add('foo')
        # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
        # required to generate the revision-id.
        with override_environ(BZR_EMAIL='me@example.com'):
            tree.commit('Added foo.', rev_id='rev1')
        return branch, tree

    def test_sendbranchmail(self):
        """Ensure sendbranchmail runs and sends email."""
        self.useBzrBranches()
        branch, tree = self.createBranch()
        mail_job = RevisionMailJob.create(
            branch, 1, 'from@example.org', 'body', 'foo')
        transaction.commit()
        retcode, stdout, stderr = run_script(
            'cronscripts/sendbranchmail.py', [])
        self.assertEqual(
            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
            'INFO    Running RevisionMailJob (ID %d) in status Waiting\n'
            'INFO    Ran 1 RevisionMailJobs.\n' % mail_job.job.id, stderr)
        self.assertEqual('', stdout)
        self.assertEqual(0, retcode)

    def test_sendbranchmail_handles_oops(self):
        """Ensure sendbranchmail runs and sends email."""
        self.useTempBzrHome()
        branch = self.factory.makeBranch()
        RevisionsAddedJob.create(
            branch, 'rev1', 'rev2', 'from@example.org')
        transaction.commit()
        retcode, stdout, stderr = run_script(
            'cronscripts/sendbranchmail.py', [])
        self.assertIn(
            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n',
            stderr)
        self.assertIn('INFO    Job resulted in OOPS:', stderr)
        self.assertIn('INFO    Ran 0 RevisionMailJobs.\n', stderr)
        self.assertEqual('', stdout)
        self.assertEqual(0, retcode)

    def test_revision_added_job(self):
        """RevisionsAddedJobs are run by sendbranchmail."""
        self.useBzrBranches()
        branch, tree = self.createBranch()
        tree.bzrdir.root_transport.put_bytes('foo', 'baz')
        # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
        # required to generate the revision-id.
        with override_environ(BZR_EMAIL='me@example.com'):
            tree.commit('Added foo.', rev_id='rev2')
        job = RevisionsAddedJob.create(
            branch, 'rev1', 'rev2', 'from@example.org')
        transaction.commit()
        retcode, stdout, stderr = run_script(
            'cronscripts/sendbranchmail.py', [])
        self.assertEqual(
            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
            'INFO    Running RevisionsAddedJob (ID %d) in status Waiting\n'
            'INFO    Ran 1 RevisionMailJobs.\n' % job.job.id,
            stderr)
        self.assertEqual('', stdout)
        self.assertEqual(0, retcode)