← Back to branch summary

~ubuntu-branches/debian/sid/sqlalchemy/sid

« back to all changes in this revision

Viewing changes to test/dialect/test_postgresql.py

  • Committer: Bazaar Package Importer
  • Author(s): Piotr Ożarowski
  • Date: 2011-10-18 00:02:50 UTC
  • mfrom: (1.4.16 upstream)
  • Revision ID: james.westby@ubuntu.com-20111018000250-prowqcleosluapxg
Tags: 0.7.3-2
http://bugs.debian.org/645697
remove build-indep from build target's dependencies (closes: 645697)

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/dialect/test_postgresql.py
174
174
                            '(data text_pattern_ops, data2 int4_ops)',
175
175
                            dialect=postgresql.dialect())
176
176
 
 
177
    def test_create_index_with_using(self):
 
178
        m = MetaData()
 
179
        tbl = Table('testtbl', m, Column('data', String))
 
180
 
 
181
        idx1 = Index('test_idx1', tbl.c.data)
 
182
        idx2 = Index('test_idx2', tbl.c.data, postgresql_using='btree')
 
183
        idx3 = Index('test_idx3', tbl.c.data, postgresql_using='hash')
 
184
 
 
185
        self.assert_compile(schema.CreateIndex(idx1),
 
186
                            'CREATE INDEX test_idx1 ON testtbl '
 
187
                            '(data)',
 
188
                            dialect=postgresql.dialect())
 
189
        self.assert_compile(schema.CreateIndex(idx2),
 
190
                            'CREATE INDEX test_idx2 ON testtbl '
 
191
                            'USING btree (data)',
 
192
                            dialect=postgresql.dialect())
 
193
        self.assert_compile(schema.CreateIndex(idx3),
 
194
                            'CREATE INDEX test_idx3 ON testtbl '
 
195
                            'USING hash (data)',
 
196
                            dialect=postgresql.dialect())
 
197
 
177
198
    @testing.uses_deprecated(r".*'postgres_where' argument has been "
178
199
                             "renamed.*")
179
200
    def test_old_create_partial_index(self):
1255
1276
            "t.b AS b FROM t) AS sq WHERE t.id = sq.id"
1256
1277
            )
1257
1278
 
 
1279
class ReflectionTest(fixtures.TestBase):
 
1280
    __only_on__ = 'postgresql'
 
1281
 
 
1282
    @testing.provide_metadata
 
1283
    def test_pg_weirdchar_reflection(self):
 
1284
        meta1 = self.metadata
 
1285
        subject = Table('subject', meta1, Column('id$', Integer,
 
1286
                        primary_key=True))
 
1287
        referer = Table('referer', meta1, Column('id', Integer,
 
1288
                        primary_key=True), Column('ref', Integer,
 
1289
                        ForeignKey('subject.id$')))
 
1290
        meta1.create_all()
 
1291
        meta2 = MetaData(testing.db)
 
1292
        subject = Table('subject', meta2, autoload=True)
 
1293
        referer = Table('referer', meta2, autoload=True)
 
1294
        self.assert_((subject.c['id$']
 
1295
                     == referer.c.ref).compare(
 
1296
                        subject.join(referer).onclause))
 
1297
 
 
1298
    @testing.provide_metadata
 
1299
    def test_renamed_sequence_reflection(self):
 
1300
        metadata = self.metadata
 
1301
        t = Table('t', metadata, Column('id', Integer, primary_key=True))
 
1302
        metadata.create_all()
 
1303
        m2 = MetaData(testing.db)
 
1304
        t2 = Table('t', m2, autoload=True, implicit_returning=False)
 
1305
        eq_(t2.c.id.server_default.arg.text,
 
1306
            "nextval('t_id_seq'::regclass)")
 
1307
        r = t2.insert().execute()
 
1308
        eq_(r.inserted_primary_key, [1])
 
1309
        testing.db.connect().execution_options(autocommit=True).\
 
1310
                execute('alter table t_id_seq rename to foobar_id_seq'
 
1311
                )
 
1312
        m3 = MetaData(testing.db)
 
1313
        t3 = Table('t', m3, autoload=True, implicit_returning=False)
 
1314
        eq_(t3.c.id.server_default.arg.text,
 
1315
            "nextval('foobar_id_seq'::regclass)")
 
