--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/testpgmanager.py Tue Dec 06 17:11:21 2011 +0100
@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+
+import unittest
+
+from config import Config
+from pgtoolkit import pgmanager
+
+
+class TestSuite(unittest.TestCase):
+
+ def setUp(self):
+ cfg = Config('tests.conf')
+ self.m = pgmanager.get_instance()
+ self.m.create_conn(**cfg)
+ self.m.create_conn('autocommit', isolation_level='autocommit', **cfg)
+
+ def tearDown(self):
+ self.m.destroy_conn()
+ self.m.destroy_conn('autocommit')
+
+ def test_connection_get_put(self):
+ conn = self.m.get_conn()
+
+ curs = conn.cursor()
+ curs.execute('SELECT %s AS t', [True])
+ d = curs.fetchone_dict()
+ self.assertTrue(d.t)
+
+ self.m.put_conn(conn)
+
+ def test_connection_with(self):
+ with self.m.cursor() as curs:
+ curs.execute('SELECT %s AS t', [True])
+ d = curs.fetchone_dict()
+ self.assertTrue(d.t)
+
+ def test_connection_autocommit(self):
+ conn = self.m.get_conn('autocommit')
+ curs = conn.cursor()
+ name = 'test'
+ curs.execute('DELETE FROM test WHERE name = %s', [name])
+ curs.execute('INSERT INTO test (name) VALUES (%s)', [name])
+ conn.close()
+
+ conn = self.m.get_conn('autocommit')
+ curs = conn.cursor()
+ curs.execute('SELECT * FROM test WHERE name = %s', [name])
+ self.assertTrue(curs.rowcount == 1)
+ self.m.put_conn(conn)
+
+if __name__ == '__main__':
+ unittest.main()
+