Added connection.pgconn_ptr and cursor.pgresult_ptr

Allow interacting with libpq in Python via ctypes.

See #782.

3 files changed, 47 insertions, 0 deletions

diff --git a/tests/test_connection.py b/tests/test_connection.py
index e450883..9bad8c3 100755
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -26,6 +26,7 @@ import re
 import os
 import sys
 import time
+import ctypes
 import threading
 import subprocess as sp
 from operator import attrgetter
@@ -346,6 +347,20 @@ class ConnectionTests(ConnectingTestCase):
         # we can't do anything else in Python
         self.assertIsNotNone(capsule)
 
+    def test_pgconn_ptr(self):
+        conn = self.connect()
+        self.assert_(conn.pgconn_ptr is not None)
+
+        f = self.libpq.PQserverVersion
+        f.argtypes = [ctypes.c_void_p]
+        f.restype = ctypes.c_int
+        ver = f(conn.pgconn_ptr)
+        self.assertEqual(ver, conn.server_version)
+
+        conn.close()
+        self.assert_(conn.pgconn_ptr is None)
+
+
 class ParseDsnTestCase(ConnectingTestCase):
     def test_parse_dsn(self):
         from psycopg2 import ProgrammingError
diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index bee469e..82338d3 100755
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -23,6 +23,7 @@
 # License for more details.
 
 import time
+import ctypes
 import pickle
 import psycopg2
 import psycopg2.extensions
@@ -650,6 +651,23 @@ class CursorTests(ConnectingTestCase):
             [(i,) for i in range(5)])
         self.assertEqual(cur.rowcount, 5)
 
+    @skip_before_postgres(9)
+    def test_pgresult_ptr(self):
+        curs = self.conn.cursor()
+        self.assert_(curs.pgresult_ptr is None)
+
+        f = self.libpq.PQcmdStatus
+        f.argtypes = [ctypes.c_void_p]
+        f.restype = ctypes.c_char_p
+
+        curs.execute("select 'x'")
+        self.assert_(curs.pgresult_ptr is not None)
+        status = f(curs.pgresult_ptr)
+        self.assertEqual(status, b'SELECT 1')
+
+        curs.close()
+        self.assert_(curs.pgresult_ptr is None)
+
 
 def test_suite():
     return unittest.TestLoader().loadTestsFromName(__name__)
diff --git a/tests/testutils.py b/tests/testutils.py
index e4b6843..5f41789 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -26,10 +26,12 @@ import re
 import os
 import sys
 import types
+import ctypes
 import select
 import platform
 import unittest
 from functools import wraps
+from ctypes.util import find_library
 from .testconfig import dsn, repl_dsn
 from psycopg2.compat import text_type
 
@@ -174,6 +176,18 @@ class ConnectingTestCase(unittest.TestCase):
             else:
                 raise Exception("Unexpected result from poll: %r", state)
 
+    _libpq = None
+
+    @property
+    def libpq(self):
+        """Return a ctypes wrapper for the libpq library"""
+        if ConnectingTestCase._libpq is not None:
+            return ConnectingTestCase._libpq
+
+        libname = find_library('pq')
+        rv = ConnectingTestCase._libpq = ctypes.pydll.LoadLibrary(libname)
+        return rv
+
 
 def decorate_all_tests(obj, *decorators):
     """