Class TestDictCursor

pymysql/tests/test_DictCursor.py:8–114

class TestDictCursor(base.PyMySQLTestCase):
    bob = {"name": "bob", "age": 21, "DOB": datetime.datetime(1990, 2, 6, 23, 4, 56)}
    jim = {"name": "jim", "age": 56, "DOB": datetime.datetime(1955, 5, 9, 13, 12, 45)}
    fred = {"name": "fred", "age": 100, "DOB": datetime.datetime(1911, 9, 12, 1, 1, 1)}

    cursor_type = pymysql.cursors.DictCursor

    def setUp(self):
        super().setUp()
        self.conn = conn = self.connect()
        c = conn.cursor(self.cursor_type)

        # create a table and some data to query
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore")
            c.execute("drop table if exists dictcursor")
            # include in filterwarnings since for unbuffered dict cursor warning for lack of table
            # will only be propagated at start of next execute() call
            c.execute(
                """CREATE TABLE dictcursor (name char(20), age int , DOB datetime)"""
            )
        data = [
            ("bob", 21, "1990-02-06 23:04:56"),
            ("jim", 56, "1955-05-09 13:12:45"),
            ("fred", 100, "1911-09-12 01:01:01"),
        ]
        c.executemany("insert into dictcursor values (%s,%s,%s)", data)

    def tearDown(self):
        c = self.conn.cursor()
        c.execute("drop table dictcursor")
        super().tearDown()

    def _ensure_cursor_expired(self, cursor):
        pass

    def test_DictCursor(self):
        bob, jim, fred = self.bob.copy(), self.jim.copy(), self.fred.copy()
        # all assert test compare to the structure as would come out from MySQLdb
        conn = self.conn
        c = conn.cursor(self.cursor_type)

        # try an update which should return no rows
        c.execute("update dictcursor set age=20 where name='bob'")
        bob["age"] = 20
        # pull back the single row dict for bob and check
        c.execute("SELECT * from dictcursor where name='bob'")
        r = c.fetchone()
        self.assertEqual(bob, r, "fetchone via DictCursor failed")
        self._ensure_cursor_expired(c)

        # same again, but via fetchall => tuple)
        c.execute("SELECT * from dictcursor where name='bob'")
        r = c.fetchall()
        self.assertEqual(
            [bob], r, "fetch a 1 row result via fetchall failed via DictCursor"
        )
        # same test again but iterate over the