- Alter the datetime in test_basic->test_datatypes to be a

fixed value without microseconds, so that no truncation
or rounding of microseconds takes place; whether they
are truncated or rounded depends on backend.
- improve the test for test_datetime to round trip the datetime
value, passing it in the INSERT as well as getting it back
from the SELECT.  Use version detection to only run the test
on compliant backends; don't catch and discard a ProgrammingError
without knowing what it is.

Author: zzzeek

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):