X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=test%2Ftest_mysql.py;h=88f014d138851c1e53f52620e8a0be982ae308b1;hb=eb25ddf86e9f278d3e39137cae613789344dcfa2;hp=6e723ef57c2ca941c49d65a3af8aaf48a57227c8;hpb=109b5189786e67a5cfaf804408e282cd5d964ca5;p=roundup.git

diff --git a/test/test_mysql.py b/test/test_mysql.py
index 6e723ef..88f014d 100644
--- a/test/test_mysql.py
+++ b/test/test_mysql.py
@@ -14,105 +14,95 @@
 # 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.1 2003-10-25 22:53:26 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
+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:
-    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, [])
-
-    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()
-
-    from roundup import backends
-    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'
@@ -120,9 +110,13 @@ def test_suite():
         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 :