From 82ae44ac3ae3bb8d474b5f4f50dd031d06f00147 Mon Sep 17 00:00:00 2001 From: Daniele Varrazzo Date: Sun, 20 May 2018 23:35:26 +0100 Subject: [PATCH] DictCursor/RealDictCursor tests split --- tests/test_extras_dictcursor.py | 160 ++++++++++++++++---------------- 1 file changed, 82 insertions(+), 78 deletions(-) diff --git a/tests/test_extras_dictcursor.py b/tests/test_extras_dictcursor.py index d43980af..13477791 100755 --- a/tests/test_extras_dictcursor.py +++ b/tests/test_extras_dictcursor.py @@ -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):