source: trunk/module/TEST_PyGreSQL_dbapi20.py @ 553

Last change on this file since 553 was 553, checked in by cito, 4 years ago

Require at least Python 2.6 for the trunk (5.x)

Support for even older Python versions is maintained in the 4.x branch.
The goal for 5.x is to be a single-source code for both Python 2 and 3,
and this is only possible by dropping support for Python 2.5 and older.
For instance, the new except .. as syntax works only since Python 2.6.
Otherwise we would need to use 2to3 and things would be very ugly.
Note that Python 2.6 is now 7 years old. We may want to drop Python 2.6
as well at some point if it turns out to be a burden.

  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 9.5 KB
Line 
1#! /usr/bin/python
2# $Id: TEST_PyGreSQL_dbapi20.py 553 2015-11-20 11:16:03Z cito $
3
4import unittest
5import dbapi20
6import pgdb
7
8# We need a database to test against.
9# If LOCAL_PyGreSQL.py exists we will get our information from that.
10# Otherwise we use the defaults.
11dbname = 'dbapi20_test'
12dbhost = ''
13dbport = 5432
14try:
15    from LOCAL_PyGreSQL import *
16except ImportError:
17    pass
18
19
20class test_PyGreSQL(dbapi20.DatabaseAPI20Test):
21
22    driver = pgdb
23    connect_args = ()
24    connect_kw_args = {'dsn': dbhost + ':' + dbname}
25
26    lower_func = 'lower'  # For stored procedure test
27
28    def setUp(self):
29        # Call superclass setUp in case this does something in the future
30        dbapi20.DatabaseAPI20Test.setUp(self)
31        try:
32            con = self._connect()
33            con.close()
34        except pgdb.Error:
35            import pg
36            pg.DB().query('create database ' + dbname)
37
38    def tearDown(self):
39        dbapi20.DatabaseAPI20Test.tearDown(self)
40
41    def test_row_factory(self):
42        class myCursor(pgdb.pgdbCursor):
43            def row_factory(self, row):
44                d = {}
45                for idx, col in enumerate(self.description):
46                    d[col[0]] = row[idx]
47                return d
48
49        con = self._connect()
50        cur = myCursor(con)
51        ret = cur.execute("select 1 as a, 2 as b")
52        self.assert_(ret is cur, 'execute() should return cursor')
53        self.assertEqual(cur.fetchone(), {'a': 1, 'b': 2})
54
55    def test_cursor_iteration(self):
56        con = self._connect()
57        cur = con.cursor()
58        cur.execute("select 1 union select 2 union select 3")
59        self.assertEqual([r[0] for r in cur], [1, 2, 3])
60
61    def test_fetch_2_rows(self):
62        Decimal = pgdb.decimal_type()
63        values = ['test', pgdb.Binary('\xff\x52\xb2'),
64            True, 5, 6, 5.7, Decimal('234.234234'), Decimal('75.45'),
65            '2011-07-17', '15:47:42', '2008-10-20 15:25:35', '15:31:05',
66            7897234]
67        table = self.table_prefix + 'booze'
68        con = self._connect()
69        try:
70            cur = con.cursor()
71            cur.execute("create table %s ("
72                "stringtest varchar,"
73                "binarytest bytea,"
74                "booltest bool,"
75                "integertest int4,"
76                "longtest int8,"
77                "floattest float8,"
78                "numerictest numeric,"
79                "moneytest money,"
80                "datetest date,"
81                "timetest time,"
82                "datetimetest timestamp,"
83                "intervaltest interval,"
84                "rowidtest oid)" % table)
85            for s in ('numeric', 'monetary', 'time'):
86                cur.execute("set lc_%s to 'C'" % s)
87            for _i in range(2):
88                cur.execute("insert into %s values ("
89                    "%%s,%%s,%%s,%%s,%%s,%%s,%%s,"
90                    "'%%s'::money,%%s,%%s,%%s,%%s,%%s)" % table, values)
91            cur.execute("select * from %s" % table)
92            rows = cur.fetchall()
93            self.assertEqual(len(rows), 2)
94            self.assertEqual(rows[0], values)
95            self.assertEqual(rows[0], rows[1])
96        finally:
97            con.close()
98
99    def test_sqlstate(self):
100        con = self._connect()
101        cur = con.cursor()
102        try:
103            cur.execute("select 1/0")
104        except pgdb.DatabaseError as error:
105            self.assert_(isinstance(error, pgdb.ProgrammingError))
106            # the SQLSTATE error code for division by zero is 22012
107            self.assertEqual(error.sqlstate, '22012')
108
109    def test_float(self):
110        nan, inf = float('nan'), float('inf')
111        from math import isnan, isinf
112        self.assert_(isnan(nan) and not isinf(nan))
113        self.assert_(isinf(inf) and not isnan(inf))
114        values = [0, 1, 0.03125, -42.53125, nan, inf, -inf]
115        table = self.table_prefix + 'booze'
116        con = self._connect()
117        try:
118            cur = con.cursor()
119            cur.execute(
120                "create table %s (n smallint, floattest float)" % table)
121            params = enumerate(values)
122            cur.executemany("insert into %s values(%%s,%%s)" % table, params)
123            cur.execute("select * from %s order by 1" % table)
124            rows = cur.fetchall()
125        finally:
126            con.close()
127        self.assertEqual(len(rows), len(values))
128        rows = [row[1] for row in rows]
129        for inval, outval in zip(values, rows):
130            if isinf(inval):
131                self.assert_(isinf(outval))
132                if inval < 0:
133                    self.assert_(outval < 0)
134                else:
135                    self.assert_(outval > 0)
136            elif isnan(inval):
137                self.assert_(isnan(outval))
138            else:
139                self.assertEqual(inval, outval)
140
141    def test_set_decimal_type(self):
142        decimal_type = pgdb.decimal_type()
143        self.assert_(decimal_type is not None and callable(decimal_type))
144        con = self._connect()
145        try:
146            cur = con.cursor()
147            self.assert_(pgdb.decimal_type(int) is int)
148            cur.execute('select 42')
149            value = cur.fetchone()[0]
150            self.assert_(isinstance(value, int))
151            self.assertEqual(value, 42)
152            self.assert_(pgdb.decimal_type(float) is float)
153            cur.execute('select 4.25')
154            value = cur.fetchone()[0]
155            self.assert_(isinstance(value, float))
156            self.assertEqual(value, 4.25)
157        finally:
158            con.close()
159            pgdb.decimal_type(decimal_type)
160        self.assert_(pgdb.decimal_type() is decimal_type)
161
162    def test_nextset(self):
163        con = self._connect()
164        cur = con.cursor()
165        self.assertRaises(con.NotSupportedError, cur.nextset)
166
167    def test_setoutputsize(self):
168        pass  # not supported
169
170    def test_connection_errors(self):
171        con = self._connect()
172        self.assertEqual(con.Error, pgdb.Error)
173        self.assertEqual(con.Warning, pgdb.Warning)
174        self.assertEqual(con.InterfaceError, pgdb.InterfaceError)
175        self.assertEqual(con.DatabaseError, pgdb.DatabaseError)
176        self.assertEqual(con.InternalError, pgdb.InternalError)
177        self.assertEqual(con.OperationalError, pgdb.OperationalError)
178        self.assertEqual(con.ProgrammingError, pgdb.ProgrammingError)
179        self.assertEqual(con.IntegrityError, pgdb.IntegrityError)
180        self.assertEqual(con.DataError, pgdb.DataError)
181        self.assertEqual(con.NotSupportedError, pgdb.NotSupportedError)
182
183    def test_connection_as_contextmanager(self):
184        table = self.table_prefix + 'booze'
185        con = self._connect()
186        try:
187            cur = con.cursor()
188            cur.execute("create table %s (n smallint check(n!=4))" % table)
189            with con:
190                cur.execute("insert into %s values (1)" % table)
191                cur.execute("insert into %s values (2)" % table)
192            try:
193                with con:
194                    cur.execute("insert into %s values (3)" % table)
195                    cur.execute("insert into %s values (4)" % table)
196            except con.ProgrammingError as error:
197                self.assertTrue('check' in str(error).lower())
198            with con:
199                cur.execute("insert into %s values (5)" % table)
200                cur.execute("insert into %s values (6)" % table)
201            try:
202                with con:
203                    cur.execute("insert into %s values (7)" % table)
204                    cur.execute("insert into %s values (8)" % table)
205                    raise ValueError('transaction should rollback')
206            except ValueError as error:
207                self.assertEqual(str(error), 'transaction should rollback')
208            with con:
209                cur.execute("insert into %s values (9)" % table)
210            cur.execute("select * from %s order by 1" % table)
211            rows = cur.fetchall()
212            rows = [row[0] for row in rows]
213        finally:
214            con.close()
215        self.assertEqual(rows, [1, 2, 5, 6, 9])
216
217    def test_cursor_connection(self):
218        con = self._connect()
219        cur = con.cursor()
220        self.assertEqual(cur.connection, con)
221        cur.close()
222
223    def test_cursor_as_contextmanager(self):
224        con = self._connect()
225        with con.cursor() as cur:
226            self.assertEqual(cur.connection, con)
227
228    def test_pgdb_type(self):
229        self.assertEqual(pgdb.STRING, pgdb.STRING)
230        self.assertNotEqual(pgdb.STRING, pgdb.INTEGER)
231        self.assertNotEqual(pgdb.STRING, pgdb.BOOL)
232        self.assertNotEqual(pgdb.BOOL, pgdb.INTEGER)
233        self.assertEqual(pgdb.INTEGER, pgdb.INTEGER)
234        self.assertNotEqual(pgdb.INTEGER, pgdb.NUMBER)
235        self.assertEqual('char', pgdb.STRING)
236        self.assertEqual('varchar', pgdb.STRING)
237        self.assertEqual('text', pgdb.STRING)
238        self.assertNotEqual('numeric', pgdb.STRING)
239        self.assertEqual('numeric', pgdb.NUMERIC)
240        self.assertEqual('numeric', pgdb.NUMBER)
241        self.assertEqual('int4', pgdb.NUMBER)
242        self.assertNotEqual('int4', pgdb.NUMERIC)
243        self.assertEqual('int2', pgdb.SMALLINT)
244        self.assertNotEqual('int4', pgdb.SMALLINT)
245        self.assertEqual('int2', pgdb.INTEGER)
246        self.assertEqual('int4', pgdb.INTEGER)
247        self.assertEqual('int8', pgdb.INTEGER)
248        self.assertNotEqual('int4', pgdb.LONG)
249        self.assertEqual('int8', pgdb.LONG)
250        self.assert_('char' in pgdb.STRING)
251        self.assert_(pgdb.NUMERIC <= pgdb.NUMBER)
252        self.assert_(pgdb.NUMBER >= pgdb.INTEGER)
253        self.assert_(pgdb.TIME <= pgdb.DATETIME)
254        self.assert_(pgdb.DATETIME >= pgdb.DATE)
255
256
257if __name__ == '__main__':
258    unittest.main()
Note: See TracBrowser for help on using the repository browser.