source: trunk/tests/test_dbapi20.py @ 848

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

Fix test with time zones under Python 3

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