diff --git a/test/test_mysql.py b/test/test_mysql.py
index 5825ff3050078e8d926362a89e46d10d926d8cf1..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.3 2003-11-02 08:44:17 richard 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'
-
- def testCreation(self):
- ae = self.assertEqual
-
- # create the instance
- init.install(self.dirname, 'templates/classic')
- init.write_select_db(self.dirname, self.backend)
- f = open(os.path.join(self.dirname, 'config.py'), 'a')
- try:
- f.write('''
-MYSQL_DBHOST = 'localhost'
-MYSQL_DBUSER = 'rounduptest'
-MYSQL_DBPASSWORD = 'rounduptest'
-MYSQL_DBNAME = 'rounduptest'
-MYSQL_DATABASE = (MYSQL_DBHOST, MYSQL_DBUSER, MYSQL_DBPASSWORD, MYSQL_DBNAME)
- ''')
- finally:
- f.close()
- init.initialise(self.dirname, 'sekrit')
-
- # check we can load the package
- instance = imp.load_package(self.dirname, self.dirname)
-
- # and open the database
- db = instance.open()
-
- # check the basics of the schema and initial data set
- l = db.priority.list()
- ae(l, ['1', '2', '3', '4', '5'])
- l = db.status.list()
- ae(l, ['1', '2', '3', '4', '5', '6', '7', '8'])
- l = db.keyword.list()
- ae(l, [])
- l = db.user.list()
- ae(l, ['1', '2'])
- l = db.msg.list()
- ae(l, [])
- l = db.file.list()
- ae(l, [])
- l = db.issue.list()
- ae(l, [])
-
- 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 :