← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python-keystoneclient/trusty-proposed

« back to all changes in this revision

Viewing changes to .pc/fix-requests-requirements.patch/tests/v3/utils.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2013-02-11 12:32:43 UTC
  • Revision ID: package-import@ubuntu.com-20130211123243-vf4o7xig8e4lbijf
Tags: 1:0.2.2-0ubuntu2
debian/patches/fix-requests-requirements.patch: Bump python-requests
to 1.1.

Show diffs side-by-side

added added

removed removed

Lines of Context:
.pc/fix-requests-requirements.patch/tests/v3/utils.py
 
1
import copy
 
2
import json
 
3
import uuid
 
4
import time
 
5
import urlparse
 
6
 
 
7
import mox
 
8
import requests
 
9
import testtools
 
10
 
 
11
from keystoneclient.v3 import client
 
12
 
 
13
 
 
14
def parameterize(ref):
 
15
    """Rewrites attributes to match the kwarg naming convention in client.
 
16
 
 
17
    >>> paramterize({'project_id': 0})
 
18
    {'project': 0}
 
19
 
 
20
    """
 
21
    params = ref.copy()
 
22
    for key in ref:
 
23
        if key[-3:] == '_id':
 
24
            params.setdefault(key[:-3], params.pop(key))
 
25
    return params
 
26
 
 
27
 
 
28
class TestClient(client.Client):
 
29
 
 
30
    def serialize(self, entity):
 
31
        return json.dumps(entity, sort_keys=True)
 
32
 
 
33
 
 
34
class TestCase(testtools.TestCase):
 
35
    TEST_TENANT_NAME = 'aTenant'
 
36
    TEST_TOKEN = 'aToken'
 
37
    TEST_USER = 'test'
 
38
    TEST_ROOT_URL = 'http://127.0.0.1:5000/'
 
39
    TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3')
 
40
    TEST_ROOT_ADMIN_URL = 'http://127.0.0.1:35357/'
 
41
    TEST_ADMIN_URL = '%s%s' % (TEST_ROOT_ADMIN_URL, 'v3')
 
42
    TEST_REQUEST_BASE = {
 
43
        'config': {'danger_mode': False},
 
44
        'verify': True,
 
45
    }
 
46
 
 
47
    def setUp(self):
 
48
        super(TestCase, self).setUp()
 
49
        self.mox = mox.Mox()
 
50
        self._original_time = time.time
 
51
        time.time = lambda: 1234
 
52
        requests.request = self.mox.CreateMockAnything()
 
53
        self.client = TestClient(username=self.TEST_USER,
 
54
                                 token=self.TEST_TOKEN,
 
55
                                 tenant_name=self.TEST_TENANT_NAME,
 
56
                                 auth_url=self.TEST_URL,
 
57
                                 endpoint=self.TEST_URL)
 
58
 
 
59
    def tearDown(self):
 
60
        time.time = self._original_time
 
61
        self.mox.UnsetStubs()
 
62
        self.mox.VerifyAll()
 
63
        super(TestCase, self).tearDown()
 
64
 
 
65
 
 
66
class UnauthenticatedTestCase(testtools.TestCase):
 
67
    """ Class used as base for unauthenticated calls """
 
68
    TEST_ROOT_URL = 'http://127.0.0.1:5000/'
 
69
    TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3')
 
70
    TEST_ROOT_ADMIN_URL = 'http://127.0.0.1:35357/'
 
71
    TEST_ADMIN_URL = '%s%s' % (TEST_ROOT_ADMIN_URL, 'v3')
 
72
    TEST_REQUEST_BASE = {
 
73
        'config': {'danger_mode': False},
 
74
        'verify': True,
 
75
    }
 
76
 
 
77
    def setUp(self):
 
78
        super(UnauthenticatedTestCase, self).setUp()
 
79
        self.mox = mox.Mox()
 
80
        self._original_time = time.time
 
81
        time.time = lambda: 1234
 
