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