1316
        r = t3.insert().execute()
 
1317
        eq_(r.inserted_primary_key, [2])
 
1318
 
 
1319
    @testing.provide_metadata
 
1320
    def test_renamed_pk_reflection(self):
 
1321
        metadata = self.metadata
 
1322
        t = Table('t', metadata, Column('id', Integer, primary_key=True))
 
1323
        metadata.create_all()
 
1324
        testing.db.connect().execution_options(autocommit=True).\
 
1325
            execute('alter table t rename id to t_id')
 
1326
        m2 = MetaData(testing.db)
 
1327
        t2 = Table('t', m2, autoload=True)
 
1328
        eq_([c.name for c in t2.primary_key], ['t_id'])
 
1329
 
 
1330
    @testing.provide_metadata
 
1331
    def test_schema_reflection(self):
 
1332
        """note: this test requires that the 'test_schema' schema be
 
1333
        separate and accessible by the test user"""
 
1334
 
 
1335
        meta1 = self.metadata
 
1336
 
 
1337
        users = Table('users', meta1, Column('user_id', Integer,
 
1338
                      primary_key=True), Column('user_name',
 
1339
                      String(30), nullable=False), schema='test_schema')
 
1340
        addresses = Table(
 
1341
            'email_addresses',
 
1342
            meta1,
 
1343
            Column('address_id', Integer, primary_key=True),
 
1344
            Column('remote_user_id', Integer,
 
1345
                   ForeignKey(users.c.user_id)),
 
1346
            Column('email_address', String(20)),
 
1347
            schema='test_schema',
 
1348
            )
 
1349
        meta1.create_all()
 
1350
        meta2 = MetaData(testing.db)
 
1351
        addresses = Table('email_addresses', meta2, autoload=True,
 
1352
                          schema='test_schema')
 
1353
        users = Table('users', meta2, mustexist=True,
 
1354
                      schema='test_schema')
 
1355
        j = join(users, addresses)
 
1356
        self.assert_((users.c.user_id
 
1357
                     == addresses.c.remote_user_id).compare(j.onclause))
 
1358
 
 
1359
    @testing.provide_metadata
 
1360
    def test_schema_reflection_2(self):
 
1361
        meta1 = self.metadata
 
1362
        subject = Table('subject', meta1, Column('id', Integer,
 
1363
                        primary_key=True))
 
1364
        referer = Table('referer', meta1, Column('id', Integer,
 
1365
                        primary_key=True), Column('ref', Integer,
 
1366
                        ForeignKey('subject.id')), schema='test_schema')
 
1367
        meta1.create_all()
 
1368
        meta2 = MetaData(testing.db)
 
1369
        subject = Table('subject', meta2, autoload=True)
 
1370
        referer = Table('referer', meta2, schema='test_schema',
 
1371
                        autoload=True)
 
1372
        self.assert_((subject.c.id
 
1373
                     == referer.c.ref).compare(
 
1374
                        subject.join(referer).onclause))
 
1375
 
 
1376
    @testing.provide_metadata
 
1377
    def test_schema_reflection_3(self):
 
1378
        meta1 = self.metadata
 
1379
        subject = Table('subject', meta1, Column('id', Integer,
 
1380
                        primary_key=True), schema='test_schema_2')
 
1381
        referer = Table('referer', meta1, Column('id', Integer,
 
1382
                        primary_key=True), Column('ref', Integer,
 
1383
                        ForeignKey('test_schema_2.subject.id')),
 
1384
                        schema='test_schema')
 
1385
        meta1.create_all()
 
1386
        meta2 = MetaData(testing.db)
 
1387
        subject = Table('subject', meta2, autoload=True,
 
1388
                        schema='test_schema_2')
 
1389
        referer = Table('referer', meta2, schema='test_schema',
 
1390
                        autoload=True)
 
1391
        self.assert_((subject.c.id
 
1392
                     == referer.c.ref).compare(
 
1393
                        subject.join(referer).onclause))
 
1394
 
 
1395
    @testing.provide_metadata
 
1396
    def test_uppercase_lowercase_table(self):
 
1397
        metadata = self.metadata
 
1398
 
 
1399
        a_table = Table('a', metadata, Column('x', Integer))
 
1400
        A_table = Table('A', metadata, Column('x', Integer))
 
