| 83 | self._ensure_cursor_expired(c) |
| 84 | |
| 85 | def test_custom_dict(self): |
| 86 | class MyDict(dict): |
| 87 | pass |
| 88 | |
| 89 | class MyDictCursor(self.cursor_type): |
| 90 | dict_type = MyDict |
| 91 | |
| 92 | keys = ["name", "age", "DOB"] |
| 93 | bob = MyDict([(k, self.bob[k]) for k in keys]) |
| 94 | jim = MyDict([(k, self.jim[k]) for k in keys]) |
| 95 | fred = MyDict([(k, self.fred[k]) for k in keys]) |
| 96 | |
| 97 | cur = self.conn.cursor(MyDictCursor) |
| 98 | cur.execute("SELECT * FROM dictcursor WHERE name='bob'") |
| 99 | r = cur.fetchone() |
| 100 | self.assertEqual(bob, r, "fetchone() returns MyDictCursor") |
| 101 | self._ensure_cursor_expired(cur) |
| 102 | |
| 103 | cur.execute("SELECT * FROM dictcursor") |
| 104 | r = cur.fetchall() |
| 105 | self.assertEqual([bob, jim, fred], r, "fetchall failed via MyDictCursor") |
| 106 | |
| 107 | cur.execute("SELECT * FROM dictcursor") |
| 108 | r = list(cur) |
| 109 | self.assertEqual([bob, jim, fred], r, "list failed via MyDictCursor") |
| 110 | |
| 111 | cur.execute("SELECT * FROM dictcursor") |
| 112 | r = cur.fetchmany(2) |
| 113 | self.assertEqual([bob, jim], r, "list failed via MyDictCursor") |
| 114 | self._ensure_cursor_expired(cur) |
| 115 | |
| 116 | |
| 117 | class TestSSDictCursor(TestDictCursor): |