source: trunk/module/TEST_PyGreSQL_classic_largeobj.py @ 553

Last change on this file since 553 was 553, checked in by cito, 4 years ago

Require at least Python 2.6 for the trunk (5.x)

Support for even older Python versions is maintained in the 4.x branch.
The goal for 5.x is to be a single-source code for both Python 2 and 3,
and this is only possible by dropping support for Python 2.5 and older.
For instance, the new except .. as syntax works only since Python 2.6.
Otherwise we would need to use 2to3 and things would be very ugly.
Note that Python 2.6 is now 7 years old. We may want to drop Python 2.6
as well at some point if it turns out to be a burden.

  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 11.3 KB
Line 
1#! /usr/bin/python
2# -*- coding: utf-8 -*-
3
4"""Test the classic PyGreSQL interface.
5
6Sub-tests for large object support.
7
8Contributed by Christoph Zwerschke.
9
10These tests need a database to test against.
11
12"""
13
14try:
15    import unittest2 as unittest  # for Python < 2.7
16except ImportError:
17    import unittest
18
19import tempfile
20
21import pg  # the module under test
22
23# We need a database to test against.  If LOCAL_PyGreSQL.py exists we will
24# get our information from that.  Otherwise we use the defaults.
25dbname = 'unittest'
26dbhost = None
27dbport = 5432
28
29try:
30    from LOCAL_PyGreSQL import *
31except ImportError:
32    pass
33
34
35def connect():
36    """Create a basic pg connection to the test database."""
37    connection = pg.connect(dbname, dbhost, dbport)
38    connection.query("set client_min_messages=warning")
39    return connection
40
41
42class TestModuleConstants(unittest.TestCase):
43    """Test the existence of the documented module constants."""
44
45    def testLargeObjectIntConstants(self):
46        names = 'INV_READ INV_WRITE SEEK_SET SEEK_CUR SEEK_END'.split()
47        for name in names:
48            try:
49                value = getattr(pg, name)
50            except AttributeError:
51                self.fail('Module constant %s is missing' % name)
52            self.assertIsInstance(value, int)
53
54
55class TestCreatingLargeObjects(unittest.TestCase):
56    """Test creating large objects using a connection."""
57
58    def setUp(self):
59        self.c = connect()
60        self.c.query('begin')
61
62    def tearDown(self):
63        self.c.query('end')
64        self.c.close()
65
66    def assertIsLargeObject(self, obj):
67        self.assertIsNotNone(obj)
68        self.assertTrue(hasattr(obj, 'open'))
69        self.assertTrue(hasattr(obj, 'close'))
70        self.assertTrue(hasattr(obj, 'oid'))
71        self.assertTrue(hasattr(obj, 'pgcnx'))
72        self.assertTrue(hasattr(obj, 'error'))
73        self.assertIsInstance(obj.oid, int)
74        self.assertNotEqual(obj.oid, 0)
75        self.assertIs(obj.pgcnx, self.c)
76        self.assertIsInstance(obj.error, str)
77        self.assertFalse(obj.error)
78
79    def testLoCreate(self):
80        large_object = self.c.locreate(pg.INV_READ | pg.INV_WRITE)
81        try:
82            self.assertIsLargeObject(large_object)
83        finally:
84            del large_object
85
86    def testGetLo(self):
87        large_object = self.c.locreate(pg.INV_READ | pg.INV_WRITE)
88        try:
89            self.assertIsLargeObject(large_object)
90            oid = large_object.oid
91        finally:
92            del large_object
93        data = 'some data to be shared'
94        large_object = self.c.getlo(oid)
95        try:
96            self.assertIsLargeObject(large_object)
97            self.assertEqual(large_object.oid, oid)
98            large_object.open(pg.INV_WRITE)
99            large_object.write(data)
100            large_object.close()
101        finally:
102            del large_object
103        large_object = self.c.getlo(oid)
104        try:
105            self.assertIsLargeObject(large_object)
106            self.assertEqual(large_object.oid, oid)
107            large_object.open(pg.INV_READ)
108            r = large_object.read(80)
109            large_object.close()
110            large_object.unlink()
111        finally:
112            del large_object
113        self.assertEqual(r, data)
114
115    def testLoImport(self):
116        f = tempfile.NamedTemporaryFile()
117        data = 'some data to be imported'
118        f.write(data)
119        f.flush()
120        f.seek(0)
121        large_object = self.c.loimport(f.name)
122        try:
123            f.close()
124            self.assertIsLargeObject(large_object)
125            large_object.open(pg.INV_READ)
126            large_object.seek(0, pg.SEEK_SET)
127            r = large_object.size()
128            self.assertIsInstance(r, int)
129            self.assertEqual(r, len(data))
130            r = large_object.read(80)
131            self.assertIsInstance(r, str)
132            self.assertEqual(r, data)
133            large_object.close()
134            large_object.unlink()
135        finally:
136            del large_object
137
138
139class TestLargeObjects(unittest.TestCase):
140    """Test the large object methods."""
141
142    def setUp(self):
143        self.pgcnx = connect()
144        self.pgcnx.query('begin')
145        self.obj = self.pgcnx.locreate(pg.INV_READ | pg.INV_WRITE)
146
147    def tearDown(self):
148        if self.obj.oid:
149            try:
150                self.obj.close()
151            except IOError:
152                pass
153            try:
154                self.obj.unlink()
155            except IOError:
156                pass
157        del self.obj
158        self.pgcnx.query('end')
159        self.pgcnx.close()
160
161    def testOid(self):
162        self.assertIsInstance(self.obj.oid, int)
163        self.assertNotEqual(self.obj.oid, 0)
164
165    def testPgcn(self):
166        self.assertIs(self.obj.pgcnx, self.pgcnx)
167
168    def testError(self):
169        self.assertIsInstance(self.obj.error, str)
170        self.assertEqual(self.obj.error, '')
171
172    def testOpen(self):
173        open = self.obj.open
174        # testing with invalid parameters
175        self.assertRaises(TypeError, open)
176        self.assertRaises(TypeError, open, pg.INV_READ, pg.INV_WRITE)
177        open(pg.INV_READ)
178        # object is already open
179        self.assertRaises(IOError, open, pg.INV_READ)
180
181    def testClose(self):
182        close = self.obj.close
183        # testing with invalid parameters
184        self.assertRaises(TypeError, close, pg.INV_READ)
185        # object is not yet open
186        self.assertRaises(IOError, close)
187        self.obj.open(pg.INV_READ)
188        close()
189        self.assertRaises(IOError, close)
190
191    def testRead(self):
192        read = self.obj.read
193        # testing with invalid parameters
194        self.assertRaises(TypeError, read)
195        self.assertRaises(ValueError, read, -1)
196        self.assertRaises(TypeError, read, 'invalid')
197        self.assertRaises(TypeError, read, 80, 'invalid')
198        # reading when object is not yet open
199        self.assertRaises(IOError, read, 80)
200        data = 'some data to be read'
201        self.obj.open(pg.INV_WRITE)
202        self.obj.write(data)
203        self.obj.close()
204        self.obj.open(pg.INV_READ)
205        r = read(80)
206        self.assertIsInstance(r, str)
207        self.assertEqual(r, data)
208        self.obj.close()
209        self.obj.open(pg.INV_READ)
210        r = read(8)
211        self.assertIsInstance(r, str)
212        self.assertEqual(r, data[:8])
213        self.obj.close()
214
215    def testWrite(self):
216        write = self.obj.write
217        # testing with invalid parameters
218        self.assertRaises(TypeError, write)
219        self.assertRaises(TypeError, write, -1)
220        self.assertRaises(TypeError, write, '', 'invalid')
221        # writing when object is not yet open
222        self.assertRaises(IOError, write, 'invalid')
223        data = 'some data to be written'
224        self.obj.open(pg.INV_WRITE)
225        write(data)
226        self.obj.close()
227        self.obj.open(pg.INV_READ)
228        r = self.obj.read(80)
229        self.assertEqual(r, data)
230
231    def testSeek(self):
232        seek = self.obj.seek
233        # testing with invalid parameters
234        self.assertRaises(TypeError, seek)
235        self.assertRaises(TypeError, seek, 0)
236        self.assertRaises(TypeError, seek, 0, pg.SEEK_SET, pg.SEEK_END)
237        self.assertRaises(TypeError, seek, 'invalid', pg.SEEK_SET)
238        self.assertRaises(TypeError, seek, 0, 'invalid')
239        # seeking when object is not yet open
240        self.assertRaises(IOError, seek, 0, pg.SEEK_SET)
241        data = 'some data to be seeked'
242        self.obj.open(pg.INV_WRITE)
243        self.obj.write(data)
244        self.obj.close()
245        self.obj.open(pg.INV_READ)
246        seek(0, pg.SEEK_SET)
247        r = self.obj.read(9)
248        self.assertEqual(r, 'some data')
249        seek(4, pg.SEEK_CUR)
250        r = self.obj.read(2)
251        self.assertEqual(r, 'be')
252        seek(-10, pg.SEEK_CUR)
253        r = self.obj.read(4)
254        self.assertEqual(r, 'data')
255        seek(0, pg.SEEK_SET)
256        r = self.obj.read(4)
257        self.assertEqual(r, 'some')
258        seek(-6, pg.SEEK_END)
259        r = self.obj.read(4)
260        self.assertEqual(r, 'seek')
261
262    def testTell(self):
263        tell = self.obj.tell
264        # testing with invalid parameters
265        self.assertRaises(TypeError, tell, 0)
266        # telling when object is not yet open
267        self.assertRaises(IOError, tell)
268        data = 'some story to be told'
269        self.obj.open(pg.INV_WRITE)
270        self.obj.write(data)
271        r = tell()
272        self.assertIsInstance(r, int)
273        self.assertEqual(r, len(data))
274        self.obj.close()
275        self.obj.open(pg.INV_READ)
276        r = tell()
277        self.assertIsInstance(r, int)
278        self.assertEqual(r, 0)
279        self.obj.seek(5, pg.SEEK_SET)
280        r = tell()
281        self.assertIsInstance(r, int)
282        self.assertEqual(r, 5)
283
284    def testUnlink(self):
285        unlink = self.obj.unlink
286        # testing with invalid parameters
287        self.assertRaises(TypeError, unlink, 0)
288        # unlinking when object is still open
289        self.obj.open(pg.INV_WRITE)
290        self.assertRaises(IOError, unlink)
291        data = 'some data to be sold'
292        self.obj.write(data)
293        self.obj.close()
294        oid = self.obj.oid
295        self.assertIsInstance(oid, int)
296        self.assertNotEqual(oid, 0)
297        obj = self.pgcnx.getlo(oid)
298        try:
299            self.assertIsNot(obj, self.obj)
300            self.assertEqual(obj.oid, oid)
301            obj.open(pg.INV_READ)
302            r = obj.read(80)
303            obj.close()
304            self.assertEqual(r, data)
305        finally:
306            del obj
307        unlink()
308        self.assertIsNone(self.obj.oid)
309
310    def testSize(self):
311        size = self.obj.size
312        # testing with invalid parameters
313        self.assertRaises(TypeError, size, 0)
314        # sizing when object is not yet open
315        self.assertRaises(IOError, size)
316        # sizing an empty object
317        self.obj.open(pg.INV_READ)
318        r = size()
319        self.obj.close()
320        self.assertIsInstance(r, int)
321        self.assertEqual(r, 0)
322        # sizing after adding some data
323        data = 'some data to be sized'
324        self.obj.open(pg.INV_WRITE)
325        self.obj.write(data)
326        self.obj.close()
327        # sizing when current position is zero
328        self.obj.open(pg.INV_READ)
329        r = size()
330        self.obj.close()
331        self.assertIsInstance(r, int)
332        self.assertEqual(r, len(data))
333        self.obj.open(pg.INV_READ)
334        # sizing when current position is not zero
335        self.obj.seek(5, pg.SEEK_SET)
336        r = size()
337        self.obj.close()
338        self.assertIsInstance(r, int)
339        self.assertEqual(r, len(data))
340        # sizing after adding more data
341        data += ' and more data'
342        self.obj.open(pg.INV_WRITE)
343        self.obj.write(data)
344        self.obj.close()
345        self.obj.open(pg.INV_READ)
346        r = size()
347        self.obj.close()
348        self.assertIsInstance(r, int)
349        self.assertEqual(r, len(data))
350
351    def testExport(self):
352        export = self.obj.export
353        # testing with invalid parameters
354        self.assertRaises(TypeError, export)
355        self.assertRaises(TypeError, export, 0)
356        self.assertRaises(TypeError, export, 'invalid', 0)
357        f = tempfile.NamedTemporaryFile()
358        data = 'some data to be exported'
359        self.obj.open(pg.INV_WRITE)
360        self.obj.write(data)
361        # exporting when object is not yet closed
362        self.assertRaises(IOError, export, f.name)
363        self.obj.close()
364        export(f.name)
365        r = f.read()
366        f.close()
367        self.assertEqual(r, data)
368
369
370if __name__ == '__main__':
371    unittest.main()
Note: See TracBrowser for help on using the repository browser.