Revision 65f480f5 snf-pithos-app/pithos/api/test/__init__.py
| b/snf-pithos-app/pithos/api/test/__init__.py | ||
|---|---|---|
| 88 | 88 |
TEST_BLOCK_SIZE = 1024 |
| 89 | 89 |
TEST_HASH_ALGORITHM = 'sha256' |
| 90 | 90 |
|
| 91 |
BACKEND_DB_CONNECTION = None |
|
| 92 |
|
|
| 93 |
|
|
| 94 |
def django_to_sqlalchemy(): |
|
| 95 |
"""Convert the django default database to sqlalchemy connection string""" |
|
| 96 |
|
|
| 97 |
global BACKEND_DB_CONNECTION |
|
| 98 |
if BACKEND_DB_CONNECTION: |
|
| 99 |
return BACKEND_DB_CONNECTION |
|
| 100 |
|
|
| 101 |
# TODO support for more complex configuration |
|
| 102 |
db = settings.DATABASES['default'] |
|
| 103 |
name = db.get('TEST_NAME', 'test_%s' % db['NAME'])
|
|
| 104 |
if db['ENGINE'] == 'django.db.backends.sqlite3': |
|
| 105 |
BACKEND_DB_CONNECTION = 'sqlite:///%s' % name |
|
| 106 |
else: |
|
| 107 |
d = dict(scheme=django_sqlalchemy_engines.get(db['ENGINE']), |
|
| 108 |
user=db['USER'], |
|
| 109 |
pwd=db['PASSWORD'], |
|
| 110 |
host=db['HOST'].lower(), |
|
| 111 |
port=int(db['PORT']) if db['PORT'] != '' else '', |
|
| 112 |
name=name) |
|
| 113 |
BACKEND_DB_CONNECTION = ( |
|
| 114 |
'%(scheme)s://%(user)s:%(pwd)s@%(host)s:%(port)s/%(name)s' % d) |
|
| 115 |
return BACKEND_DB_CONNECTION |
|
| 116 |
|
|
| 91 | 117 |
|
| 92 | 118 |
class PithosAPITest(TestCase): |
| 93 | 119 |
def setUp(self): |
| ... | ... | |
| 115 | 141 |
|
| 116 | 142 |
self._clean_account() |
| 117 | 143 |
|
| 144 |
def _clean_account(self): |
|
| 145 |
for c in self.list_containers(): |
|
| 146 |
self.delete_container_content(c['name']) |
|
| 147 |
self.delete_container(c['name']) |
|
| 148 |
|
|
| 118 | 149 |
def head(self, url, user='user', *args, **kwargs): |
| 119 | 150 |
with astakos_user(user): |
| 120 | 151 |
response = self.client.head(url, *args, **kwargs) |
| ... | ... | |
| 142 | 173 |
response = self.client.put(url, *args, **kwargs) |
| 143 | 174 |
return response |
| 144 | 175 |
|
| 145 |
def _clean_account(self): |
|
| 146 |
for c in self.list_containers(): |
|
| 147 |
self.delete_container_content(c['name']) |
|
| 148 |
self.delete_container(c['name']) |
|
| 149 |
|
|
| 150 | 176 |
def update_account_meta(self, meta): |
| 151 | 177 |
kwargs = dict( |
| 152 | 178 |
('HTTP_X_ACCOUNT_META_%s' % k, str(v)) for k, v in meta.items())
|
| ... | ... | |
| 468 | 494 |
'django.db.backends.oracle': 'oracle'} |
| 469 | 495 |
|
| 470 | 496 |
|
| 471 |
def django_to_sqlalchemy(): |
|
| 472 |
"""Convert the django default database to sqlalchemy connection string""" |
|
| 473 |
# TODO support for more complex configuration |
|
| 474 |
db = settings.DATABASES['default'] |
|
| 475 |
name = db.get('TEST_NAME', 'test_%s' % db['NAME'])
|
|
| 476 |
if db['ENGINE'] == 'django.db.backends.sqlite3': |
|
| 477 |
db.get('TEST_NAME', db['NAME'])
|
|
| 478 |
return 'sqlite:///%s' % name |
|
| 479 |
else: |
|
| 480 |
d = dict(scheme=django_sqlalchemy_engines.get(db['ENGINE']), |
|
| 481 |
user=db['USER'], |
|
| 482 |
pwd=db['PASSWORD'], |
|
| 483 |
host=db['HOST'].lower(), |
|
| 484 |
port=int(db['PORT']) if db['PORT'] != '' else '', |
|
| 485 |
name=name) |
|
| 486 |
return '%(scheme)s://%(user)s:%(pwd)s@%(host)s:%(port)s/%(name)s' % d |
|
| 487 |
|
|
| 488 |
|
|
| 489 | 497 |
def test_concurrently(times=2): |
| 490 | 498 |
""" |
| 491 | 499 |
Add this decorator to small pieces of code that you want to test |
Also available in: Unified diff