Ticket #1235: mysql_timestamp_workaround.patch

File mysql_timestamp_workaround.patch, 10.0 kB (added by Felix.Schwarz, 2 years ago)

This patch obsoletes test_mysql_incompatibility.py too

  • turbogears/database.py

    old new  
    11"""Provides convenient access to an SQLObject-managed database.""" 
    22 
    33import sys 
     4import time 
    45import logging 
    56 
    67try: 
    78    import sqlobject 
    89    from sqlobject.dbconnection import ConnectionHub, Transaction, TheURIOpener 
     10    from sqlobject.util.threadinglocal import local as threading_local 
    911except ImportError: 
    1012    sqlobject = None 
    1113 
    …… 
    7173 
    7274 
    7375if sqlobject: 
     76    def _mysql_timestamp_converter(raw): 
     77        """Convert a MySQL TIMESTAMP to a floating point number representing 
     78        the seconds since the Un*x Epoch. It uses custom code the input seems  
     79        to be the new (MySQL 4.1+) timestamp format, otherwise code from the  
     80        MySQLdb module is used.""" 
     81        if raw[4] == '-': 
     82            return time.mktime(time.strptime(raw, '%Y-%m-%d %H:%M:%S')) 
     83        else: 
     84            import MySQLdb.converters 
     85            return MySQLdb.converters.mysql_timestamp_converter(raw) 
     86             
     87 
    7488    class AutoConnectHub(ConnectionHub): 
    7589        """Connects to the database once per thread. The AutoConnectHub also 
    7690        provides convenient methods for managing transactions.""" 
    …… 
    8599            hub_registry.add(self) 
    86100            ConnectionHub.__init__(self) 
    87101 
     102        def _is_interesting_version(self): 
     103            "Return True only if version of MySQLdb <= 1.0." 
     104            import MySQLdb 
     105            module_version = MySQLdb.version_info[0:2] 
     106            major = module_version[0] 
     107            minor = module_version[1] 
     108            # we can't use Decimal here because it is only available for Python 2.4 
     109            return (major < 1 or (major == 1 and minor < 2)) 
     110 
     111        def _enable_timestamp_workaround(self, connection): 
     112            """Enable a workaround for an incompatible timestamp format change  
     113            in MySQL 4.1 when using an old version of MySQLdb. See trac ticket  
     114            #1235 - http://trac.turbogears.org/ticket/1235 for details.""" 
     115            # precondition: connection is a MySQLConnection 
     116            import MySQLdb 
     117            import MySQLdb.converters 
     118            if self._is_interesting_version(): 
     119                conversions = MySQLdb.converters.conversions.copy() 
     120                conversions[MySQLdb.constants.FIELD_TYPE.TIMESTAMP] = \ 
     121                    _mysql_timestamp_converter 
     122                # There is no method to use custom keywords when using  
     123                # "connectionForURI" in sqlobject so we have to insert the 
     124                # conversions afterwards. 
     125                connection.kw["conv"] = conversions 
     126 
    88127        def getConnection(self): 
    89128            try: 
    90129                conn = self.threadingLocal.connection 
    …… 
    97136                    # and the cache causes problems with sqlite. 
    98137                    if self.uri.startswith("sqlite"): 
    99138                        TheURIOpener.cachedURIs = {} 
     139                    elif self.uri.startswith("mysql") and \ 
     140                         config.get("turbogears.enable_mysql41_timestamp_workaround", False): 
     141                        self._enable_timestamp_workaround(conn) 
    100142                    self.threadingLocal.connection = conn 
    101143                    return self.begin(conn) 
    102144                raise AttributeError( 

  • turbogears/tests/test_mysql_incompatibility.py

    old new  
     1#!/usr/bin/env python 
     2# -*- coding: UTF-8 -*- 
     3"""Unit test for incompatibility between MySQLdb 1.0 and MySQL 4.1+ as seen 
     4on RHEL 4 (Update 4). The main problem is that the Timestamp format changed 
     5in MySQL 4.1 and MySQLdb 1.0 does not support this MySQL version. 
     6 
     7A detailled explanation can be found in the ticket #1235: 
     8http://trac.turbogears.org/ticket/1235 
     9 
     10This test requires a proper environment (RHEL 4.4) and a real MySQL  
     11environment which makes this a bit tedious to test. 
     12"""  
     13 
     14import unittest 
     15 
     16from sqlobject import MixedCaseStyle, SQLObject 
     17 
     18from turbogears.database import PackageHub, _mysql_timestamp_converter 
     19import turbogears 
     20 
     21user = "root" 
     22host = "localhost" 
     23database_name = "test" 
     24 
     25connection_string = "mysql://%s@%s/%s" % (user, host, database_name) 
     26workaround_optionname = 'turbogears.enable_mysql41_timestamp_workaround' 
     27 
     28MySQLdb = None 
     29__connection__ = None 
     30FooBarKlass = None 
     31 
     32 
     33class TestTimestampWorkaround(unittest.TestCase): 
     34    """Test that TurboGears has workarounds for incompatible MySQL 
     35    libraries."""  
     36 
     37    def delete_table(self): 
     38        "Delete the table if it exists." 
     39        self.cursor.execute("DROP TABLE IF EXISTS `FooBarObject`") 
     40 
     41 
     42    def setup_database(self): 
     43        "Create the table manually and inserts one item." 
     44        if MySQLdb != None: 
     45            self.connection = MySQLdb.connect(user=user, db=database_name,  
     46                                              host=host) 
     47            self.cursor = self.connection.cursor() 
     48            self.delete_table() 
     49            self.cursor.execute("CREATE TABLE FooBarObject (" +  
     50                                "ID int(11) NOT NULL auto_increment PRIMARY KEY, "+ 
     51                                "time timestamp(14) NOT NULL)") 
     52            self.cursor.execute("INSERT INTO FooBarObject SET time=NOW()") 
     53 
     54 
     55    def define_class(self): 
     56        """Define the SQLObject class dynamically so that the database  
     57        configuration can be done first in the setUp method (SQLObject has 
     58        a class init method which tries to connect to the database as soon 
     59        the class is being defined which causes problems in this test.""" 
     60        global FooBarKlass 
     61        if FooBarKlass == None: 
     62            class FooBarObject(SQLObject): 
     63                class sqlmeta: 
     64                    fromDatabase = True 
     65                    style = MixedCaseStyle() 
     66            FooBarKlass = FooBarObject  
     67 
     68 
     69    def load_mysql_module(self): 
     70        "Loads the MySQLdb module but catches ImportErrors" 
     71        global MySQLdb 
     72        if MySQLdb is None: 
     73            try: 
     74                import MySQLdb 
     75            except ImportError: 
     76                # no MySQLdb module, do not execute this test  
     77                pass 
     78 
     79 
     80    def prepare_mysql(self): 
     81        """Install a package hub, setup a database, define the test class. 
     82        This method should not be called from setUp but from the actual test 
     83        case because the PackageHub does connect immediately and the  
     84        workaround configuration option must be set at this time already.""" 
     85        global __connection__ 
     86        if MySQLdb != None: 
     87            self.hub = PackageHub("turbogears.mysql") 
     88            __connection__ = self.hub 
     89            self.setup_database() 
     90            self.define_class() 
     91 
     92             
     93    def setUp(self): # pylint: disable-msg=C0103,C0111 
     94        self.load_mysql_module() 
     95        self._original_configuration = \ 
     96            turbogears.config.get(workaround_optionname, False) 
     97        self._original_dburi = turbogears.config.get('turbogears.mysql.dburi', None) 
     98        turbogears.config.update({'turbogears.mysql.dburi': connection_string }) 
     99        self.cursor = None 
     100        self.hub = None 
     101        
     102 
     103    def tearDown(self): # pylint: disable-msg=C0103,C0111 
     104        if MySQLdb != None: 
     105            if self.cursor != None: 
     106                self.delete_table() 
     107                self.cursor.close() 
     108                self.connection.close() 
     109            if self.cursor != None: 
     110                # Felix Schwarz: accessing private members is ugly, I know. But 
     111                # we have to disable SQLObject's pooling in order to force it 
     112                # really to make a new connection for every test case here! 
     113                connection = self.hub.threadingLocal.connection._dbConnection 
     114                connection._pool = None 
     115                self.hub.reset() 
     116            turbogears.config.update( 
     117                {'turbogears.mysql.dburi': self._original_dburi, 
     118                 workaround_optionname: self._original_configuration }) 
     119 
     120 
     121    def is_interesting_version(self): 
     122        "Return True only if version of MySQLdb <= 1.0." 
     123        module_version = MySQLdb.version_info[0:2] 
     124        major = module_version[0] 
     125        minor = module_version[1] 
     126        # we can't use Decimal here because it is only available for Python 2.4 
     127        return (major < 1 or (major == 1 and minor < 2)) 
     128 
     129     
     130    def test_timestamp_workarounds(self): 
     131        """Test that TurboGears has custom converters for incompatible MySQL 
     132        timestamp formats.""" 
     133        if MySQLdb != None and self.is_interesting_version(): 
     134            turbogears.config.update({ workaround_optionname: True }) 
     135            self.prepare_mysql() 
     136            for i in FooBarKlass.select(): 
     137                pass 
     138                 
     139    def test_timestamp_configuration(self): 
     140        "Test that the configuration must be explicitely enabled." 
     141        self.prepare_mysql() 
     142        if MySQLdb != None and self.is_interesting_version(): 
     143            try: 
     144                for i in FooBarKlass.select(): pass 
     145                self.fail("No exception although no workaround was enabled.") 
     146            except ValueError, e: 
     147                self.assertEqual("invalid literal for int(): 1-", e.args[0] ) 
     148 
     149    def test_converter(self): 
     150        "Test that the actual converter function is ok." 
     151        if MySQLdb != None: 
     152            self.assertEquals(1168789874.0,  
     153                              _mysql_timestamp_converter("2007-01-14 16:51:14")) 
     154            # Felix Schwarz: I'm not quite sure about this format, never saw it  
     155            # myself but got it from http://bugs.mysql.com/bug.php?id=20288 so I 
     156            # think it is correct.  
     157            # The second assertion ensures that connecting to old databases is  
     158            # still okay. 
     159            self.assertEquals(1168789874.0,  
     160                              _mysql_timestamp_converter("20070114165114")) 
     161         
     162 
     163if __name__ == "__main__": 
     164    unittest.main()