UCS4 fixes; printf fixes;

Fixes found with UCS4 Python and UCS2 SQLWCHAR (Fedora 13 64-bit).

Connection code now uses common SQLWChar class too.

Added pyodbc.UNICODE_SIZE and pyodbc.SQLWCHAR_SIZE to help troubleshoot problems.

Fixed some printfs 64-bit problems.

src/connection.cpp

@@ -16,6 +16,7 @@
 #include "errors.h"
 #include "wrapper.h"
 #include "cnxninfo.h"
+#include "sqlwchar.h"
 
 static char connection_doc[] =
     "Connection objects manage connections to the database.\n"
@@ -71,22 +72,9 @@ static bool Connect(PyObject* pConnectString, HDBC hdbc, bool fAnsi)
 
     if (!fAnsi)
     {
-        SQLWCHAR szConnectW[cchMax];
-        if (PyUnicode_Check(pConnectString))
-        {
-            Py_UNICODE* p = PyUnicode_AS_UNICODE(pConnectString);
-            for (Py_ssize_t i = 0, c = PyUnicode_GET_SIZE(pConnectString); i <= c; i++)
-                szConnectW[i] = (SQLWCHAR)p[i];
-        }
-        else
-        {
-            const char* p = PyString_AS_STRING(pConnectString);
-            for (Py_ssize_t i = 0, c = PyString_GET_SIZE(pConnectString); i <= c; i++)
-                szConnectW[i] = (SQLWCHAR)p[i];
-        }
-
+        SQLWChar connectString(pConnectString);
         Py_BEGIN_ALLOW_THREADS
-        ret = SQLDriverConnectW(hdbc, 0, szConnectW, SQL_NTS, 0, 0, 0, SQL_DRIVER_NOPROMPT);
+        ret = SQLDriverConnectW(hdbc, 0, connectString, connectString.size(), 0, 0, 0, SQL_DRIVER_NOPROMPT);
         Py_END_ALLOW_THREADS
         if (SQL_SUCCEEDED(ret))
             return true;

src/errors.cpp

@@ -244,14 +244,14 @@ PyObject* GetErrorFromHandle(const char* szFunction, HDBC hdbc, HSTMT hstmt)
 
                 memcpy(sqlstate, sqlstateT, sizeof(sqlstate[0]) * _countof(sqlstate));
 
-                pMsg = PyString_FromFormat("[%s] %s (%ld) (%s)", sqlstateT, szMsg, nNativeError, szFunction);
+                pMsg = PyString_FromFormat("[%s] %s (%ld) (%s)", sqlstateT, szMsg, (long)nNativeError, szFunction);
                 if (pMsg == 0)
                     return 0;
             }
             else
             {
                 // This is not the first error message, so append to the existing one.
-                pMsgPart = PyString_FromFormat("; [%s] %s (%ld)", sqlstateT, szMsg, nNativeError);
+                pMsgPart = PyString_FromFormat("; [%s] %s (%ld)", sqlstateT, szMsg, (long)nNativeError);
                 if (pMsgPart == 0)
                 {
                     Py_XDECREF(pMsg);

src/pyodbc.h

@@ -141,7 +141,7 @@ inline void _strlwr(char* name)
 #ifdef PYODBC_TRACE
 void __cdecl DebugTrace(const char* szFmt, ...);
 #else  
-inline void __cdecl DebugTrace(const char* szFmt, ...) { UNUSED(szFmt); }
+inline void DebugTrace(const char* szFmt, ...) { UNUSED(szFmt); }
 #endif
 #define TRACE DebugTrace
 

src/pyodbcmodule.cpp

@@ -341,7 +341,7 @@ static PyObject* mod_connect(PyObject* self, PyObject* args, PyObject* kwargs)
                 return PyErr_Format(PyExc_TypeError, "'%s' is not a string or unicode value'", szKey);
 
             // Map DB API recommended names to ODBC names (e.g. user --> uid).
-            for (int i = 0; i < _countof(keywordmaps); i++)
+            for (size_t i = 0; i < _countof(keywordmaps); i++)
             {
                 if (_strcmpi(szKey, keywordmaps[i].oldname) == 0)
                 {
@@ -925,6 +925,9 @@ initpyodbc()
     PyModule_AddObject(pModule, "Binary", (PyObject*)&PyBuffer_Type);
     Py_INCREF((PyObject*)&PyBuffer_Type);
 
+    PyModule_AddIntConstant(pModule, "UNICODE_SIZE", sizeof(Py_UNICODE));
+    PyModule_AddIntConstant(pModule, "SQLWCHAR_SIZE", sizeof(SQLWCHAR));
+
     if (PyErr_Occurred())
         ErrorCleanup();
 }

src/sqlwchar.cpp

@@ -37,7 +37,7 @@ bool SQLWChar::Convert(PyObject* o)
         return false;
     }
 
-    Py_UNICODE* pU   = (SQLWCHAR*)PyUnicode_AS_UNICODE(o);
+    Py_UNICODE* pU   = (Py_UNICODE*)PyUnicode_AS_UNICODE(o);
     Py_ssize_t  lenT = PyUnicode_GET_SIZE(o);
 
     if (sizeof(SQLWCHAR) == Py_UNICODE_SIZE)
@@ -88,7 +88,7 @@ bool sqlwchar_copy(SQLWCHAR* pdest, const Py_UNICODE* psrc, Py_ssize_t len)
 PyObject* PyUnicode_FromSQLWCHAR(const SQLWCHAR* sz, Py_ssize_t cch)
 {
     if (sizeof(SQLWCHAR) == Py_UNICODE_SIZE)
-        return PyUnicode_FromUnicode(sz, cch);
+        return PyUnicode_FromUnicode((const Py_UNICODE*)sz, cch);
 
 #if HAVE_WCHAR_H
     if (sizeof(wchar_t) == sizeof(SQLWCHAR))
@@ -116,3 +116,35 @@ PyObject* PyUnicode_FromSQLWCHAR(const SQLWCHAR* sz, Py_ssize_t cch)
 
     return result.Detach();
 }
+
+void SQLWChar::dump()
+{
+    printf("sqlwchar=%ld pch=%p len=%ld owns=%d\n", sizeof(SQLWCHAR), pch, len, (int)owns_memory);
+    if (pch && len)
+    {
+        int i = 0;
+        while (i < len)
+        {
+            int stop = min(i + 10, len);
+
+            for (int x = i; x < stop; x++)
+            {
+                for (int byteindex = (int)sizeof(SQLWCHAR)-1; byteindex >= 0; byteindex--)
+                {
+                    int byte = (pch[x] >> (byteindex * 8)) & 0xFF;
+                    printf("%02x", byte);
+                }
+                printf(" ");
+            }
+
+            for (int x = i; x < stop; x++)
+                printf("%c", (char)pch[x]);
+
+            printf("\n");
+
+            i += 10;
+        }
+
+        printf("\n\n");
+    }
+}

src/sqlwchar.h

@@ -28,6 +28,8 @@ class SQLWChar
         Free();
     }
 
+    void dump();
+
     operator SQLWCHAR*() { return pch; }
     operator const SQLWCHAR*() const { return pch; }
     operator bool() const { return pch != 0; }

tests/pgtests.py

@@ -41,12 +41,13 @@ class PGTestCase(unittest.TestCase):
     SMALL_STRING = _generate_test_string(SMALL_READ)
     LARGE_STRING = _generate_test_string(LARGE_READ)
 
-    def __init__(self, connection_string, method_name):
+    def __init__(self, connection_string, ansi, method_name):
         unittest.TestCase.__init__(self, method_name)
         self.connection_string = connection_string
+        self.ansi = ansi
 
     def setUp(self):
-        self.cnxn   = pyodbc.connect(self.connection_string)
+        self.cnxn   = pyodbc.connect(self.connection_string, ansi=self.ansi)
         self.cursor = self.cnxn.cursor()
 
         for i in range(3):
@@ -372,6 +373,7 @@ def main():
     parser.add_option("-v", "--verbose", action="count", help="Increment test verbosity (can be used multiple times)")
     parser.add_option("-d", "--debug", action="store_true", default=False, help="Print debugging items")
     parser.add_option("-t", "--test", help="Run only the named test")
+    parser.add_option('-a', '--ansi', help='ANSI only', default=False, action='store_true')
 
     (options, args) = parser.parse_args()
 
@@ -388,26 +390,26 @@ def main():
         connection_string = args[0]
 
     if options.verbose:
-        cnxn = pyodbc.connect(connection_string)
-
+        cnxn = pyodbc.connect(connection_string, ansi=options.ansi)
         print 'library:', os.path.abspath(pyodbc.__file__)
         print 'odbc:    %s' % cnxn.getinfo(pyodbc.SQL_ODBC_VER)
         print 'driver:  %s %s' % (cnxn.getinfo(pyodbc.SQL_DRIVER_NAME), cnxn.getinfo(pyodbc.SQL_DRIVER_VER))
         print 'driver supports ODBC version %s' % cnxn.getinfo(pyodbc.SQL_DRIVER_ODBC_VER)
+        print 'unicode:', pyodbc.UNICODE_SIZE, 'sqlwchar:', pyodbc.SQLWCHAR_SIZE
         cnxn.close()
 
     if options.test:
         # Run a single test
         if not options.test.startswith('test_'):
             options.test = 'test_%s' % (options.test)
 
-        s = unittest.TestSuite([ PGTestCase(connection_string, options.test) ])
+        s = unittest.TestSuite([ PGTestCase(connection_string, options.ansi, options.test) ])
     else:
         # Run all tests in the class
 
         methods = [ m for m in dir(PGTestCase) if m.startswith('test_') ]
         methods.sort()
-        s = unittest.TestSuite([ PGTestCase(connection_string, m) for m in methods ])
+        s = unittest.TestSuite([ PGTestCase(connection_string, options.ansi, m) for m in methods ])
 
     testRunner = unittest.TextTestRunner(verbosity=options.verbose)
     result = testRunner.run(s)