Cleaning and changin some methods name in sqlwrapper

Database/sqlwrapper.py

@@ -62,7 +62,7 @@ class SqlWrapper(object):
         self.r_dbconf = read_db
         self.w_dbconf = write_db
 
-        self.checkConf() #raise if errors in configuration
+        self._checkConf() #raise if errors in configuration
 
         if self.name in self.__class__.wrapinstance:
             logger.warning("A SqlWrapper with the name "+self.name+" allready exist. Replacing the old one by the new one")
@@ -82,9 +82,12 @@ class SqlWrapper(object):
         pass
 
     @property
-    def cfg(self): return self.__class__.config;
+    def cfg(self):
+        """ Return the SqlWrapper.config dict """
+        return self.__class__.config;
     @property
-    def engines_cfg(self): return self.__class__.ENGINES;
+    def _engines_cfg(self):
+        return self.__class__.ENGINES;
 
     @property
     def meta(self):
@@ -104,18 +107,19 @@ class SqlWrapper(object):
         """ Return the read connection
             @warning Do not store the connection, call this method each time you need it
         """
-        return self.getConnection(True)
+        return self._getConnection(True)
     @property
     def wconn(self):
         """ Return the write connection
             @warning Do not store the connection, call this method each time you need it
         """
-        return self.getConnection(False)
+        return self._getConnection(False)
 
-    def getConnection(self, read):
+    def _getConnection(self, read):
         """ Return an opened connection
             @param read bool: If true return the reading connection
             @return A sqlAlchemy db connection
+            @private
         """
         if read:
             r = self.r_conn
@@ -125,7 +129,7 @@ class SqlWrapper(object):
         if r == None:
             #Connection not yet opened
             self.connect(read)
-            r = self.getConnection(read) #TODO : Un truc plus safe/propre qu'un appel reccursif ?
+            r = self._getConnection(read) #TODO : Un truc plus safe/propre qu'un appel reccursif ?
         return r
 
 
@@ -175,6 +179,9 @@ class SqlWrapper(object):
 
     @classmethod
     def reconnectAll(c, read = None):
+        """ Reconnect all the wrappers 
+            @static
+        """
         for wname in c.wrapinstance:
             c.wrapinstance[wname].reconnect(read)
             
@@ -199,7 +206,7 @@ class SqlWrapper(object):
         #Loading confs
         cfg = self.cfg['db'][self.r_dbconf if read else self.w_dbconf]
 
-        edata = self.engines_cfg[cfg['ENGINE']] #engine infos
+        edata = self._engines_cfg[cfg['ENGINE']] #engine infos
         conn_str = ""
 
         if cfg['ENGINE'] == 'sqlite':
@@ -242,7 +249,7 @@ class SqlWrapper(object):
             raise KeyError("No wrapper named '"+name+"' exists")
         return c.wrapinstance[name]
 
-    def checkConf(self):
+    def _checkConf(self):
         """ Class method that check the configuration
             
             Configuration looks like
@@ -269,7 +276,7 @@ class SqlWrapper(object):
                     if 'ENGINE' not in db:
                         err.append('Missing "ENGINE" in database "'+db+'"')
                     else:
-                        if db['ENGINE'] not in self.engines_cfg:
+                        if db['ENGINE'] not in self._engines_cfg:
                             err.append('Unknown engine "'+db['ENGINE']+'"')
                         elif db['ENGINE'] != 'sqlite' and 'USER' not in db:
                             err.append('Missing "User" in configuration of database "'+dbname+'"')
@@ -300,6 +307,8 @@ class SqlWrapper(object):
 
         logger.info("Running function createAllFromConf")
         for i,table in enumerate(schema):
+            if not isinstance(table, dict):
+                raise TypeError("Excepted a list of dict but got a "+str(type(schema))+" in the list")
             self.createTable(**table)
 
         self.meta_crea.create_all(bind = self.w_engine)
@@ -370,7 +379,6 @@ class SqlWrapper(object):
             kwargs['primary_key'] = kwargs['primarykey']
             del kwargs['primarykey']
 
-
         res = sqla.Column(**kwargs)
 
         if fk != None:
@@ -397,29 +405,6 @@ class SqlWrapper(object):
         column_length = int(check_length.groups()[0]) if check_length else None
         return sqla.VARCHAR(length=column_length)
      
-    @classmethod
-    def engineFamily(c, engine):
-        """ Given an engine return the db family
-            @see SqlWrapper::ENGINES
-            @return A str or None
-        """
-        for fam in c.ENGINES:
-            if engine.driver == c.ENGINES[fam]['driver']:
-                return fam
-        return None
-
-    @property
-    def wEngineFamily(self):
-        """ Return the db family of the write engine
-            @return a string or None
-        """
-        return self.__class__.engineFamily(self.w_engine)
-    @property   
-    def rEngineFamily(self):
-        """ Return the db family of the read engine
-            @return a string or None
-        """
-        return self.__class__.engineFamily(self.r_engine)
 
     def dropColumn(self, tname, colname):
         """ Drop a column from a table