82
        requests.request = self.mox.CreateMockAnything()
 
83
 
 
84
    def tearDown(self):
 
85
        time.time = self._original_time
 
86
        self.mox.UnsetStubs()
 
87
        self.mox.VerifyAll()
 
88
        super(UnauthenticatedTestCase, self).tearDown()
 
89
 
 
90
 
 
91
class CrudTests(testtools.TestCase):
 
92
    key = None
 
93
    collection_key = None
 
94
    model = None
 
95
    manager = None
 
96
 
 
97
    def new_ref(self, **kwargs):
 
98
        kwargs.setdefault('id', uuid.uuid4().hex)
 
99
        return kwargs
 
100
 
 
101
    def additionalSetUp(self):
 
102
        self.headers = {
 
103
            'GET': {
 
104
                'X-Auth-Token': 'aToken',
 
105
                'User-Agent': 'python-keystoneclient',
 
106
            }
 
107
        }
 
108
 
 
109
        self.headers['HEAD'] = self.headers['GET'].copy()
 
110
        self.headers['DELETE'] = self.headers['GET'].copy()
 
111
        self.headers['PUT'] = self.headers['GET'].copy()
 
112
        self.headers['POST'] = self.headers['GET'].copy()
 
113
        self.headers['POST']['Content-Type'] = 'application/json'
 
114
        self.headers['PATCH'] = self.headers['POST'].copy()
 
115
 
 
116
    def serialize(self, entity):
 
117
        if isinstance(entity, dict):
 
118
            return json.dumps({self.key: entity}, sort_keys=True)
 
119
        if isinstance(entity, list):
 
120
            return json.dumps({self.collection_key: entity}, sort_keys=True)
 
121
        raise NotImplementedError('Are you sure you want to serialize that?')
 
122
 
 
123
    def test_create(self, ref=None):
 
124
        ref = ref or self.new_ref()
 
125
        resp = TestResponse({
 
126
            "status_code": 201,
 
127
            "text": self.serialize(ref),
 
128
        })
 
129
 
 
130
        method = 'POST'
 
131
        req_ref = ref.copy()
 
132
        req_ref.pop('id')
 
133
        kwargs = copy.copy(self.TEST_REQUEST_BASE)
 
134
        kwargs['headers'] = self.headers[method]
 
135
        kwargs['data'] = self.serialize(req_ref)
 
136
        requests.request(
 
137
            method,
 
138
            urlparse.urljoin(
 
139
                self.TEST_URL,
 
140
                'v3/%s' % self.collection_key),
 
141
            **kwargs).AndReturn((resp))
 
142
        self.mox.ReplayAll()
 
143
 
 
144
        returned = self.manager.create(**parameterize(req_ref))
 
145
        self.assertTrue(isinstance(returned, self.model))
 
146
        for attr in ref:
 
147
            self.assertEqual(
 
148
                getattr(returned, attr),
 
149
                ref[attr],
 
150
                'Expected different %s' % attr)
 
151
 
 
152
    def test_get(self, ref=None):
 
153
        ref = ref or self.new_ref()
 
154
        resp = TestResponse({
 
155
            "status_code": 200,
 
156
            "text": self.serialize(ref),
 
157
        })
 
158
 
 
159
        method = 'GET'
 
160
        kwargs = copy.copy(self.TEST_REQUEST_BASE)
 
161
        kwargs['headers'] = self.headers[method]
 
162
        requests.request(
 
163
            method,
 
164
            urlparse.urljoin(
 
165
                self.TEST_URL,
 
166
                'v3/%s/%s' % (self.collection_key, ref['id'])),
 
167
            **kwargs).AndReturn((resp))
 
168
        self.mox.ReplayAll()
 
169
 
 
170
        returned = self.manager.get(ref['id'])
 
171
        self.assertTrue(isinstance(returned, self.model))
 
172
        for attr in ref:
 
173
            self.assertEqual(
 
174
                getattr(returned, attr),
 
175
                ref[attr],
 
176
                'Expected different %s' % attr)
 
