Revision d6a92fa0 snf-pithos-app/pithos/api/test/__init__.py

b/snf-pithos-app/pithos/api/test/__init__.py
42 42
from pithos.backends.random_word import get_random_word
43 43
from pithos.api import settings as pithos_settings
44 44

  
45
from synnefo.lib.services import get_service_path
46
from synnefo.lib import join_urls
47

  
45 48
from django.test import TestCase
46 49
from django.utils.http import urlencode
47 50
from django.conf import settings
......
87 90
    #TODO unauthorized request
88 91
    def setUp(self):
89 92
        pithos_settings.BACKEND_DB_MODULE = 'pithos.backends.lib.sqlalchemy'
90
        pithos_settings.BACKEND_DB_CONNECTION = construct_db_connection()
93
        pithos_settings.BACKEND_DB_CONNECTION = django_to_sqlalchemy()
91 94
        pithos_settings.BACKEND_POOL_SIZE = 1
92 95
        self.user = 'user'
96
        self.pithos_path = join_urls(get_service_path(
97
            pithos_settings.pithos_services, 'object-store'))
93 98

  
94 99
    def tearDown(self):
95 100
        #delete additionally created metadata
......
137 142
    def update_account_meta(self, meta):
138 143
        kwargs = dict(
139 144
            ('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
140
        r = self.post('/v1/%s?update=' % self.user, **kwargs)
145
        url = join_urls(self.pithos_path, self.user)
146
        r = self.post('%s?update=' % url, **kwargs)
141 147
        self.assertEqual(r.status_code, 202)
142 148
        account_meta = self.get_account_meta()
143 149
        (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
......
148 154
    def reset_account_meta(self, meta):
149 155
        kwargs = dict(
150 156
            ('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
151
        r = self.post('/v1/%s' % self.user, **kwargs)
157
        url = join_urls(self.pithos_path, self.user)
158
        r = self.post(url, **kwargs)
152 159
        self.assertEqual(r.status_code, 202)
153 160
        account_meta = self.get_account_meta()
154 161
        (self.assertTrue('X-Account-Meta-%s' % k in account_meta) for
......
159 166
    def delete_account_meta(self, meta):
160 167
        transform = lambda k: 'HTTP_%s' % k.replace('-', '_').upper()
161 168
        kwargs = dict((transform(k), '') for k, v in meta.items())
162
        r = self.post('/v1/%s?update=' % self.user, **kwargs)
169
        url = join_urls(self.pithos_path, self.user)
170
        r = self.post('%s?update=' % url, **kwargs)
163 171
        self.assertEqual(r.status_code, 202)
164 172
        account_meta = self.get_account_meta()
165 173
        (self.assertTrue('X-Account-Meta-%s' % k not in account_meta) for
......
167 175
        return r
168 176

  
169 177
    def delete_account_groups(self, groups):
170
        r = self.post('/v1/%s?update=' % self.user, **groups)
178
        url = join_urls(self.pithos_path, self.user)
179
        r = self.post('%s?update=' % url, **groups)
171 180
        self.assertEqual(r.status_code, 202)
172 181
        return r
173 182

  
174 183
    def get_account_info(self, until=None):
175
        url = '/v1/%s' % self.user
184
        url = join_urls(self.pithos_path, self.user)
176 185
        if until is not None:
177 186
            parts = list(urlsplit(url))
178 187
            parts[3] = urlencode({
......
200 209
        return headers
201 210

  
202 211
    def list_containers(self, format='json', headers={}, **params):
203
        url = '/v1/%s' % self.user
204
        parts = list(urlsplit(url))
212
        _url = join_urls(self.pithos_path, self.user)
213
        parts = list(urlsplit(_url))
205 214
        params['format'] = format
206 215
        parts[3] = urlencode(params)
207 216
        url = urlunsplit(parts)
......
224 233
            return minidom.parseString(r.content)
225 234

  
226 235
    def delete_container_content(self, cname):
227
        r = self.delete('/v1/%s/%s?delimiter=/' % (self.user, cname))
236
        url = join_urls(self.pithos_path, self.user, cname)
237
        r = self.delete('%s?delimiter=/' % url)
228 238
        self.assertEqual(r.status_code, 204)
229 239
        return r
230 240

  
231 241
    def delete_container(self, cname):
232
        r = self.delete('/v1/%s/%s' % (self.user, cname))
242
        url = join_urls(self.pithos_path, self.user, cname)
243
        r = self.delete(url)
233 244
        self.assertEqual(r.status_code, 204)
234 245
        return r
235 246

  
236 247
    def create_container(self, cname):
237
        r = self.put('/v1/%s/%s' % (self.user, cname), data='')
248
        url = join_urls(self.pithos_path, self.user, cname)
249
        r = self.put(url, data='')
238 250
        self.assertTrue(r.status_code in (202, 201))
239 251
        return r
240 252

  
......
243 255
        data = get_random_word(length=random.randint(1, 1024))
244 256
        headers = dict(('HTTP_X_OBJECT_META_%s' % k.upper(), v)
245 257
                       for k, v in meta.iteritems())
246
        r = self.put('/v1/%s/%s/%s' % (
247
            self.user, cname, oname), data=data, **headers)
258
        url = join_urls(self.pithos_path, self.user, cname, oname)
259
        r = self.put(url, data=data, **headers)
248 260
        self.assertEqual(r.status_code, 201)
249 261
        return oname, data, r
250 262

  
251 263
    def create_folder(self, cname, oname=get_random_word(8), **headers):
252
        r = self.put('/v1/%s/%s/%s' % (
253
            self.user, cname, oname), data='',
254
            content_type='application/directory',
255
            **headers)
264
        url = join_urls(self.pithos_path, self.user, cname, oname)
265
        r = self.put(url, data='', content_type='application/directory',
266
                     **headers)
256 267
        self.assertEqual(r.status_code, 201)
257 268
        return oname, r
258 269

  
259 270
    def list_objects(self, cname):
260
        r = self.get('/v1/%s/%s?format=json' % (self.user, cname))
271
        url = join_urls(self.pithos_path, self.user, cname)
272
        r = self.get('%s?format=json' % url)
261 273
        self.assertTrue(r.status_code in (200, 204))
262 274
        try:
263 275
            objects = json.loads(r.content)
......
331 343
    'django.db.backends.oracle': 'oracle'}
332 344

  
333 345

  
334
def construct_db_connection():
335
    """Convert the django default database to an sqlalchemy connection
336
       string"""
346
def django_to_sqlalchemy():
347
    """Convert the django default database to sqlalchemy connection string"""
348
    # TODO support for more complex configuration
337 349
    db = settings.DATABASES['default']
350
    name = db.get('TEST_NAME', 'test_%s' % db['NAME'])
338 351
    if db['ENGINE'] == 'django.db.backends.sqlite3':
339
        return 'sqlite://'
352
        db.get('TEST_NAME', db['NAME'])
353
        return 'sqlite:///%s' % name
340 354
    else:
341 355
        d = dict(scheme=django_sqlalchemy_engines.get(db['ENGINE']),
342 356
                 user=db['USER'],
343 357
                 pwd=db['PASSWORD'],
344 358
                 host=db['HOST'].lower(),
345 359
                 port=int(db['PORT']) if db['PORT'] != '' else '',
346
                 name=db['NAME'])
360
                 name=name)
347 361
        return '%(scheme)s://%(user)s:%(pwd)s@%(host)s:%(port)s/%(name)s' % d
348 362

  
349 363

  

Also available in: Unified diff