path: root/tests/test_import_emails.py

import os
import shutil
import tempfile

import import_emails

from django.conf import settings
from django.test import TestCase
from django.contrib.auth.models import User
from patchwork.models import Comment, Patch, Person, Project, State
from linaro_metrics.models import Team, TeamCredit, TeamMembership

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 tearDown(self):
        super(TestImportEmail, self).tearDown()
        Person.objects.all().delete()
        User.objects.all().delete()

    def setUp(self):
        super(TestImportEmail, self).setUp()

        self.tmpdir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.tmpdir)

        self.user = User.objects.create(
            email='user.name@linaro.org',
            username='user.name',
            first_name='User',
            last_name='Name',
            is_active=True
        )

        self.user2 = User.objects.create(
            email='peter.maydell@linaro.org',
            username='peter.maydell',
            first_name='Peter',
            last_name='Maydell',
            is_active=True
        )

        self.person = Person.objects.create(
            email='user.name@linaro.org',
            name='User Name',
            user=self.user
        )

        self.person2 = Person.objects.create(
            email='user.name@kernel.org',
            name='User Name',
            user=self.user
        )

        self.person3 = Person.objects.create(
            email='peter.maydell@linaro.org',
            name='Peter Maydell',
            user=self.user2
        )

        self.person_other = Person.objects.create(
            email='robh@kernel.org',
            name='Rob Herring',
            user=self.user
        )

        self.person_olof = Person.objects.create(
            email='olof@lixom.net',
            name='Olof Johansson',
            user=self.user2
        )

        self.notlinaro_person = Person.objects.create(
            email='not@notlinaro.org',
            name='Not Linaro',
            user=None
        )

        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())

    def test_monkey_patch_linaro_only(self):
        '''Test that monkey patching rejects non-linaro patches'''
        Project.objects.all().delete()
        with import_emails.get_monkey_patcher()(import_emails.parser):
            self._import_patch('non_linaro.mbox')
            self.assertEqual(0, Patch.objects.all().count())

    def test_monkey_patch_author_check_user_created(self):
        '''Test that we can find the author from the patch comment. The
        author is linaro but doesn't exist locally'''
        Project.objects.all().delete()
        # junk
        self.assertEqual(1, User.objects.filter(username='user.name').count())
        self.assertEqual(
            Person.objects.filter(email='user.name@linaro.org').first().user,
            User.objects.filter(username='user.name').first()
        )
        # junk

        with import_emails.get_monkey_patcher()(import_emails.parser):
            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)

    def test_monkey_patch_submitter_is_linaro(self):
        '''A valid Linaro User may submit patches on behalf of a user. We have
        users that want to track these patches in our instance, but we
        SHOULD NOT contribute these to "team credits"'''
        Project.objects.all().delete()
        with import_emails.get_monkey_patcher()(import_emails.parser):
            self._import_patch('applied-patch.mbox')
            self.assertEqual(0, Patch.objects.all().count())
            tcs = TeamCredit.objects.all()
            self.assertEqual(0, tcs.count())

    def test_monkey_patch_maintainer_applied(self):
        '''Don't give a patch credit to a maintainer applying a patch to a
        tree'''

        Project.objects.all().delete()
        with import_emails.get_monkey_patcher()(import_emails.parser):
            self._import_patch('author_not_linaro.mbox')
            self.assertEqual(1, Patch.objects.all().count())
            tcs = TeamCredit.objects.all()
            self.assertEqual(0, tcs.count())

            patches = Patch.objects.all()
            self.assertEqual(1, patches.count())
            p = Project.objects.get(linkname=settings.DEFAULT_PROJECT)
            self.assertEqual(p, patches[0].project)

    def _test_patch_auth(self, patch, author):
        Project.objects.all().delete()
        self.auth_found = None

        person = Person.objects.filter(email=author)
        if person.count() > 0:
            self.auth_found = person[0].email

        with import_emails.get_monkey_patcher()(import_emails.parser):
            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')

    def test_monkey_patch_comment(self):
        '''Ensure non-linaro commments are added to Linaro patches'''
        self._test_patch_auth(
            'cp8859.mbox', 'user.name@linaro.org')

        with import_emails.get_monkey_patcher()(import_emails.parser):
            self._import_patch('cp8859_comment.mbox')
            self.assertEqual(1, Comment.objects.all().count())

    def test_monkey_patch_user_is_linaro(self):
        '''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')

        teams = [
            Team.objects.create(name='foo'),
            Team.objects.create(name='bar'),
        ]
        for t in teams:
            TeamMembership.objects.create(team=t, user=self.user)

        with import_emails.get_monkey_patcher()(import_emails.parser):
            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)

    def test_monkey_patch_no_listid(self):
        Project.objects.all().delete()
        qemu = Project.objects.create(
            name='qemu-devel', linkname='qemu-devel',
            listemail='qemu-devel@nongnu.org', listid='qemu-devel.nongnu.org')

        with import_emails.get_monkey_patcher()(import_emails.parser):
            self._import_patch('no-list-id.mbox')
            self.assertEqual(1, Patch.objects.all().count())
            self.assertEqual(qemu, Patch.objects.all()[0].project)