Initial commit.

mail/spamassassin/pyzor-0.7.0/tests/functional/test_mysql.py

@@ -0,0 +1,108 @@
+import unittest
+import ConfigParser
+
+from tests.util import *
+
+try:
+    import MySQLdb
+except ImportError:
+    MySQLdb = None
+    
+schema = """
+    CREATE TABLE IF NOT EXISTS `%s` (
+    `digest` char(40) default NULL,
+    `r_count` int(11) default NULL,
+    `wl_count` int(11) default NULL,
+    `r_entered` datetime default NULL,
+    `wl_entered` datetime default NULL,
+    `r_updated` datetime default NULL,
+    `wl_updated` datetime default NULL,
+    PRIMARY KEY  (`digest`)
+    )
+"""
+
+@unittest.skipIf(not os.path.exists("./test.conf"),
+                 "test.conf is not available")
+@unittest.skipIf(MySQLdb == None, "MySQLdb library not available")
+class MySQLdbPyzorTest(PyzorTest, PyzorTestBase):
+    """Test the mysql engine."""
+    dsn = None
+    engine = "mysql"
+    
+    @classmethod
+    def setUpClass(cls):
+        conf = ConfigParser.ConfigParser()    
+        conf.read("./test.conf")
+        table = conf.get("test", "table")        
+        db = MySQLdb.Connect(host=conf.get("test", "host"),
+                             user=conf.get("test", "user"),
+                             passwd=conf.get("test", "passwd"),
+                             db=conf.get("test", "db")) 
+        c = db.cursor()
+        c.execute(schema % table)
+        c.close()
+        db.close()
+        cls.dsn = "%s,%s,%s,%s,%s" % (conf.get("test", "host"),
+                                      conf.get("test", "user"),
+                                      conf.get("test", "passwd"),
+                                      conf.get("test", "db"),
+                                      conf.get("test", "table"))
+        super(MySQLdbPyzorTest, cls).setUpClass()
+        
+    @classmethod
+    def tearDownClass(cls):
+        super(MySQLdbPyzorTest, cls).tearDownClass()
+        try:
+            conf = ConfigParser.ConfigParser()    
+            conf.read("./test.conf")
+            table = conf.get("test", "table")        
+            db = MySQLdb.Connect(host=conf.get("test", "host"),
+                                 user=conf.get("test", "user"),
+                                 passwd=conf.get("test", "passwd"),
+                                 db=conf.get("test", "db"))
+            c = db.cursor()
+            c.execute("DROP TABLE %s" % table)
+            c.close()
+            db.close()
+        except:
+            pass
+
+class ThreadsMySQLdbPyzorTest(MySQLdbPyzorTest):
+    """Test the mysql engine with threads activated."""
+    threads = "True"
+    max_threads = "0"
+
+class BoundedThreadsMySQLdbPyzorTest(MySQLdbPyzorTest):
+    """Test the mysql engine with threads and DBConnections set."""
+    threads = "True"
+    max_threads = "0"
+    db_connections = "10"
+
+class MaxThreadsMySQLdbPyzorTest(MySQLdbPyzorTest):
+    """Test the mysql engine with threads and MaxThreads set."""
+    threads = "True"
+    max_threads = "10"
+
+class BoundedMaxThreadsMySQLdbPyzorTest(MySQLdbPyzorTest):
+    """Test the mysql engine with threads, MaxThreads and DBConnections set."""
+    threads = "True"
+    max_threads = "10"
+    db_connections = "10"
+    
+class ProcessesMySQLdbPyzorTest(MySQLdbPyzorTest):
+    processes = "True"
+    max_processes = "10"
+
+def suite():
+    """Gather all the tests from this module in a test suite."""
+    test_suite = unittest.TestSuite()
+    test_suite.addTest(unittest.makeSuite(MySQLdbPyzorTest))
+    test_suite.addTest(unittest.makeSuite(ThreadsMySQLdbPyzorTest))
+    test_suite.addTest(unittest.makeSuite(BoundedThreadsMySQLdbPyzorTest))
+    test_suite.addTest(unittest.makeSuite(MaxThreadsMySQLdbPyzorTest))
+    test_suite.addTest(unittest.makeSuite(BoundedMaxThreadsMySQLdbPyzorTest))
+    test_suite.addTest(unittest.makeSuite(ProcessesMySQLdbPyzorTest))
+    return test_suite
+        
+if __name__ == '__main__':
+    unittest.main()