Merge pull request #293 from zzzeek/repair_datetime_comparison

Repair test_basic -> test_datatypes for MySQL 5.6 re: date rounding

Author: methane

pymysql/tests/base.py

@@ -1,6 +1,7 @@
 import os
 import json
 import pymysql
+import re
 try:
     import unittest2 as unittest
 except ImportError:
@@ -19,6 +20,23 @@ class PyMySQLTestCase(unittest.TestCase):
              "passwd":"","db":"test_pymysql", "use_unicode": True, 'local_infile': True},
             {"host":"localhost","user":"root","passwd":"","db":"test_pymysql2"}]
 
+    def mysql_server_is(self, conn, version_tuple):
+        """Return True if the given connection is on the version given or
+        greater.
+
+        e.g.::
+
+            if self.mysql_server_is(conn, (5, 6, 4)):
+                # do something for MySQL 5.6.4 and above
+        """
+        server_version = conn.get_server_info()
+        server_version_tuple = tuple(
+            (int(dig) if dig is not None else 0)
+            for dig in
+            re.match(r'(\d+)\.(\d+)\.(\d+)', server_version).group(1, 2, 3)
+        )
+        return server_version_tuple >= version_tuple
+
     def setUp(self):
         self.connections = []
         for params in self.databases:

pymysql/tests/test_basic.py

@@ -8,6 +8,12 @@
 import datetime
 import warnings
 
+try:
+    from unittest2 import SkipTest
+except:
+    from unittest import SkipTest
+
+
 __all__ = ["TestConversion", "TestCursor", "TestBulkInserts"]
 
 
@@ -19,16 +25,13 @@ def test_datatypes(self):
         c.execute("create table test_datatypes (b bit, i int, l bigint, f real, s varchar(32), u varchar(32), bb blob, d date, dt datetime, ts timestamp, td time, t time, st datetime)")
         try:
             # insert values
-            v = (True, -3, 123456789012, 5.7, "hello'\" world", u"Espa\xc3\xb1ol", "binary\x00data".encode(conn.charset), datetime.date(1988,2,2), datetime.datetime.now(), datetime.timedelta(5,6), datetime.time(16,32), time.localtime())
+
+            v = (True, -3, 123456789012, 5.7, "hello'\" world", u"Espa\xc3\xb1ol", "binary\x00data".encode(conn.charset), datetime.date(1988,2,2), datetime.datetime(2014, 5, 15, 7, 45, 57), datetime.timedelta(5,6), datetime.time(16,32), time.localtime())
             c.execute("insert into test_datatypes (b,i,l,f,s,u,bb,d,dt,td,t,st) values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", v)
             c.execute("select b,i,l,f,s,u,bb,d,dt,td,t,st from test_datatypes")
             r = c.fetchone()
             self.assertEqual(util.int2byte(1), r[0])
-            self.assertEqual(v[1:8], r[1:8])
-            # mysql throws away microseconds so we need to check datetimes
-            # specially. additionally times are turned into timedeltas.
-            self.assertEqual(datetime.datetime(*v[8].timetuple()[:6]), r[8])
-            self.assertEqual(v[9], r[9]) # just timedeltas
+            self.assertEqual(v[1:10], r[1:10])
             self.assertEqual(datetime.timedelta(0, 60 * (v[10].hour * 60 + v[10].minute)), r[10])
             self.assertEqual(datetime.datetime(*v[-1][:6]), r[-1])
 
@@ -123,23 +126,24 @@ def test_timedelta(self):
                           -datetime.timedelta(0, 1800)),
                          c.fetchone())
 
-    def test_datetime(self):
-        """ test datetime conversion """
+    def test_datetime_microseconds(self):
+        """ test datetime conversion w microseconds"""
+
         conn = self.connections[0]
+        if not self.mysql_server_is(conn, (5, 6, 4)):
+            raise SkipTest("target backend does not support microseconds")
         c = conn.cursor()
-        dt = datetime.datetime(2013,11,12,9,9,9,123450)
+        dt = datetime.datetime(2013, 11, 12, 9, 9, 9, 123450)
+        c.execute("create table test_datetime (id int, ts datetime(6))")
         try:
-            c.execute("create table test_datetime (id int, ts datetime(6))")
-            c.execute("insert into test_datetime values (1,'2013-11-12 09:09:09.12345')")
+            c.execute(
+                "insert into test_datetime values (%s, %s)",
+                (1, dt)
+            )
             c.execute("select ts from test_datetime")
-            self.assertEqual((dt,),c.fetchone())
-        except ProgrammingError:
-            # User is running a version of MySQL that doesn't support msecs within datetime
-            pass
+            self.assertEqual((dt,), c.fetchone())
         finally:
-            with warnings.catch_warnings():
-                warnings.filterwarnings("ignore")
-                c.execute("drop table if exists test_datetime")
+            c.execute("drop table test_datetime")
 
 
 class TestCursor(base.PyMySQLTestCase):