Changeset 744 for trunk/tests


Ignore:
Timestamp:
Jan 14, 2016, 3:26:30 PM (4 years ago)
Author:
cito
Message:

Use cleanup feature in unittests

Location:
trunk/tests
Files:
2 edited

Legend:

Unmodified
Added
Removed
  • trunk/tests/test_classic_connection.py

    r730 r744  
    265265
    266266    def tearDown(self):
     267        self.doCleanups()
    267268        self.c.close()
    268269
     
    507508        query = self.c.query
    508509        query("drop table if exists test_table")
     510        self.addCleanup(query, "drop table test_table")
    509511        q = "create table test_table (n integer) with oids"
    510512        r = query(q)
     
    537539        self.assertIsInstance(r, str)
    538540        self.assertEqual(r, '5')
    539         query("drop table test_table")
    540541
    541542
     
    12051206
    12061207    def tearDown(self):
     1208        self.doCleanups()
    12071209        self.c.close()
    12081210
     
    12511253
    12521254    def testNoticeReceiver(self):
     1255        self.addCleanup(self.c.query, 'drop function bilbo_notice();')
    12531256        self.c.query('''create function bilbo_notice() returns void AS $$
    12541257            begin
     
    12561259            end;
    12571260            $$ language plpgsql''')
    1258         try:
    1259             received = {}
    1260 
    1261             def notice_receiver(notice):
    1262                 for attr in dir(notice):
    1263                     if attr.startswith('__'):
    1264                         continue
    1265                     value = getattr(notice, attr)
    1266                     if isinstance(value, str):
    1267                         value = value.replace('WARNUNG', 'WARNING')
    1268                     received[attr] = value
    1269 
    1270             self.c.set_notice_receiver(notice_receiver)
    1271             self.c.query('''select bilbo_notice()''')
    1272             self.assertEqual(received, dict(
    1273                 pgcnx=self.c, message='WARNING:  Bilbo was here!\n',
    1274                 severity='WARNING', primary='Bilbo was here!',
    1275                 detail=None, hint=None))
    1276         finally:
    1277             self.c.query('''drop function bilbo_notice();''')
     1261        received = {}
     1262
     1263        def notice_receiver(notice):
     1264            for attr in dir(notice):
     1265                if attr.startswith('__'):
     1266                    continue
     1267                value = getattr(notice, attr)
     1268                if isinstance(value, str):
     1269                    value = value.replace('WARNUNG', 'WARNING')
     1270                received[attr] = value
     1271
     1272        self.c.set_notice_receiver(notice_receiver)
     1273        self.c.query('select bilbo_notice()')
     1274        self.assertEqual(received, dict(
     1275            pgcnx=self.c, message='WARNING:  Bilbo was here!\n',
     1276            severity='WARNING', primary='Bilbo was here!',
     1277            detail=None, hint=None))
    12781278
    12791279
  • trunk/tests/test_classic_dbwrapper.py

    r743 r744  
    310310
    311311    def tearDown(self):
     312        self.doCleanups()
    312313        self.db.close()
    313314
     
    416417        query = self.db.query
    417418        query("drop table if exists test_table")
     419        self.addCleanup(query, "drop table test_table")
    418420        query("create table test_table("
    419421            " n int, alpha smallint, beta bool,"
     
    424426            n='int', alpha='int', beta='bool',
    425427            gamma='text', tau='text', v='text'))
    426         query("drop table test_table")
    427428
    428429    def testGetAttnamesWithQuotes(self):
     
    431432        table = 'test table for get_attnames()'
    432433        query('drop table if exists "%s"' % table)
     434        self.addCleanup(query, 'drop table "%s"' % table)
    433435        query('create table "%s"('
    434436            '"Prime!" smallint,'
     
    438440        self.assertEquals(r, {
    439441            'Prime!': 'int', 'much space': 'int', 'Questions?': 'text'})
    440         query('drop table "%s"' % table)
    441442
    442443    def testGetAttnamesWithRegtypes(self):
     
    444445        query = self.db.query
    445446        query("drop table if exists test_table")
     447        self.addCleanup(query, "drop table test_table")
    446448        query("create table test_table("
    447449            " n int, alpha smallint, beta bool,"
     
    456458            n='integer', alpha='smallint', beta='boolean',
    457459            gamma='character', tau='text', v='character varying'))
    458         query("drop table test_table")
    459460
    460461    def testGetAttnamesIsCached(self):
     
    462463        query = self.db.query
    463464        query("drop table if exists test_table")
     465        self.addCleanup(query, "drop table if exists test_table")
    464466        query("create table test_table(col int)")
    465467        r = get_attnames("test_table")
     
    482484        query = self.db.query
    483485        query("drop table if exists test_table")
     486        self.addCleanup(query, "drop table test_table")
    484487        query("create table test_table("
    485488            " n int, alpha smallint, v varchar(3),"
     
    490493            ('n', 'int'), ('alpha', 'int'), ('v', 'text'),
    491494            ('gamma', 'text'), ('tau', 'text'), ('beta', 'bool')]))
    492         query("drop table test_table")
    493495        if OrderedDict is dict:
    494496            self.skipTest('OrderedDict is not supported')
     
    499501        query = self.db.query
    500502        query("drop table if exists test_table")
     503        self.addCleanup(query, "drop table test_table")
    501504        q = "create table test_table (n integer) with oids"
    502505        r = query(q)
     
    528531        self.assertIsInstance(r, str)
    529532        self.assertEqual(r, '5')
    530         query("drop table test_table")
    531533
    532534    def testMultipleQueries(self):
     
    539541        query = self.db.query
    540542        query("drop table if exists test_table")
     543        self.addCleanup(query, "drop table test_table")
    541544        q = "create table test_table (n1 integer, n2 integer) with oids"
    542545        query(q)
     
    562565        r = query(q, 4)
    563566        self.assertEqual(r, '3')
    564         query("drop table test_table")
    565567
    566568    def testEmptyQuery(self):
     
    579581            for n in range(7):
    580582                query('drop table if exists "%s%d"' % (t, n))
     583                self.addCleanup(query, 'drop table "%s%d"' % (t, n))
    581584            query('create table "%s0" ('
    582585                "a smallint)" % t)
     
    620623            # we get the changed primary key when the cache is flushed
    621624            self.assertEqual(pkey('%s1' % t, flush=True), 'x')
    622             for n in range(7):
    623                 query('drop table "%s%d"' % (t, n))
    624625
    625626    def testGetDatabases(self):
     
    693694        for table in ('attnames_test_table', 'test table for attnames'):
    694695            self.db.query('drop table if exists "%s"' % table)
     696            self.addCleanup(self.db.query, 'drop table "%s"' % table)
    695697            self.db.query('create table "%s" ('
    696698                'a smallint, b integer, c bigint, '
     
    707709                'y': 'int', 'x': 'int', 'z': 'int', 'oid': 'int'}
    708710            self.assertEqual(attributes, result)
    709             self.db.query('drop table "%s"' % table)
    710711
    711712    def testHasTablePrivilege(self):
     
    728729        table = 'get_test_table'
    729730        query('drop table if exists "%s"' % table)
     731        self.addCleanup(query, 'drop table "%s"' % table)
    730732        query('create table "%s" ('
    731733            "n integer, t text) with oids" % table)
     
    761763        r['n'] = 2
    762764        self.assertEqual(get(table, r)['t'], 'y')
    763         query('drop table "%s"' % table)
    764765
    765766    def testGetWithCompositeKey(self):
     
    768769        table = 'get_test_table_1'
    769770        query('drop table if exists "%s"' % table)
     771        self.addCleanup(query, 'drop table "%s"' % table)
    770772        query('create table "%s" ('
    771773            "n integer, t text, primary key (n))" % table)
     
    774776                "%d, '%s')" % (table, n + 1, t))
    775777        self.assertEqual(get(table, 2)['t'], 'b')
    776         query('drop table "%s"' % table)
    777778        table = 'get_test_table_2'
    778779        query('drop table if exists "%s"' % table)
     780        self.addCleanup(query, 'drop table "%s"' % table)
    779781        query('create table "%s" ('
    780782            "n integer, m integer, t text, primary key (n, m))" % table)
     
    790792        r = get(table, dict(n=3, m=2), frozenset(['n', 'm']))
    791793        self.assertEqual(r['t'], 'f')
    792         query('drop table "%s"' % table)
    793794
    794795    def testGetWithQuotedNames(self):
     
    797798        table = 'test table for get()'
    798799        query('drop table if exists "%s"' % table)
     800        self.addCleanup(query, 'drop table "%s"' % table)
    799801        query('create table "%s" ('
    800802            '"Prime!" smallint primary key,'
     
    807809        self.assertEqual(r['much space'], 1001)
    808810        self.assertEqual(r['Questions?'], 'No!')
    809         query('drop table "%s"' % table)
    810811
    811812    def testGetFromView(self):
     
    821822        query = self.db.query
    822823        query("drop table if exists test_students")
     824        self.addCleanup(query, "drop table test_students")
    823825        query("create table test_students (firstname varchar primary key,"
    824826            " nickname varchar, grade char(2))")
     
    854856        self.assertEqual(len(r), 3)
    855857        self.assertEqual(r[1][2], 'D-')
    856         query('drop table test_students')
    857858
    858859    def testInsert(self):
     
    863864        table = 'insert_test_table'
    864865        query('drop table if exists "%s"' % table)
     866        self.addCleanup(query, 'drop table "%s"' % table)
    865867        query('create table "%s" ('
    866868            "i2 smallint, i4 integer, i8 bigint,"
     
    958960            self.assertEqual(data, expect)
    959961            query('delete from "%s"' % table)
    960         query('drop table "%s"' % table)
    961962
    962963    def testInsertWithQuotedNames(self):
     
    965966        table = 'test table for insert()'
    966967        query('drop table if exists "%s"' % table)
     968        self.addCleanup(query, 'drop table "%s"' % table)
    967969        query('create table "%s" ('
    968970            '"Prime!" smallint primary key,'
     
    980982        self.assertEqual(r['much space'], 2002)
    981983        self.assertEqual(r['Questions?'], 'What?')
    982         query('drop table "%s"' % table)
    983984
    984985    def testUpdate(self):
     
    987988        table = 'update_test_table'
    988989        query('drop table if exists "%s"' % table)
     990        self.addCleanup(query, 'drop table "%s"' % table)
    989991        query('create table "%s" ('
    990992            "n integer, t text) with oids" % table)
     
    10001002        r = query(q).getresult()[0][0]
    10011003        self.assertEqual(r, 'u')
    1002         query('drop table "%s"' % table)
    10031004
    10041005    def testUpdateWithCompositeKey(self):
     
    10071008        table = 'update_test_table_1'
    10081009        query('drop table if exists "%s"' % table)
     1010        self.addCleanup(query, 'drop table if exists "%s"' % table)
    10091011        query('create table "%s" ('
    10101012            "n integer, t text, primary key (n))" % table)
     
    10491051        r = [r[0] for r in query(q).getresult()]
    10501052        self.assertEqual(r, ['c', 'x'])
    1051         query('drop table "%s"' % table)
    10521053
    10531054    def testUpdateWithQuotedNames(self):
     
    10561057        table = 'test table for update()'
    10571058        query('drop table if exists "%s"' % table)
     1059        self.addCleanup(query, 'drop table "%s"' % table)
    10581060        query('create table "%s" ('
    10591061            '"Prime!" smallint primary key,'
     
    10731075        self.assertEqual(r['much space'], 7007)
    10741076        self.assertEqual(r['Questions?'], 'When?')
    1075         query('drop table "%s"' % table)
    10761077
    10771078    def testUpsert(self):
     
    10801081        table = 'upsert_test_table'
    10811082        query('drop table if exists "%s"' % table)
     1083        self.addCleanup(query, 'drop table "%s"' % table)
    10821084        query('create table "%s" ('
    10831085            "n integer primary key, t text) with oids" % table)
     
    11421144        r = query(q).getresult()
    11431145        self.assertEqual(r, [(1, 'x2'), (2, 'y3')])
    1144         query('drop table "%s"' % table)
    11451146
    11461147    def testUpsertWithCompositeKey(self):
     
    11491150        table = 'upsert_test_table_2'
    11501151        query('drop table if exists "%s"' % table)
     1152        self.addCleanup(query, 'drop table "%s"' % table)
    11511153        query('create table "%s" ('
    11521154            "n integer, m integer, t text, primary key (n, m))" % table)
     
    12111213        r = query(q).getresult()
    12121214        self.assertEqual(r, [(1, 2, 'x'), (1, 3, 'nm'), (2, 3, 'y')])
    1213         query('drop table "%s"' % table)
    12141215
    12151216    def testUpsertWithQuotedNames(self):
     
    12181219        table = 'test table for upsert()'
    12191220        query('drop table if exists "%s"' % table)
     1221        self.addCleanup(query, 'drop table "%s"' % table)
    12201222        query('create table "%s" ('
    12211223            '"Prime!" smallint primary key,'
     
    12501252        table = 'clear_test_table'
    12511253        query('drop table if exists "%s"' % table)
     1254        self.addCleanup(query, 'drop table "%s"' % table)
    12521255        query('create table "%s" ('
    12531256            "n integer, b boolean, d date, t text)" % table)
     
    12631266            'oid': long(1)}
    12641267        self.assertEqual(r, result)
    1265         query('drop table "%s"' % table)
    12661268
    12671269    def testClearWithQuotedNames(self):
     
    12701272        table = 'test table for clear()'
    12711273        query('drop table if exists "%s"' % table)
     1274        self.addCleanup(query, 'drop table "%s"' % table)
    12721275        query('create table "%s" ('
    12731276            '"Prime!" smallint primary key,'
     
    12781281        self.assertEqual(r['much space'], 0)
    12791282        self.assertEqual(r['Questions?'], '')
    1280         query('drop table "%s"' % table)
    12811283
    12821284    def testDelete(self):
     
    12851287        table = 'delete_test_table'
    12861288        query('drop table if exists "%s"' % table)
     1289        self.addCleanup(query, 'drop table "%s"' % table)
    12871290        query('create table "%s" ('
    12881291            "n integer, t text) with oids" % table)
     
    13101313        self.assertEqual(s, 0)
    13111314        self.assertRaises(pg.DatabaseError, self.db.get, table, 2, 'n')
    1312         query('drop table "%s"' % table)
    13131315
    13141316    def testDeleteWithCompositeKey(self):
     
    13161318        table = 'delete_test_table_1'
    13171319        query('drop table if exists "%s"' % table)
     1320        self.addCleanup(query, 'drop table "%s"' % table)
    13181321        query('create table "%s" ('
    13191322            "n integer, t text, primary key (n))" % table)
     
    13311334                  ).getresult()[0][0]
    13321335        self.assertEqual(r, 'c')
    1333         query('drop table "%s"' % table)
    13341336        table = 'delete_test_table_2'
    13351337        query('drop table if exists "%s"' % table)
     1338        self.addCleanup(query, 'drop table "%s"' % table)
    13361339        query('create table "%s" ('
    13371340            "n integer, m integer, t text, primary key (n, m))" % table)
     
    13551358            ' order by m' % table).getresult()]
    13561359        self.assertEqual(r, ['f'])
    1357         query('drop table "%s"' % table)
    13581360
    13591361    def testDeleteWithQuotedNames(self):
     
    13621364        table = 'test table for delete()'
    13631365        query('drop table if exists "%s"' % table)
     1366        self.addCleanup(query, 'drop table "%s"' % table)
    13641367        query('create table "%s" ('
    13651368            '"Prime!" smallint primary key,'
     
    13771380        r = query('select count(*) from "%s"' % table).getresult()
    13781381        self.assertEqual(r[0][0], 0)
    1379         query('drop table "%s"' % table)
    13801382
    13811383    def testTransaction(self):
    13821384        query = self.db.query
    13831385        query("drop table if exists test_table")
     1386        self.addCleanup(query, "drop table test_table")
    13841387        query("create table test_table (n integer)")
    13851388        self.db.begin()
     
    14101413            "select * from test_table order by 1").getresult()]
    14111414        self.assertEqual(r, [1, 2, 5, 7, 9])
    1412         query("drop table test_table")
    14131415
    14141416    def testContextManager(self):
    14151417        query = self.db.query
    14161418        query("drop table if exists test_table")
     1419        self.addCleanup(query, "drop table test_table")
    14171420        query("create table test_table (n integer check(n>0))")
    14181421        with self.db:
     
    14391442            "select * from test_table order by 1").getresult()]
    14401443        self.assertEqual(r, [1, 2, 5, 7])
    1441         query("drop table test_table")
    14421444
    14431445    def testBytea(self):
    14441446        query = self.db.query
    14451447        query('drop table if exists bytea_test')
     1448        self.addCleanup(query, 'drop table bytea_test')
    14461449        query('create table bytea_test (n smallint primary key, data bytea)')
    14471450        s = b"It's all \\ kinds \x00 of\r nasty \xff stuff!\n"
     
    14581461        self.assertIsInstance(r, bytes)
    14591462        self.assertEqual(r, s)
    1460         query('drop table bytea_test')
    14611463
    14621464    def testInsertUpdateGetBytea(self):
    14631465        query = self.db.query
    14641466        query('drop table if exists bytea_test')
     1467        self.addCleanup(query, 'drop table bytea_test')
    14651468        query('create table bytea_test (n smallint primary key, data bytea)')
    14661469        # insert null value
     
    15201523        self.assertIsInstance(r, bytes)
    15211524        self.assertEqual(r, s)
    1522         query('drop table bytea_test')
    15231525
    15241526    def testDebugWithCallable(self):
     
    16151617
    16161618    def tearDown(self):
     1619        self.doCleanups()
    16171620        self.db.close()
    16181621
     
    16371640        self.assertEqual(r, result)
    16381641        query("drop table if exists s3.t3m")
     1642        self.addCleanup(query, "drop table s3.t3m")
    16391643        query("create table s3.t3m with oids as select 1 as m")
    16401644        result_m = {'oid': 'int', 'm': 'int'}
     
    16461650        r = get_attnames("t3m")
    16471651        self.assertEqual(r, result_m)
    1648         query("drop table s3.t3m")
    16491652
    16501653    def testGet(self):
Note: See TracChangeset for help on using the changeset viewer.