source: branches/4.x/module/TEST_PyGreSQL_classic_largeobj.py @ 552

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

unittest2 is needed for Python 2.6 as well

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