summaryrefslogtreecommitdiff
path: root/linaro_metrics/tests/test_crowd.py
blob: 9e5cc34d95dc51f1510b6a1625edee42984fe9d9 (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
import json
import os
import tempfile
import time
import unittest

from linaro_metrics.crowd import Crowd


class TestCrowdCache(unittest.TestCase):
    def setUp(self):
        _, self.tmpfile = tempfile.mkstemp(prefix='crowdtest')
        self.addCleanup(os.unlink, self.tmpfile)
        self.crowd = Crowd('user', 'pass', 'https://foo/bar/server')

        def fake_get(api_url):
            return '''{"email": "email"}'''
        self.crowd._get = fake_get

    def test_unicode(self):
        '''Ensure we can handle unicode characters in an email address'''
        # a real commit in linux.git where the author has a unicode character:
        #  https://git.kernel.org/cgit/linux/kernel/git/torvalds/linux.git/
        #   commit/?id=e82661e23c60fc41424ca138820d729d8e4a2226
        self.crowd.get_user_no_cache(u'samuel.pitoiset\u0153gmail.com')

    def test_corrupted_cache(self):
        with open(self.tmpfile, 'w') as f:
            f.write('invalid json')
        with self.crowd.cached(self.tmpfile):
            self.assertTrue(self.crowd.user_valid('foo@bar.com'))
        with open(self.tmpfile) as f:
            self.assertIn('foo@bar.com', json.load(f))

    def test_cache_hit(self):
        data = {
            'foo@bar.com': {
                'email': 'foo',
                'valid': False,
                'expires': time.time() + 100,
            }
        }
        with open(self.tmpfile, 'w') as f:
            json.dump(data, f)
        with self.crowd.cached(self.tmpfile):
            self.assertFalse(self.crowd.user_valid('foo@bar.com'))

    def test_cache_miss(self):
        data = {
            'foo@bar.com': {
                'email': 'foo',
                'expires': time.time() - 1,
            }
        }
        with open(self.tmpfile, 'w') as f:
            json.dump(data, f)
        with self.crowd.cached(self.tmpfile):
            self.assertTrue(self.crowd.user_valid('foo@bar.com'))

    def test_cache_clean(self):
        data = {
            'foo@bar.com': {
                'email': 'email',
                'expires': time.time() - 1,
            },
            'FOO@bar.com': {
                'email': 'email',
                'expires': time.time() + 100,
            }
        }
        with open(self.tmpfile, 'w') as f:
            json.dump(data, f)
        with self.crowd.cached(self.tmpfile):
            pass
        with open(self.tmpfile) as f:
            self.assertEqual(1, len(json.load(f).keys()))