177
 
 
178
    def test_list(self, ref_list=None, expected_path=None, **filter_kwargs):
 
179
        ref_list = ref_list or [self.new_ref(), self.new_ref()]
 
180
        resp = TestResponse({
 
181
            "status_code": 200,
 
182
            "text": self.serialize(ref_list),
 
183
        })
 
184
 
 
185
        method = 'GET'
 
186
        kwargs = copy.copy(self.TEST_REQUEST_BASE)
 
187
        kwargs['headers'] = self.headers[method]
 
188
        requests.request(
 
189
            method,
 
190
            urlparse.urljoin(
 
191
                self.TEST_URL,
 
192
                expected_path or 'v3/%s' % self.collection_key),
 
193
            **kwargs).AndReturn((resp))
 
194
        self.mox.ReplayAll()
 
195
 
 
196
        returned_list = self.manager.list(**filter_kwargs)
 
197
        self.assertTrue(len(returned_list))
 
198
        [self.assertTrue(isinstance(r, self.model)) for r in returned_list]
 
199
 
 
200
    def test_update(self, ref=None):
 
201
        ref = ref or self.new_ref()
 
202
        req_ref = ref.copy()
 
203
        del req_ref['id']
 
204
        resp = TestResponse({
 
205
            "status_code": 200,
 
206
            "text": self.serialize(ref),
 
207
        })
 
208
 
 
209
        method = 'PATCH'
 
210
        kwargs = copy.copy(self.TEST_REQUEST_BASE)
 
211
        kwargs['headers'] = self.headers[method]
 
212
        kwargs['data'] = self.serialize(req_ref)
 
213
        requests.request(
 
214
            method,
 
215
            urlparse.urljoin(
 
216
                self.TEST_URL,
 
217
                'v3/%s/%s' % (self.collection_key, ref['id'])),
 
218
            **kwargs).AndReturn((resp))
 
219
        self.mox.ReplayAll()
 
220
 
 
221
        returned = self.manager.update(ref['id'], **parameterize(req_ref))
 
222
        self.assertTrue(isinstance(returned, self.model))
 
223
        for attr in ref:
 
224
            self.assertEqual(
 
225
                getattr(returned, attr),
 
226
                ref[attr],
 
227
                'Expected different %s' % attr)
 
228
 
 
229
    def test_delete(self, ref=None):
 
230
        ref = ref or self.new_ref()
 
231
        resp = TestResponse({
 
232
            "status_code": 204,
 
233
            "text": '',
 
234
        })
 
235
 
 
236
        method = 'DELETE'
 
237
        kwargs = copy.copy(self.TEST_REQUEST_BASE)
 
238
        kwargs['headers'] = self.headers[method]
 
239
        requests.request(
 
240
            method,
 
241
            urlparse.urljoin(
 
242
                self.TEST_URL,
 
243
                'v3/%s/%s' % (self.collection_key, ref['id'])),
 
244
            **kwargs).AndReturn((resp))
 
245
        self.mox.ReplayAll()
 
246
 
 
247
        self.manager.delete(ref['id'])
 
248
 
 
249
 
 
250
class TestResponse(requests.Response):
 
251
    """ Class used to wrap requests.Response and provide some
 
252
        convenience to initialize with a dict """
 
253
 
 
254
    def __init__(self, data):
 
255
        self._text = None
 
256
        super(TestResponse, self)
 
257
        if isinstance(data, dict):
 
258
            self.status_code = data.get('status_code', None)
 
259
            self.headers = data.get('headers', None)
 
260
            # Fake the text attribute to streamline Response creation
 
261
            self._text = data.get('text', None)
 
262
        else:
 
263
            self.status_code = data
 
264
 
 
265
    def __eq__(self, other):
 
266
        return self.__dict__ == other.__dict__
 
267
 
 
268
    @property
 
269
    def text(self):
 
270
        return self._text