OXIESEC PANEL

Name	Size	Modified	Perms
📁 ..	-	03/16/2023 12:45:13 PM	rwxr-xr-x
📄 __init__.py	600 bytes	09/01/2016 11:09:00 AM	rw-r--r--
📄 __init__.pyc	773 bytes	04/21/2022 01:39:17 PM	rw-r--r--
📄 __init__.pyo	773 bytes	04/21/2022 01:39:17 PM	rw-r--r--
📄 base.py	2.64 KB	05/18/2016 09:03:00 AM	rw-r--r--
📄 base.pyc	3.64 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 base.pyo	3.64 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_DictCursor.py	4.51 KB	08/29/2016 08:21:29 AM	rw-r--r--
📄 test_DictCursor.pyc	5.21 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_DictCursor.pyo	5.21 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_SSCursor.py	3.83 KB	05/18/2016 09:03:00 AM	rw-r--r--
📄 test_SSCursor.pyc	3.29 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_SSCursor.pyo	3.29 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_basic.py	14.54 KB	12/20/2017 09:29:49 AM	rw-r--r--
📄 test_basic.pyc	14.09 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_basic.pyo	14.09 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_connection.py	23.71 KB	05/18/2016 09:03:00 AM	rw-r--r--
📄 test_connection.pyc	24.2 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_connection.pyo	24.16 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_converters.py	2.14 KB	08/29/2016 08:21:29 AM	rw-r--r--
📄 test_converters.pyc	3.51 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_converters.pyo	3.51 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_cursor.py	4.09 KB	12/20/2017 09:29:49 AM	rw-r--r--
📄 test_cursor.pyc	4.25 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_cursor.pyo	4.21 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_err.py	670 bytes	09/01/2016 11:09:00 AM	rw-r--r--
📄 test_err.pyc	1.29 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_err.pyo	1.29 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_issues.py	19.8 KB	12/20/2017 12:16:54 PM	rw-r--r--
📄 test_issues.pyc	19.88 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_issues.pyo	19.82 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_load_local.py	3.5 KB	09/01/2016 11:09:00 AM	rw-r--r--
📄 test_load_local.pyc	3.93 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_load_local.pyo	3.93 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_nextset.py	1.81 KB	06/29/2017 07:34:50 AM	rw-r--r--
📄 test_nextset.pyc	2.88 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_nextset.pyo	2.88 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_optionfile.py	737 bytes	08/30/2017 10:38:31 AM	rw-r--r--
📄 test_optionfile.pyc	1.25 KB	04/21/2022 01:39:17 PM	rw-r--r--
📄 test_optionfile.pyo	1.25 KB	04/21/2022 01:39:17 PM	rw-r--r--
📁 thirdparty	-	03/16/2023 12:45:13 PM	rwxr-xr-x

Editing: test_DictCursor.py

from pymysql.tests import base
import pymysql.cursors

import datetime
import warnings

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(TestDictCursor, self).setUp()
        self.conn = conn = self.connections[0]
        c = conn.cursor(self.cursor_type)

# create a table ane 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(TestDictCursor, self).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
        c.execute("SELECT * from dictcursor where name='bob'")
        for r in c:
            self.assertEqual(bob, r, "fetch a 1 row result via iteration failed via DictCursor")
        # get all 3 row via fetchall
        c.execute("SELECT * from dictcursor")
        r = c.fetchall()
        self.assertEqual([bob,jim,fred], r, "fetchall failed via DictCursor")
        #same test again but do a list comprehension
        c.execute("SELECT * from dictcursor")
        r = list(c)
        self.assertEqual([bob,jim,fred], r, "DictCursor should be iterable")
        # get all 2 row via fetchmany
        c.execute("SELECT * from dictcursor")
        r = c.fetchmany(2)
        self.assertEqual([bob, jim], r, "fetchmany failed via DictCursor")
        self._ensure_cursor_expired(c)

def test_custom_dict(self):
        class MyDict(dict): pass

class MyDictCursor(self.cursor_type):
            dict_type = MyDict

keys = ['name', 'age', 'DOB']
        bob = MyDict([(k, self.bob[k]) for k in keys])
        jim = MyDict([(k, self.jim[k]) for k in keys])
        fred = MyDict([(k, self.fred[k]) for k in keys])

cur = self.conn.cursor(MyDictCursor)
        cur.execute("SELECT * FROM dictcursor WHERE name='bob'")
        r = cur.fetchone()
        self.assertEqual(bob, r, "fetchone() returns MyDictCursor")
        self._ensure_cursor_expired(cur)

cur.execute("SELECT * FROM dictcursor")
        r = cur.fetchall()
        self.assertEqual([bob, jim, fred], r,
                         "fetchall failed via MyDictCursor")

cur.execute("SELECT * FROM dictcursor")
        r = list(cur)
        self.assertEqual([bob, jim, fred], r,
                         "list failed via MyDictCursor")

cur.execute("SELECT * FROM dictcursor")
        r = cur.fetchmany(2)
        self.assertEqual([bob, jim], r,
                         "list failed via MyDictCursor")
        self._ensure_cursor_expired(cur)

class TestSSDictCursor(TestDictCursor):
    cursor_type = pymysql.cursors.SSDictCursor

def _ensure_cursor_expired(self, cursor):
        list(cursor.fetchall_unbuffered())

if __name__ == "__main__":
    import unittest
    unittest.main()