tests/test_pgmanager.py
changeset 50 f71d3abbb18f
parent 49 08e4dfe1b0cb
child 104 d8ff52a0390f
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/test_pgmanager.py	Sat Sep 29 14:01:47 2012 +0200
@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+
+from tests.config import Config
+from pgtoolkit import pgmanager
+
+import unittest
+
+
+class TestSuite(unittest.TestCase):
+
+    def setUp(self):
+        cfg = Config('pgtoolkit.conf')
+        test_db_conn_params = cfg['databases']['test']
+        self.m = pgmanager.get_instance()
+        self.m.create_conn(dsn=test_db_conn_params)
+        self.m.create_conn('autocommit', isolation_level='autocommit', dsn=test_db_conn_params)
+
+    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()
+