tests/testpgmanager.py
changeset 50 f71d3abbb18f
parent 49 08e4dfe1b0cb
child 51 bdc44f96cb0b
equal deleted inserted replaced
49:08e4dfe1b0cb 50:f71d3abbb18f
     1 #!/usr/bin/env python3
       
     2 
       
     3 from tests.config import Config
       
     4 from pgtoolkit import pgmanager
       
     5 
       
     6 import unittest
       
     7 
       
     8 
       
     9 class TestSuite(unittest.TestCase):
       
    10 
       
    11     def setUp(self):
       
    12         cfg = Config('pgtoolkit.conf')
       
    13         test_db_conn_params = cfg['databases']['test']
       
    14         self.m = pgmanager.get_instance()
       
    15         self.m.create_conn(dsn=test_db_conn_params)
       
    16         self.m.create_conn('autocommit', isolation_level='autocommit', dsn=test_db_conn_params)
       
    17 
       
    18     def tearDown(self):
       
    19         self.m.destroy_conn()
       
    20         self.m.destroy_conn('autocommit')
       
    21 
       
    22     def test_connection_get_put(self):
       
    23         conn = self.m.get_conn()
       
    24 
       
    25         curs = conn.cursor()
       
    26         curs.execute('SELECT %s AS t', [True])
       
    27         d = curs.fetchone_dict()
       
    28         self.assertTrue(d.t)
       
    29 
       
    30         self.m.put_conn(conn)
       
    31 
       
    32     def test_connection_with(self):
       
    33         with self.m.cursor() as curs:
       
    34             curs.execute('SELECT %s AS t', [True])
       
    35             d = curs.fetchone_dict()
       
    36             self.assertTrue(d.t)
       
    37 
       
    38     def test_connection_autocommit(self):
       
    39         conn = self.m.get_conn('autocommit')
       
    40         curs = conn.cursor()
       
    41         name = 'test'
       
    42         curs.execute('DELETE FROM test WHERE name = %s', [name])
       
    43         curs.execute('INSERT INTO test (name) VALUES (%s)', [name])
       
    44         conn.close()
       
    45 
       
    46         conn = self.m.get_conn('autocommit')
       
    47         curs = conn.cursor()
       
    48         curs.execute('SELECT * FROM test WHERE name = %s', [name])
       
    49         self.assertTrue(curs.rowcount == 1)
       
    50         self.m.put_conn(conn)
       
    51 
       
    52 
       
    53 if __name__ == '__main__':
       
    54     unittest.main()
       
    55