source: trunk/tests/test_dbapi20.py @ 849

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

Make timetz and timestamptz work properly with Python 2

Python 2 has no concrete timezone class, therefore we had so far returned only
naive datetimes in this case. This patch adds a simple concrete timezone class,
so we can return timetz and timestamptz with timezones in Python 2 as well.

  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 47.6 KB
Line 
1#! /usr/bin/python
2# -*- coding: utf-8 -*-
3# $Id: test_dbapi20.py 849 2016-02-09 13:14:26Z cito $
4
5try:
6    import unittest2 as unittest  # for Python < 2.7
7except ImportError:
8    import unittest
9
10import pgdb
11
12try:
13    from . import dbapi20
14except (ImportError, ValueError, SystemError):
15    import dbapi20
16
17# We need a database to test against.
18# If LOCAL_PyGreSQL.py exists we will get our information from that.
19# Otherwise we use the defaults.
20dbname = 'dbapi20_test'
21dbhost = ''
22dbport = 5432
23try:
24    from .LOCAL_PyGreSQL import *
25except (ImportError, ValueError):
26    try:
27        from LOCAL_PyGreSQL import *
28    except ImportError:
29        pass
30
31from datetime import date, time, datetime, timedelta
32from uuid import UUID as Uuid
33
34try:
35    long
36except NameError:  # Python >= 3.0
37    long = int
38
39try:
40    from collections import OrderedDict
41except ImportError:  # Python 2.6 or 3.0
42    OrderedDict = None
43
44
45class PgBitString:
46    """Test object with a PostgreSQL representation as Bit String."""
47
48    def __init__(self, value):
49        self.value = value
50
51    def __pg_repr__(self):
52         return "B'{0:b}'".format(self.value)
53
54
55class test_PyGreSQL(dbapi20.DatabaseAPI20Test):
56
57    driver = pgdb
58    connect_args = ()
59    connect_kw_args = {'database': dbname,
60        'host': '%s:%d' % (dbhost or '', dbport or -1)}
61
62    lower_func = 'lower'  # For stored procedure test
63
64    def setUp(self):
65        # Call superclass setUp in case this does something in the future
66        dbapi20.DatabaseAPI20Test.setUp(self)
67        try:
68            con = self._connect()
69            con.close()
70        except pgdb.Error:  # try to create a missing database
71            import pg
72            try:  # first try to log in as superuser
73                db = pg.DB('postgres', dbhost or None, dbport or -1,
74                    user='postgres')
75            except Exception:  # then try to log in as current user
76                db = pg.DB('postgres', dbhost or None, dbport or -1)
77            db.query('create database ' + dbname)
78
79    def tearDown(self):
80        dbapi20.DatabaseAPI20Test.tearDown(self)
81
82    def testVersion(self):
83        v = pgdb.version
84        self.assertIsInstance(v, str)
85        self.assertIn('.', v)
86        self.assertEqual(pgdb.__version__, v)
87
88    def test_percent_sign(self):
89        con = self._connect()
90        cur = con.cursor()
91        cur.execute("select %s, 'a %% sign'", ('a % sign',))
92        self.assertEqual(cur.fetchone(), ('a % sign', 'a % sign'))
93        cur.execute("select 'a % sign'")
94        self.assertEqual(cur.fetchone(), ('a % sign',))
95        cur.execute("select 'a %% sign'")
96        self.assertEqual(cur.fetchone(), ('a % sign',))
97
98    def test_callproc_no_params(self):
99        con = self._connect()
100        cur = con.cursor()
101        # note that now() does not change within a transaction
102        cur.execute('select now()')
103        now = cur.fetchone()[0]
104        res = cur.callproc('now')
105        self.assertIsNone(res)
106        res = cur.fetchone()[0]
107        self.assertEqual(res, now)
108
109    def test_callproc_bad_params(self):
110        con = self._connect()
111        cur = con.cursor()
112        self.assertRaises(TypeError, cur.callproc, 'lower', 42)
113        self.assertRaises(pgdb.ProgrammingError, cur.callproc, 'lower', (42,))
114
115    def test_callproc_one_param(self):
116        con = self._connect()
117        cur = con.cursor()
118        params = (42.4382,)
119        res = cur.callproc("round", params)
120        self.assertIs(res, params)
121        res = cur.fetchone()[0]
122        self.assertEqual(res, 42)
123
124    def test_callproc_two_params(self):
125        con = self._connect()
126        cur = con.cursor()
127        params = (9, 4)
128        res = cur.callproc("div", params)
129        self.assertIs(res, params)
130        res = cur.fetchone()[0]
131        self.assertEqual(res, 2)
132
133    def test_cursor_type(self):
134
135        class TestCursor(pgdb.Cursor):
136            pass
137
138        con = self._connect()
139        self.assertIs(con.cursor_type, pgdb.Cursor)
140        cur = con.cursor()
141        self.assertIsInstance(cur, pgdb.Cursor)
142        self.assertNotIsInstance(cur, TestCursor)
143        con.cursor_type = TestCursor
144        cur = con.cursor()
145        self.assertIsInstance(cur, TestCursor)
146        cur = con.cursor()
147        self.assertIsInstance(cur, TestCursor)
148        con = self._connect()
149        self.assertIs(con.cursor_type, pgdb.Cursor)
150        cur = con.cursor()
151        self.assertIsInstance(cur, pgdb.Cursor)
152        self.assertNotIsInstance(cur, TestCursor)
153
154    def test_row_factory(self):
155
156        class TestCursor(pgdb.Cursor):
157
158            def row_factory(self, row):
159                return dict(('column %s' % desc[0], value)
160                    for desc, value in zip(self.description, row))
161
162        con = self._connect()
163        con.cursor_type = TestCursor
164        cur = con.cursor()
165        self.assertIsInstance(cur, TestCursor)
166        res = cur.execute("select 1 as a, 2 as b")
167        self.assertIs(res, cur, 'execute() should return cursor')
168        res = cur.fetchone()
169        self.assertIsInstance(res, dict)
170        self.assertEqual(res, {'column a': 1, 'column b': 2})
171        cur.execute("select 1 as a, 2 as b union select 3, 4 order by 1")
172        res = cur.fetchall()
173        self.assertIsInstance(res, list)
174        self.assertEqual(len(res), 2)
175        self.assertIsInstance(res[0], dict)
176        self.assertEqual(res[0], {'column a': 1, 'column b': 2})
177        self.assertIsInstance(res[1], dict)
178        self.assertEqual(res[1], {'column a': 3, 'column b': 4})
179
180    def test_build_row_factory(self):
181
182        class TestCursor(pgdb.Cursor):
183
184            def build_row_factory(self):
185                keys = [desc[0] for desc in self.description]
186                return lambda row: dict((key, value)
187                    for key, value in zip(keys, row))
188
189        con = self._connect()
190        con.cursor_type = TestCursor
191        cur = con.cursor()
192        self.assertIsInstance(cur, TestCursor)
193        cur.execute("select 1 as a, 2 as b")
194        res = cur.fetchone()
195        self.assertIsInstance(res, dict)
196        self.assertEqual(res, {'a': 1, 'b': 2})
197        cur.execute("select 1 as a, 2 as b union select 3, 4 order by 1")
198        res = cur.fetchall()
199        self.assertIsInstance(res, list)
200        self.assertEqual(len(res), 2)
201        self.assertIsInstance(res[0], dict)
202        self.assertEqual(res[0], {'a': 1, 'b': 2})
203        self.assertIsInstance(res[1], dict)
204        self.assertEqual(res[1], {'a': 3, 'b': 4})
205
206    def test_cursor_with_named_columns(self):
207        con = self._connect()
208        cur = con.cursor()
209        res = cur.execute("select 1 as abc, 2 as de, 3 as f")
210        self.assertIs(res, cur, 'execute() should return cursor')
211        res = cur.fetchone()
212        self.assertIsInstance(res, tuple)
213        self.assertEqual(res, (1, 2, 3))
214        self.assertEqual(res._fields, ('abc', 'de', 'f'))
215        self.assertEqual(res.abc, 1)
216        self.assertEqual(res.de, 2)
217        self.assertEqual(res.f, 3)
218        cur.execute("select 1 as one, 2 as two union select 3, 4 order by 1")
219        res = cur.fetchall()
220        self.assertIsInstance(res, list)
221        self.assertEqual(len(res), 2)
222        self.assertIsInstance(res[0], tuple)
223        self.assertEqual(res[0], (1, 2))
224        self.assertEqual(res[0]._fields, ('one', 'two'))
225        self.assertIsInstance(res[1], tuple)
226        self.assertEqual(res[1], (3, 4))
227        self.assertEqual(res[1]._fields, ('one', 'two'))
228
229    def test_cursor_with_unnamed_columns(self):
230        con = self._connect()
231        cur = con.cursor()
232        cur.execute("select 1, 2, 3")
233        res = cur.fetchone()
234        self.assertIsInstance(res, tuple)
235        self.assertEqual(res, (1, 2, 3))
236        old_py = OrderedDict is None  # Python 2.6 or 3.0
237        # old Python versions cannot rename tuple fields with underscore
238        if old_py:
239            self.assertEqual(res._fields, ('column_0', 'column_1', 'column_2'))
240        else:
241            self.assertEqual(res._fields, ('_0', '_1', '_2'))
242        cur.execute("select 1 as one, 2, 3 as three")
243        res = cur.fetchone()
244        self.assertIsInstance(res, tuple)
245        self.assertEqual(res, (1, 2, 3))
246        if old_py:  # cannot auto rename with underscore
247            self.assertEqual(res._fields, ('one', 'column_1', 'three'))
248        else:
249            self.assertEqual(res._fields, ('one', '_1', 'three'))
250        cur.execute("select 1 as abc, 2 as def")
251        res = cur.fetchone()
252        self.assertIsInstance(res, tuple)
253        self.assertEqual(res, (1, 2))
254        if old_py:
255            self.assertEqual(res._fields, ('column_0', 'column_1'))
256        else:
257            self.assertEqual(res._fields, ('abc', '_1'))
258
259    def test_colnames(self):
260        con = self._connect()
261        cur = con.cursor()
262        cur.execute("select 1, 2, 3")
263        names = cur.colnames
264        self.assertIsInstance(names, list)
265        self.assertEqual(names, ['?column?', '?column?', '?column?'])
266        cur.execute("select 1 as a, 2 as bc, 3 as def, 4 as g")
267        names = cur.colnames
268        self.assertIsInstance(names, list)
269        self.assertEqual(names, ['a', 'bc', 'def', 'g'])
270
271    def test_coltypes(self):
272        con = self._connect()
273        cur = con.cursor()
274        cur.execute("select 1::int2, 2::int4, 3::int8")
275        types = cur.coltypes
276        self.assertIsInstance(types, list)
277        self.assertEqual(types, ['int2', 'int4', 'int8'])
278
279    def test_description_fields(self):
280        con = self._connect()
281        cur = con.cursor()
282        cur.execute("select 123456789::int8 col0,"
283            " 123456.789::numeric(41, 13) as col1,"
284            " 'foobar'::char(39) as col2")
285        desc = cur.description
286        self.assertIsInstance(desc, list)
287        self.assertEqual(len(desc), 3)
288        cols = [('int8', 8, None), ('numeric', 41, 13), ('bpchar', 39, None)]
289        for i in range(3):
290            c, d = cols[i], desc[i]
291            self.assertIsInstance(d, tuple)
292            self.assertEqual(len(d), 7)
293            self.assertIsInstance(d.name, str)
294            self.assertEqual(d.name, 'col%d' % i)
295            self.assertIsInstance(d.type_code, str)
296            self.assertEqual(d.type_code, c[0])
297            self.assertIsNone(d.display_size)
298            self.assertIsInstance(d.internal_size, int)
299            self.assertEqual(d.internal_size, c[1])
300            if c[2] is not None:
301                self.assertIsInstance(d.precision, int)
302                self.assertEqual(d.precision, c[1])
303                self.assertIsInstance(d.scale, int)
304                self.assertEqual(d.scale, c[2])
305            else:
306                self.assertIsNone(d.precision)
307                self.assertIsNone(d.scale)
308            self.assertIsNone(d.null_ok)
309
310    def test_type_cache_info(self):
311        con = self._connect()
312        try:
313            cur = con.cursor()
314            type_cache = con.type_cache
315            self.assertNotIn('numeric', type_cache)
316            type_info = type_cache['numeric']
317            self.assertIn('numeric', type_cache)
318            self.assertEqual(type_info, 'numeric')
319            self.assertEqual(type_info.oid, 1700)
320            self.assertEqual(type_info.len, -1)
321            self.assertEqual(type_info.type, 'b')  # base
322            self.assertEqual(type_info.category, 'N')  # numeric
323            self.assertEqual(type_info.delim, ',')
324            self.assertEqual(type_info.relid, 0)
325            self.assertIs(con.type_cache[1700], type_info)
326            self.assertNotIn('pg_type', type_cache)
327            type_info = type_cache['pg_type']
328            self.assertIn('pg_type', type_cache)
329            self.assertEqual(type_info.type, 'c')  # composite
330            self.assertEqual(type_info.category, 'C')  # composite
331            cols = type_cache.get_fields('pg_type')
332            self.assertEqual(cols[0].name, 'typname')
333            typname = type_cache[cols[0].type]
334            self.assertEqual(typname, 'name')
335            self.assertEqual(typname.type, 'b')  # base
336            self.assertEqual(typname.category, 'S')  # string
337            self.assertEqual(cols[3].name, 'typlen')
338            typlen = type_cache[cols[3].type]
339            self.assertEqual(typlen, 'int2')
340            self.assertEqual(typlen.type, 'b')  # base
341            self.assertEqual(typlen.category, 'N')  # numeric
342            cur.close()
343            cur = con.cursor()
344            type_cache = con.type_cache
345            self.assertIn('numeric', type_cache)
346            cur.close()
347        finally:
348            con.close()
349        con = self._connect()
350        try:
351            cur = con.cursor()
352            type_cache = con.type_cache
353            self.assertNotIn('pg_type', type_cache)
354            self.assertEqual(type_cache.get('pg_type'), type_info)
355            self.assertIn('pg_type', type_cache)
356            self.assertIsNone(type_cache.get(
357                self.table_prefix + '_surely_does_not_exist'))
358            cur.close()
359        finally:
360            con.close()
361
362    def test_type_cache_typecast(self):
363        con = self._connect()
364        try:
365            cur = con.cursor()
366            type_cache = con.type_cache
367            self.assertIs(type_cache.get_typecast('int4'), int)
368            cast_int = lambda v: 'int(%s)' % v
369            type_cache.set_typecast('int4', cast_int)
370            query = 'select 2::int2, 4::int4, 8::int8'
371            cur.execute(query)
372            i2, i4, i8 = cur.fetchone()
373            self.assertEqual(i2, 2)
374            self.assertEqual(i4, 'int(4)')
375            self.assertEqual(i8, 8)
376            self.assertEqual(type_cache.typecast(42, 'int4'), 'int(42)')
377            type_cache.set_typecast(['int2', 'int8'], cast_int)
378            cur.execute(query)
379            i2, i4, i8 = cur.fetchone()
380            self.assertEqual(i2, 'int(2)')
381            self.assertEqual(i4, 'int(4)')
382            self.assertEqual(i8, 'int(8)')
383            type_cache.reset_typecast('int4')
384            cur.execute(query)
385            i2, i4, i8 = cur.fetchone()
386            self.assertEqual(i2, 'int(2)')
387            self.assertEqual(i4, 4)
388            self.assertEqual(i8, 'int(8)')
389            type_cache.reset_typecast(['int2', 'int8'])
390            cur.execute(query)
391            i2, i4, i8 = cur.fetchone()
392            self.assertEqual(i2, 2)
393            self.assertEqual(i4, 4)
394            self.assertEqual(i8, 8)
395            type_cache.set_typecast(['int2', 'int8'], cast_int)
396            cur.execute(query)
397            i2, i4, i8 = cur.fetchone()
398            self.assertEqual(i2, 'int(2)')
399            self.assertEqual(i4, 4)
400            self.assertEqual(i8, 'int(8)')
401            type_cache.reset_typecast()
402            cur.execute(query)
403            i2, i4, i8 = cur.fetchone()
404            self.assertEqual(i2, 2)
405            self.assertEqual(i4, 4)
406            self.assertEqual(i8, 8)
407            cur.close()
408        finally:
409            con.close()
410
411    def test_cursor_iteration(self):
412        con = self._connect()
413        cur = con.cursor()
414        cur.execute("select 1 union select 2 union select 3")
415        self.assertEqual([r[0] for r in cur], [1, 2, 3])
416
417    def test_cursor_invalidation(self):
418        con = self._connect()
419        cur = con.cursor()
420        cur.execute("select 1 union select 2")
421        self.assertEqual(cur.fetchone(), (1,))
422        self.assertFalse(con.closed)
423        con.close()
424        self.assertTrue(con.closed)
425        self.assertRaises(pgdb.OperationalError, cur.fetchone)
426
427    def test_fetch_2_rows(self):
428        Decimal = pgdb.decimal_type()
429        values = ('test', pgdb.Binary(b'\xff\x52\xb2'),
430            True, 5, 6, 5.7, Decimal('234.234234'), Decimal('75.45'),
431            pgdb.Date(2011, 7, 17), pgdb.Time(15, 47, 42),
432            pgdb.Timestamp(2008, 10, 20, 15, 25, 35),
433            pgdb.Interval(15, 31, 5), 7897234)
434        table = self.table_prefix + 'booze'
435        con = self._connect()
436        try:
437            cur = con.cursor()
438            cur.execute("set datestyle to iso")
439            cur.execute("create table %s ("
440                "stringtest varchar,"
441                "binarytest bytea,"
442                "booltest bool,"
443                "integertest int4,"
444                "longtest int8,"
445                "floattest float8,"
446                "numerictest numeric,"
447                "moneytest money,"
448                "datetest date,"
449                "timetest time,"
450                "datetimetest timestamp,"
451                "intervaltest interval,"
452                "rowidtest oid)" % table)
453            cur.execute("set standard_conforming_strings to on")
454            for s in ('numeric', 'monetary', 'time'):
455                cur.execute("set lc_%s to 'C'" % s)
456            for _i in range(2):
457                cur.execute("insert into %s values ("
458                    "%%s,%%s,%%s,%%s,%%s,%%s,%%s,"
459                    "'%%s'::money,%%s,%%s,%%s,%%s,%%s)" % table, values)
460            cur.execute("select * from %s" % table)
461            rows = cur.fetchall()
462            self.assertEqual(len(rows), 2)
463            row0 = rows[0]
464            self.assertEqual(row0, values)
465            self.assertEqual(row0, rows[1])
466            self.assertIsInstance(row0[0], str)
467            self.assertIsInstance(row0[1], bytes)
468            self.assertIsInstance(row0[2], bool)
469            self.assertIsInstance(row0[3], int)
470            self.assertIsInstance(row0[4], long)
471            self.assertIsInstance(row0[5], float)
472            self.assertIsInstance(row0[6], Decimal)
473            self.assertIsInstance(row0[7], Decimal)
474            self.assertIsInstance(row0[8], date)
475            self.assertIsInstance(row0[9], time)
476            self.assertIsInstance(row0[10], datetime)
477            self.assertIsInstance(row0[11], timedelta)
478        finally:
479            con.close()
480
481    def test_integrity_error(self):
482        table = self.table_prefix + 'booze'
483        con = self._connect()
484        try:
485            cur = con.cursor()
486            cur.execute("set client_min_messages = warning")
487            cur.execute("create table %s (i int primary key)" % table)
488            cur.execute("insert into %s values (1)" % table)
489            cur.execute("insert into %s values (2)" % table)
490            self.assertRaises(pgdb.IntegrityError, cur.execute,
491                "insert into %s values (1)" % table)
492        finally:
493            con.close()
494
495    def test_update_rowcount(self):
496        table = self.table_prefix + 'booze'
497        con = self._connect()
498        try:
499            cur = con.cursor()
500            cur.execute("create table %s (i int)" % table)
501            cur.execute("insert into %s values (1)" % table)
502            cur.execute("update %s set i=2 where i=2 returning i" % table)
503            self.assertEqual(cur.rowcount, 0)
504            cur.execute("update %s set i=2 where i=1 returning i" % table)
505            self.assertEqual(cur.rowcount, 1)
506            cur.close()
507            # keep rowcount even if cursor is closed (needed by SQLAlchemy)
508            self.assertEqual(cur.rowcount, 1)
509        finally:
510            con.close()
511
512    def test_sqlstate(self):
513        con = self._connect()
514        cur = con.cursor()
515        try:
516            cur.execute("select 1/0")
517        except pgdb.DatabaseError as error:
518            self.assertTrue(isinstance(error, pgdb.DataError))
519            # the SQLSTATE error code for division by zero is 22012
520            self.assertEqual(error.sqlstate, '22012')
521
522    def test_float(self):
523        nan, inf = float('nan'), float('inf')
524        from math import isnan, isinf
525        self.assertTrue(isnan(nan) and not isinf(nan))
526        self.assertTrue(isinf(inf) and not isnan(inf))
527        values = [0, 1, 0.03125, -42.53125, nan, inf, -inf,
528            'nan', 'inf', '-inf', 'NaN', 'Infinity', '-Infinity']
529        table = self.table_prefix + 'booze'
530        con = self._connect()
531        try:
532            cur = con.cursor()
533            cur.execute(
534                "create table %s (n smallint, floattest float)" % table)
535            params = enumerate(values)
536            cur.executemany("insert into %s values (%%d,%%s)" % table, params)
537            cur.execute("select floattest from %s order by n" % table)
538            rows = cur.fetchall()
539            self.assertEqual(cur.description[0].type_code, pgdb.FLOAT)
540            self.assertNotEqual(cur.description[0].type_code, pgdb.ARRAY)
541            self.assertNotEqual(cur.description[0].type_code, pgdb.RECORD)
542        finally:
543            con.close()
544        self.assertEqual(len(rows), len(values))
545        rows = [row[0] for row in rows]
546        for inval, outval in zip(values, rows):
547            if inval in ('inf', 'Infinity'):
548                inval = inf
549            elif inval in ('-inf', '-Infinity'):
550                inval = -inf
551            elif inval in ('nan', 'NaN'):
552                inval = nan
553            if isinf(inval):
554                self.assertTrue(isinf(outval))
555                if inval < 0:
556                    self.assertTrue(outval < 0)
557                else:
558                    self.assertTrue(outval > 0)
559            elif isnan(inval):
560                self.assertTrue(isnan(outval))
561            else:
562                self.assertEqual(inval, outval)
563
564    def test_datetime(self):
565        dt = datetime(2011, 7, 17, 15, 47, 42, 317509)
566        table = self.table_prefix + 'booze'
567        con = self._connect()
568        try:
569            cur = con.cursor()
570            cur.execute("create table %s ("
571                "d date, t time,  ts timestamp,"
572                "tz timetz, tsz timestamptz)" % table)
573            for n in range(3):
574                values = [dt.date(), dt.time(), dt,
575                    dt.time(), dt]
576                values[3] = values[3].replace(tzinfo=pgdb.timezone.utc)
577                values[4] = values[4].replace(tzinfo=pgdb.timezone.utc)
578                if n == 0:  # input as objects
579                    params = values
580                if n == 1:  # input as text
581                    params = [v.isoformat() for v in values]  # as text
582                elif n == 2:  # input using type helpers
583                    d = (dt.year, dt.month, dt.day)
584                    t = (dt.hour, dt.minute, dt.second, dt.microsecond)
585                    z = (pgdb.timezone.utc,)
586                    params = [pgdb.Date(*d), pgdb.Time(*t),
587                            pgdb.Timestamp(*(d + t)), pgdb.Time(*(t + z)),
588                            pgdb.Timestamp(*(d + t + z))]
589                for datestyle in ('iso', 'postgres, mdy', 'postgres, dmy',
590                        'sql, mdy', 'sql, dmy', 'german'):
591                    cur.execute("set datestyle to %s" % datestyle)
592                    if n != 1:
593                        cur.execute("select %s,%s,%s,%s,%s", params)
594                        row = cur.fetchone()
595                        self.assertEqual(row, tuple(values))
596                    cur.execute("insert into %s"
597                        " values (%%s,%%s,%%s,%%s,%%s)" % table, params)
598                    cur.execute("select * from %s" % table)
599                    d = cur.description
600                    for i in range(5):
601                        self.assertEqual(d[i].type_code, pgdb.DATETIME)
602                        self.assertNotEqual(d[i].type_code, pgdb.STRING)
603                        self.assertNotEqual(d[i].type_code, pgdb.ARRAY)
604                        self.assertNotEqual(d[i].type_code, pgdb.RECORD)
605                    self.assertEqual(d[0].type_code, pgdb.DATE)
606                    self.assertEqual(d[1].type_code, pgdb.TIME)
607                    self.assertEqual(d[2].type_code, pgdb.TIMESTAMP)
608                    self.assertEqual(d[3].type_code, pgdb.TIME)
609                    self.assertEqual(d[4].type_code, pgdb.TIMESTAMP)
610                    row = cur.fetchone()
611                    self.assertEqual(row, tuple(values))
612                    cur.execute("delete from %s" % table)
613        finally:
614            con.close()
615
616    def test_interval(self):
617        td = datetime(2011, 7, 17, 15, 47, 42, 317509) - datetime(1970, 1, 1)
618        table = self.table_prefix + 'booze'
619        con = self._connect()
620        try:
621            cur = con.cursor()
622            cur.execute("create table %s (i interval)" % table)
623            for n in range(3):
624                if n == 0:  # input as objects
625                    param = td
626                if n == 1:  # input as text
627                    param = '%d days %d seconds %d microseconds ' % (
628                        td.days, td.seconds, td.microseconds)
629                elif n == 2:  # input using type helpers
630                    param = pgdb.Interval(
631                        td.days, 0, 0, td.seconds, td.microseconds)
632                for intervalstyle in ('sql_standard ', 'postgres',
633                        'postgres_verbose', 'iso_8601'):
634                    cur.execute("set intervalstyle to %s" % intervalstyle)
635                    cur.execute("insert into %s"
636                        " values (%%s)" % table, [param])
637                    cur.execute("select * from %s" % table)
638                    tc = cur.description[0].type_code
639                    self.assertEqual(tc, pgdb.DATETIME)
640                    self.assertNotEqual(tc, pgdb.STRING)
641                    self.assertNotEqual(tc, pgdb.ARRAY)
642                    self.assertNotEqual(tc, pgdb.RECORD)
643                    self.assertEqual(tc, pgdb.INTERVAL)
644                    row = cur.fetchone()
645                    self.assertEqual(row, (td,))
646                    cur.execute("delete from %s" % table)
647        finally:
648            con.close()
649
650    def test_hstore(self):
651        con = self._connect()
652        try:
653            cur = con.cursor()
654            cur.execute("select 'k=>v'::hstore")
655        except pgdb.DatabaseError:
656            try:
657                cur.execute("create extension hstore")
658            except pgdb.DatabaseError:
659                self.skipTest("hstore extension not enabled")
660        finally:
661            con.close()
662        d = {'k': 'v', 'foo': 'bar', 'baz': 'whatever', 'back\\': '\\slash',
663            '1a': 'anything at all', '2=b': 'value = 2', '3>c': 'value > 3',
664            '4"c': 'value " 4', "5'c": "value ' 5", 'hello, world': '"hi!"',
665            'None': None, 'NULL': 'NULL', 'empty': ''}
666        con = self._connect()
667        try:
668            cur = con.cursor()
669            cur.execute("select %s::hstore", (pgdb.Hstore(d),))
670            result = cur.fetchone()[0]
671        finally:
672            con.close()
673        self.assertIsInstance(result, dict)
674        self.assertEqual(result, d)
675
676    def test_uuid(self):
677        self.assertIs(Uuid, pgdb.Uuid)
678        d = Uuid('{12345678-1234-5678-1234-567812345678}')
679        con = self._connect()
680        try:
681            cur = con.cursor()
682            cur.execute("select %s::uuid", (d,))
683            result = cur.fetchone()[0]
684        finally:
685            con.close()
686        self.assertIsInstance(result, Uuid)
687        self.assertEqual(result, d)
688
689    def test_insert_array(self):
690        values = [(None, None), ([], []), ([None], [[None], ['null']]),
691            ([1, 2, 3], [['a', 'b'], ['c', 'd']]),
692            ([20000, 25000, 25000, 30000],
693            [['breakfast', 'consulting'], ['meeting', 'lunch']]),
694            ([0, 1, -1], [['Hello, World!', '"Hi!"'], ['{x,y}', ' x y ']])]
695        table = self.table_prefix + 'booze'
696        con = self._connect()
697        try:
698            cur = con.cursor()
699            cur.execute("create table %s"
700                " (n smallint, i int[], t text[][])" % table)
701            params = [(n, v[0], v[1]) for n, v in enumerate(values)]
702            # Note that we must explicit casts because we are inserting
703            # empty arrays.  Otherwise this is not necessary.
704            cur.executemany("insert into %s values"
705                " (%%d,%%s::int[],%%s::text[][])" % table, params)
706            cur.execute("select i, t from %s order by n" % table)
707            d = cur.description
708            self.assertEqual(d[0].type_code, pgdb.ARRAY)
709            self.assertNotEqual(d[0].type_code, pgdb.RECORD)
710            self.assertEqual(d[0].type_code, pgdb.NUMBER)
711            self.assertEqual(d[0].type_code, pgdb.INTEGER)
712            self.assertEqual(d[1].type_code, pgdb.ARRAY)
713            self.assertNotEqual(d[1].type_code, pgdb.RECORD)
714            self.assertEqual(d[1].type_code, pgdb.STRING)
715            rows = cur.fetchall()
716        finally:
717            con.close()
718        self.assertEqual(rows, values)
719
720    def test_select_array(self):
721        values = ([1, 2, 3, None], ['a', 'b', 'c', None])
722        con = self._connect()
723        try:
724            cur = con.cursor()
725            cur.execute("select %s::int[], %s::text[]", values)
726            row = cur.fetchone()
727        finally:
728            con.close()
729        self.assertEqual(row, values)
730
731    def test_unicode_list_and_tuple(self):
732        value = (u'KÀse', u'WÃŒrstchen')
733        con = self._connect()
734        try:
735            cur = con.cursor()
736            try:
737                cur.execute("select %s, %s", value)
738            except pgdb.DatabaseError:
739                self.skipTest('database does not support latin-1')
740            row = cur.fetchone()
741            cur.execute("select %s, %s", (list(value), tuple(value)))
742            as_list, as_tuple = cur.fetchone()
743        finally:
744            con.close()
745        self.assertEqual(as_list, list(row))
746        self.assertEqual(as_tuple, tuple(row))
747
748    def test_insert_record(self):
749        values = [('John', 61), ('Jane', 63),
750                  ('Fred', None), ('Wilma', None),
751                  (None, 42), (None, None)]
752        table = self.table_prefix + 'booze'
753        record = self.table_prefix + 'munch'
754        con = self._connect()
755        try:
756            cur = con.cursor()
757            cur.execute("create type %s as (name varchar, age int)" % record)
758            cur.execute("create table %s (n smallint, r %s)" % (table, record))
759            params = enumerate(values)
760            cur.executemany("insert into %s values (%%d,%%s)" % table, params)
761            cur.execute("select r from %s order by n" % table)
762            type_code = cur.description[0].type_code
763            self.assertEqual(type_code, record)
764            self.assertEqual(type_code, pgdb.RECORD)
765            self.assertNotEqual(type_code, pgdb.ARRAY)
766            columns = con.type_cache.get_fields(type_code)
767            self.assertEqual(columns[0].name, 'name')
768            self.assertEqual(columns[1].name, 'age')
769            self.assertEqual(con.type_cache[columns[0].type], 'varchar')
770            self.assertEqual(con.type_cache[columns[1].type], 'int4')
771            rows = cur.fetchall()
772        finally:
773            cur.execute('drop table %s' % table)
774            cur.execute('drop type %s' % record)
775            con.close()
776        self.assertEqual(len(rows), len(values))
777        rows = [row[0] for row in rows]
778        self.assertEqual(rows, values)
779        self.assertEqual(rows[0].name, 'John')
780        self.assertEqual(rows[0].age, 61)
781
782    def test_select_record(self):
783        value = (1, 25000, 2.5, 'hello', 'Hello World!', 'Hello, World!',
784            '(test)', '(x,y)', ' x y ', 'null', None)
785        con = self._connect()
786        try:
787            cur = con.cursor()
788            cur.execute("select %s as test_record", [value])
789            self.assertEqual(cur.description[0].name, 'test_record')
790            self.assertEqual(cur.description[0].type_code, 'record')
791            row = cur.fetchone()[0]
792        finally:
793            con.close()
794        # Note that the element types get lost since we created an
795        # untyped record (an anonymous composite type). For the same
796        # reason this is also a normal tuple, not a named tuple.
797        text_row = tuple(None if v is None else str(v) for v in value)
798        self.assertEqual(row, text_row)
799
800    def test_custom_type(self):
801        values = [3, 5, 65]
802        values = list(map(PgBitString, values))
803        table = self.table_prefix + 'booze'
804        con = self._connect()
805        try:
806            cur = con.cursor()
807            params = enumerate(values)  # params have __pg_repr__ method
808            cur.execute(
809                'create table "%s" (n smallint, b bit varying(7))' % table)
810            cur.executemany("insert into %s values (%%s,%%s)" % table, params)
811            cur.execute("select * from %s" % table)
812            rows = cur.fetchall()
813        finally:
814            con.close()
815        self.assertEqual(len(rows), len(values))
816        con = self._connect()
817        try:
818            cur = con.cursor()
819            params = (1, object())  # an object that cannot be handled
820            self.assertRaises(pgdb.InterfaceError, cur.execute,
821                "insert into %s values (%%s,%%s)" % table, params)
822        finally:
823            con.close()
824
825    def test_set_decimal_type(self):
826        decimal_type = pgdb.decimal_type()
827        self.assertTrue(decimal_type is not None and callable(decimal_type))
828        con = self._connect()
829        try:
830            cur = con.cursor()
831            # change decimal type globally to int
832            int_type = lambda v: int(float(v))
833            self.assertTrue(pgdb.decimal_type(int_type) is int_type)
834            cur.execute('select 4.25')
835            self.assertEqual(cur.description[0].type_code, pgdb.NUMBER)
836            value = cur.fetchone()[0]
837            self.assertTrue(isinstance(value, int))
838            self.assertEqual(value, 4)
839            # change decimal type again to float
840            self.assertTrue(pgdb.decimal_type(float) is float)
841            cur.execute('select 4.25')
842            self.assertEqual(cur.description[0].type_code, pgdb.NUMBER)
843            value = cur.fetchone()[0]
844            # the connection still uses the old setting
845            self.assertTrue(isinstance(value, int))
846            # bust the cache for type functions for the connection
847            con.type_cache.reset_typecast()
848            cur.execute('select 4.25')
849            self.assertEqual(cur.description[0].type_code, pgdb.NUMBER)
850            value = cur.fetchone()[0]
851            # now the connection uses the new setting
852            self.assertTrue(isinstance(value, float))
853            self.assertEqual(value, 4.25)
854        finally:
855            con.close()
856            pgdb.decimal_type(decimal_type)
857        self.assertTrue(pgdb.decimal_type() is decimal_type)
858
859    def test_global_typecast(self):
860        try:
861            query = 'select 2::int2, 4::int4, 8::int8'
862            self.assertIs(pgdb.get_typecast('int4'), int)
863            cast_int = lambda v: 'int(%s)' % v
864            pgdb.set_typecast('int4', cast_int)
865            con = self._connect()
866            try:
867                i2, i4, i8 = con.cursor().execute(query).fetchone()
868            finally:
869                con.close()
870            self.assertEqual(i2, 2)
871            self.assertEqual(i4, 'int(4)')
872            self.assertEqual(i8, 8)
873            pgdb.set_typecast(['int2', 'int8'], cast_int)
874            con = self._connect()
875            try:
876                i2, i4, i8 = con.cursor().execute(query).fetchone()
877            finally:
878                con.close()
879            self.assertEqual(i2, 'int(2)')
880            self.assertEqual(i4, 'int(4)')
881            self.assertEqual(i8, 'int(8)')
882            pgdb.reset_typecast('int4')
883            con = self._connect()
884            try:
885                i2, i4, i8 = con.cursor().execute(query).fetchone()
886            finally:
887                con.close()
888            self.assertEqual(i2, 'int(2)')
889            self.assertEqual(i4, 4)
890            self.assertEqual(i8, 'int(8)')
891            pgdb.reset_typecast(['int2', 'int8'])
892            con = self._connect()
893            try:
894                i2, i4, i8 = con.cursor().execute(query).fetchone()
895            finally:
896                con.close()
897            self.assertEqual(i2, 2)
898            self.assertEqual(i4, 4)
899            self.assertEqual(i8, 8)
900            pgdb.set_typecast(['int2', 'int8'], cast_int)
901            con = self._connect()
902            try:
903                i2, i4, i8 = con.cursor().execute(query).fetchone()
904            finally:
905                con.close()
906            self.assertEqual(i2, 'int(2)')
907            self.assertEqual(i4, 4)
908            self.assertEqual(i8, 'int(8)')
909        finally:
910            pgdb.reset_typecast()
911        con = self._connect()
912        try:
913            i2, i4, i8 = con.cursor().execute(query).fetchone()
914        finally:
915            con.close()
916        self.assertEqual(i2, 2)
917        self.assertEqual(i4, 4)
918        self.assertEqual(i8, 8)
919
920    def test_unicode_with_utf8(self):
921        table = self.table_prefix + 'booze'
922        input = u"He wes Leovenaðes sone — liðe him be Drihten"
923        con = self._connect()
924        try:
925            cur = con.cursor()
926            cur.execute("create table %s (t text)" % table)
927            try:
928                cur.execute("set client_encoding=utf8")
929                cur.execute(u"select '%s'" % input)
930            except Exception:
931                self.skipTest("database does not support utf8")
932            output1 = cur.fetchone()[0]
933            cur.execute("insert into %s values (%%s)" % table, (input,))
934            cur.execute("select * from %s" % table)
935            output2 = cur.fetchone()[0]
936            cur.execute("select t = '%s' from %s" % (input, table))
937            output3 = cur.fetchone()[0]
938            cur.execute("select t = %%s from %s" % table, (input,))
939            output4 = cur.fetchone()[0]
940        finally:
941            con.close()
942        if str is bytes:  # Python < 3.0
943            input = input.encode('utf8')
944        self.assertIsInstance(output1, str)
945        self.assertEqual(output1, input)
946        self.assertIsInstance(output2, str)
947        self.assertEqual(output2, input)
948        self.assertIsInstance(output3, bool)
949        self.assertTrue(output3)
950        self.assertIsInstance(output4, bool)
951        self.assertTrue(output4)
952
953    def test_unicode_with_latin1(self):
954        table = self.table_prefix + 'booze'
955        input = u"Ehrt den König seine WÃŒrde, ehret uns der HÀnde Fleiß."
956        con = self._connect()
957        try:
958            cur = con.cursor()
959            cur.execute("create table %s (t text)" % table)
960            try:
961                cur.execute("set client_encoding=latin1")
962                cur.execute(u"select '%s'" % input)
963            except Exception:
964                self.skipTest("database does not support latin1")
965            output1 = cur.fetchone()[0]
966            cur.execute("insert into %s values (%%s)" % table, (input,))
967            cur.execute("select * from %s" % table)
968            output2 = cur.fetchone()[0]
969            cur.execute("select t = '%s' from %s" % (input, table))
970            output3 = cur.fetchone()[0]
971            cur.execute("select t = %%s from %s" % table, (input,))
972            output4 = cur.fetchone()[0]
973        finally:
974            con.close()
975        if str is bytes:  # Python < 3.0
976            input = input.encode('latin1')
977        self.assertIsInstance(output1, str)
978        self.assertEqual(output1, input)
979        self.assertIsInstance(output2, str)
980        self.assertEqual(output2, input)
981        self.assertIsInstance(output3, bool)
982        self.assertTrue(output3)
983        self.assertIsInstance(output4, bool)
984        self.assertTrue(output4)
985
986    def test_bool(self):
987        values = [False, True, None, 't', 'f', 'true', 'false']
988        table = self.table_prefix + 'booze'
989        con = self._connect()
990        try:
991            cur = con.cursor()
992            cur.execute(
993                "create table %s (n smallint, booltest bool)" % table)
994            params = enumerate(values)
995            cur.executemany("insert into %s values (%%s,%%s)" % table, params)
996            cur.execute("select booltest from %s order by n" % table)
997            rows = cur.fetchall()
998            self.assertEqual(cur.description[0].type_code, pgdb.BOOL)
999        finally:
1000            con.close()
1001        rows = [row[0] for row in rows]
1002        values[3] = values[5] = True
1003        values[4] = values[6] = False
1004        self.assertEqual(rows, values)
1005
1006    def test_literal(self):
1007        con = self._connect()
1008        try:
1009            cur = con.cursor()
1010            value = "lower('Hello')"
1011            cur.execute("select %s, %s", (value, pgdb.Literal(value)))
1012            row = cur.fetchone()
1013        finally:
1014            con.close()
1015        self.assertEqual(row, (value, 'hello'))
1016
1017
1018    def test_json(self):
1019        inval = {"employees":
1020            [{"firstName": "John", "lastName": "Doe", "age": 61}]}
1021        table = self.table_prefix + 'booze'
1022        con = self._connect()
1023        try:
1024            cur = con.cursor()
1025            try:
1026                cur.execute("create table %s (jsontest json)" % table)
1027            except pgdb.ProgrammingError:
1028                self.skipTest('database does not support json')
1029            params = (pgdb.Json(inval),)
1030            cur.execute("insert into %s values (%%s)" % table, params)
1031            cur.execute("select jsontest from %s" % table)
1032            outval = cur.fetchone()[0]
1033            self.assertEqual(cur.description[0].type_code, pgdb.JSON)
1034        finally:
1035            con.close()
1036        self.assertEqual(inval, outval)
1037
1038    def test_jsonb(self):
1039        inval = {"employees":
1040            [{"firstName": "John", "lastName": "Doe", "age": 61}]}
1041        table = self.table_prefix + 'booze'
1042        con = self._connect()
1043        try:
1044            cur = con.cursor()
1045            try:
1046                cur.execute("create table %s (jsonbtest jsonb)" % table)
1047            except pgdb.ProgrammingError:
1048                self.skipTest('database does not support jsonb')
1049            params = (pgdb.Json(inval),)
1050            cur.execute("insert into %s values (%%s)" % table, params)
1051            cur.execute("select jsonbtest from %s" % table)
1052            outval = cur.fetchone()[0]
1053            self.assertEqual(cur.description[0].type_code, pgdb.JSON)
1054        finally:
1055            con.close()
1056        self.assertEqual(inval, outval)
1057
1058    def test_execute_edge_cases(self):
1059        con = self._connect()
1060        try:
1061            cur = con.cursor()
1062            sql = 'invalid'  # should be ignored with empty parameter list
1063            cur.executemany(sql, [])
1064            sql = 'select %d + 1'
1065            cur.execute(sql, [(1,), (2,)])  # deprecated use of execute()
1066            self.assertEqual(cur.fetchone()[0], 3)
1067            sql = 'select 1/0'  # cannot be executed
1068            self.assertRaises(pgdb.DataError, cur.execute, sql)
1069            cur.close()
1070            con.rollback()
1071            if pgdb.shortcutmethods:
1072                res = con.execute('select %d', (1,)).fetchone()
1073                self.assertEqual(res, (1,))
1074                res = con.executemany('select %d', [(1,), (2,)]).fetchone()
1075                self.assertEqual(res, (2,))
1076        finally:
1077            con.close()
1078        sql = 'select 1'  # cannot be executed after connection is closed
1079        self.assertRaises(pgdb.OperationalError, cur.execute, sql)
1080
1081    def test_fetchmany_with_keep(self):
1082        con = self._connect()
1083        try:
1084            cur = con.cursor()
1085            self.assertEqual(cur.arraysize, 1)
1086            cur.execute('select * from generate_series(1, 25)')
1087            self.assertEqual(len(cur.fetchmany()), 1)
1088            self.assertEqual(len(cur.fetchmany()), 1)
1089            self.assertEqual(cur.arraysize, 1)
1090            cur.arraysize = 3
1091            self.assertEqual(len(cur.fetchmany()), 3)
1092            self.assertEqual(len(cur.fetchmany()), 3)
1093            self.assertEqual(cur.arraysize, 3)
1094            self.assertEqual(len(cur.fetchmany(size=2)), 2)
1095            self.assertEqual(cur.arraysize, 3)
1096            self.assertEqual(len(cur.fetchmany()), 3)
1097            self.assertEqual(len(cur.fetchmany()), 3)
1098            self.assertEqual(len(cur.fetchmany(size=2, keep=True)), 2)
1099            self.assertEqual(cur.arraysize, 2)
1100            self.assertEqual(len(cur.fetchmany()), 2)
1101            self.assertEqual(len(cur.fetchmany()), 2)
1102            self.assertEqual(len(cur.fetchmany(25)), 3)
1103        finally:
1104            con.close()
1105
1106    def test_nextset(self):
1107        con = self._connect()
1108        cur = con.cursor()
1109        self.assertRaises(con.NotSupportedError, cur.nextset)
1110
1111    def test_setoutputsize(self):
1112        pass  # not supported
1113
1114    def test_connection_errors(self):
1115        con = self._connect()
1116        self.assertEqual(con.Error, pgdb.Error)
1117        self.assertEqual(con.Warning, pgdb.Warning)
1118        self.assertEqual(con.InterfaceError, pgdb.InterfaceError)
1119        self.assertEqual(con.DatabaseError, pgdb.DatabaseError)
1120        self.assertEqual(con.InternalError, pgdb.InternalError)
1121        self.assertEqual(con.OperationalError, pgdb.OperationalError)
1122        self.assertEqual(con.ProgrammingError, pgdb.ProgrammingError)
1123        self.assertEqual(con.IntegrityError, pgdb.IntegrityError)
1124        self.assertEqual(con.DataError, pgdb.DataError)
1125        self.assertEqual(con.NotSupportedError, pgdb.NotSupportedError)
1126
1127    def test_connection_as_contextmanager(self):
1128        table = self.table_prefix + 'booze'
1129        con = self._connect()
1130        try:
1131            cur = con.cursor()
1132            cur.execute("create table %s (n smallint check(n!=4))" % table)
1133            with con:
1134                cur.execute("insert into %s values (1)" % table)
1135                cur.execute("insert into %s values (2)" % table)
1136            try:
1137                with con:
1138                    cur.execute("insert into %s values (3)" % table)
1139                    cur.execute("insert into %s values (4)" % table)
1140            except con.IntegrityError as error:
1141                self.assertTrue('check' in str(error).lower())
1142            with con:
1143                cur.execute("insert into %s values (5)" % table)
1144                cur.execute("insert into %s values (6)" % table)
1145            try:
1146                with con:
1147                    cur.execute("insert into %s values (7)" % table)
1148                    cur.execute("insert into %s values (8)" % table)
1149                    raise ValueError('transaction should rollback')
1150            except ValueError as error:
1151                self.assertEqual(str(error), 'transaction should rollback')
1152            with con:
1153                cur.execute("insert into %s values (9)" % table)
1154            cur.execute("select * from %s order by 1" % table)
1155            rows = cur.fetchall()
1156            rows = [row[0] for row in rows]
1157        finally:
1158            con.close()
1159        self.assertEqual(rows, [1, 2, 5, 6, 9])
1160
1161    def test_cursor_connection(self):
1162        con = self._connect()
1163        cur = con.cursor()
1164        self.assertEqual(cur.connection, con)
1165        cur.close()
1166
1167    def test_cursor_as_contextmanager(self):
1168        con = self._connect()
1169        with con.cursor() as cur:
1170            self.assertEqual(cur.connection, con)
1171
1172    def test_pgdb_type(self):
1173        self.assertEqual(pgdb.STRING, pgdb.STRING)
1174        self.assertNotEqual(pgdb.STRING, pgdb.INTEGER)
1175        self.assertNotEqual(pgdb.STRING, pgdb.BOOL)
1176        self.assertNotEqual(pgdb.BOOL, pgdb.INTEGER)
1177        self.assertEqual(pgdb.INTEGER, pgdb.INTEGER)
1178        self.assertNotEqual(pgdb.INTEGER, pgdb.NUMBER)
1179        self.assertEqual('char', pgdb.STRING)
1180        self.assertEqual('varchar', pgdb.STRING)
1181        self.assertEqual('text', pgdb.STRING)
1182        self.assertNotEqual('numeric', pgdb.STRING)
1183        self.assertEqual('numeric', pgdb.NUMERIC)
1184        self.assertEqual('numeric', pgdb.NUMBER)
1185        self.assertEqual('int4', pgdb.NUMBER)
1186        self.assertNotEqual('int4', pgdb.NUMERIC)
1187        self.assertEqual('int2', pgdb.SMALLINT)
1188        self.assertNotEqual('int4', pgdb.SMALLINT)
1189        self.assertEqual('int2', pgdb.INTEGER)
1190        self.assertEqual('int4', pgdb.INTEGER)
1191        self.assertEqual('int8', pgdb.INTEGER)
1192        self.assertNotEqual('int4', pgdb.LONG)
1193        self.assertEqual('int8', pgdb.LONG)
1194        self.assertTrue('char' in pgdb.STRING)
1195        self.assertTrue(pgdb.NUMERIC <= pgdb.NUMBER)
1196        self.assertTrue(pgdb.NUMBER >= pgdb.INTEGER)
1197        self.assertTrue(pgdb.TIME <= pgdb.DATETIME)
1198        self.assertTrue(pgdb.DATETIME >= pgdb.DATE)
1199        self.assertEqual(pgdb.ARRAY, pgdb.ARRAY)
1200        self.assertNotEqual(pgdb.ARRAY, pgdb.STRING)
1201        self.assertEqual('_char', pgdb.ARRAY)
1202        self.assertNotEqual('char', pgdb.ARRAY)
1203        self.assertEqual(pgdb.RECORD, pgdb.RECORD)
1204        self.assertNotEqual(pgdb.RECORD, pgdb.STRING)
1205        self.assertNotEqual(pgdb.RECORD, pgdb.ARRAY)
1206        self.assertEqual('record', pgdb.RECORD)
1207        self.assertNotEqual('_record', pgdb.RECORD)
1208
1209
1210if __name__ == '__main__':
1211    unittest.main()
Note: See TracBrowser for help on using the repository browser.