Author: cito
Date: Sun Nov 22 14:09:49 2015
New Revision: 606
Log:
Amend the large obj tests for Windows
Modified:
trunk/module/TEST_PyGreSQL_classic_largeobj.py
Modified: trunk/module/TEST_PyGreSQL_classic_largeobj.py
==============================================================================
--- trunk/module/TEST_PyGreSQL_classic_largeobj.py Sun Nov 22 13:20:09
2015 (r605)
+++ trunk/module/TEST_PyGreSQL_classic_largeobj.py Sun Nov 22 14:09:49
2015 (r606)
@@ -16,6 +16,7 @@
except ImportError:
import unittest
import tempfile
+import os
import pg # the module under test
@@ -30,6 +31,8 @@
except ImportError:
pass
+windows = os.name == 'nt'
+
def connect():
"""Create a basic pg connection to the test database."""
@@ -59,7 +62,7 @@
self.c.query('begin')
def tearDown(self):
- self.c.query('end')
+ self.c.query('rollback')
self.c.close()
def assertIsLargeObject(self, obj):
@@ -113,14 +116,26 @@
self.assertEqual(r, data)
def testLoImport(self):
- f = tempfile.NamedTemporaryFile()
+ if windows:
+ # NamedTemporaryFiles don't work well here
+ fname = 'temp_test_pg_largeobj_import.txt'
+ f = open(fname, 'wb')
+ else:
+ f = tempfile.NamedTemporaryFile()
+ fname = f.name
data = b'some data to be imported'
f.write(data)
- f.flush()
- f.seek(0)
+ if windows:
+ f.close()
+ f = open(fname, 'rb')
+ else:
+ f.flush()
+ f.seek(0)
large_object = self.c.loimport(f.name)
try:
f.close()
+ if windows:
+ os.remove(fname)
self.assertIsLargeObject(large_object)
large_object.open(pg.INV_READ)
large_object.seek(0, pg.SEEK_SET)
@@ -148,14 +163,17 @@
if self.obj.oid:
try:
self.obj.close()
- except IOError:
+ except (SystemError, IOError):
pass
try:
self.obj.unlink()
- except IOError:
+ except (SystemError, IOError):
pass
del self.obj
- self.pgcnx.query('end')
+ try:
+ self.pgcnx.query('rollback')
+ except SystemError:
+ pass
self.pgcnx.close()
def testClassName(self):
@@ -363,7 +381,10 @@
finally:
del obj
unlink()
- self.assertIsNone(self.obj.oid)
+ try:
+ self.assertIsNone(self.obj.oid)
+ except SystemError:
+ pass
def testSize(self):
size = self.obj.size
@@ -412,16 +433,27 @@
self.assertRaises(TypeError, export)
self.assertRaises(TypeError, export, 0)
self.assertRaises(TypeError, export, 'invalid', 0)
- f = tempfile.NamedTemporaryFile()
+ if windows:
+ # NamedTemporaryFiles don't work well here
+ fname = 'temp_test_pg_largeobj_export.txt'
+ f = open(fname, 'wb')
+ else:
+ f = tempfile.NamedTemporaryFile()
+ fname = f.name
data = b'some data to be exported'
self.obj.open(pg.INV_WRITE)
self.obj.write(data)
# exporting when object is not yet closed
self.assertRaises(IOError, export, f.name)
self.obj.close()
- export(f.name)
+ export(fname)
+ if windows:
+ f.close()
+ f = open(fname, 'rb')
r = f.read()
f.close()
+ if windows:
+ os.remove(fname)
self.assertIsInstance(r, bytes)
self.assertEqual(r, data)
_______________________________________________
PyGreSQL mailing list
[email protected]
https://mail.vex.net/mailman/listinfo.cgi/pygresql
