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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
import os
import shutil
import tempfile
import import_emails
from django.conf import settings
from django.test import TestCase
from patchwork.models import Patch, Person, Project, State
from linaro_metrics.models import Team, TeamCredit, TeamMembership
from linaro_metrics.sync_teams import get_or_create_person
import mock
EMAIL_DIR = os.path.join(os.path.dirname(__file__), 'data')
class ImapFake(object):
def __init__(self, emails):
self.emails = emails
def select(self, folder):
return 'OK', None
def uid(self, command, *args):
if command == 'search':
return 'OK', [' '.join(self.emails)]
elif command == 'fetch':
with open(os.path.join(EMAIL_DIR, args[0])) as f:
msg = f.read()
return 'OK', [[None, msg]]
elif command == 'copy':
# make sure an error didn't occur
assert args[1] != 'retry'
return 'OK', None
elif command == 'store':
return 'OK', None
else:
raise RuntimeError('Invalid command: ' + command)
class TestImportEmail(TestCase):
fixtures = ['default_states']
def setUp(self):
super(TestImportEmail, self).setUp()
self.tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.tmpdir)
p = Project.objects.create(listid='lng-odp.lists.linaro.org')
self.addCleanup(p.delete)
@mock.patch('import_emails.process_message')
def test_process_inbox(self, process_message):
'''Test that we can iterate through emails from fake imap server'''
mail = ImapFake(['odp_1781.mbox', 'odp_1792.mbox', 'odp_2040.mbox'])
import_emails.process_inbox(mail)
self.assertEqual(len(mail.emails), process_message.call_count)
process_message.reset_mock()
import_emails.process_inbox(mail, 2)
self.assertEqual(2, process_message.call_count)
def _import_patch(self, fname):
with open(os.path.join(EMAIL_DIR, fname)) as f:
mail = f.read()
import_emails.process_message(mail)
def test_process_message(self):
'''Test that we can process multiple version of a patch'''
self._import_patch('odp_1781.mbox')
patches = Patch.objects.all()
self.assertEqual(1, patches.count())
self.assertEqual(State.objects.get(name='New'), patches[0].state)
# now add a newer version
self._import_patch('odp_1792.mbox')
patches = Patch.objects.all()
self.assertEqual(2, patches.count())
self.assertEqual(
State.objects.get(name='Superseded'), patches[0].state)
self.assertEqual(State.objects.get(name='New'), patches[1].state)
@mock.patch('import_emails.settings')
def test_get_monkey_patcher(self, settings):
settings.PARSEMAIL_MONKEY_PATCHER = None
self.assertIsNone(import_emails.get_monkey_patcher())
settings.PARSEMAIL_MONKEY_PATCHER = 'tests.test_import_emails.ImapFake'
self.assertEqual(ImapFake, import_emails.get_monkey_patcher())
@mock.patch('linaro_metrics.parsemail.Crowd')
def test_monkey_patch_linaro_only(self, crowd):
'''Test that monkey patching rejects non-linaro patches'''
crowd().user_valid.return_value = False
Project.objects.all().delete()
with import_emails.get_monkey_patcher()(import_emails.parsemail):
self._import_patch('non_linaro.mbox')
self.assertEqual(0, Patch.objects.all().count())
@mock.patch('linaro_metrics.parsemail.Crowd')
def test_monkey_patch_author_check_user_created(self, crowd):
'''Test that we can find the author from the patch comment. The
author is linaro but doesn't exist locally'''
Project.objects.all().delete()
crowd().get_user_no_cache.return_value = {'display-name': 'User Name'}
with import_emails.get_monkey_patcher()(import_emails.parsemail):
self._import_patch('author_submitter_differ.mbox')
self.assertEqual(1, Patch.objects.all().count())
tcs = TeamCredit.objects.all()
self.assertEqual(1, tcs.count())
self.assertEqual(settings.DEFAULT_TEAM, tcs[0].team.name)
patches = Patch.objects.all()
self.assertEqual(1, patches.count())
p = Project.objects.get(linkname=settings.DEFAULT_PROJECT)
self.assertEqual(p, patches[0].project)
@mock.patch('linaro_metrics.parsemail.Crowd')
def _test_patch_auth(self, patch, author, crowd):
Project.objects.all().delete()
Person.objects.all().delete()
self.auth_found = None
def get_user_no_cache(email):
self.auth_found = email
return {'display-name': 'User Name'}
crowd().get_user_no_cache = get_user_no_cache
with import_emails.get_monkey_patcher()(import_emails.parsemail):
self._import_patch(patch)
self.assertEqual(1, Patch.objects.all().count())
self.assertEqual(author, self.auth_found)
def test_monkey_patch_author_check_user_exists(self):
'''Test that we can find the author from a patch comment.'''
self._test_patch_auth(
'author_submitter_differ.mbox', 'user.name@linaro.org')
def test_monkey_patch_encodings(self):
'''Ensure we can handle various email encodings that we receive.'''
self._test_patch_auth(
'cp8859.mbox', 'user.name@linaro.org')
self._test_patch_auth(
'odd_encoding.mbox', 'olof@lixom.net')
@mock.patch('linaro_metrics.parsemail.Crowd')
def test_monkey_patch_user_is_linaro(self, crowd):
'''A valid Linaro User may submit patches from a non-linaro looking
Person'''
Project.objects.all().delete()
def user_valid(email):
return email.endswith('@linaro.org')
crowd().user_valid = user_valid
crowd().get_user_no_cache.return_value = {'display-name': 'User Name'}
person = get_or_create_person(crowd(), 'user.name@linaro.org')
teams = [
Team.objects.create(name='foo'),
Team.objects.create(name='bar'),
]
for t in teams:
TeamMembership.objects.create(team=t, user=person.user)
Person.objects.create(email='robh@kernel.org', user=person.user)
with import_emails.get_monkey_patcher()(import_emails.parsemail):
self._import_patch('user_linaro_not_person.mbox')
self.assertEqual(1, Patch.objects.all().count())
tcs = [x.team for x in TeamCredit.objects.all()]
self.assertEqual(teams, tcs)
|
