~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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# Copyright 2011 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

from zope.security.proxy import removeSecurityProxy

from canonical.testing.layers import LaunchpadZopelessLayer
from lp.archivepublisher.utils import get_ppa_reference
from lp.services.mail.sendmail import format_address_for_person
from lp.soyuz.adapters.notification import (
    calculate_subject,
    fetch_information,
    person_to_email,
    notify,
    )
from lp.soyuz.enums import PackageUploadCustomFormat
from lp.testing import TestCaseWithFactory
from lp.testing.mail_helpers import pop_notifications


class TestNotification(TestCaseWithFactory):

    layer = LaunchpadZopelessLayer

    def test_fetch_information_changes(self):
        changes = {
            'Date': '2001-01-01',
            'Changed-By': 'Foo Bar <foo.bar@canonical.com>',
            'Maintainer': 'Foo Bar <foo.bar@canonical.com>',
            'Changes': ' * Foo!'
            }
        (changesfile, date, changedby, maintainer) = fetch_information(
            None, None, changes)
        self.assertEqual('2001-01-01', date)
        self.assertEqual(' * Foo!', changesfile)
        for field in (changedby, maintainer):
            self.assertEqual('Foo Bar <foo.bar@canonical.com>', field)

    def test_fetch_information_spr(self):
        spr = self.factory.makeSourcePackageRelease()
        (changesfile, date, changedby, maintainer) = fetch_information(
            spr, None, None)
        self.assertEqual(date, spr.dateuploaded)
        self.assertEqual(changesfile, spr.changelog_entry)
        self.assertEqual(changedby, format_address_for_person(spr.creator))
        self.assertEqual(
            maintainer, format_address_for_person(spr.maintainer))

    def test_fetch_information_bprs(self):
        bpr = self.factory.makeBinaryPackageRelease()
        (changesfile, date, changedby, maintainer) = fetch_information(
            None, [bpr], None)
        spr = bpr.build.source_package_release
        self.assertEqual(date, spr.dateuploaded)
        self.assertEqual(changesfile, spr.changelog_entry)
        self.assertEqual(changedby, format_address_for_person(spr.creator))
        self.assertEqual(
            maintainer, format_address_for_person(spr.maintainer))

    def test_calculate_subject_spr(self):
        spr = self.factory.makeSourcePackageRelease()
        archive = self.factory.makeArchive()
        pocket = self.factory.getAnyPocket()
        distroseries = self.factory.makeDistroSeries()
        expected_subject = '[PPA %s] [%s/%s] %s %s (Accepted)' % (
            get_ppa_reference(archive), distroseries.distribution.name,
            distroseries.getSuite(pocket), spr.name, spr.version)  
        subject = calculate_subject(
            spr, [], [], archive, distroseries, pocket, 'accepted')
        self.assertEqual(expected_subject, subject)

    def test_calculate_subject_bprs(self):
        bpr = self.factory.makeBinaryPackageRelease()
        archive = self.factory.makeArchive()
        pocket = self.factory.getAnyPocket()
        distroseries = self.factory.makeDistroSeries()
        expected_subject = '[PPA %s] [%s/%s] %s %s (Accepted)' % (
            get_ppa_reference(archive), distroseries.distribution.name,
            distroseries.getSuite(pocket),
            bpr.build.source_package_release.name, bpr.version)
        subject = calculate_subject(
            None, [bpr], [], archive, distroseries, pocket, 'accepted')
        self.assertEqual(expected_subject, subject)

    def test_calculate_subject_customfile(self):
        lfa = self.factory.makeLibraryFileAlias()
        package_upload = self.factory.makePackageUpload()
        customfile = package_upload.addCustom(
            lfa, PackageUploadCustomFormat.DEBIAN_INSTALLER)
        archive = self.factory.makeArchive()
        pocket = self.factory.getAnyPocket()
        distroseries = self.factory.makeDistroSeries()
        expected_subject = '[PPA %s] [%s/%s] %s - (Accepted)' % (
            get_ppa_reference(archive), distroseries.distribution.name,
            distroseries.getSuite(pocket), lfa.filename)
        subject = calculate_subject(
            None, [], [customfile], archive, distroseries, pocket,
            'accepted')
        self.assertEqual(expected_subject, subject)

    def test_notify_bpr(self):
        # If we notify about an accepted bpr with no source, it is from a
        # build, and no notification is sent.
        bpr = self.factory.makeBinaryPackageRelease()
        archive = self.factory.makeArchive()
        pocket = self.factory.getAnyPocket()
        distroseries = self.factory.makeDistroSeries()
        person = self.factory.makePerson()
        notify(
            person, None, [bpr], [], archive, distroseries, pocket,
            action='accepted')
        notifications = pop_notifications()
        self.assertEqual(0, len(notifications))

    def test_notify_bpr_rejected(self):
        # If we notify about a rejected bpr with no source, a notification is
        # sent.
        bpr = self.factory.makeBinaryPackageRelease()
        removeSecurityProxy(
            bpr.build.source_package_release).changelog_entry = '* Foo!'
        archive = self.factory.makeArchive()
        pocket = self.factory.getAnyPocket()
        distroseries = self.factory.makeDistroSeries()
        person = self.factory.makePerson()
        notify(
            person, None, [bpr], [], archive, distroseries, pocket,
            action='rejected')
        [notification] = pop_notifications()
        body = notification.as_string()
        self.assertEqual(person_to_email(person), notification['To'])
        self.assertIn('Rejected by archive administrator.\n\n* Foo!\n', body)