1 #!/usr/bin/env python

     2 

     3 import unittest

     4 

     5 from config import Config

     6 from pgtoolkit import pgmanager

     7 

     8 

     9 class TestSuite(unittest.TestCase):

    10 

    11     def setUp(self):

    12         cfg = Config('tests.conf')

    13         self.m = pgmanager.get_instance()

    14         self.m.create_conn(**cfg)

    15         self.m.create_conn('autocommit', isolation_level='autocommit', **cfg)

    16 

    17     def tearDown(self):

    18         self.m.destroy_conn()

    19         self.m.destroy_conn('autocommit')

    20 

    21     def test_connection_get_put(self):

    22         conn = self.m.get_conn()

    23 

    24         curs = conn.cursor()

    25         curs.execute('SELECT %s AS t', [True])

    26         d = curs.fetchone_dict()

    27         self.assertTrue(d.t)

    28 

    29         self.m.put_conn(conn)

    30 

    31     def test_connection_with(self):

    32         with self.m.cursor() as curs:

    33             curs.execute('SELECT %s AS t', [True])

    34             d = curs.fetchone_dict()

    35             self.assertTrue(d.t)

    36 

    37     def test_connection_autocommit(self):

    38         conn = self.m.get_conn('autocommit')

    39         curs = conn.cursor()

    40         name = 'test'

    41         curs.execute('DELETE FROM test WHERE name = %s', [name])

    42         curs.execute('INSERT INTO test (name) VALUES (%s)', [name])

    43         conn.close()

    44 

    45         conn = self.m.get_conn('autocommit')

    46         curs = conn.cursor()

    47         curs.execute('SELECT * FROM test WHERE name = %s', [name])

    48         self.assertTrue(curs.rowcount == 1)

    49         self.m.put_conn(conn)

    50 

    51 if __name__ == '__main__':

    52     unittest.main()

    53