Lubuntu
/
britney2-ubuntu
mirror of https://github.com/lubuntu-team/britney2-ubuntu.git
8
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '858775a3c1'
${ noResults }
britney2-ubuntu/tests/test_email.py

400 lines
15 KiB
Raw Blame History

#!/usr/bin/python3
# (C) 2017 Canonical Ltd.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
from collections import defaultdict
import fileinput
import json
import os
import pprint
import sys
import unittest
import yaml
from unittest.mock import DEFAULT, patch, call
PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, PROJECT_DIR)
from britney2.policies.policy import PolicyVerdict
from britney2.policies.email import EmailPolicy, person_chooser, address_chooser
from tests.test_sourceppa import FakeOptions
from tests import TestBase
from tests.mock_smtpd import FakeSMTPServer
# Example of a direct upload by core dev: openstack-doc-tools 1.5.0-0ubuntu1
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7524835
UPLOAD = dict(
creator_link=None,
package_creator_link='https://api.launchpad.net/1.0/~zulcss',
package_signer_link='https://api.launchpad.net/1.0/~zulcss',
sponsor_link=None,
)
# Example of a sponsored upload: kerneloops 0.12+git20140509-2ubuntu1
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7013597
SPONSORED_UPLOAD = dict(
creator_link=None,
package_creator_link='https://api.launchpad.net/1.0/~smb',
package_signer_link='https://api.launchpad.net/1.0/~apw',
sponsor_link=None,
)
# Example of a bileto upload: autopilot 1.6.0+17.04.20170302-0ubuntu1
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7525085
# (dobey clicked 'build' and sil2100 clicked 'publish')
BILETO = dict(
creator_link='https://api.launchpad.net/1.0/~sil2100',
package_creator_link='https://api.launchpad.net/1.0/~dobey',
package_signer_link='https://api.launchpad.net/1.0/~ci-train-bot',
sponsor_link='https://api.launchpad.net/1.0/~ubuntu-archive-robot',
)
# Example of a non-sponsored copy: linux 4.10.0-11.13
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7522481
# (the upload to the PPA was sponsored but the copy was done directly)
UNSPONSORED_COPY = dict(
creator_link='https://api.launchpad.net/1.0/~timg-tpi',
package_creator_link='https://api.launchpad.net/1.0/~sforshee',
package_signer_link='https://api.launchpad.net/1.0/~timg-tpi',
sponsor_link=None,
)
# Example of a sponsored copy: pagein 0.00.03-1
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7533336
SPONSORED_COPY = dict(
creator_link='https://api.launchpad.net/1.0/~colin-king',
package_creator_link='https://api.launchpad.net/1.0/~colin-king',
package_signer_link=None,
sponsor_link='https://api.launchpad.net/1.0/~mapreri',
)
# Example of a manual debian sync: systemd 232-19
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7522736
MANUAL_SYNC = dict(
creator_link='https://api.launchpad.net/1.0/~costamagnagianfranco',
package_creator_link='https://api.launchpad.net/1.0/~pkg-systemd-maintainers',
package_signer_link=None,
sponsor_link=None,
)
# Example of a sponsored manual debian sync: python-pymysql 0.7.9-2
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7487820
SPONSORED_MANUAL_SYNC = dict(
creator_link='https://api.launchpad.net/1.0/~lars-tangvald',
package_creator_link='https://api.launchpad.net/1.0/~openstack-1.0',
package_signer_link=None,
sponsor_link='https://api.launchpad.net/1.0/~racb',
)
# Example of an automatic debian sync: gem2deb 0.33.1
# https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/7255529
AUTO_SYNC = dict(
creator_link='https://api.launchpad.net/1.0/~katie',
package_creator_link='https://api.launchpad.net/1.0/~pkg-ruby-extras-maintainers',
package_signer_link=None,
sponsor_link='https://api.launchpad.net/1.0/~ubuntu-archive-robot',
)
# address lists
UBUNTU = ['personal@gmail.com', 'ubuntu@ubuntu.com', 'work@canonical.com']
CANONICAL = ['personal@gmail.com', 'work@canonical.com']
COMMUNITY = ['personal@gmail.com', 'other@gmail.com']
def retvals(retvals):
"""Return different retvals on different calls of mock."""
def returner(*args, **kwargs):
return retvals.pop()
return returner
class FakeSourceData:
version = '55.0'
class FakeExcuse:
is_valid = True
daysold = 0
reason = []
current_policy_verdict = PolicyVerdict.PASS
class T(unittest.TestCase):
maxDiff = None
def test_person_chooser(self):
"""Find the correct person to blame for an upload."""
self.assertEqual(person_chooser(UPLOAD), {
'https://api.launchpad.net/1.0/~zulcss',
})
self.assertEqual(person_chooser(SPONSORED_UPLOAD), {
'https://api.launchpad.net/1.0/~apw',
'https://api.launchpad.net/1.0/~smb'
})
self.assertEqual(person_chooser(BILETO), {
'https://api.launchpad.net/1.0/~dobey',
'https://api.launchpad.net/1.0/~sil2100',
})
self.assertEqual(person_chooser(UNSPONSORED_COPY), {
'https://api.launchpad.net/1.0/~timg-tpi',
})
self.assertEqual(person_chooser(SPONSORED_COPY), {
'https://api.launchpad.net/1.0/~colin-king',
'https://api.launchpad.net/1.0/~mapreri',
})
self.assertEqual(person_chooser(MANUAL_SYNC), {
'https://api.launchpad.net/1.0/~costamagnagianfranco',
})
self.assertSequenceEqual(person_chooser(SPONSORED_MANUAL_SYNC), {
'https://api.launchpad.net/1.0/~lars-tangvald',
'https://api.launchpad.net/1.0/~racb',
})
self.assertEqual(person_chooser(AUTO_SYNC), set())
def test_address_chooser(self):
"""Prioritize email addresses correctly."""
self.assertEqual(address_chooser(UBUNTU), 'ubuntu@ubuntu.com')
self.assertEqual(address_chooser(CANONICAL), 'work@canonical.com')
self.assertEqual(address_chooser(COMMUNITY), 'personal@gmail.com')
@patch('britney2.policies.email.EmailPolicy.query_rest_api')
@patch('britney2.policies.email.EmailPolicy.query_lp_rest_api')
def test_email_scraping(self, lp, rest):
"""Poke correct REST APIs to find email addresses."""
lp.side_effect = retvals([
dict(entries=[dict(fingerprint='DEFACED_ED1F1CE')]),
dict(entries=[UPLOAD]),
])
rest.return_value = 'uid:Defaced Edifice <ex@example.com>:12345::'
e = EmailPolicy(FakeOptions, None)
self.assertEqual(e.lp_get_emails('openstack-doct-tools', '1.5.0-0ubuntu1'), ['ex@example.com'])
self.assertSequenceEqual(lp.mock_calls, [
call('testbuntu/+archive/primary', {
'distro_series': '/testbuntu/zazzy',
'exact_match': 'true',
'order_by_date': 'true',
'pocket': 'Proposed',
'source_name': 'openstack-doct-tools',
'version': '1.5.0-0ubuntu1',
'ws.op': 'getPublishedSources',
}),
call('https://api.launchpad.net/1.0/~zulcss/gpg_keys', {})
])
self.assertSequenceEqual(rest.mock_calls, [
call('http://keyserver.ubuntu.com/pks/lookup', {
'exact': 'on',
'op': 'index',
'options': 'mr',
'search': '0xDEFACED_ED1F1CE',
})
])
@patch('britney2.policies.email.EmailPolicy.lp_get_emails')
@patch('britney2.policies.email.smtplib')
def test_smtp_not_sent(self, smtp, lp):
"""Know when not to send any emails."""
lp.return_value = ['example@email.com']
e = EmailPolicy(FakeOptions, None)
FakeExcuse.daysold = 0.002
e.apply_policy_impl(None, None, 'chromium-browser', None, FakeSourceData, FakeExcuse)
FakeExcuse.daysold = 2.98
e.apply_policy_impl(None, None, 'chromium-browser', None, FakeSourceData, FakeExcuse)
# Would email but no address found
FakeExcuse.daysold = 10.12
lp.return_value = []
e.apply_policy_impl(None, None, 'chromium-browser', None, FakeSourceData, FakeExcuse)
self.assertEqual(smtp.mock_calls, [])
@patch('britney2.policies.email.EmailPolicy.lp_get_emails')
@patch('britney2.policies.email.smtplib')
def test_smtp_sent(self, smtp, lp):
"""Send emails correctly."""
lp.return_value = ['email@address.com']
e = EmailPolicy(FakeOptions, None)
FakeExcuse.is_valid = False
FakeExcuse.daysold = 100
e.apply_policy_impl(None, None, 'chromium-browser', None, FakeSourceData, FakeExcuse)
smtp.SMTP.assert_called_once_with('localhost')
@patch('britney2.policies.email.EmailPolicy.lp_get_emails')
@patch('britney2.policies.email.smtplib', autospec=True)
def test_smtp_repetition(self, smtp, lp):
"""Resend mails periodically, with decreasing frequency."""
lp.return_value = ['email@address.com']
sendmail = smtp.SMTP().sendmail
e = EmailPolicy(FakeOptions, None)
called = []
e.cache = {}
for hours in range(0, 5000):
previous = sendmail.call_count
age = hours / 24
FakeExcuse.daysold = age
e.apply_policy_impl(None, None, 'unity8', None, FakeSourceData, FakeExcuse)
if sendmail.call_count > previous:
e.initialise(None) # Refill e.cache from disk
called.append(age)
# Emails were sent when daysold reached these values:
self.assertSequenceEqual(called, [
3.0, 6.0, 12.0, 24.0, 48.0, 78.0, 108.0, 138.0, 168.0, 198.0
])
class ET(TestBase):
''' Test sending mail through a mocked SMTP server '''
@classmethod
def setUpClass(cls):
cls.smtpd = FakeSMTPServer('localhost', 1337)
cls.smtpd.run()
@classmethod
def tearDownClass(cls):
cls.smtpd.close()
def setUp(self):
super().setUp()
# disable ADT, not relevant for us
for line in fileinput.input(self.britney_conf, inplace=True):
if line.startswith('ADT_ENABLE'):
print('ADT_ENABLE = no')
elif line.startswith('MINDAYS_EMERGENCY'):
print('MINDAYS_EMERGENCY = 10')
elif not line.startswith('ADT_AMQP') and not line.startswith('ADT_SWIFT_URL'):
sys.stdout.write(line)
# and set up a fake smtpd
with open(self.britney_conf, 'a') as f:
f.write('EMAIL_HOST = localhost:1337')
self.sourceppa_cache = {}
self.email_cache = {}
self.data.add('libc6', False)
def do_test(self, unstable_add, expect_emails):
'''Run britney with some unstable packages and verify excuses.
unstable_add is a list of (binpkgname, field_dict, daysold, emails)
expect_emails is a list that is checked against the emails sent during
this do_test run.
Return (output, excuses_dict, excuses_html, emails).
'''
ET.smtpd.emails.clear()
age_file = os.path.join(self.data.path,
'data',
'series',
'Dates')
email_cache_file = os.path.join(self.data.path,
'data',
'series-proposed',
'EmailCache')
for (pkg, fields, daysold, emails) in unstable_add:
self.data.add(pkg, True, fields, True, None)
self.sourceppa_cache.setdefault(pkg, {})
if fields['Version'] not in self.sourceppa_cache[pkg]:
self.sourceppa_cache[pkg][fields['Version']] = ''
with open(age_file, 'w') as f:
import time
do = time.time() - (60 * 60 * 24 * daysold)
f.write('%s %s %d' % (pkg, fields['Version'], do))
with open(email_cache_file, 'w') as f:
d = defaultdict(dict)
d[pkg][fields['Version']] = (emails, 0)
f.write(json.dumps(d))
# Set up sourceppa cache for testing
sourceppa_path = os.path.join(self.data.dirs[True], 'SourcePPA')
with open(sourceppa_path, 'w', encoding='utf-8') as sourceppa:
sourceppa.write(json.dumps(self.sourceppa_cache))
(excuses_yaml, excuses_html, out) = self.run_britney()
# convert excuses to source indexed dict
excuses_dict = {}
for s in yaml.load(excuses_yaml)['sources']:
excuses_dict[s['source']] = s
if 'SHOW_EXCUSES' in os.environ:
print('------- excuses -----')
pprint.pprint(excuses_dict, width=200)
if 'SHOW_HTML' in os.environ:
print('------- excuses.html -----\n%s\n' % excuses_html)
if 'SHOW_OUTPUT' in os.environ:
print('------- output -----\n%s\n' % out)
self.assertNotIn('FIXME', out)
# check all the emails that we asked for are there
for email in expect_emails:
self.assertIn(email, ET.smtpd.get_emails())
self.assertEqual(len(ET.smtpd.get_emails()), len(expect_emails))
return (out, excuses_dict, excuses_html, ET.smtpd.emails)
def test_email_sent(self):
'''Test that an email is sent through the SMTP server'''
pkg = ('libc6', {'Version': '2',
'Depends': 'notavailable (>= 2)'},
6,
['foo@bar.com'])
self.do_test([pkg], ["foo@bar.com"])
def test_email_not_sent_block_all_source(self):
'''Test that an email is not sent if the package is blocked by a
block-all source hint'''
self.create_hint('freeze', 'block-all source')
pkg = ('libc6', {'Version': '2'},
6, # daysold
['foo@bar.com'])
self.do_test([pkg], [])
def test_email_not_sent_blocked(self):
'''Test that an email is not sent if the package is blocked by a block hint'''
self.create_hint('freeze', 'block libc6')
pkg = ('libc6', {'Version': '2'},
6, # daysold
['foo@bar.com'])
self.do_test([pkg], [])
def test_email_sent_unblocked(self):
'''Test that an email is sent if the package is unblocked'''
self.create_hint('freeze', 'block libc6')
self.create_hint('laney', 'unblock libc6/2')
pkg = ('libc6', {'Version': '2',
'Depends': 'notavailable (>= 2)'},
6, # daysold
['foo@bar.com'])
self.do_test([pkg], ['foo@bar.com'])
def test_email_not_sent_rejected_temporarily(self):
'''Test that an email is not sent if the package is REJECTED_TEMPORARILY'''
urgency_file = os.path.join(self.data.path,
'data',
'series',
'Urgency')
with open(urgency_file, 'w') as f:
# we specified in setUp() that emergency has a 10 day delay, and
# age rejections are REJECTED_TEMPORARILY
f.write('libc6 2 emergency')
pkg = ('libc6', {'Version': '2',
'Depends': 'notavailable (>= 2)'},
6, # daysold
['foo@bar.com'])
self.do_test([pkg], [])
if __name__ == '__main__':
unittest.main()
Reference in new issue View Git Blame Copy Permalink