I found my original set of tests and added them to head (patch
attached).
This covers async connections and select queries, which I believe are
the most useful use cases. I should add tests for DDL and DML
statements, too.
I'm not really sure how to test getnonblocking() and isnonblocking() in
a useful way. I really only included them because they're documented in
the asynch section of the libpq manual.
--
Patrick TJ McPhee <pmcp...@givex.com>
Index: tests/test_classic_connection.py
===================================================================
--- tests/test_classic_connection.py (revision 895)
+++ tests/test_classic_connection.py (working copy)
@@ -120,15 +120,23 @@
self.assertEqual(attributes, connection_attributes)
def testAllConnectMethods(self):
- methods = '''cancel close date_format endcopy
+ methods = '''cancel close connectpoll date_format endcopy
escape_bytea escape_identifier escape_literal escape_string
fileno get_cast_hook get_notice_receiver getline getlo getnotify
- inserttable locreate loimport parameter putline query reset
- set_cast_hook set_notice_receiver source transaction'''.split()
+ inserttable isnonblocking locreate loimport parameter putline query reset
+ sendquery set_cast_hook set_notice_receiver setnonblocking source transaction'''.split()
connection_methods = [a for a in dir(self.connection)
if not a.startswith('__') and self.is_method(a)]
self.assertEqual(methods, connection_methods)
+ def testAsyncConnect(self):
+ self.connection.close()
+ self.connection = pg.connect(dbname, dbhost, dbport, nowait=True)
+ rc = self.connection.connectpoll()
+ while rc not in (pg.PGRES_POLLING_OK, pg.PGRES_POLLING_FAILED):
+ rc = self.connection.connectpoll()
+ self.assertEqual(rc, pg.PGRES_POLLING_OK)
+
def testAttributeDb(self):
self.assertEqual(self.connection.db, dbname)
@@ -183,6 +191,48 @@
def testMethodQueryEmpty(self):
self.assertRaises(ValueError, self.connection.query, '')
+ def testMethodSendQuery(self):
+ query = self.connection.sendquery
+ for qs,args, result in (
+ ("select 1+1 as a", (), 2),
+ ("select 1+$1 as a", ((1,),), 2),
+ ("select 1+$1+$2 as a", ((2, 3),), 6),
+ ("select 1+$1+$2 as a", ([2, 3],), 6),):
+ pgq = query(qs, *args)
+ self.assertEqual(self.connection.transaction(), pg.TRANS_ACTIVE)
+ self.assertEqual(pgq.getresult()[0][0], result)
+ self.assertEqual(self.connection.transaction(), pg.TRANS_ACTIVE)
+ self.assertIsNone(pgq.getresult())
+ self.assertEqual(self.connection.transaction(), pg.TRANS_IDLE)
+
+ pgq = query(qs, *args)
+ self.assertEqual(pgq.namedresult()[0].a, result)
+ self.assertIsNone(pgq.namedresult())
+
+ pgq = query(qs, *args)
+ self.assertEqual(pgq.dictresult()[0]['a'], result)
+ self.assertIsNone(pgq.dictresult())
+
+
+ pgq = query("select 1+1; select 'pg';")
+ self.assertEqual(pgq.getresult()[0][0], 2)
+ self.assertEqual(pgq.getresult()[0][0], 'pg')
+ self.assertIsNone(pgq.getresult())
+
+ pgq = query("select 1+1 as a; select 'pg' as a;")
+ self.assertEqual(pgq.namedresult()[0].a, 2)
+ self.assertEqual(pgq.namedresult()[0].a, 'pg')
+ self.assertIsNone(pgq.namedresult())
+
+ pgq = query("select 1+1 as a; select 'pg' as a;")
+ self.assertEqual(pgq.dictresult()[0]['a'], 2)
+ self.assertEqual(pgq.dictresult()[0]['a'], 'pg')
+ self.assertIsNone(pgq.dictresult())
+
+ def testMethodSendQueryEmpty(self):
+ pgq = self.connection.sendquery('')
+ self.assertRaises(ValueError, pgq.getresult)
+
def testMethodEndcopy(self):
try:
self.connection.endcopy()
_______________________________________________
PyGreSQL mailing list
PyGreSQL@vex.net
https://mail.vex.net/mailman/listinfo.cgi/pygresql