1401
 
 
1402
        a_table.create()
 
1403
        assert testing.db.has_table("a")
 
1404
        assert not testing.db.has_table("A")
 
1405
        A_table.create(checkfirst=True)
 
1406
        assert testing.db.has_table("A")
 
1407
 
 
1408
    def test_uppercase_lowercase_sequence(self):
 
1409
 
 
1410
        a_seq = Sequence('a')
 
1411
        A_seq = Sequence('A')
 
1412
 
 
1413
        a_seq.create(testing.db)
 
1414
        assert testing.db.dialect.has_sequence(testing.db, "a")
 
1415
        assert not testing.db.dialect.has_sequence(testing.db, "A")
 
1416
        A_seq.create(testing.db, checkfirst=True)
 
1417
        assert testing.db.dialect.has_sequence(testing.db, "A")
 
1418
 
 
1419
        a_seq.drop(testing.db)
 
1420
        A_seq.drop(testing.db)
 
1421
 
 
1422
    def test_schema_reflection_multi_search_path(self):
 
1423
        """test the 'set the same schema' rule when
 
1424
        multiple schemas/search paths are in effect."""
 
1425
 
 
1426
        db = engines.testing_engine()
 
1427
        conn = db.connect()
 
1428
        trans = conn.begin()
 
1429
        try:
 
1430
            conn.execute("set search_path to test_schema_2, "
 
1431
                                "test_schema, public")
 
1432
            conn.dialect.default_schema_name = "test_schema_2"
 
1433
 
 
1434
            conn.execute("""
 
1435
            CREATE TABLE test_schema.some_table (
 
1436
                id SERIAL not null primary key
 
1437
            )
 
1438
            """)
 
1439
 
 
1440
            conn.execute("""
 
1441
            CREATE TABLE test_schema_2.some_other_table (
 
1442
                id SERIAL not null primary key,
 
1443
                sid INTEGER REFERENCES test_schema.some_table(id)
 
1444
            )
 
1445
            """)
 
1446
 
 
1447
            m1 = MetaData()
 
1448
 
 
1449
            t2_schema = Table('some_other_table', 
 
1450
                                m1, 
 
1451
                                schema="test_schema_2", 
 
1452
                                autoload=True, 
 
1453
                                autoload_with=conn)
 
1454
            t1_schema = Table('some_table', 
 
1455
                                m1, 
 
1456
                                schema="test_schema", 
 
1457
                                autoload=True,
 
1458
                                autoload_with=conn)
 
1459
 
 
1460
            t2_no_schema = Table('some_other_table', 
 
1461
                                m1, 
 
1462
                                autoload=True, 
 
1463
                                autoload_with=conn)
 
1464
 
 
1465
            t1_no_schema = Table('some_table', 
 
1466
                                m1, 
 
1467
                                autoload=True, 
 
1468
                                autoload_with=conn)
 
1469
 
 
1470
            # OK, this because, "test_schema" is 
 
1471
            # in the search path, and might as well be
 
1472
            # the default too.  why would we assign
 
1473
            # a "schema" to the Table ?
 
1474
            assert t2_schema.c.sid.references(
 
1475
                                t1_no_schema.c.id)
 
1476
 
 
1477
            assert t2_no_schema.c.sid.references(
 
1478
                                t1_no_schema.c.id)
 
1479
 
 
1480
        finally:
 
1481
            trans.rollback()
 
1482
            conn.close()
 
1483
            db.dispose()
 
1484
 
 
1485
    @testing.provide_metadata
 
1486
    def test_index_reflection(self):
 
1487
        """ Reflecting partial & expression-based indexes should warn
 
1488
        """
 
1489
 
 
1490
        metadata = self.metadata
 
1491
 
 
1492
        t1 = Table('party', metadata, Column('id', String(10),
 
1493
                   nullable=False), Column('name', String(20),
 
1494
                   index=True), Column('aname', String(20)))
 
1495
        metadata.create_all()
 
1496
        testing.db.execute("""
 
1497
          create index idx1 on party ((id || name))
 
1498
        """)
 
1499
        testing.db.execute("""
 
1500
          create unique index idx2 on party (id) where name = 'test'
 
1501
        """)
 
1502
        testing.db.execute("""
 
1503
            create index idx3 on party using btree
 
1504
                (lower(name::text), lower(aname::text))
 
1505
        """)
 
