python-digitalocean/digitalocean/tests/test_baseapi.py at b8a974c760fa4d5aaa8a9e74ec507c87da7d771f · koalalorenzo/python-digitalocean · GitHub

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
import os

from requests.structures import CaseInsensitiveDict

from digitalocean.baseapi import BaseAPI
try:
    import mock
except ImportError:
    from unittest import mock

import random
import responses
import requests
import digitalocean

from .BaseTest import BaseTest


class TestBaseAPI(BaseTest):

    def setUp(self):
        super(TestBaseAPI, self).setUp()
        self.manager = digitalocean.Manager(token=self.token)
        self.user_agent = "{0}/{1} {2}/{3}".format('python-digitalocean',
                                                   digitalocean.__version__,
                                                   requests.__name__,
                                                   requests.__version__)

    @responses.activate
    def test_user_agent(self):
        data = self.load_from_file('account/account.json')

        url = self.base_url + 'account/'
        responses.add(responses.GET, url,
                      body=data,
                      status=200,
                      content_type='application/json')

        self.manager.get_account()

        self.assertEqual(responses.calls[0].request.headers['User-Agent'],
                         self.user_agent)

    @responses.activate
    def test_customize_session(self):
        data = self.load_from_file('account/account.json')
        url = self.base_url + 'account/'
        responses.add(responses.GET, url,
                      body=data,
                      status=200,
                      content_type='application/json')

        self.manager._session.proxies['https'] = 'https://127.0.0.1:3128'
        self.manager.get_account()

    def test_custom_endpoint(self):
        custom_endpoint = 'http://example.com/'

        with mock.patch.dict(os.environ,
                            {'DIGITALOCEAN_END_POINT': custom_endpoint},
                            clear=True):
            base_api = digitalocean.baseapi.BaseAPI()

            self.assertEqual(base_api.end_point, custom_endpoint)

    def test_invalid_custom_endpoint(self):
        custom_endpoint = 'not a valid endpoint'

        with mock.patch.dict(os.environ,
                            {'DIGITALOCEAN_END_POINT': custom_endpoint},
                            clear=True):
            self.assertRaises(digitalocean.EndPointError, digitalocean.baseapi.BaseAPI)

    def test_get_data_error_response_no_body(self):
        with mock.patch.object(self.manager, '_BaseAPI__perform_request') as mock_4xx_response:
            mock_4xx_response.return_value = requests.Response()
            mock_4xx_response.return_value._content = b''
            mock_4xx_response.return_value.status_code = random.randint(400, 499) # random 4xx status code

            self.assertRaises(requests.HTTPError, self.manager.get_data, 'test')

        with mock.patch.object(self.manager, '_BaseAPI__perform_request') as mock_5xx_response:
            mock_5xx_response.return_value = requests.Response()
            mock_5xx_response.return_value._content = b''
            mock_5xx_response.return_value.status_code = random.randint(500, 599) # random 5xx status code

            self.assertRaises(requests.HTTPError, self.manager.get_data, 'test')

    def test_get_data_rate_limit_case_error(self):
        with mock.patch.object(self.manager, '_BaseAPI__perform_request') as mock_429:
            mock_429.return_value = requests.Response()
            mock_429.return_value._content = b''
            mock_429.return_value.status_code = 429
            mock_429.return_value.headers = CaseInsensitiveDict(data={
                'ratelimit-limit': "1200",
                'ratelimit-remaining': "1193",
                'rateLimit-reset': "1402425459"
            })

            self.assertRaises(requests.HTTPError, self.manager.get_data, 'test')

            self.assertEqual(self.manager.ratelimit_limit, 1200)
            self.assertEqual(self.manager.ratelimit_remaining, 1193)
            self.assertEqual(self.manager.ratelimit_reset, 1402425459)

    def test_get_data_rate_limit_case_ok(self):
        with mock.patch.object(self.manager, '_BaseAPI__perform_request') as mock_200:
            mock_200.return_value = requests.Response()
            mock_200.return_value._content = b'{}'
            mock_200.return_value.status_code = 200
            mock_200.return_value.headers = CaseInsensitiveDict(data={
                'ratelimit-limit': "1200",
                'ratelimit-remaining': "1193",
                'rateLimit-reset': "1402425459"
            })

            self.manager.get_data('test')

            self.assertEqual(self.manager.ratelimit_limit, 1200)
            self.assertEqual(self.manager.ratelimit_remaining, 1193)
            self.assertEqual(self.manager.ratelimit_reset, 1402425459)