

def test_fetchmany(self):
    con = self._connect()
    try:
        cur = con.cursor()
        self.assertRaises(self.driver.Error, cur.fetchmany, 4)
        self.executeDDL1(cur)
        for sql in self._populate():
            cur.execute(sql)
        cur.execute(('select name from %sbooze' % self.table_prefix))
        r = cur.fetchmany()
        self.assertEqual(len(r), 1, 'cursor.fetchmany retrieved incorrect number of rows, default of arraysize is one.')
        cur.arraysize = 10
        r = cur.fetchmany(3)
        self.assertEqual(len(r), 3, 'cursor.fetchmany retrieved incorrect number of rows')
        r = cur.fetchmany(4)
        self.assertEqual(len(r), 2, 'cursor.fetchmany retrieved incorrect number of rows')
        r = cur.fetchmany(4)
        self.assertEqual(len(r), 0, 'cursor.fetchmany should return an empty sequence after results are exhausted')
        self.assertTrue((cur.rowcount in ((- 1), 6)))
        cur.arraysize = 4
        cur.execute(('select name from %sbooze' % self.table_prefix))
        r = cur.fetchmany()
        self.assertEqual(len(r), 4, 'cursor.arraysize not being honoured by fetchmany')
        r = cur.fetchmany()
        self.assertEqual(len(r), 2)
        r = cur.fetchmany()
        self.assertEqual(len(r), 0)
        self.assertTrue((cur.rowcount in ((- 1), 6)))
        cur.arraysize = 6
        cur.execute(('select name from %sbooze' % self.table_prefix))
        rows = cur.fetchmany()
        self.assertTrue((cur.rowcount in ((- 1), 6)))
        self.assertEqual(len(rows), 6)
        self.assertEqual(len(rows), 6)
        rows = [r[0] for r in rows]
        rows.sort()
        for i in range(0, 6):
            self.assertEqual(rows[i], self.samples[i], 'incorrect data retrieved by cursor.fetchmany')
        rows = cur.fetchmany()
        self.assertEqual(len(rows), 0, 'cursor.fetchmany should return an empty sequence if called after the whole result set has been fetched')
        self.assertTrue((cur.rowcount in ((- 1), 6)))
        self.executeDDL2(cur)
        cur.execute(('select name from %sbarflys' % self.table_prefix))
        r = cur.fetchmany()
        self.assertEqual(len(r), 0, 'cursor.fetchmany should return an empty sequence if query retrieved no rows')
        self.assertTrue((cur.rowcount in ((- 1), 0)))
    finally:
        con.close()

dataset/ETHPy150Open cloudera/hue/desktop/core/ext-py/MySQL-python-1.2.5/tests/dbapi20.py DatabaseAPI20Test.test_fetchmany/original