File: test_utils.py

package info (click to toggle)
python-discogs-client 2.7.1-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 568 kB
  • sloc: python: 1,685; sh: 83; makefile: 18
file content (135 lines) | stat: -rw-r--r-- 4,742 bytes parent folder | download
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
import unittest
from unittest.mock import patch, MagicMock
from discogs_client.exceptions import TooManyAttemptsError
from datetime import datetime, timezone, timedelta
from . import DiscogsClientTestCase
from discogs_client import utils
from dateutil.tz import tzutc, tzoffset


class UtilsTestCase(DiscogsClientTestCase):
    def test_update_qs(self):
        """update_qs helper works as intended"""
        u = utils.update_qs
        self.assertEqual(u('http://example.com', {'foo': 'bar'}), 'http://example.com?foo=bar')
        self.assertEqual(u('http://example.com?foo=bar', {'foo': 'baz'}), 'http://example.com?foo=bar&foo=baz')
        # be careful for dict iteration order is not deterministic
        result = u('http://example.com?c=3&a=yep', {'a': 1, 'b': '1'})
        try:
            self.assertEqual(result, 'http://example.com?c=3&a=yep&a=1&b=1')
        except AssertionError:
            self.assertEqual(result, 'http://example.com?c=3&a=yep&b=1&a=1')

        self.assertEqual(u('http://example.com', {'a': 't\xe9st'}),
                         'http://example.com?a=t%C3%A9st')

    def test_omit_none(self):
        o = utils.omit_none
        self.assertEqual(o({
            'foo': None,
            'baz': 'bat',
            'qux': None,
            'flan': 0,
        }), {
            'baz': 'bat',
            'flan': 0,
        })

        self.assertEqual(o({k: None for k in ('qux', 'quux', 'quuux')}), {})
        self.assertEqual(o({'nope': 'yep'}), {'nope': 'yep'})
        self.assertEqual(o({}), {})

    def test_parse_timestamp(self):
        p = utils.parse_timestamp
        self.assertEqual(
            p('2016-07-27T08:11:29-07:00'),
            datetime(2016, 7, 27, 8, 11, 29, tzinfo=tzoffset(None, -25200))
        )
        self.assertEqual(
            p('2055-07-27T08:11:29-07:00'),
            datetime(2055, 7, 27, 8, 11, 29, tzinfo=timezone(timedelta(hours=-7)))
        )
        self.assertEqual(
            p('1930-07-27T08:11:29-00:00'),
            datetime(1930, 7, 27, 8, 11, 29, tzinfo=timezone.utc)
        )
        self.assertEqual(
            p('1930-07-27T08:11:29-00:00'),
            datetime(1930, 7, 27, 8, 11, 29, tzinfo=tzutc())
        )

    def test_condition(self):
        self.assertRaises(TypeError, lambda: utils.Condition())
        self.assertEqual(utils.Condition.MINT, 'Mint (M)')
        self.assertEqual(utils.Condition.NEAR_MINT, 'Near Mint (NM or M-)')

    def test_status(self):
        self.assertRaises(TypeError, lambda: utils.Status())
        self.assertEqual(utils.Status.DRAFT, 'Draft')
        self.assertEqual(utils.Status.FOR_SALE, 'For Sale')

    def test_sort(self):
        self.assertRaises(TypeError, lambda: utils.Sort())


    @patch('discogs_client.utils.get_backoff_duration')
    def test_backed_off_when_rate_limit_reached(self, patched_duration):
        # Mock sleep duration returned so it doesn't effect tests speed
        patched_duration.return_value = 0

        backoff = utils.backoff

        mock_ratelimited_response = MagicMock()
        mock_ratelimited_response.status_code = 429

        mock_ok_response = MagicMock()
        mock_ok_response.status_code = 200

        call_count = 0

        class BackoffTestClass:
            def __init__(self):
                self.backoff_enabled = True

            @backoff
            def always_fails(self):
                return mock_ratelimited_response

            @backoff
            def returns_non_ratelimit_status_code(self):
                return mock_ok_response

            @backoff
            def succeeds_after_x_calls(self):
                nonlocal call_count
                call_count += 1
                if call_count < 3:
                    return mock_ratelimited_response
                else:
                    return mock_ok_response

            @backoff
            def function_not_decorated_when_disabled(self):
                return mock_ratelimited_response

        test_class = BackoffTestClass()

        with self.assertRaises(TooManyAttemptsError):
            test_class.always_fails()

        self.assertEqual(mock_ok_response, test_class.returns_non_ratelimit_status_code())
        self.assertEqual(mock_ok_response, test_class.succeeds_after_x_calls())

        with patch('discogs_client.utils.get_backoff_duration') as patched_duration:
            test_class.backoff_enabled = False
            test_class.function_not_decorated_when_disabled()
            self.assertEqual(0, patched_duration.call_count)


def suite():
    suite = unittest.TestSuite()
    suite = unittest.TestLoader().loadTestsFromTestCase(UtilsTestCase)
    return suite

if __name__ == '__main__':
    unittest.main(defaultTest='suite')