@@ -450,10 +435,10 @@ class SqlWrapper(object):
 
             @return True if query success False if it fails
         """
-        newcol = self.createColumn(name=colname, type_ = coltype)
         if tname not in self.meta.tables: #Useless ?
             raise NameError("The table '"+tname+"' dont exist")
         table = self.Table(tname)
+        newcol = self.createColumn(name=colname, type_ = coltype)
 
         ddl = AddColumn(table, newcol)
         sql = ddl.compile(dialect=self.w_engine.dialect)
@@ -473,20 +458,17 @@ class SqlWrapper(object):
             @return True if query successs False if it fails
         """
         
-        if self.wEngineFamily == 'sqlite':
-            raise NotImplementedError('AlterColumn not yet implemented for sqlite engines')
-            
+        if tname not in self.meta.tables: #Useless ?
+            raise NameError("The table '"+tname+"' dont exist")
 
         col = self.createColumn(name=colname, type_=col_newtype)
         table = self.Table(tname)
 
-        typepref = 'TYPE ' if self.wEngineFamily == 'postgresql' else ''
-
-        query = 'ALTER TABLE %s ALTER COLUMN %s %s'%(table.name, col.name, typepref+col.type)
-
-        logger.debug("Executing SQL : '"+query+"'")
-
-        ret = bool(self.wconn.execute(query))
+        ddl = AlterColumn(table, newcol)
+        sql = ddl.compile(dialect=self.w_engine.dialect)
+        sql = str(sql)
+        logger.debug("Executing SQL  : '"+sql+"'")
+        ret = bool(self.wconn.execute(sql))
 
         self.renewMetaData()
         return ret

Database/test/test_sqlwrapper.py