1506
 
 
1507
        def go():
 
1508
            m2 = MetaData(testing.db)
 
1509
            t2 = Table('party', m2, autoload=True)
 
1510
            assert len(t2.indexes) == 2
 
1511
 
 
1512
            # Make sure indexes are in the order we expect them in
 
1513
 
 
1514
            tmp = [(idx.name, idx) for idx in t2.indexes]
 
1515
            tmp.sort()
 
1516
            r1, r2 = [idx[1] for idx in tmp]
 
1517
            assert r1.name == 'idx2'
 
1518
            assert r1.unique == True
 
1519
            assert r2.unique == False
 
1520
            assert [t2.c.id] == r1.columns
 
1521
            assert [t2.c.name] == r2.columns
 
1522
 
 
1523
        testing.assert_warnings(go,
 
1524
            [
 
1525
                'Skipped unsupported reflection of '
 
1526
                'expression-based index idx1',
 
1527
                'Predicate of partial index idx2 ignored during '
 
1528
                'reflection',
 
1529
                'Skipped unsupported reflection of '
 
1530
                'expression-based index idx3'
 
1531
            ])
 
1532
 
 
1533
    @testing.provide_metadata
 
1534
    def test_index_reflection_modified(self):
 
1535
        """reflect indexes when a column name has changed - PG 9 
 
1536
        does not update the name of the column in the index def.
 
1537
        [ticket:2141]
 
1538
 
 
1539
        """
 
1540
 
 
1541
        metadata = self.metadata
 
1542
 
 
1543
        t1 = Table('t', metadata,
 
1544
            Column('id', Integer, primary_key=True),
 
1545
            Column('x', Integer)
 
1546
        )
 
1547
        metadata.create_all()
 
1548
        conn = testing.db.connect().execution_options(autocommit=True)
 
1549
        conn.execute("CREATE INDEX idx1 ON t (x)")
 
1550
        conn.execute("ALTER TABLE t RENAME COLUMN x to y")
 
1551
 
 
1552
        ind = testing.db.dialect.get_indexes(conn, "t", None)
 
1553
        eq_(ind, [{'unique': False, 'column_names': [u'y'], 'name': u'idx1'}])
 
1554
        conn.close()
1258
1555
 
1259
1556
class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
1260
1557
 
1325
1622
        assert 'will create implicit sequence' in msgs
1326
1623
        assert 'will create implicit index' in msgs
1327
1624
 
1328
 
    def test_pg_weirdchar_reflection(self):
1329
 
        meta1 = MetaData(testing.db)
1330
 
        subject = Table('subject', meta1, Column('id$', Integer,
1331
 
                        primary_key=True))
1332
 
        referer = Table('referer', meta1, Column('id', Integer,
1333
 
                        primary_key=True), Column('ref', Integer,
1334
 
                        ForeignKey('subject.id$')))
1335
 
        meta1.create_all()
1336
 
        try:
1337
 
            meta2 = MetaData(testing.db)
1338
 
            subject = Table('subject', meta2, autoload=True)
1339
 
            referer = Table('referer', meta2, autoload=True)
1340
 
            print str(subject.join(referer).onclause)
1341
 
            self.assert_((subject.c['id$']
1342
 
                         == referer.c.ref).compare(
1343
 
                            subject.join(referer).onclause))
1344
 
        finally:
1345
 
            meta1.drop_all()
 
1625
    @testing.only_on('postgresql+psycopg2', 'psycopg2-specific feature')
 
1626
    @engines.close_open_connections
 
1627
    def test_client_encoding(self):
 
1628
        c = testing.db.connect()
 
1629
        current_encoding = c.connection.connection.encoding
 
1630
        c.close()
 
1631
 
 
1632
        # attempt to use an encoding that's not 
 
1633
        # already set
 
1634
        if current_encoding == 'UTF8':
 
1635
            test_encoding = 'LATIN1'
 
1636
        else:
 
1637
            test_encoding = 'UTF8'
 
1638
 
 
1639
        e = engines.testing_engine(
 
1640
                        options={'client_encoding':test_encoding}
 
1641
                    )
 
1642
        c = e.connect()
 
1643
        eq_(c.connection.connection.encoding, test_encoding)
1346
1644
 
