Add basic tests for large object code. Fix lobject.seek() to actually

work.

psycopg/lobject_int.c

@@ -255,10 +255,14 @@ lobject_seek(lobjectObject *self, int pos, int whence)
     char *error = NULL;
     int where;
 
+    Dprintf("lobject_seek: fd = %d, pos = %d, whence = %d",
+            self->fd, pos, whence);
+
     Py_BEGIN_ALLOW_THREADS;
     pthread_mutex_lock(&(self->conn->lock));
 
     where = lo_lseek(self->conn->pgconn, self->fd, pos, whence);
+    Dprintf("lobject_seek: where = %d", where);
     if (where < 0)
         collect_error(self->conn, &error);
 
@@ -279,10 +283,13 @@ lobject_tell(lobjectObject *self)
     char *error = NULL;
     int where;
 
+    Dprintf("lobject_tell: fd = %d", self->fd);
+
     Py_BEGIN_ALLOW_THREADS;
     pthread_mutex_lock(&(self->conn->lock));
 
     where = lo_tell(self->conn->pgconn, self->fd);
+    Dprintf("lobject_tell: where = %d", where);
     if (where < 0)
         collect_error(self->conn, &error);
 

psycopg/lobject_type.c

@@ -140,7 +140,7 @@ psyco_lobj_seek(lobjectObject *self, PyObject *args)
     EXC_IF_LOBJ_LEVEL0(self);
     EXC_IF_LOBJ_UNMARKED(self);
  
-    if ((pos = lobject_seek(self, pos, whence)) < 0)
+    if ((pos = lobject_seek(self, offset, whence)) < 0)
     	return NULL;
     
     return PyInt_FromLong((long)pos);

tests/__init__.py

@@ -24,6 +24,7 @@ import test_quote
 import test_connection
 import test_transaction
 import types_basic
+import test_lobject
 
 def test_suite():
     suite = unittest.TestSuite()
@@ -35,6 +36,7 @@ def test_suite():
     suite.addTest(test_connection.test_suite())
     suite.addTest(test_transaction.test_suite())
     suite.addTest(types_basic.test_suite())
+    suite.addTest(test_lobject.test_suite())
     return suite
 
 if __name__ == '__main__':

tests/test_lobject.py

@@ -0,0 +1,80 @@
+import unittest
+
+import psycopg2
+import psycopg2.extensions
+import tests
+
+
+class LargeObjectTests(unittest.TestCase):
+
+    def setUp(self):
+        self.conn = psycopg2.connect(tests.dsn)
+        self.lo_oid = None
+
+    def tearDown(self):
+        if self.lo_oid is not None:
+            self.conn.lobject(self.lo_oid).unlink()
+
+    def test_create(self):
+        lo = self.conn.lobject()
+        self.assertNotEqual(lo, None)
+        self.assertEqual(lo.mode, "w")
+
+    def test_open_non_existent(self):
+        # This test will give a false negative if Oid 42 is a large object.
+        self.assertRaises(psycopg2.OperationalError, self.conn.lobject, 42)
+
+    def test_open_existing(self):
+        lo = self.conn.lobject()
+        lo2 = self.conn.lobject(lo.oid)
+        self.assertNotEqual(lo2, None)
+        self.assertEqual(lo2.oid, lo.oid)
+        self.assertEqual(lo2.mode, "r")
+
+    def test_close(self):
+        lo = self.conn.lobject()
+        self.assertEqual(lo.closed, False)
+        lo.close()
+        self.assertEqual(lo.closed, True)
+
+    def test_write(self):
+        lo = self.conn.lobject()
+        self.assertEqual(lo.write("some data"), len("some data"))
+
+    def test_seek_tell(self):
+        lo = self.conn.lobject()
+        length = lo.write("some data")
+        self.assertEqual(lo.tell(), length)
+        lo.close()
+        lo = self.conn.lobject(lo.oid)
+
+        self.assertEqual(lo.seek(5, 0), 5)
+        self.assertEqual(lo.tell(), 5)
+
+        # SEEK_CUR: relative current location
+        self.assertEqual(lo.seek(2, 1), 7)
+        self.assertEqual(lo.tell(), 7)
+
+        # SEEK_END: relative to end of file
+        self.assertEqual(lo.seek(-2, 2), length - 2)
+
+    def test_read(self):
+        lo = self.conn.lobject()
+        length = lo.write("some data")
+        lo.close()
+
+        lo = self.conn.lobject(lo.oid)
+        self.assertEqual(lo.read(4), "some")
+        self.assertEqual(lo.read(), " data")
+
+    def test_unlink(self):
+        lo = self.conn.lobject()
+        lo.close()
+        lo.unlink()
+
+        # the object doesn't exist now, so we can't reopen it.
+        self.assertRaises(psycopg2.OperationalError, self.conn.lobject, lo.oid)
+
+
+def test_suite():
+    return unittest.TestLoader().loadTestsFromName(__name__)