@@ -1,519 +0,0 @@
-import os
-import logging
-import random
-
-from unittest import TestCase
-from unittest.mock import MagicMock, Mock, patch, call
-import unittest
-
-import sqlalchemy
-from Database.sqlwrapper import SqlWrapper
-
-from django.conf import settings
-from Database.sqlsettings import SQLSettings
-
-os.environ.setdefault("DJANGO_SETTINGS_MODULE", "Lodel.settings")
-
-BADNAMES = ['',  "  \t  ", "bad--", "%", "*", "`", '"', "'", "\0", "hello`World", 'Hello"world', 'foo%', '*bar.*', '%%', "hello\0world", print, 42 ]
-
-for c in ('*','%','`', "'", '"', "\\"):
-    for _ in range(16):
-        c = "\\"+c
-        BADNAMES.append(c)
-BADNAMES+=[chr(i) for i in list(range(0,0x09))+[0x0b, 0x0c]+list(range(0x0e, 0x01f))]
-
-class SqlWrapperTests(TestCase):
-
-    def setUp(self):
-        #Overwriting database conf
-        SQLSettings.DB_READ_CONNECTION_NAME = 'testdb'
-        SQLSettings.DB_WRITE_CONNECTION_NAME = 'testdb'
-        
-
-        self.testdb = os.path.join('/tmp/', 'lodel2_testdb.sqlite3')
-
-        settings.DATABASES['testdb'] = {
-            'ENGINE': 'sqlite',
-            'NAME': self.testdb,
-        }
-        #Disable logging but CRITICAL
-        logging.basicConfig(level=logging.CRITICAL)
-        pass
-
-    def tearDown(self):
-        try:
-            os.unlink(self.testdb) #Removing the test database
-        except FileNotFoundError: pass
-    
-    """ Testing standart instanciation of sqlwrapper """
-    def test_init_sqlwrapper(self):
-        sw = SqlWrapper()
-        self.assertIsInstance(sw, SqlWrapper)
-        sw2 = SqlWrapper()
-        self.assertIsInstance(sw2, SqlWrapper)
-
-    def test_get_engine(self):
-        sw = SqlWrapper()
-        engine = sw.get_engine(SQLSettings.DB_READ_CONNECTION_NAME)
-        self.assertIsInstance(engine, sqlalchemy.engine.base.Engine)
-    
-    def test_get_engine_badargs(self):
-        sw = SqlWrapper()
-        with self.assertRaises(Exception):
-            sw.get_engine(0)
-        with self.assertRaises(Exception):
-            sw.get_engine('')
-        with self.assertRaises(Exception):
-            sw.get_engine('           ')
-
-    
-    @patch.object(SqlWrapper, 'get_read_engine')
-    def test_execute_queries(self, mock_read):
-        queries = [ 'SELECT * FROM foo', 'SELECT * FROM bar', 'SELECT foo.id, bar.id FROM foo, bar WHERE foo.id = 42 AND bar.name = \'hello world !\'' ]
-
-        mock_read.return_value = MagicMock()
-        mock_engine = mock_read.return_value
-        mock_engine.connect.return_value = MagicMock()
-        mock_conn = mock_engine.connect.return_value
-        sw = SqlWrapper()
-
-        #One query execution
-        sw.execute(queries[0], 'read')
-        mock_conn.execute.assert_called_with(queries[0])
-        mock_conn.reset_mock()
-
-        #multiple queries execution
-        sw.execute(queries, 'read')
-
-        except_calls = [ call(q) for q in queries ]
-        self.assertEqual(except_calls, mock_conn.execute.mock_calls)
-
-    @patch.object(sqlalchemy.ForeignKeyConstraint, '__init__')
-    def test_create_fk_constraint_mock(self, mock_fk):
-        mock_fk.return_value = None
-
-        sw = SqlWrapper()
-        sw.create_foreign_key_constraint_object('foo','bar', 'foobar')
-        mock_fk.asser_called_with(['foo'], ['bar'], 'foobar')
-
-    def test_create_fk_constraint(self):
-        sw = SqlWrapper()
-        fk = sw.create_foreign_key_constraint_object('foo', 'bar', 'foobar')
-        self.assertIsInstance(fk, sqlalchemy.ForeignKeyConstraint)
-        
-    @unittest.skip('Dev') #TODO remove it
-    def test_create_fk_constraint_badargs(self):
-        sw = SqlWrapper()
-
-        
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(['foo', 'bar'], 'foofoo', 'babar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('bar', 2, 'babar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(None, 'foo', 'bar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('bar', None, 'babar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(None, None, 'babar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(print, 'foo', 'babar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('foo', print, 'babar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('foo', 'bar', print)
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('bar', 'foo', 42)
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('bar', 'foo', '   ')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(" \t  ", 'foo')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('bar', 'foo', "  \t ")
-
-        foocol = sqlalchemy.Column('foo',sqlalchemy.INTEGER)
-
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(foocol, foocol)
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object(foocol, 'bar')
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('foo', foocol)
-        with self.assertRaises(Exception):
-            sw.create_foreign_key_constraint_object('foo', 'bar', foocol)
-    
-    @patch.object(sqlalchemy.Column, '__init__')
-    def test_create_column(self, mockcol):
-        mockcol.return_value = None
-        sw = SqlWrapper()
-        foo = sw.create_column_object('foo', 'INTEGER')
-        mockcol.assert_called_once_with('foo', sqlalchemy.INTEGER)
-        mockcol.reset_mock()
-
-        foo = sw.create_column_object('foo', 'VARCHAR(50)')
-        self.assertEqual(mockcol.call_args[0][0], 'foo')
-        self.assertIsInstance(mockcol.call_args[0][1], sqlalchemy.VARCHAR)
-        self.assertEqual(mockcol.call_args[0][1].length, 50)
-        mockcol.reset_mock()
-
-        foo = sw.create_column_object('foo', 'TEXT')
-        mockcol.assert_called_once_with('foo', sqlalchemy.TEXT)
-        mockcol.reset_mock()
-
-        foo = sw.create_column_object('foo', 'DATE')
-        mockcol.assert_called_once_with('foo', sqlalchemy.DATE)
-        mockcol.reset_mock()
-
-        foo = sw.create_column_object('foo', 'BOOLEAN')
-        mockcol.assert_called_once_with('foo', sqlalchemy.BOOLEAN)
-        mockcol.reset_mock()
- 
-        xtrmlongname = ''
-        for _ in range(200):
-            xtrmlongname += 'veryvery'
-        xtrmlongname += 'longname'
-
-        foo = sw.create_column_object(xtrmlongname, 'TEXT')
-        mockcol.assert_called_once_with(xtrmlongname, sqlalchemy.TEXT)
-        mockcol.reset_mock()
-
-    def test_create_column_extra_fk(self):
-        sw = SqlWrapper()
-
-        extra = { 'foreignkey': 'bar' }
-        rescol = sw.create_column_object('foo', 'INTEGER', extra)
-        self.assertIsInstance(rescol, sqlalchemy.Column)
-        fk = rescol.foreign_keys.pop()
-        self.assertIsInstance(fk, sqlalchemy.ForeignKey)
-        self.assertEqual(fk._colspec, 'bar')
-
-    def test_create_column_extra_default(self):
-        sw = SqlWrapper()
-
-
-        extra = { 'default': None, 'nullable': True }
-        rescol = sw.create_column_object('foo', 'INTEGER', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, None)
-
-        extra = { 'default': "NULL", 'nullable': True }
-        rescol = sw.create_column_object('foo', 'INTEGER', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, "NULL")
-
-        extra = { 'default': 'null', 'nullable': True }
-        rescol = sw.create_column_object('foo', 'INTEGER', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 'null')
-
-        extra = { 'default': 42 }
-        rescol = sw.create_column_object('foo', 'INTEGER', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 42)
-
-        extra = { 'default': 'foobardefault' }
-        rescol = sw.create_column_object('foo', 'VARCHAR(50)', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 'foobardefault')
-
-        extra = { 'default': 'foodefault' }
-        rescol = sw.create_column_object('foo', 'TEXT', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 'foodefault')
-
-        extra = { 'default': True }
-        rescol = sw.create_column_object('foo', 'BOOLEAN', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, True)
-
-        extra = { 'default': False }
-        rescol = sw.create_column_object('foo', 'BOOLEAN', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, False)
-
-        extra = { 'default': "true" }
-        rescol = sw.create_column_object('foo', 'BOOLEAN', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, "true")
-
-        extra = { 'default': 0 }
-        rescol = sw.create_column_object('foo', 'BOOLEAN', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 0)
-
-        extra = { 'default': 1 }
-        rescol = sw.create_column_object('foo', 'BOOLEAN', extra)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 1)
-
-
-    def test_create_column_extra_pknull(self):
-        sw = SqlWrapper()
-
-        for b in (True,False):
-            extra = { 'primarykey': b }
-            rescol = sw.create_column_object('foo', 'INTEGER', extra)
-            self.assertIsInstance(rescol, sqlalchemy.Column)
-            self.assertEqual(rescol.primary_key, b)
-
-            extra = { 'nullable': b }
-            rescol = sw.create_column_object('foo', 'INTEGER', extra)
-            self.assertIsInstance(rescol, sqlalchemy.Column)
-            self.assertEqual(rescol.nullable, b)
-
-            extra = { 'primarykey' : b, 'nullable': not b }
-            rescol = sw.create_column_object('foo', 'INTEGER', extra)
-            self.assertIsInstance(rescol, sqlalchemy.Column)
-            self.assertEqual(rescol.primary_key, b)
-            self.assertEqual(rescol.nullable, not b)
-
-            extra = { 'primarykey' : b, 'nullable': b }
-            rescol = sw.create_column_object('foo', 'INTEGER', extra)
-            self.assertIsInstance(rescol, sqlalchemy.Column)
-            self.assertEqual(rescol.primary_key, b)
-            self.assertEqual(rescol.nullable, b)
-
-
-
-    def test_create_column_extra_all(self):
-        sw = SqlWrapper()
-
-        extra = { 'primarykey': False, 'nullable': False, 'default':42, 'foreignkey': 'foobar'}
-        rescol = sw.create_column_object('foo', 'INTEGER', extra)
-        self.assertIsInstance(rescol, sqlalchemy.Column)
-        self.assertEqual(rescol.primary_key, False)
-        self.assertEqual(rescol.nullable, False)
-        self.assertIsInstance(rescol.default, sqlalchemy.ColumnDefault)
-        self.assertEqual(rescol.default.arg, 42)
-        fk = rescol.foreign_keys.pop()
-        self.assertIsInstance(fk, sqlalchemy.ForeignKey)
-        self.assertEqual(fk._colspec, 'foobar')
-
-    @unittest.skip('Dev') #TODO remove it
-    def test_create_column_badargs(self):
-        sw = SqlWrapper()
-        
-        cc = sw.create_column_object
-        ain = self.assertIsNone
-        
-        for bname in BADNAMES:
-            ain(cc(bname, 'INTEGER'))
-
-        ain(cc('c', 'INNNTEGER'))
-        ain(cc(" \t\t  ", 'TEXT'))
-        ain(cc('c', '             '))
-        ain(cc('c', 'VARCHAR(foo)'))
-        ain(cc('supercol', 'SUPERNOTTYPEDVARCHARSTR'))
-
-        ain(cc('c', None))
-        ain(cc(None, None))
-
-    @unittest.skip('Dev') #TODO remove it
-    def test_create_column_badextra(self):
-        sw = SqlWrapper()
-
-        cc = sw.create_column_object
-        ain = self.assertIsNone
-        
-        #Put junk in extra datas
-        for xtra_name in [ 'primarykey', 'nullable', 'foreignkey' ]:
-            for junk in [ print, sqlalchemy, SqlWrapper, '   ' ]+BADNAMES:
-                ain(cc('c', 'TEXT', { xtra_name: junk }))
-
-        for junk in [True, False, 52, '   ']+BADNAMES:
-            ain(cc('c', 'TEXT', { 'foreignkey': junk }))
-
-        for xtra_name in ('primarykey', 'nullalble'):
-            for junk in [4096, 'HELLO WORLD !', '   ']+BADNAMES:
-                ain(cc('c', 'TEXT', { xtra_name, junk } ))
-
-    @patch.object(sqlalchemy.Table, '__init__')
-    def test_create_table(self, mock_table):
-        #TODO check constraints
-        """ Create tables and check that the names are ok """
-        mock_table.return_value = None
-        sw = SqlWrapper()
-        
-        cols = [
-            { 'name': 'supercol', 'type': 'INTEGER', 'extra': { 'nullable': True } },
-            { 'name': 'superpk', 'type': 'INTEGER', 'extra': { 'primarykey': True} },
-            { 'name': 'col2', 'type': 'TEXT' }
-        ]
-        
-        for table_name in ['supertable', 'besttableever', 'foo-table']:
-
-            params = { 'name': table_name, 'columns': cols, 'constraints':dict() }
-
-            self.assertTrue(sw.create_table(params))
-            self.assertEqual(mock_table.call_args[0][0], table_name)
-        pass
-    
-    @patch.object(sqlalchemy.Table, 'append_column')
-    def test_create_table_col(self, mock_append):
-        #TODO check constraints
-        """ Create a table and check that the columns are OK """
-        sw = SqlWrapper()
-        
-        table_name = 'supertablecol'
-
-        cols = [
-            { 'name': 'supercol', 'type': 'INTEGER', 'extra': { 'nullable': True } },
-            { 'name': 'superpk', 'type': 'INTEGER', 'extra': { 'primarykey': True} },
-            { 'name': 'col2', 'type': 'TEXT' }
-        ]
-
-        except_coltype = [ sqlalchemy.INTEGER, sqlalchemy.INTEGER, sqlalchemy.TEXT ]
-
-        params = { 'name': table_name, 'columns': cols, 'constraints':dict() }
-    
-        sw.create_table(params) #This call return false because of the mock on append_column
-        
-        self.assertEqual(len(mock_append.mock_calls), len(cols))
-
-        for n,c in enumerate(mock_append.mock_calls):
-            self.assertEqual(c[1][0].name, cols[n]['name'])
-            self.assertIsInstance(c[1][0].type, except_coltype[n])
-        pass
-    
-    def test_create_table_recreate(self):
-        #TODO check constraints
-        """ Try to create the same table 2 times (except a False return the second time) """
-        sw = SqlWrapper()
-
-        params = { 'name': 'redondant', 'columns': [{'name':'foocol', 'type': 'INTEGER'}]}
-
-        self.assertTrue(sw.create_table(params))
-
-        params['columns'] = [{'name':'barcol', 'type': 'INTEGER'}]
-
-        self.assertFalse(sw.create_table(params))
-
-    @unittest.skip('dev') #TODO remove it
-    def test_create_table_badargs(self):
-        sw = SqlWrapper()
-
-        af = self.assertFalse
-        ct = sw.create_table
-
-        foocol = {'name': 'foocol', 'type': 'INTEGER'}
-
-        p = {'name': 'foo'}
-        af(ct(p)) #no columns
-        for bname in BADNAMES: #bad column name
-            p['columns'] = bname
-            af(ct(p))
-        p['columns']=[]; af(ct(p)) #empty columns TODO Or return True is normal ???
-
-
-        p = {'columns':[foocol]}
-        af(ct(p)) #no name
-        for bname in BADNAMES:
-            p['name'] = bname
-            af(ct(p))
-        pass
-
-    
-    def create_test_table(self, sw):
-        """ Create a table for test purpose """
-        table_name = 'testtable'
-        cols = [
-            { 'name': 'pk', 'type': 'INTEGER', 'extra': {'primarykey': True} },
-            { 'name': 'testtxt', 'type': 'TEXT', 'extra': {'nullable': True, 'default': 'hello'} },
-            { 'name': 'testchr', 'type': 'VARCHAR(50)', 'extra': {'nullable': True, 'default': 'hello world'} },
-        ]
-        
-        sw.create_table( { 'name': table_name, 'columns': cols} )
-        pass
-    
-    def test_get_table(self):
-        """ Try to get the testtable (check the name and type of return) """
-        sw = SqlWrapper()
-
-        self.create_test_table(sw)
-
-        rt = sw.get_table('testtable')
-        self.assertIsInstance(rt, sqlalchemy.Table)
-        self.assertEqual(rt.name, 'testtable')
-
-        rt = sw.get_table('testtable', 'write')
-        self.assertIsInstance(rt, sqlalchemy.Table)
-        self.assertEqual(rt.name, 'testtable')
-
-        rt = sw.get_table('testtable', 'read')
-        self.assertIsInstance(rt, sqlalchemy.Table)
-        self.assertEqual(rt.name, 'testtable')
-        pass
-
-    @unittest.skip('dev') #TODO remove skip
-    def test_get_table_badargs(self):
-        sw = SqlWrapper()
-
-        self.create_test_table(sw)
-
-        for badname in BADNAMES:
-            with self.assertRaises((sqlalchemy.exc.NoSuchTableError, Exception)):
-                sw.get_table(badname)
-
-        with self.assertRaises(sqlalchemy.exc.NoSuchTableError):
-            sw.get_table('FOOBAR')
-        with self.assertRaises(Exception):
-            sw.get_table(print)
-        with self.assertRaises(Exception):
-            sw.get_table(1)
-
-        #bad action types
-        with self.assertRaises(Exception):
-            sw.get_table('testtable', print)
-        with self.assertRaises(Exception):
-            sw.get_table('testtable', 'foobar')
-        with self.assertRaises(Exception):
-            sw.get_table('testtable', 42)
-        pass
-
-    def test_drop_table(self):
-        sw = SqlWrapper()
-        
-        self.create_test_table(sw)
-
-        self.assertTrue(sw.drop_table('testtable'))
-        self.assertFalse(sw.drop_table('testtable'))
-        self.assertFalse(sw.drop_table('nonexisting'))
-        pass
-
-    def test_drop_table_badargs(self):
-        sw = SqlWrapper()
-
-        self.create_test_table(sw)
-
-        af = self.assertFalse
-
-        for bname in BADNAMES:
-            self.assertFalse(sw.drop_table(bname))
-
-    def test_create_get_drop_table(self):
-        sw = SqlWrapper()
-        
-        funkynames = ('standart', "wow-nice", "-really-", "test_test-test", "_test", "test_", "test42", "foobar_123", "foobar-123", "foofoo--babar")
-        types = ['INTEGER', 'VARCHAR(5)', 'VARCHAR(128)', 'TEXT', 'BOOLEAN']
-        params = dict()
-
-        cols = []
-        for i,name in enumerate(funkynames):
-            cols.append({'name':name, 'type':types[i%len(types)]})
-        params['columns'] = cols
-
-
-        for name in random.sample(funkynames, len(funkynames)):
-            params['name'] = name
-            self.assertTrue(sw.create_table(params))
-
-        for name in random.sample(funkynames, len(funkynames)):
-            rt = sw.get_table(name)
-            self.assertIsInstance(rt, sqlalchemy.Table)#do we get a table ?
-            self.assertEqual(rt.name, name)
-
-        for name in random.sample(funkynames, len(funkynames)):
-            self.assertTrue(sw.drop_table(name))
-        pass
-        
-

Database/test/test_sqlwrapper_querystring.py

@@ -1,175 +0,0 @@
-import os
-import logging
-import random
-
-from unittest import TestCase
-import unittest
-
-from Database.sqlwrapper import SqlWrapper
-
-from django.conf import settings
-from Database.sqlsettings import SQLSettings
-
-os.environ.setdefault("DJANGO_SETTINGS_MODULE", "Lodel.settings")
-
-#Bad strings for injection tests
-INJECTIONS = [ 'foo UNION SELECT 1,2,3,4--', "foo' UNION SELECT 1,2,3--", 'foo" UNION SELECT 1,2,3--', 'foo` UNION SELECT 1,2,3,4--', '--', 'foo`; SELECT 1,2,3', 'foo"; SELECT 1,2,3', "foo'; SELECT 1,2,3", "; SELECT 1,2,3" ]
-NAMEINJECT = INJECTIONS + [ '%', '*', "\0", "\b\b\b\b\b\b\b\b\b" ]
-
-#Valid SQL types
-VTYPES = [ 'integer', 'varchar(1)', 'varchar(50)', 'text', 'boolean' ]
-
-class SqlWrapperQueryStrTests(TestCase):
-    
-    def setUp(self):
-        #creating a test table
-        sw = SqlWrapper()
-        self.ttn = 'testtable'
-        self.cols = [
-            { 'name': 'pk', 'type': 'INTEGER', 'extra': {'primarykey': True} },
-            { 'name': 'testtxt', 'type': 'TEXT', 'extra': {'nullable': True, 'default': 'hello'} },
-            { 'name': 'testchr', 'type': 'VARCHAR(50)', 'extra': {'nullable': True, 'default': 'hello world'} },
-            { 'name': 'testbool', 'type': 'BOOLEAN', 'extra': {'nullable':False, 'default': False}},
-        ]
-        
-        sw.create_table( { 'name': self.ttn, 'columns': self.cols} )
-
-
-        #Disable logging but CRITICAL
-        logging.basicConfig(level=logging.CRITICAL)
-        pass
-
-    def tearDown(self):
-        sw = SqlWrapper()
-        sw.drop_table(self.ttn)
-
-    @unittest.skip('dev') #TODO remove skip
-    def test_get_querystring(self):
-        sw = SqlWrapper()
-
-        actions = [ 'add_column', 'alter_column', 'drop_column' ]
-        dialects = [ 'default', 'mysql', 'postgresql' ]
-
-        for action in actions:
-            for dialect in dialects:
-                r = sw.get_querystring(action, dialect)
-                self.assertIsInstance(r, str)
-
-    @unittest.skip('dev') #TODO remove skip
-    def test_get_querystring_badargs(self):
-        sw = SqlWrapper()
-        
-        actions = [ 1, -1, print, [], 'foo']
-        dialects = actions
-        for action in actions:
-            for dialect in dialects:
-                with self.assertRaises(ValueError):
-                    r = sw.get_querystring(action, dialect)
-    
-    @unittest.skip('dev') #TODO remove skip
-    def test_add_column(self):
-        sw = SqlWrapper()
-
-        colnames = [ 'taddcol1', 'test-add-col', 'test_add_col', '-test', '_add', '__col__' ]
-
-        for i, name in enumerate(colnames):
-            col = { 'name': name, 'type': VTYPES[i%len(VTYPES)] }
-            self.assertTrue(sw.add_column(self.ttn, col))
-        pass
-
-    @unittest.skip('dev') #TODO remove skip
-    def test_add_column_badargs(self):
-        sw = SqlWrapper()
-
-        coolname = 'cool'
-        i=0
-        self.assertFalse(sw.add_column(self.ttn, {'type': 'INTEGER'}))
-        self.assertFalse(sw.add_column(self.ttn, {'name': 'foo'}))
-        self.assertFalse(sw.add_column(self.ttn, dict()))
-        self.assertFalse(sw.add_column(self.ttn, print))
-        self.assertFalse(sw.add_column(self.ttn, ['foo', 'integer']))
-        self.assertFalse(sw.add_column(self.ttn, None))
-        self.assertFalse(sw.add_column(self.ttn, 42))
-        self.assertFalse(sw.add_column(1, {'name':'foo', 'type':'integer'}))
-        self.assertFalse(sw.add_column(print, {'name':'foo', 'type':'integer'}))
-        self.assertFalse(sw.add_column([], {'name':'foo', 'type':'integer'}))
-        self.assertFalse(sw.add_column(dict(), {'name':'foo', 'type':'integer'}))
-        for badname in NAMEINJECT:
-            self.assertFalse(sw.add_column(self.ttn, {'name':badname, 'type':'INTEGER'}))
-            self.assertFalse(sw.add_column(self.ttn, {'name':coolname+str(i), 'type':badname}))
-            self.assertFalse(sw.add_column(badname, {'name':coolname+str(i), 'type':'INTEGER'}))
-            i+=1
-        
-    @unittest.skip('dev') #TODO remove skip
-    def test_alter_column(self):
-        sw = SqlWrapper()
-
-        colnames = ['talter', 'talter1', 'test_alter', 'test-alter-col', '-test_alter', '__test_alter__']
-        for i,name in enumerate(random.sample(colnames, len(colnames))):
-            col = { 'name': name, 'type': VTYPES[i%len(VTYPES)] }
-            self.assertTrue(sw.add_column( self.ttn, col))
-
-        for i,name in enumerate(random.sample(colnames, len(colnames))):
-            col = {'name': name, 'type': VTYPES[i%len(VTYPES)]}
-            self.assertTrue(self.ttn, col)
-        pass
-
-    @unittest.skip('dev') #TODO remove skip
-    def test_alter_column_badargs(self):
-        sw = SqlWrapper()
-
-        colnames = ['tabad', 'tabad1']
-
-        for i,name in enumerate(colnames):
-            col = { 'name': name, 'type': VTYPES[i%len(VTYPES)] }
-            self.assertTrue(sw.add_column(self.ttn, col))
-
-        for i,badname in enumerate(NAMEINJECT):
-            col = { 'name': badname, 'type': VTYPES[i%len(VTYPES)] }
-            self.assertFalse(sw.alter_column(self.ttn, col))
-
-            col = { 'name': colnames[i%len(colnames)], 'type': badname}
-            self.assertFalse(sw.alter_column(self.ttn, col))
-
-            col = { 'name': badname, 'type': NAMEINJECT[random.randint(0,len(NAMEINJECT))]}
-            self.assertFalse(sw.alter_column(self.ttn, col))
-
-            col = { 'name': colnames[i%len(colnames)], 'type': VTYPES[i%len(VTYPES)] }
-            self.assertFalse(sw.alter_column(badname, col))
-
-    def test_insert(self):
-        sw = SqlWrapper()
-
-        records = [
-            {   'pk': 0,
-                'testchr': 'Hello world !',
-                'testtext': 'Wow ! Super text... I\'m amazed',
-                'testbool': False
-            },
-            {   'pk': 1,
-                'testchr': 'Hello"world...--',
-                'testtext': 'Another wonderfull text. But this time with spécials chars@;,:--*/+\'{}]{[|~&ù^$*µ$£ê;<ç>\/*-+',
-                'testbool': True
-            },
-            {   'pk': 2 }, #default values for others
-            {   'pk': '3',
-                'testchr': None,
-                'testtext': None,
-                'testbool': 'true'
-            },
-            {   'pk': 4,
-                'testchr': '',
-                'testtext': '',
-                'testbool': 'false'
-            },
-            {   'pk': 5,
-                'testbool': 0
-            },
-            {   'pk': 6,
-                'testbool': 1
-            },
-            {   'pk':1024,
-                'testbool': False
-            },
-        ]
-                

Database/tests.py

@@ -1,3 +0,0 @@
-from django.test import TestCase
-
-# Create your tests here.