← Back to branch summary

~ubuntu-branches/ubuntu/trusty/mysql-connector-python/trusty

« back to all changes in this revision

Viewing changes to python3/tests/test_mysql_datatypes.py

  • Committer: Package Import Robot
  • Author(s): Sandro Tosi, Jakub Wilk, Sandro Tosi
  • Date: 2014-02-18 16:26:46 UTC
  • mfrom: (1.2.1) (2.1.5 experimental)
  • Revision ID: package-import@ubuntu.com-20140218162646-1mvqa9g0ypnbqsq6
Tags: 1.1.5-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Sandro Tosi ]
* New upstream release
* debian/watch
  - updated to reflect upstream website changes
* debian/copyright
  - fix license, it's only GPLv2
  - update upstream copyright years
  - extend packaging copyright years
  - add reference to local GPL-2 file for packaging copyright notice
* debian/control
  - switch me to Maintainer (team to Uploaders)
  - bump Standards-Version to 3.9.5 (no changes needed)
* debian/rules
  - Disable test execution, currently broken upstream - see 739189
* Switch to dh_python2

Show diffs side-by-side

added added

removed removed

Lines of Context:
python3/tests/test_mysql_datatypes.py
1
 
# MySQL Connector/Python - MySQL driver written in Python.
2
 
# Copyright (c) 2009,2010, Oracle and/or its affiliates. All rights reserved.
3
 
# Use is subject to license terms. (See COPYING)
4
 
 
5
 
# This program is free software; you can redistribute it and/or modify
6
 
# it under the terms of the GNU General Public License as published by
7
 
# the Free Software Foundation.
8
 
# 
9
 
# There are special exceptions to the terms and conditions of the GNU
10
 
# General Public License as it is applied to this software. View the
11
 
# full text of the exception in file EXCEPTIONS-CLIENT in the directory
12
 
# of this software distribution or see the FOSS License Exception at
13
 
# www.mysql.com.
14
 
# 
15
 
# This program is distributed in the hope that it will be useful,
16
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
17
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
18
 
# GNU General Public License for more details.
19
 
# 
20
 
# You should have received a copy of the GNU General Public License
21
 
# along with this program; if not, write to the Free Software
22
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
23
 
 
24
 
"""Unittests for MySQL data types
25
 
"""
26
 
 
27
 
from decimal import Decimal
28
 
import time
29
 
import datetime
30
 
 
31
 
from mysql.connector import connection
32
 
import tests
33
 
 
34
 
class TestsDataTypes(tests.MySQLConnectorTests):
35
 
    
36
 
    tables = {
37
 
        'bit': 'myconnpy_mysql_bit',
38
 
        'int': 'myconnpy_mysql_int',
39
 
        'bool': 'myconnpy_mysql_bool',
40
 
        'float': 'myconnpy_mysql_float',
41
 
        'decimal': 'myconnpy_mysql_decimal',
42
 
        'temporal': 'myconnpy_mysql_temporal',
43
 
    }
44
 
    
45
 
    def _get_insert_stmt(self, tbl, cols):
46
 
        insert = """insert into %s (%s) values (%s)""" % (
47
 
            tbl, ','.join(cols),
48
 
            ','.join(['%s']*len(cols)))
49
 
        return insert
50
 
        
51
 
    def _get_select_stmt(self, tbl, cols):
52
 
        select = "SELECT %s FROM %s ORDER BY id" % (
53
 
            ','.join(cols), tbl)
54
 
        return select
55
 
        
56
 
class TestsCursor(TestsDataTypes):
57
 
    
58
 
    def setUp(self):
59
 
        config = self.getMySQLConfig()
60
 
        self.db = connection.MySQLConnection(**config)
61
 
        c = self.db.cursor()
62
 
        tblNames = self.tables.values()
63
 
        c.execute("DROP TABLE IF EXISTS %s" % (','.join(tblNames)))
64
 
        c.close()
