evilny0/fpdb
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Moved SHOW TABLES query to query dict as list_tables

Browse Source 
Added failing tests for Postgres until table deletion and table listing is fixed.
This commit is contained in:
Worros 2008-10-08 03:12:38 +08:00
parent 5dd9dbc86f
commit 6aca36b564
3 changed files with 36 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
pyfpdb/FpdbSQLQueries.py
View File

@ -36,6 +36,17 @@ class FpdbSQLQueries:
# elif(self.dbname == 'PostgreSQL'):
# elif(self.dbname == 'SQLite'):
################################
# List tables
################################
if(self.dbname == 'MySQL InnoDB'):
self.query['list_tables'] = """SHOW TABLES"""
elif(self.dbname == 'PostgreSQL'):
self.query['list_tables'] = """SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"""
elif(self.dbname == 'SQLite'):
self.query['list_tables'] = """ """
##################################################################
# Drop Tables - MySQL, PostgreSQL and SQLite all share same syntax
##################################################################
@ -48,9 +59,6 @@ class FpdbSQLQueries:
# Create Tables
################################
################################
# Create Settings
################################

21
pyfpdb/RegressionTest.py
View File

@ -40,10 +40,23 @@ class TestSequenceFunctions(unittest.TestCase):
self.mysql_settings['db-password'])
self.mysqldict = FpdbSQLQueries.FpdbSQLQueries('MySQL InnoDB')
"""Configure Postgres settings/database and establish connection"""
self.pg_settings={ 'db-host':"localhost", 'db-backend':3, 'db-databaseName':"fpdbtest", 'db-user':"fpdb", 'db-password':"fpdb"}
self.pg_db = fpdb_db.fpdb_db()
self.pg_db.connect(self.pg_settings['db-backend'], self.pg_settings['db-host'],
self.pg_settings['db-databaseName'], self.pg_settings['db-user'],
self.pg_settings['db-password'])
self.pgdict = FpdbSQLQueries.FpdbSQLQueries('PostgreSQL')
def testDatabaseConnection(self):
"""Test all supported DBs"""
self.result = self.mysql_db.cursor.execute("SHOW TABLES")
self.result = self.mysql_db.cursor.execute(self.mysqldict.query['list_tables'])
self.failUnless(self.result==13, "Number of tables in database incorrect. Expected 13 got " + str(self.result))
print self.pgdict.query['list_tables']
self.result = self.pg_db.cursor.execute(self.pgdict.query['list_tables'])
self.failUnless(self.result==13, "Number of tables in database incorrect. Expected 13 got " + str(self.result))
def testMySQLRecreateTables(self):
@ -52,6 +65,12 @@ class TestSequenceFunctions(unittest.TestCase):
self.result = self.mysql_db.cursor.execute("SHOW TABLES")
self.failUnless(self.result==13, "Number of tables in database incorrect. Expected 13 got " + str(self.result))
def testPostgresSQLRecreateTables(self):
"""Test droping then recreating fpdb table schema"""
self.pg_db.recreate_tables()
self.result = self.pg_db.cursor.execute(self.pgdict.query['list_tables'])
self.failUnless(self.result==13, "Number of tables in database incorrect. Expected 13 got " + str(self.result))
if __name__ == '__main__':
unittest.main()

9
pyfpdb/fpdb_db.py
View File

@ -104,12 +104,13 @@ class fpdb_db:
self.drop_referencial_integrity()
# Query the DB to see what tables exist
self.cursor.execute('SHOW TABLES')
self.cursor.execute(self.sql.query['list_tables'])
for table in self.cursor:
self.cursor.execute(self.sql.query['drop_table'] + table[0])
elif(self.get_backend_name() == 'PostgreSQL'):
#todo: postgres version here
print "Empty function here"
self.cursor.execute(self.sql.query['list_tables'])
for table in self.cursor:
print table
elif(self.get_backend_name() == 'SQLite'):
#todo: sqlite version here
print "Empty function here"
@ -120,7 +121,7 @@ class fpdb_db:
def drop_referencial_integrity(self):
"""Update all tables to remove foreign keys"""
self.cursor.execute('SHOW TABLES') # todo: move to FpdbSQLQueries
self.cursor.execute(self.sql.query['list_tables'])
result = self.cursor.fetchall()
for i in range(len(result)):