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