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