psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-23 01:16:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'devel'

Browse Source
This commit is contained in:
Federico Di Gregorio 2011-05-11 09:58:49 +02:00
commit ab685c2fc0
16 changed files with 458 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
NEWS
View File

@ -1,3 +1,18 @@
What's new in psycopg 2.4.1
---------------------------
- Use own parser for bytea output, not requiring anymore the libpq 9.0
to parse the hex format.
- Don't fail connection if the client encoding is a non-normalized
variant. Issue reported by Peter Eisentraut.
- Correctly detect an empty query sent to the backend (ticket #46).
- Fixed a SystemError clobbering libpq errors raised without SQLSTATE.
Bug vivisectioned by Eric Snow.
- Fixed interaction between NamedTuple and server-side cursors.
- Allow to specify --static-libpq on setup.py command line instead of
just in 'setup.cfg'. Patch provided by Matthew Ryan (ticket #48).
What's new in psycopg 2.4
-------------------------
@ -163,6 +178,8 @@ doc/html/advanced.html.
- Fixed TimestampFromTicks() and TimeFromTicks() for seconds >= 59.5.
- Fixed spurious exception raised when calling C typecasters from Python
ones.
What's new in psycopg 2.0.14
----------------------------

2
ZPsycopgDA/DA.py
View File

@ -16,7 +16,7 @@
# their work without bothering about the module dependencies.
ALLOWED_PSYCOPG_VERSIONS = ('2.4-beta1', '2.4-beta2', '2.4')
ALLOWED_PSYCOPG_VERSIONS = ('2.4-beta1', '2.4-beta2', '2.4', '2.4.1')
import sys
import time

4
doc/src/faq.rst
View File

@ -97,7 +97,9 @@ Psycopg converts :sql:`decimal`\/\ :sql:`numeric` database types into Python `!D
Transferring binary data from PostgreSQL 9.0 doesn't work.
PostgreSQL 9.0 uses by default `the "hex" format`__ to transfer
:sql:`bytea` data: the format can't be parsed by the libpq 8.4 and
earlier. Three options to solve the problem are:
earlier. The problem is solved in Psycopg 2.4.1, that uses its own parser
for the :sql:`bytea` format. For previous Psycopg releases, three options
to solve the problem are:
- set the bytea_output__ parameter to ``escape`` in the server;
- execute the database command ``SET bytea_output TO escape;`` in the

23
doc/src/usage.rst
View File

@ -271,6 +271,10 @@ the SQL string that would be sent to the database.
.. versionchanged:: 2.4
only strings were supported before.
.. versionchanged:: 2.4.1
can parse the 'hex' format from 9.0 servers without relying on the
version of the client library.
.. note::
In Python 2, if you have binary data in a `!str` object, you can pass them
@ -282,18 +286,15 @@ the SQL string that would be sent to the database.
.. warning::
PostgreSQL 9 uses by default `a new "hex" format`__ to emit :sql:`bytea`
fields. Unfortunately this format can't be parsed by libpq versions
before 9.0. This means that using a library client with version lesser
than 9.0 to talk with a server 9.0 or later you may have problems
receiving :sql:`bytea` data. To work around this problem you can set the
`bytea_output`__ parameter to ``escape``, either in the server
configuration or in the client session using a query such as ``SET
bytea_output TO escape;`` before trying to receive binary data.
Since version 9.0 PostgreSQL uses by default `a new "hex" format`__ to
emit :sql:`bytea` fields. Starting from Psycopg 2.4.1 the format is
correctly supported. If you use a previous version you will need some
extra care when receiving bytea from PostgreSQL: you must have at least
the libpq 9.0 installed on the client or alternatively you can set the
`bytea_output`__ configutation parameter to ``escape``, either in the
server configuration file or in the client session (using a query such as
``SET bytea_output TO escape;``) before receiving binary data.
Starting from Psycopg 2.4 this condition is detected and signaled with a
`~psycopg2.InterfaceError`.
.. __: http://www.postgresql.org/docs/9.0/static/datatype-binary.html
.. __: http://www.postgresql.org/docs/9.0/static/runtime-config-client.html#GUC-BYTEA-OUTPUT

14
lib/extras.py
View File

@ -28,7 +28,6 @@ and classes untill a better place in the distribution is found.
import os
import sys
import time
import codecs
import warnings
import re as regex
@ -291,21 +290,28 @@ class NamedTupleCursor(_cursor):
return nt(*t)
def fetchmany(self, size=None):
ts = _cursor.fetchmany(self, size)
nt = self.Record
if nt is None:
nt = self.Record = self._make_nt()
ts = _cursor.fetchmany(self, size)
return [nt(*t) for t in ts]
def fetchall(self):
ts = _cursor.fetchall(self)
nt = self.Record
if nt is None:
nt = self.Record = self._make_nt()
ts = _cursor.fetchall(self)
return [nt(*t) for t in ts]
def __iter__(self):
return iter(self.fetchall())
# Invoking _cursor.__iter__(self) goes to infinite recursion,
# so we do pagination by hand
while 1:
recs = self.fetchmany(self.itersize)
if not recs:
return
for rec in recs:
yield rec
try:
from collections import namedtuple

70
psycopg/connection_int.c
View File

@ -236,10 +236,45 @@ conn_get_standard_conforming_strings(PGconn *pgconn)
return equote;
}
/* Remove irrelevant chars from encoding name and turn it uppercase.
*
* Return a buffer allocated on Python heap,
* NULL and set an exception on error.
*/
static char *
clean_encoding_name(const char *enc)
{
const char *i = enc;
char *rv, *j;
/* convert to upper case and remove '-' and '_' from string */
if (!(j = rv = PyMem_Malloc(strlen(enc) + 1))) {
PyErr_NoMemory();
return NULL;
}
while (*i) {
if (!isalnum(*i)) {
++i;
}
else {
*j++ = toupper(*i++);
}
}
*j = '\0';
Dprintf("clean_encoding_name: %s -> %s", enc, rv);
return rv;
}
/* Convert a PostgreSQL encoding to a Python codec.
*
* Return a new copy of the codec name allocated on the Python heap,
* NULL with exception in case of error.
*
* 'enc' should be already normalized (uppercase, no - or _).
*/
static char *
conn_encoding_to_codec(const char *enc)
@ -285,7 +320,7 @@ exit:
static int
conn_read_encoding(connectionObject *self, PGconn *pgconn)
{
char *enc = NULL, *codec = NULL, *j;
char *enc = NULL, *codec = NULL;
const char *tmp;
int rv = -1;
@ -297,16 +332,10 @@ conn_read_encoding(connectionObject *self, PGconn *pgconn)
goto exit;
}
if (!(enc = PyMem_Malloc(strlen(tmp)+1))) {
PyErr_NoMemory();
if (!(enc = clean_encoding_name(tmp))) {
goto exit;
}
/* turn encoding in uppercase */
j = enc;
while (*tmp) { *j++ = toupper(*tmp++); }
*j = '\0';
/* Look for this encoding in Python codecs. */
if (!(codec = conn_encoding_to_codec(enc))) {
goto exit;
@ -965,21 +994,23 @@ conn_set_client_encoding(connectionObject *self, const char *enc)
PGresult *pgres = NULL;
char *error = NULL;
char query[48];
int res = 0;
char *codec;
int res = 1;
char *codec = NULL;
char *clean_enc = NULL;
/* If the current encoding is equal to the requested one we don't
issue any query to the backend */
if (strcmp(self->encoding, enc) == 0) return 0;
/* We must know what python codec this encoding is. */
if (!(codec = conn_encoding_to_codec(enc))) { return -1; }
if (!(clean_enc = clean_encoding_name(enc))) { goto exit; }
if (!(codec = conn_encoding_to_codec(clean_enc))) { goto exit; }
Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&self->lock);
/* set encoding, no encoding string is longer than 24 bytes */
PyOS_snprintf(query, 47, "SET client_encoding = '%s'", enc);
PyOS_snprintf(query, 47, "SET client_encoding = '%s'", clean_enc);
/* abort the current transaction, to set the encoding ouside of
transactions */
@ -994,21 +1025,18 @@ conn_set_client_encoding(connectionObject *self, const char *enc)
/* no error, we can proceeed and store the new encoding */
{
char *tmp = self->encoding;
self->encoding = NULL;
self->encoding = clean_enc;
PyMem_Free(tmp);
}
if (!(self->encoding = psycopg_strdup(enc, 0))) {
res = 1; /* don't call pq_complete_error below */
goto endlock;
clean_enc = NULL;
}
/* Store the python codec too. */
{
char *tmp = self->codec;
self->codec = NULL;
self->codec = codec;
PyMem_Free(tmp);
codec = NULL;
}
self->codec = codec;
Dprintf("conn_set_client_encoding: set encoding to %s (codec: %s)",
self->encoding, self->codec);
@ -1021,6 +1049,10 @@ endlock:
if (res < 0)
pq_complete_error(self, &pgres, &error);
exit:
PyMem_Free(clean_enc);
PyMem_Free(codec);
return res;
}

22
psycopg/connection_type.c
View File

@ -423,36 +423,18 @@ static PyObject *
psyco_conn_set_client_encoding(connectionObject *self, PyObject *args)
{
const char *enc;
char *buffer, *dest;
PyObject *rv = NULL;
Py_ssize_t len;
EXC_IF_CONN_CLOSED(self);
EXC_IF_CONN_ASYNC(self, set_client_encoding);
EXC_IF_TPC_PREPARED(self, set_client_encoding);
if (!PyArg_ParseTuple(args, "s#", &enc, &len)) return NULL;
if (!PyArg_ParseTuple(args, "s", &enc)) return NULL;
/* convert to upper case and remove '-' and '_' from string */
if (!(dest = buffer = PyMem_Malloc(len+1))) {
return PyErr_NoMemory();
}
while (*enc) {
if (*enc == '_' || *enc == '-') {
++enc;
}
else {
*dest++ = toupper(*enc++);
}
}
*dest = '\0';
if (conn_set_client_encoding(self, buffer) == 0) {
if (conn_set_client_encoding(self, enc) == 0) {
Py_INCREF(Py_None);
rv = Py_None;
}
PyMem_Free(buffer);
return rv;
}

20
psycopg/pqpath.c
View File

@ -172,16 +172,19 @@ pq_raise(connectionObject *conn, cursorObject *curs, PGresult *pgres)
if (pgres) {
err = PQresultErrorMessage(pgres);
if (err != NULL) {
Dprintf("pq_raise: PQresultErrorMessage: err=%s", err);
code = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
}
}
if (err == NULL)
if (err == NULL) {
err = PQerrorMessage(conn->pgconn);
Dprintf("pq_raise: PQerrorMessage: err=%s", err);
}
/* if the is no error message we probably called pq_raise without reason:
we need to set an exception anyway because the caller will probably
raise and a meaningful message is better than an empty one */
if (err == NULL) {
if (err == NULL || err[0] == '\0') {
PyErr_SetString(Error, "psycopg went psycotic without error set");
return;
}
@ -191,9 +194,15 @@ pq_raise(connectionObject *conn, cursorObject *curs, PGresult *pgres)
if (code != NULL) {
exc = exception_from_sqlstate(code);
}
else {
/* Fallback if there is no exception code (reported happening e.g.
* when the connection is closed). */
exc = DatabaseError;
}
/* try to remove the initial "ERROR: " part from the postgresql error */
err2 = strip_severity(err);
Dprintf("pq_raise: err2=%s", err2);
psyco_set_error(exc, curs, err2, err, code);
}
@ -1355,6 +1364,13 @@ pq_fetch(cursorObject *curs)
/* don't clear curs->pgres, because it contains the results! */
break;
case PGRES_EMPTY_QUERY:
PyErr_SetString(ProgrammingError,
"can't execute an empty query");
IFCLEARPGRES(curs->pgres);
ex = -1;
break;
default:
Dprintf("pq_fetch: uh-oh, something FAILED: pgconn = %p", curs->conn);
pq_raise(curs->conn, curs, NULL);

232
psycopg/typecast_binary.c
View File

@ -40,7 +40,7 @@ chunk_dealloc(chunkObject *self)
FORMAT_CODE_PY_SSIZE_T,
self->base, self->len
);
PQfreemem(self->base);
PyMem_Free(self->base);
Py_TYPE(self)->tp_free((PyObject *)self);
}
@ -127,95 +127,185 @@ PyTypeObject chunkType = {
chunk_doc /* tp_doc */
};
static PyObject *
static char *psycopg_parse_hex(
const char *bufin, Py_ssize_t sizein, Py_ssize_t *sizeout);
static char *psycopg_parse_escape(
const char *bufin, Py_ssize_t sizein, Py_ssize_t *sizeout);
/* The function is not static and not hidden as we use ctypes to test it. */
PyObject *
typecast_BINARY_cast(const char *s, Py_ssize_t l, PyObject *curs)
{
chunkObject *chunk = NULL;
PyObject *res = NULL;
char *str = NULL, *buffer = NULL;
size_t len;
char *buffer = NULL;
Py_ssize_t len;
if (s == NULL) {Py_INCREF(Py_None); return Py_None;}
/* PQunescapeBytea absolutely wants a 0-terminated string and we don't
want to copy the whole buffer, right? Wrong, but there isn't any other
way <g> */
if (s[l] != '\0') {
if ((buffer = PyMem_Malloc(l+1)) == NULL) {
PyErr_NoMemory();
goto fail;
if (s[0] == '\\' && s[1] == 'x') {
/* This is a buffer escaped in hex format: libpq before 9.0 can't
* parse it and we can't detect reliably the libpq version at runtime.
* So the only robust option is to parse it ourselves - luckily it's
* an easy format.
*/
if (NULL == (buffer = psycopg_parse_hex(s, l, &len))) {
goto exit;
}
/* Py_ssize_t->size_t cast is safe, as long as the Py_ssize_t is
* >= 0: */
assert (l >= 0);
strncpy(buffer, s, (size_t) l);
buffer[l] = '\0';
s = buffer;
}
str = (char*)PQunescapeBytea((unsigned char*)s, &len);
Dprintf("typecast_BINARY_cast: unescaped " FORMAT_CODE_SIZE_T " bytes",
len);
/* The type of the second parameter to PQunescapeBytea is size_t *, so it's
* possible (especially with Python < 2.5) to get a return value too large
* to fit into a Python container. */
if (len > (size_t) PY_SSIZE_T_MAX) {
PyErr_SetString(PyExc_IndexError, "PG buffer too large to fit in Python"
" buffer.");
goto fail;
}
/* Check the escaping was successful */
if (s[0] == '\\' && s[1] == 'x' /* input encoded in hex format */
&& str[0] == 'x' /* output resulted in an 'x' */
&& s[2] != '7' && s[3] != '8') /* input wasn't really an x (0x78) */
{
PyErr_SetString(InterfaceError,
"can't receive bytea data from server >= 9.0 with the current "
"libpq client library: please update the libpq to at least 9.0 "
"or set bytea_output to 'escape' in the server config "
"or with a query");
goto fail;
else {
/* This is a buffer in the classic bytea format. So we can handle it
* to the PQunescapeBytea to have it parsed, rignt? ...Wrong. We
* could, but then we'd have to record whether buffer was allocated by
* Python or by the libpq to dispose it properly. Furthermore the
* PQunescapeBytea interface is not the most brilliant as it wants a
* null-terminated string even if we have known its length thus
* requiring a useless memcpy and strlen.
* So we'll just have our better integrated parser, let's finish this
* story.
*/
if (NULL == (buffer = psycopg_parse_escape(s, l, &len))) {
goto exit;
}
}
chunk = (chunkObject *) PyObject_New(chunkObject, &chunkType);
if (chunk == NULL) goto fail;
if (chunk == NULL) goto exit;
/* **Transfer** ownership of str's memory to the chunkObject: */
chunk->base = str;
str = NULL;
/* **Transfer** ownership of buffer's memory to the chunkObject: */
chunk->base = buffer;
buffer = NULL;
chunk->len = (Py_ssize_t)len;
/* size_t->Py_ssize_t cast was validated above: */
chunk->len = (Py_ssize_t) len;
#if PY_MAJOR_VERSION < 3
if ((res = PyBuffer_FromObject((PyObject *)chunk, 0, chunk->len)) == NULL)
goto fail;
goto exit;
#else
if ((res = PyMemoryView_FromObject((PyObject*)chunk)) == NULL)
goto fail;
goto exit;
#endif
/* PyBuffer_FromObject() created a new reference. We'll release our
* reference held in 'chunk' in the 'cleanup' clause. */
goto cleanup;
fail:
assert (PyErr_Occurred());
if (res != NULL) {
Py_DECREF(res);
res = NULL;
}
/* Fall through to cleanup: */
cleanup:
if (chunk != NULL) {
Py_DECREF((PyObject *) chunk);
}
if (str != NULL) {
/* str's mem was allocated by PQunescapeBytea; must use PQfreemem: */
PQfreemem(str);
}
/* We allocated buffer with PyMem_Malloc; must use PyMem_Free: */
PyMem_Free(buffer);
exit:
Py_XDECREF((PyObject *)chunk);
PyMem_Free(buffer);
return res;
return res;
}
static const char hex_lut[128] = {
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, -1, -1, -1, -1, -1, -1,
-1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
};
/* Parse a bytea output buffer encoded in 'hex' format.
*
* the format is described in
* http://www.postgresql.org/docs/9.0/static/datatype-binary.html
*
* Parse the buffer in 'bufin', whose length is 'sizein'.
* Return a new buffer allocated by PyMem_Malloc and set 'sizeout' to its size.
* In case of error set an exception and return NULL.
*/
static char *
psycopg_parse_hex(const char *bufin, Py_ssize_t sizein, Py_ssize_t *sizeout)
{
char *ret = NULL;
const char *bufend = bufin + sizein;
const char *pi = bufin + 2; /* past the \x */
char *bufout;
char *po;
po = bufout = PyMem_Malloc((sizein - 2) >> 1); /* output size upper bound */
if (NULL == bufout) {
PyErr_NoMemory();
goto exit;
}
/* Implementation note: we call this function upon database response, not
* user input (because we are parsing the output format of a buffer) so we
* don't expect errors. On bad input we reserve the right to return a bad
* output, not an error.
*/
while (pi < bufend) {
char c;
while (-1 == (c = hex_lut[*pi++ & '\x7f'])) {
if (pi >= bufend) { goto endloop; }
}
*po = c << 4;
while (-1 == (c = hex_lut[*pi++ & '\x7f'])) {
if (pi >= bufend) { goto endloop; }
}
*po++ |= c;
}
endloop:
ret = bufout;
*sizeout = po - bufout;
exit:
return ret;
}
/* Parse a bytea output buffer encoded in 'escape' format.
*
* the format is described in
* http://www.postgresql.org/docs/9.0/static/datatype-binary.html
*
* Parse the buffer in 'bufin', whose length is 'sizein'.
* Return a new buffer allocated by PyMem_Malloc and set 'sizeout' to its size.
* In case of error set an exception and return NULL.
*/
static char *
psycopg_parse_escape(const char *bufin, Py_ssize_t sizein, Py_ssize_t *sizeout)
{
char *ret = NULL;
const char *bufend = bufin + sizein;
const char *pi = bufin;
char *bufout;
char *po;
po = bufout = PyMem_Malloc(sizein); /* output size upper bound */
if (NULL == bufout) {
PyErr_NoMemory();
goto exit;
}
while (pi < bufend) {
if (*pi != '\\') {
/* Unescaped char */
*po++ = *pi++;
continue;
}
if ((pi[1] >= '0' && pi[1] <= '3') &&
(pi[2] >= '0' && pi[2] <= '7') &&
(pi[3] >= '0' && pi[3] <= '7'))
{
/* Escaped octal value */
*po++ = ((pi[1] - '0') << 6) |
((pi[2] - '0') << 3) |
((pi[3] - '0'));
pi += 4;
}
else {
/* Escaped char */
*po++ = pi[1];
pi += 2;
}
}
ret = bufout;
*sizeout = po - bufout;
exit:
return ret;
}

5
setup.py
View File

@ -79,7 +79,7 @@ except ImportError:
# Take a look at http://www.python.org/dev/peps/pep-0386/
# for a consistent versioning pattern.
PSYCOPG_VERSION = '2.4'
PSYCOPG_VERSION = '2.4.1'
version_flags = ['dt', 'dec']
@ -133,6 +133,7 @@ class psycopg_build_ext(build_ext):
self.mx_include_dir = None
self.use_pydatetime = 1
self.have_ssl = have_ssl
self.static_libpq = static_libpq
self.pg_config = None
def get_compiler(self):
@ -263,7 +264,7 @@ or with the pg_config option in 'setup.cfg'.
sys.exit(1)
self.include_dirs.append(".")
if static_libpq:
if self.static_libpq:
if not self.link_objects: self.link_objects = []
self.link_objects.append(
os.path.join(self.get_pg_config("libdir"), "libpq.a"))

22
tests/__init__.py
View File

@ -27,17 +27,6 @@ import sys
from testconfig import dsn
from testutils import unittest
# If connection to test db fails, bail out early.
import psycopg2
try:
cnn = psycopg2.connect(dsn)
except Exception, e:
print "Failed connection to test db:", e.__class__.__name__, e
print "Please set env vars 'PSYCOPG2_TESTDB*' to valid values."
sys.exit(1)
else:
cnn.close()
import bug_gc
import bugX000
import extras_dictcursor
@ -57,6 +46,17 @@ import test_green
import test_cancel
def test_suite():
# If connection to test db fails, bail out early.
import psycopg2
try:
cnn = psycopg2.connect(dsn)
except Exception, e:
print "Failed connection to test db:", e.__class__.__name__, e
print "Please set env vars 'PSYCOPG2_TESTDB*' to valid values."
sys.exit(1)
else:
cnn.close()
suite = unittest.TestSuite()
suite.addTest(bug_gc.test_suite())
suite.addTest(bugX000.test_suite())

51
tests/extras_dictcursor.py
View File

@ -14,9 +14,11 @@
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import time
from datetime import timedelta
import psycopg2
import psycopg2.extras
from testutils import unittest, skip_if_no_namedtuple
from testutils import unittest, skip_before_postgres, skip_if_no_namedtuple
from testconfig import dsn
@ -261,6 +263,53 @@ class NamedTupleCursorTest(unittest.TestCase):
finally:
NamedTupleCursor._make_nt = f_orig
@skip_if_no_namedtuple
@skip_before_postgres(8, 0)
def test_named(self):
curs = self.conn.cursor('tmp')
curs.execute("""select i from generate_series(0,9) i""")
recs = []
recs.extend(curs.fetchmany(5))
recs.append(curs.fetchone())
recs.extend(curs.fetchall())
self.assertEqual(range(10), [t.i for t in recs])
@skip_if_no_namedtuple
def test_named_fetchone(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
t = curs.fetchone()
self.assertEqual(t.i, 42)
@skip_if_no_namedtuple
def test_named_fetchmany(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
recs = curs.fetchmany(10)
self.assertEqual(recs[0].i, 42)
@skip_if_no_namedtuple
def test_named_fetchall(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
recs = curs.fetchall()
self.assertEqual(recs[0].i, 42)
@skip_if_no_namedtuple
@skip_before_postgres(8, 2)
def test_not_greedy(self):
curs = self.conn.cursor('tmp')
curs.itersize = 2
curs.execute("""select clock_timestamp() as ts from generate_series(1,3)""")
recs = []
for t in curs:
time.sleep(0.01)
recs.append(t)
# check that the dataset was not fetched in a single gulp
self.assert_(recs[1].ts - recs[0].ts < timedelta(seconds=0.005))
self.assert_(recs[2].ts - recs[1].ts > timedelta(seconds=0.0099))
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

14
tests/test_connection.py
View File

@ -22,6 +22,7 @@
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
import os
import time
import threading
from testutils import unittest, decorate_all_tests, skip_before_postgres
@ -141,6 +142,19 @@ class ConnectionTests(unittest.TestCase):
cur.execute("select 'foo'::text;")
self.assertEqual(cur.fetchone()[0], u'foo')
def test_connect_nonnormal_envvar(self):
# We must perform encoding normalization at connection time
self.conn.close()
oldenc = os.environ.get('PGCLIENTENCODING')
os.environ['PGCLIENTENCODING'] = 'utf-8' # malformed spelling
try:
self.conn = psycopg2.connect(dsn)
finally:
if oldenc is not None:
os.environ['PGCLIENTENCODING'] = oldenc
else:
del os.environ['PGCLIENTENCODING']
def test_weakref(self):
from weakref import ref
conn = psycopg2.connect(dsn)

6
tests/test_cursor.py
View File

@ -37,6 +37,12 @@ class CursorTests(unittest.TestCase):
def tearDown(self):
self.conn.close()
def test_empty_query(self):
cur = self.conn.cursor()
self.assertRaises(psycopg2.ProgrammingError, cur.execute, "")
self.assertRaises(psycopg2.ProgrammingError, cur.execute, " ")
self.assertRaises(psycopg2.ProgrammingError, cur.execute, ";")
def test_executemany_propagate_exceptions(self):
conn = self.conn
cur = conn.cursor()

18
tests/testutils.py
View File

@ -140,24 +140,6 @@ def skip_if_no_namedtuple(f):
return skip_if_no_namedtuple_
def skip_if_broken_hex_binary(f):
"""Decorator to detect libpq < 9.0 unable to parse bytea in hex format"""
def cope_with_hex_binary_(self):
from psycopg2 import InterfaceError
try:
return f(self)
except InterfaceError, e:
if '9.0' in str(e) and self.conn.server_version >= 90000:
return self.skipTest(
# FIXME: we are only assuming the libpq is older here,
# but we don't have a reliable way to detect the libpq
# version, not pre-9 at least.
"bytea broken with server >= 9.0, libpq < 9")
else:
raise
return cope_with_hex_binary_
def skip_if_no_iobase(f):
"""Skip a test if io.TextIOBase is not available."""
def skip_if_no_iobase_(self):

108
tests/types_basic.py
View File

@ -28,7 +28,7 @@ except:
pass
import sys
import testutils
from testutils import unittest, skip_if_broken_hex_binary
from testutils import unittest, decorate_all_tests
from testconfig import dsn
import psycopg2
@ -116,7 +116,6 @@ class TypesBasicTests(unittest.TestCase):
s = self.execute("SELECT %s AS foo", (float("-inf"),))
self.failUnless(str(s) == "-inf", "wrong float quoting: " + str(s))
@skip_if_broken_hex_binary
def testBinary(self):
if sys.version_info[0] < 3:
s = ''.join([chr(x) for x in range(256)])
@ -143,7 +142,6 @@ class TypesBasicTests(unittest.TestCase):
b = psycopg2.Binary(bytes([]))
self.assertEqual(str(b), "''::bytea")
@skip_if_broken_hex_binary
def testBinaryRoundTrip(self):
# test to make sure buffers returned by psycopg2 are
# understood by execute:
@ -191,7 +189,6 @@ class TypesBasicTests(unittest.TestCase):
s = self.execute("SELECT '{}'::text AS foo")
self.failUnlessEqual(s, "{}")
@skip_if_broken_hex_binary
@testutils.skip_from_python(3)
def testTypeRoundtripBuffer(self):
o1 = buffer("".join(map(chr, range(256))))
@ -204,7 +201,6 @@ class TypesBasicTests(unittest.TestCase):
self.assertEqual(type(o1), type(o2))
self.assertEqual(str(o1), str(o2))
@skip_if_broken_hex_binary
@testutils.skip_from_python(3)
def testTypeRoundtripBufferArray(self):
o1 = buffer("".join(map(chr, range(256))))
@ -213,7 +209,6 @@ class TypesBasicTests(unittest.TestCase):
self.assertEqual(type(o1[0]), type(o2[0]))
self.assertEqual(str(o1[0]), str(o2[0]))
@skip_if_broken_hex_binary
@testutils.skip_before_python(3)
def testTypeRoundtripBytes(self):
o1 = bytes(range(256))
@ -225,7 +220,6 @@ class TypesBasicTests(unittest.TestCase):
o2 = self.execute("select %s;", (o1,))
self.assertEqual(memoryview, type(o2))
@skip_if_broken_hex_binary
@testutils.skip_before_python(3)
def testTypeRoundtripBytesArray(self):
o1 = bytes(range(256))
@ -233,7 +227,6 @@ class TypesBasicTests(unittest.TestCase):
o2 = self.execute("select %s;", (o1,))
self.assertEqual(memoryview, type(o2[0]))
@skip_if_broken_hex_binary
@testutils.skip_before_python(2, 6)
def testAdaptBytearray(self):
o1 = bytearray(range(256))
@ -258,7 +251,6 @@ class TypesBasicTests(unittest.TestCase):
else:
self.assertEqual(memoryview, type(o2))
@skip_if_broken_hex_binary
@testutils.skip_before_python(2, 7)
def testAdaptMemoryview(self):
o1 = memoryview(bytearray(range(256)))
@ -335,6 +327,104 @@ class AdaptSubclassTest(unittest.TestCase):
del psycopg2.extensions.adapters[A, psycopg2.extensions.ISQLQuote]
class ByteaParserTest(unittest.TestCase):
"""Unit test for our bytea format parser."""
def setUp(self):
try:
self._cast = self._import_cast()
except Exception, e:
self._cast = None
self._exc = e
def _import_cast(self):
"""Use ctypes to access the C function.
Raise any sort of error: we just support this where ctypes works as
expected.
"""
import ctypes
lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__)
cast = lib.typecast_BINARY_cast
cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object]
cast.restype = ctypes.py_object
return cast
def cast(self, buffer):
"""Cast a buffer from the output format"""
l = buffer and len(buffer) or 0
rv = self._cast(buffer, l, None)
if rv is None:
return None
if sys.version_info[0] < 3:
return str(rv)
else:
return rv.tobytes()
def test_null(self):
rv = self.cast(None)
self.assertEqual(rv, None)
def test_blank(self):
rv = self.cast(b(''))
self.assertEqual(rv, b(''))
def test_blank_hex(self):
# Reported as problematic in ticket #48
rv = self.cast(b('\\x'))
self.assertEqual(rv, b(''))
def test_full_hex(self, upper=False):
buf = ''.join(("%02x" % i) for i in range(256))
if upper: buf = buf.upper()
buf = '\\x' + buf
rv = self.cast(b(buf))
if sys.version_info[0] < 3:
self.assertEqual(rv, ''.join(map(chr, range(256))))
else:
self.assertEqual(rv, bytes(range(256)))
def test_full_hex_upper(self):
return self.test_full_hex(upper=True)
def test_full_escaped_octal(self):
buf = ''.join(("\\%03o" % i) for i in range(256))
rv = self.cast(b(buf))
if sys.version_info[0] < 3:
self.assertEqual(rv, ''.join(map(chr, range(256))))
else:
self.assertEqual(rv, bytes(range(256)))
def test_escaped_mixed(self):
import string
buf = ''.join(("\\%03o" % i) for i in range(32))
buf += string.ascii_letters
buf += ''.join('\\' + c for c in string.ascii_letters)
buf += '\\\\'
rv = self.cast(b(buf))
if sys.version_info[0] < 3:
tgt = ''.join(map(chr, range(32))) \
+ string.ascii_letters * 2 + '\\'
else:
tgt = bytes(range(32)) + \
(string.ascii_letters * 2 + '\\').encode('ascii')
self.assertEqual(rv, tgt)
def skip_if_cant_cast(f):
def skip_if_cant_cast_(self, *args, **kwargs):
if self._cast is None:
return self.skipTest("can't test bytea parser: %s - %s"
% (self._exc.__class__.__name__, self._exc))
return f(self, *args, **kwargs)
return skip_if_cant_cast_
decorate_all_tests(ByteaParserTest, skip_if_cant_cast)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)