1347
1645
    @testing.fails_on('+zxjdbc',
1348
1646
                      "Can't infer the SQL type to use for an instance "
1369
1667
        finally:
1370
1668
            t.drop(checkfirst=True)
1371
1669
 
1372
 
    @testing.provide_metadata
1373
 
    def test_renamed_sequence_reflection(self):
1374
 
        metadata = self.metadata
1375
 
        t = Table('t', metadata, Column('id', Integer, primary_key=True))
1376
 
        metadata.create_all()
1377
 
        m2 = MetaData(testing.db)
1378
 
        t2 = Table('t', m2, autoload=True, implicit_returning=False)
1379
 
        eq_(t2.c.id.server_default.arg.text,
1380
 
            "nextval('t_id_seq'::regclass)")
1381
 
        r = t2.insert().execute()
1382
 
        eq_(r.inserted_primary_key, [1])
1383
 
        testing.db.connect().execution_options(autocommit=True).\
1384
 
                execute('alter table t_id_seq rename to foobar_id_seq'
1385
 
                )
1386
 
        m3 = MetaData(testing.db)
1387
 
        t3 = Table('t', m3, autoload=True, implicit_returning=False)
1388
 
        eq_(t3.c.id.server_default.arg.text,
1389
 
            "nextval('foobar_id_seq'::regclass)")
1390
 
        r = t3.insert().execute()
1391
 
        eq_(r.inserted_primary_key, [2])
1392
 
 
1393
 
    def test_schema_reflection(self):
1394
 
        """note: this test requires that the 'test_schema' schema be
1395
 
        separate and accessible by the test user"""
1396
 
 
1397
 
        meta1 = MetaData(testing.db)
1398
 
        users = Table('users', meta1, Column('user_id', Integer,
1399
 
                      primary_key=True), Column('user_name',
1400
 
                      String(30), nullable=False), schema='test_schema')
1401
 
        addresses = Table(
1402
 
            'email_addresses',
1403
 
            meta1,
1404
 
            Column('address_id', Integer, primary_key=True),
1405
 
            Column('remote_user_id', Integer,
1406
 
                   ForeignKey(users.c.user_id)),
1407
 
            Column('email_address', String(20)),
1408
 
            schema='test_schema',
1409
 
            )
1410
 
        meta1.create_all()
1411
 
        try:
1412
 
            meta2 = MetaData(testing.db)
1413
 
            addresses = Table('email_addresses', meta2, autoload=True,
1414
 
                              schema='test_schema')
1415
 
            users = Table('users', meta2, mustexist=True,
1416
 
                          schema='test_schema')
1417
 
            print users
1418
 
            print addresses
1419
 
            j = join(users, addresses)
1420
 
            print str(j.onclause)
1421
 
            self.assert_((users.c.user_id
1422
 
                         == addresses.c.remote_user_id).compare(j.onclause))
1423
 
        finally:
1424
 
            meta1.drop_all()
1425
 
 
1426
 
    def test_schema_reflection_2(self):
1427
 
        meta1 = MetaData(testing.db)
1428
 
        subject = Table('subject', meta1, Column('id', Integer,
1429
 
                        primary_key=True))
1430
 
        referer = Table('referer', meta1, Column('id', Integer,
1431
 
                        primary_key=True), Column('ref', Integer,
1432
 
                        ForeignKey('subject.id')), schema='test_schema')
1433
 
        meta1.create_all()
1434
 
        try:
1435
 
            meta2 = MetaData(testing.db)
1436
 
            subject = Table('subject', meta2, autoload=True)
1437
 
            referer = Table('referer', meta2, schema='test_schema',
1438
 
                            autoload=True)
1439
 
            print str(subject.join(referer).onclause)
1440
 
            self.assert_((subject.c.id
1441
 
                         == referer.c.ref).compare(
1442
 
                            subject.join(referer).onclause))
1443
 
        finally:
1444
 
            meta1.drop_all()
1445
 
 
1446
 
    def test_schema_reflection_3(self):
1447
 
        meta1 = MetaData(testing.db)
1448
 
        subject = Table('subject', meta1, Column('id', Integer,
1449
 
                        primary_key=True), schema='test_schema_2')
1450
 
        referer = Table('referer', meta1, Column('id', Integer,
1451
 
                        primary_key=True), Column('ref', Integer,
1452
 
                        ForeignKey('test_schema_2.subject.id')),
1453
 
                        schema='test_schema')
1454
 
        meta1.create_all()
1455
 
        try:
1456
 
            meta2 = MetaData(testing.db)
1457
 
            subject = Table('subject', meta2, autoload=True,
1458
 
                            schema='test_schema_2')
1459
 
            referer = Table('referer', meta2, schema='test_schema',
1460
 
                            autoload=True)
1461
 
            print str(subject.join(referer).onclause)
1462
 
            self.assert_((subject.c.id
1463
 
                         == referer.c.ref).compare(
1464
 
                            subject.join(referer).onclause))
1465
 
        finally:
1466
 
            meta1.drop_all()
1467
 
 
1468
1670
    def test_schema_roundtrips(self):
1469
1671
        meta = MetaData(testing.db)
1470
1672
        users = Table('users', meta, Column('id', Integer,
1515
1717
        finally:
1516
1718
            testing.db.execute('drop table speedy_users')
1517
1719
 
1518
 
    @testing.provide_metadata
1519
 
    def test_index_reflection(self):
1520
 
        """ Reflecting partial & expression-based indexes should warn
1521
 
        """
1522
 
 
1523
 
        metadata = self.metadata
1524
 
 
1525
 
        t1 = Table('party', metadata, Column('id', String(10),
1526
 
                   nullable=False), Column('name', String(20),
1527
 
                   index=True), Column('aname', String(20)))
1528
 
        metadata.create_all()
1529
 
        testing.db.execute("""
1530
 
          create index idx1 on party ((id || name))
1531
 
        """)
1532
 
        testing.db.execute("""
1533
 
          create unique index idx2 on party (id) where name = 'test'
1534
 
        """)
1535
 
        testing.db.execute("""
1536
 
            create index idx3 on party using btree
1537
 
                (lower(name::text), lower(aname::text))
1538
 
        """)
1539
 
 
1540
 
        def go():
1541
 
            m2 = MetaData(testing.db)
1542
 
            t2 = Table('party', m2, autoload=True)
1543
 
            assert len(t2.indexes) == 2
1544
 
 
1545
 
            # Make sure indexes are in the order we expect them in
1546
 
 
1547
 
            tmp = [(idx.name, idx) for idx in t2.indexes]
1548
 
            tmp.sort()
1549
 
            r1, r2 = [idx[1] for idx in tmp]
1550
 
            assert r1.name == 'idx2'
1551
 
            assert r1.unique == True
1552
 
            assert r2.unique == False
1553
 
            assert [t2.c.id] == r1.columns
1554
 
            assert [t2.c.name] == r2.columns
1555
 
 
1556
 
        testing.assert_warnings(go,
1557
 
            [
1558
 
                'Skipped unsupported reflection of '
1559
 
                'expression-based index idx1',
1560
 
                'Predicate of partial index idx2 ignored during '
1561
 
                'reflection',
1562
 
                'Skipped unsupported reflection of '
1563
 
                'expression-based index idx3'
1564
 
            ])
1565
 
 
1566
 
    @testing.provide_metadata
1567
 
    def test_index_reflection_modified(self):
1568
 
        """reflect indexes when a column name has changed - PG 9 
1569
 
        does not update the name of the column in the index def.
1570
 
        [ticket:2141]
1571
 
 
1572
 
        """
1573
 
 
1574
 
        metadata = self.metadata
1575
 
 
1576
 
        t1 = Table('t', metadata,
1577
 
            Column('id', Integer, primary_key=True),
1578
 
            Column('x', Integer)
1579
 
        )
1580
 
        metadata.create_all()
1581
 
        conn = testing.db.connect().execution_options(autocommit=True)
1582
 
        conn.execute("CREATE INDEX idx1 ON t (x)")
1583
 
        conn.execute("ALTER TABLE t RENAME COLUMN x to y")
1584
 
 
1585
 
        ind = testing.db.dialect.get_indexes(conn, "t", None)
1586
 
        eq_(ind, [{'unique': False, 'column_names': [u'y'], 'name': u'idx1'}])
1587
 
        conn.close()
1588
1720
 
1589
1721
    @testing.fails_on('+zxjdbc', 'psycopg2/pg8000 specific assertion')
1590
1722
    @testing.fails_on('pypostgresql',