65
 
    
66
 
    def tearDown(self):
67
 
        
68
 
        c = self.db.cursor()
69
 
        tblNames = self.tables.values()
70
 
        c.execute("DROP TABLE IF EXISTS %s" % (','.join(tblNames)))
71
 
        c.close()
72
 
        
73
 
        self.db.close()
74
 
    
75
 
    def test_numeric_int(self):
76
 
        """MySQL numeric integer data types"""
77
 
        c = self.db.cursor()
78
 
        columns = [
79
 
            'tinyint_signed',
80
 
            'tinyint_unsigned',
81
 
            'bool_signed',
82
 
            'smallint_signed',
83
 
            'smallint_unsigned',
84
 
            'mediumint_signed',
85
 
            'mediumint_unsigned',
86
 
            'int_signed',
87
 
            'int_unsigned',
88
 
            'bigint_signed',
89
 
            'bigint_unsigned',
90
 
        ]
91
 
        c.execute("""CREATE TABLE %s (
92
 
          `id` TINYINT UNSIGNED NOT NULL AUTO_INCREMENT,
93
 
          `tinyint_signed` TINYINT SIGNED,
94
 
          `tinyint_unsigned` TINYINT UNSIGNED,
95
 
          `bool_signed` BOOL,
96
 
          `smallint_signed` SMALLINT SIGNED,
97
 
          `smallint_unsigned` SMALLINT UNSIGNED,
98
 
          `mediumint_signed` MEDIUMINT SIGNED,
99
 
          `mediumint_unsigned` MEDIUMINT UNSIGNED,
100
 
          `int_signed` INT SIGNED,
101
 
          `int_unsigned` INT UNSIGNED,
102
 
          `bigint_signed` BIGINT SIGNED,
103
 
          `bigint_unsigned` BIGINT UNSIGNED,
104
 
          PRIMARY KEY (id)
105
 
          )
106
 
        """ % (self.tables['int']))
107
 
        
108
 
        data = [
109
 
            (
110
 
            -128, # tinyint signed
111
 
            0, # tinyint unsigned
112
 
            0, # boolean
113
 
            -32768, # smallint signed
114
 
            0, # smallint unsigned
115
 
            -8388608, # mediumint signed
116
 
            0, # mediumint unsigned
117
 
            -2147483648, # int signed
118
 
            0, # int unsigned
119
 
            -9223372036854775808, # big signed
120
 
            0, # big unsigned
121
 
            ),
122
 
            (
123
 
            127, # tinyint signed
124
 
            255, # tinyint unsigned
125
 
            127, # boolean
126
 
            32767, # smallint signed
127
 
            65535, # smallint unsigned
128
 
            8388607, # mediumint signed
129
 
            16777215, # mediumint unsigned
130
 
            2147483647, # int signed
131
 
            4294967295, # int unsigned
132
 
            9223372036854775807, # big signed
133
 
            18446744073709551615, # big unsigned
134
 
            )
135
 
        ]
136
 
 
137
 
        insert = self._get_insert_stmt(self.tables['int'],columns)
138
 
        select = self._get_select_stmt(self.tables['int'],columns)
139
 
        
140
 
        c.executemany(insert, data)
141
 
        c.execute(select)
142
 
        rows = c.fetchall()
143
 
        
144
 
        def compare(name, d, r):
145
 
           self.assertEqual(d,r,"%s  %s != %s" % (name,d,r))
146
 
            
147
 
        for i,col in enumerate(columns):
148
 
            compare(col,data[0][i],rows[0][i])
149
 
            compare(col,data[1][i],rows[1][i])
150
 
        
151
 
        c.close()
152
 
    
153
 
    def test_numeric_bit(self):
154
 
        """MySQL numeric bit data type"""
155
 
        c = self.db.cursor()
156
 
        columns = [
157
 
             'c8','c16','c24','c32',
158
 
            'c40','c48','c56','c63',
159
 
            'c64']
