summaryrefslogtreecommitdiff
path: root/linaro_metrics/tests/test_sync_gerrit_changes.py
blob: 075227ea913276654a62fbf0d8c51ee4c7d977e4 (plain)
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
import json
import mock

from datetime import datetime

from django.contrib.auth.models import User
from django.test import TestCase

from patchwork.models import Patch, Person, Project, State

from linaro_metrics import sync_gerrit_changes
from linaro_metrics.models import Team, TeamCredit, TeamMembership


class TestSyncGerritChanges(TestCase):
    fixtures = ['default_states']

    @mock.patch('urllib2.urlopen')
    def test_get_user_changes_simple(self, urlopen):
        resp = mock.Mock()
        items = [
            {'foo': 'bar'}, {'bam': 'bang'},
        ]
        resp.read.return_value = ")]}'" + json.dumps(items)
        urlopen.return_value = resp
        changes = list(sync_gerrit_changes.get_user_changes('foo@bar.com', ''))
        self.assertEqual(items, changes)

    @mock.patch('urllib2.urlopen')
    def test_get_user_changes_continue(self, urlopen):
        resp = mock.Mock()
        responses = [
            [{'foo': 'bar'}, {'bam': 'bang', '_more_changes': True}],
            [{'x': 'y'}],
        ]

        def fake_read():
            r = responses[0]
            responses.remove(r)
            return ")]}'" + json.dumps(r)
        resp.read = fake_read
        urlopen.return_value = resp
        changes = list(sync_gerrit_changes.get_user_changes('foo@bar.com', ''))
        self.assertEqual(3, len(changes))

    def test_get_patch_content(self):
        url_base = sync_gerrit_changes.GERRIT_PROJECTS['AOSP']
        c = sync_gerrit_changes.get_patch_content(
            url_base, 'AOSP', {'project': 'foo', '_number': 123})
        self.assertIn(url_base, c)

    def test_patch_NEW_UPDATE(self):
        project = 'AOSP'
        url_base = sync_gerrit_changes.GERRIT_PROJECTS[project]
        u = User.objects.create(email='foo@foo.org')
        p = Person.objects.create(email='foo@bar.org', user=u)
        t = Team.objects.create(name='foo team')
        TeamMembership.objects.create(team=t, user=u)
        Project.objects.create(name=project)

        # create a patch
        change = {
            '_number': 123,
            'change_id': 'CHANGE_ID',
            'subject': 'patch subject',
            'project': 'test_proj',
            'status': 'NEW',
            'created': '2015-08-09 16:48:08.366000000',
            'updated': '2015-08-09 16:48:08.366000000',
        }
        self.assertTrue(sync_gerrit_changes.create_or_update(
            url_base, project, p.email, change))
        msgid = '%s@%s' % (change['_number'], change['change_id'])
        patch = Patch.objects.get(msgid=msgid)
        self.assertEqual(State.objects.get(name='New'), patch.state)
        t = datetime.strptime(change['created'], '%Y-%m-%d %H:%M:%S.%f000')
        self.assertEqual(t, patch.date)
        t = datetime.strptime(change['updated'], '%Y-%m-%d %H:%M:%S.%f000')

        tcs = TeamCredit.objects.all()
        self.assertEqual(1, tcs.count())
        self.assertEqual(t, tcs[0].last_state_change)

        # make sure we can handle no change
        self.assertFalse(sync_gerrit_changes.create_or_update(
            url_base, project, p.email, change))

        # now update the patch
        change['status'] = 'MERGED'
        change['updated'] = '2015-08-10 16:48:08.366000000'
        self.assertTrue(sync_gerrit_changes.create_or_update(
            url_base, project, p.email, change))
        patch = Patch.objects.get(msgid=msgid)
        self.assertEqual(State.objects.get(name='Accepted'), patch.state)
        t = datetime.strptime(change['created'], '%Y-%m-%d %H:%M:%S.%f000')
        self.assertEqual(t, patch.date)
        t = datetime.strptime(change['updated'], '%Y-%m-%d %H:%M:%S.%f000')

        tcs = TeamCredit.objects.all()
        self.assertEqual(1, tcs.count())
        self.assertEqual(t, tcs[0].last_state_change)