mirror of
https://github.com/psycopg/psycopg2.git
synced 2024-11-22 08:56:34 +03:00
DictCursor/RealDictCursor tests split
This commit is contained in:
Daniele Varrazzo
parent
f947c0e6be
commit
82ae44ac3a
|
|
@ -22,9 +22,7 @@ import unittest
|
|||
from .testutils import ConnectingTestCase, skip_before_postgres, skip_before_python
|
||||
|
||||
|
||||
class ExtrasDictCursorTests(ConnectingTestCase):
|
||||
"""Test if DictCursor extension class works."""
|
||||
|
||||
class _DictCursorBase(ConnectingTestCase):
|
||||
def setUp(self):
|
||||
ConnectingTestCase.setUp(self)
|
||||
curs = self.conn.cursor()
|
||||
|
|
@ -32,6 +30,30 @@ class ExtrasDictCursorTests(ConnectingTestCase):
|
|||
curs.execute("INSERT INTO ExtrasDictCursorTests VALUES ('bar')")
|
||||
self.conn.commit()
|
||||
|
||||
def _testIterRowNumber(self, curs):
|
||||
# Only checking for dataset < itersize:
|
||||
# see CursorTests.test_iter_named_cursor_rownumber
|
||||
curs.itersize = 20
|
||||
curs.execute("""select * from generate_series(1,10)""")
|
||||
for i, r in enumerate(curs):
|
||||
self.assertEqual(i + 1, curs.rownumber)
|
||||
|
||||
def _testNamedCursorNotGreedy(self, curs):
|
||||
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))
|
||||
|
||||
|
||||
class ExtrasDictCursorTests(_DictCursorBase):
|
||||
"""Test if DictCursor extension class works."""
|
||||
|
||||
def testDictConnCursorArgs(self):
|
||||
self.conn.close()
|
||||
self.conn = self.connect(connection_factory=psycopg2.extras.DictConnection)
|
||||
|
|
@ -81,35 +103,6 @@ class ExtrasDictCursorTests(ConnectingTestCase):
|
|||
self.failUnless(row[0] == 'bar')
|
||||
return row
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchOne(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchone())
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchMany(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchmany(100)[0])
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchManyNoarg(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchmany()[0])
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchAll(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchall()[0])
|
||||
|
||||
def testDictCursorWithPlainCursorRealIter(self):
|
||||
def getter(curs):
|
||||
for row in curs:
|
||||
return row
|
||||
self._testWithPlainCursorReal(getter)
|
||||
|
||||
@skip_before_postgres(8, 0)
|
||||
def testDictCursorWithPlainCursorRealIterRowNumber(self):
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||||
self._testIterRowNumber(curs)
|
||||
|
||||
def _testWithPlainCursorReal(self, getter):
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||||
curs.execute("SELECT * FROM ExtrasDictCursorTests")
|
||||
row = getter(curs)
|
||||
self.failUnless(row['foo'] == 'bar')
|
||||
|
||||
def testDictCursorWithNamedCursorFetchOne(self):
|
||||
self._testWithNamedCursor(lambda curs: curs.fetchone())
|
||||
|
||||
|
|
@ -145,6 +138,63 @@ class ExtrasDictCursorTests(ConnectingTestCase):
|
|||
self.failUnless(row['foo'] == 'bar')
|
||||
self.failUnless(row[0] == 'bar')
|
||||
|
||||
def testPickleDictRow(self):
|
||||
import pickle
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
||||
curs.execute("select 10 as a, 20 as b")
|
||||
r = curs.fetchone()
|
||||
d = pickle.dumps(r)
|
||||
r1 = pickle.loads(d)
|
||||
self.assertEqual(r, r1)
|
||||
self.assertEqual(r[0], r1[0])
|
||||
self.assertEqual(r[1], r1[1])
|
||||
self.assertEqual(r['a'], r1['a'])
|
||||
self.assertEqual(r['b'], r1['b'])
|
||||
self.assertEqual(r._index, r1._index)
|
||||
|
||||
|
||||
class ExtrasDictCursorRealTests(_DictCursorBase):
|
||||
def testDictCursorWithPlainCursorRealFetchOne(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchone())
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchMany(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchmany(100)[0])
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchManyNoarg(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchmany()[0])
|
||||
|
||||
def testDictCursorWithPlainCursorRealFetchAll(self):
|
||||
self._testWithPlainCursorReal(lambda curs: curs.fetchall()[0])
|
||||
|
||||
def testDictCursorWithPlainCursorRealIter(self):
|
||||
def getter(curs):
|
||||
for row in curs:
|
||||
return row
|
||||
self._testWithPlainCursorReal(getter)
|
||||
|
||||
@skip_before_postgres(8, 0)
|
||||
def testDictCursorWithPlainCursorRealIterRowNumber(self):
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||||
self._testIterRowNumber(curs)
|
||||
|
||||
def _testWithPlainCursorReal(self, getter):
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||||
curs.execute("SELECT * FROM ExtrasDictCursorTests")
|
||||
row = getter(curs)
|
||||
self.failUnless(row['foo'] == 'bar')
|
||||
|
||||
def testPickleRealDictRow(self):
|
||||
import pickle
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||||
curs.execute("select 10 as a, 20 as b")
|
||||
r = curs.fetchone()
|
||||
d = pickle.dumps(r)
|
||||
r1 = pickle.loads(d)
|
||||
self.assertEqual(r, r1)
|
||||
self.assertEqual(r['a'], r1['a'])
|
||||
self.assertEqual(r['b'], r1['b'])
|
||||
self.assertEqual(r._column_mapping, r1._column_mapping)
|
||||
|
||||
def testDictCursorRealWithNamedCursorFetchOne(self):
|
||||
self._testWithNamedCursorReal(lambda curs: curs.fetchone())
|
||||
|
||||
|
|
@ -180,52 +230,6 @@ class ExtrasDictCursorTests(ConnectingTestCase):
|
|||
row = getter(curs)
|
||||
self.failUnless(row['foo'] == 'bar')
|
||||
|
||||
def _testNamedCursorNotGreedy(self, curs):
|
||||
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 _testIterRowNumber(self, curs):
|
||||
# Only checking for dataset < itersize:
|
||||
# see CursorTests.test_iter_named_cursor_rownumber
|
||||
curs.itersize = 20
|
||||
curs.execute("""select * from generate_series(1,10)""")
|
||||
for i, r in enumerate(curs):
|
||||
self.assertEqual(i + 1, curs.rownumber)
|
||||
|
||||
def testPickleDictRow(self):
|
||||
import pickle
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
||||
curs.execute("select 10 as a, 20 as b")
|
||||
r = curs.fetchone()
|
||||
d = pickle.dumps(r)
|
||||
r1 = pickle.loads(d)
|
||||
self.assertEqual(r, r1)
|
||||
self.assertEqual(r[0], r1[0])
|
||||
self.assertEqual(r[1], r1[1])
|
||||
self.assertEqual(r['a'], r1['a'])
|
||||
self.assertEqual(r['b'], r1['b'])
|
||||
self.assertEqual(r._index, r1._index)
|
||||
|
||||
def testPickleRealDictRow(self):
|
||||
import pickle
|
||||
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
|
||||
curs.execute("select 10 as a, 20 as b")
|
||||
r = curs.fetchone()
|
||||
d = pickle.dumps(r)
|
||||
r1 = pickle.loads(d)
|
||||
self.assertEqual(r, r1)
|
||||
self.assertEqual(r['a'], r1['a'])
|
||||
self.assertEqual(r['b'], r1['b'])
|
||||
self.assertEqual(r._column_mapping, r1._column_mapping)
|
||||
|
||||
|
||||
class NamedTupleCursorTest(ConnectingTestCase):
|
||||
def setUp(self):
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user