| 669 | con.close() |
| 670 | |
| 671 | def test_mixedfetch(self): |
| 672 | con = self._connect() |
| 673 | try: |
| 674 | cur = con.cursor() |
| 675 | self.executeDDL1(cur) |
| 676 | for sql in self._populate(): |
| 677 | cur.execute(sql) |
| 678 | |
| 679 | cur.execute('select name from %sbooze' % self.table_prefix) |
| 680 | rows1 = cur.fetchone() |
| 681 | rows23 = cur.fetchmany(2) |
| 682 | rows4 = cur.fetchone() |
| 683 | rows56 = cur.fetchall() |
| 684 | self.assertTrue(cur.rowcount in (-1,6)) |
| 685 | self.assertEqual(len(rows23),2, |
| 686 | 'fetchmany returned incorrect number of rows' |
| 687 | ) |
| 688 | self.assertEqual(len(rows56),2, |
| 689 | 'fetchall returned incorrect number of rows' |
| 690 | ) |
| 691 | |
| 692 | rows = [rows1[0]] |
| 693 | rows.extend([rows23[0][0],rows23[1][0]]) |
| 694 | rows.append(rows4[0]) |
| 695 | rows.extend([rows56[0][0],rows56[1][0]]) |
| 696 | rows.sort() |
| 697 | for i in range(0,len(self.samples)): |
| 698 | self.assertEqual(rows[i],self.samples[i], |
| 699 | 'incorrect data retrieved or inserted' |
| 700 | ) |
| 701 | finally: |
| 702 | con.close() |
| 703 | |
| 704 | def help_nextset_setUp(self,cur): |
| 705 | ''' Should create a procedure called deleteme |