160
 
        c.execute("""CREATE TABLE %s (
161
 
          `id` int NOT NULL AUTO_INCREMENT,
162
 
          `c8` bit(8) DEFAULT NULL,
163
 
          `c16` bit(16) DEFAULT NULL,
164
 
          `c24` bit(24) DEFAULT NULL,
165
 
          `c32` bit(32) DEFAULT NULL,
166
 
          `c40` bit(40) DEFAULT NULL,
167
 
          `c48` bit(48) DEFAULT NULL,
168
 
          `c56` bit(56) DEFAULT NULL,
169
 
          `c63` bit(63) DEFAULT NULL,
170
 
          `c64` bit(64) DEFAULT NULL,
171
 
          PRIMARY KEY (id)
172
 
        )
173
 
        """ % self.tables['bit'])
174
 
        
175
 
        insert = self._get_insert_stmt(self.tables['bit'],columns)
176
 
        select = self._get_select_stmt(self.tables['bit'],columns)
177
 
        
178
 
        data = list()
179
 
        data.append(tuple([0]*len(columns)))
180
 
        
181
 
        values = list()
182
 
        for col in columns:
183
 
            values.append( 1 << int(col.replace('c',''))-1)
184
 
        data.append(tuple(values))
185
 
        
186
 
        values = list()
187
 
        for col in columns:
188
 
            values.append( (1 << int(col.replace('c',''))) -1)
189
 
        data.append(tuple(values))
190
 
        
191
 
        c.executemany(insert, data)
192
 
        c.execute(select)
193
 
        rows = c.fetchall()
194
 
 
195
 
        self.assertEqual(rows, data)
196
 
        c.close()
197
 
 
198
 
    def test_numeric_float(self):
199
 
        """MySQL numeric float data type"""
200
 
        c = self.db.cursor()
201
 
        columns = [
202
 
            'float_signed',
203
 
            'float_unsigned',
204
 
            'double_signed',
205
 
            'double_unsigned',
206
 
        ]
207
 
        c.execute("""CREATE TABLE %s (
208
 
            `id` int NOT NULL AUTO_INCREMENT,
209
 
            `float_signed` FLOAT(6,5) SIGNED,
210
 
            `float_unsigned` FLOAT(6,5) UNSIGNED,
211
 
            `double_signed` DOUBLE(15,10) SIGNED,
212
 
            `double_unsigned` DOUBLE(15,10) UNSIGNED,
213
 
            PRIMARY KEY (id)
214
 
        )""" % (self.tables['float']))
215
 
        
216
 
        insert = self._get_insert_stmt(self.tables['float'],columns)
217
 
        select = self._get_select_stmt(self.tables['float'],columns)
218
 
        
219
 
        data = [
220
 
            (-3.402823466,0,-1.7976931348623157,0,),
221
 
            (-1.175494351,3.402823466,1.7976931348623157,2.2250738585072014),
222
 
            (-1.23455678,2.999999,-1.3999999999999999,1.9999999999999999),
223
 
        ]
224
 
        c.executemany(insert, data)
225
 
        c.execute(select)
226
 
        rows = c.fetchall()
227
 
        
228
 
        def compare(name, d, r):
229
 
           self.assertEqual(d,r,"%s  %s != %s" % (name,d,r))
230
 
        
231
 
        for j in (range(0,len(data))):
232
 
            for i,col in enumerate(columns[0:2]):
233
 
                compare(col,round(data[j][i],5),rows[j][i])
234
 
            for i,col in enumerate(columns[2:2]):
235
 
                compare(col,round(data[j][i],10),rows[j][i])
236
 
        c.close()
237
 
    
238
 
    def test_numeric_decimal(self):
239
 
        """MySQL numeric decimal data type"""
240
 
        c = self.db.cursor()
241
 
        columns = [
242
 
            'decimal_signed',
243
 
            'decimal_unsigned',
244
 
        ]
