diff --git a/test/test_mysql.py b/test/test_mysql.py
index fe045f6bdd55d74070fb5f3e3b7175a01897c1de..88f014d138851c1e53f52620e8a0be982ae308b1 100644 (file)
--- a/test/test_mysql.py
+++ b/test/test_mysql.py
# FOR A PARTICULAR PURPOSE. THE CODE PROVIDED HEREUNDER IS ON AN "AS IS"
# BASIS, AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
# SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
-#
-# $Id: test_mysql.py,v 1.4 2003-11-05 21:54:57 jlgijsbers Exp $
+#
+# $Id: test_mysql.py,v 1.15 2004-11-10 22:22:59 richard Exp $
import unittest, os, shutil, time, imp
from roundup.hyperdb import DatabaseError
-from roundup import init, backends
+from roundup.backends import get_backend, have_backend
+
+from db_test_base import DBTest, ROTest, config, SchemaTest, ClassicInitTest
+from db_test_base import ConcurrentDBTest, FilterCacheTest
-from db_test_base import DBTest, ROTest, config, SchemaTest, nodbconfig, \
- ClassicInitTest
class mysqlOpener:
- if hasattr(backends, 'mysql'):
- from roundup.backends import mysql as module
+ if have_backend('mysql'):
+ module = get_backend('mysql')
+
+ def setUp(self):
+ self.module.db_nuke(config)
def tearDown(self):
self.db.close()
+ self.nuke_database()
+
+ def nuke_database(self):
self.module.db_nuke(config)
class mysqlDBTest(mysqlOpener, DBTest):
- pass
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ DBTest.setUp(self)
class mysqlROTest(mysqlOpener, ROTest):
- pass
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ ROTest.setUp(self)
class mysqlSchemaTest(mysqlOpener, SchemaTest):
- pass
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ SchemaTest.setUp(self)
-class mysqlClassicInitTest(ClassicInitTest):
+class mysqlClassicInitTest(mysqlOpener, ClassicInitTest):
backend = 'mysql'
- extra_config = '''
-MYSQL_DBHOST = 'localhost'
-MYSQL_DBUSER = 'rounduptest'
-MYSQL_DBPASSWORD = 'rounduptest'
-MYSQL_DBNAME = 'rounduptest'
-MYSQL_DATABASE = (MYSQL_DBHOST, MYSQL_DBUSER, MYSQL_DBPASSWORD, MYSQL_DBNAME)
-'''
- if hasattr(backends, 'mysql'):
- from roundup.backends import mysql as module
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ ClassicInitTest.setUp(self)
def tearDown(self):
ClassicInitTest.tearDown(self)
- self.module.db_nuke(config)
+ self.nuke_database()
+
+class mysqlConcurrencyTest(mysqlOpener, ConcurrentDBTest):
+ backend = 'mysql'
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ ConcurrentDBTest.setUp(self)
+ def tearDown(self):
+ ConcurrentDBTest.tearDown(self)
+ self.nuke_database()
+
+class mysqlFilterCacheTest(mysqlOpener, FilterCacheTest):
+ backend = 'mysql'
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ FilterCacheTest.setUp(self)
+ def tearDown(self):
+ FilterCacheTest.tearDown(self)
+ self.nuke_database()
+
+from session_common import RDBMSTest
+class mysqlSessionTest(mysqlOpener, RDBMSTest):
+ def setUp(self):
+ mysqlOpener.setUp(self)
+ RDBMSTest.setUp(self)
+ def tearDown(self):
+ RDBMSTest.tearDown(self)
+ mysqlOpener.tearDown(self)
def test_suite():
suite = unittest.TestSuite()
- if not hasattr(backends, 'mysql'):
+ if not have_backend('mysql'):
+ print "Skipping mysql tests"
return suite
- from roundup.backends import mysql
+ import MySQLdb
try:
- # Check if we can run mysql tests
- import MySQLdb
- db = mysql.Database(nodbconfig, 'admin')
- db.conn.select_db(config.MYSQL_DBNAME)
- db.sql("SHOW TABLES");
- tables = db.sql_fetchall()
- if 0: #tables:
- # Database should be empty. We don't dare to delete any data
- raise DatabaseError, "Database %s contains tables"%\
- config.MYSQL_DBNAME
- db.sql("DROP DATABASE %s" % config.MYSQL_DBNAME)
- db.sql("CREATE DATABASE %s" % config.MYSQL_DBNAME)
- db.close()
- except (MySQLdb.ProgrammingError, DatabaseError), msg:
+ # Check if we can connect to the server.
+ # use db_exists() to make a connection, ignore it's return value
+ mysqlOpener.module.db_exists(config)
+ except (MySQLdb.MySQLError, DatabaseError), msg:
print "Skipping mysql tests (%s)"%msg
else:
print 'Including mysql tests'
suite.addTest(unittest.makeSuite(mysqlROTest))
suite.addTest(unittest.makeSuite(mysqlSchemaTest))
suite.addTest(unittest.makeSuite(mysqlClassicInitTest))
+ suite.addTest(unittest.makeSuite(mysqlSessionTest))
+ suite.addTest(unittest.makeSuite(mysqlConcurrencyTest))
+ suite.addTest(unittest.makeSuite(mysqlFilterCacheTest))
return suite
if __name__ == '__main__':
runner = unittest.TextTestRunner()
unittest.main(testRunner=runner)
+# vim: set et sts=4 sw=4 :