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