245
 
        c.execute("""CREATE TABLE %s (
246
 
            `id` int NOT NULL AUTO_INCREMENT,
247
 
            `decimal_signed` DECIMAL(65,30) SIGNED,
248
 
            `decimal_unsigned` DECIMAL(65,30) UNSIGNED,
249
 
            PRIMARY KEY (id)
250
 
        )""" % (self.tables['decimal']))
251
 
        
252
 
        insert = self._get_insert_stmt(self.tables['decimal'],columns)
253
 
        select = self._get_select_stmt(self.tables['decimal'],columns)
254
 
        
255
 
        data = [
256
 
         (Decimal('-9999999999999999999999999.999999999999999999999999999999'),
257
 
          Decimal('+9999999999999999999999999.999999999999999999999999999999')),
258
 
         (Decimal('-1234567.1234'),
259
 
          Decimal('+123456789012345.123456789012345678901')),
260
 
         (Decimal('-1234567890123456789012345.123456789012345678901234567890'),
261
 
          Decimal('+1234567890123456789012345.123456789012345678901234567890')),
262
 
        ]
263
 
        c.executemany(insert, data)
264
 
        c.execute(select)
265
 
        rows = c.fetchall()
266
 
        
267
 
        self.assertEqual(data,rows)
268
 
        
269
 
        c.close()
270
 
    
271
 
    def test_temporal_datetime(self):
272
 
        """MySQL temporal date/time data types"""
273
 
        c = self.db.cursor()
274
 
        c.execute("SET SESSION time_zone = '+00:00'")
275
 
        columns = [
276
 
            't_date',
277
 
            't_datetime',
278
 
            't_time',
279
 
            't_timestamp',
280
 
            't_year_2',
281
 
            't_year_4',
282
 
        ]
283
 
        c.execute("""CREATE TABLE %s (
284
 
            `id` int NOT NULL AUTO_INCREMENT,
285
 
            `t_date` DATE,
286
 
            `t_datetime` DATETIME,
287
 
            `t_time` TIME,
288
 
            `t_timestamp` TIMESTAMP DEFAULT 0,
289
 
            `t_year_2` YEAR(2),
290
 
            `t_year_4` YEAR(4),
291
 
            PRIMARY KEY (id)
292
 
        )""" % (self.tables['temporal']))
293
 
        
294
 
        insert = self._get_insert_stmt(self.tables['temporal'],columns)
295
 
        select = self._get_select_stmt(self.tables['temporal'],columns)
296
 
        
297
 
        data = [
298
 
            (datetime.date(2010,1,17),
299
 
             datetime.datetime(2010,1,17,19,31,12),
300
 
             datetime.timedelta(hours=43,minutes=32,seconds=21),
301
 
             datetime.datetime(2010,1,17,19,31,12),
302
 
             10,
303
 
             0),
304
 
            (datetime.date(1000,1,1),
305
 
             datetime.datetime(1000,1,1,0,0,0),
306
 
             datetime.timedelta(hours=-838,minutes=59,seconds=59),
307
 
             datetime.datetime(*time.gmtime(1)[:6]),
308
 
             70,
309
 
             1901),
310
 
            (datetime.date(9999,12,31),
311
 
             datetime.datetime(9999,12,31,23,59,59),
312
 
             datetime.timedelta(hours=838,minutes=59,seconds=59),
313
 
             datetime.datetime(2038,1,19,3,14,7),
314
 
             69,
315
 
             2155),
316
 
        ]
317
 
        
318
 
        c.executemany(insert, data)
319
 
        c.execute(select)
320
 
        rows = c.fetchall()
321
 
        
322
 
        def compare(name, d, r):
323
 
           self.assertEqual(d,r,"%s  %s != %s" % (name,d,r))
324
 
        
325
 
        for j in (range(0,len(data))):
326
 
            for i,col in enumerate(columns):
327
 
                compare("%s (data[%d])" % (col,j),data[j][i],rows[j][i])
328
 
        
329
 
        c.close()