File Manager

X7ROOT File Manager

Current Path: /lib/python2.7/site-packages/pymysql/tests

lib / python2.7 / site-packages / pymysql / tests /

📁 ..
📄 __init__.py(600 B)
📄 __init__.pyc(773 B)
📄 __init__.pyo(773 B)
📄 base.py(2.64 KB)
📄 base.pyc(3.64 KB)
📄 base.pyo(3.64 KB)
📄 test_DictCursor.py(4.51 KB)
📄 test_DictCursor.pyc(5.21 KB)
📄 test_DictCursor.pyo(5.21 KB)
📄 test_SSCursor.py(3.83 KB)
📄 test_SSCursor.pyc(3.29 KB)
📄 test_SSCursor.pyo(3.29 KB)
📄 test_basic.py(14.54 KB)
📄 test_basic.pyc(14.09 KB)
📄 test_basic.pyo(14.09 KB)
📄 test_connection.py(23.71 KB)
📄 test_connection.pyc(24.2 KB)
📄 test_connection.pyo(24.16 KB)
📄 test_converters.py(2.14 KB)
📄 test_converters.pyc(3.51 KB)
📄 test_converters.pyo(3.51 KB)
📄 test_cursor.py(4.09 KB)
📄 test_cursor.pyc(4.25 KB)
📄 test_cursor.pyo(4.21 KB)
📄 test_err.py(670 B)
📄 test_err.pyc(1.29 KB)
📄 test_err.pyo(1.29 KB)
📄 test_issues.py(19.8 KB)
📄 test_issues.pyc(19.88 KB)
📄 test_issues.pyo(19.82 KB)
📄 test_load_local.py(3.5 KB)
📄 test_load_local.pyc(3.93 KB)
📄 test_load_local.pyo(3.93 KB)
📄 test_nextset.py(1.81 KB)
📄 test_nextset.pyc(2.88 KB)
📄 test_nextset.pyo(2.88 KB)
📄 test_optionfile.py(737 B)
📄 test_optionfile.pyc(1.25 KB)
📄 test_optionfile.pyo(1.25 KB)
📁 thirdparty

Editing: test_cursor.py

import warnings

from pymysql.tests import base
import pymysql.cursors

class CursorTest(base.PyMySQLTestCase):
    def setUp(self):
        super(CursorTest, self).setUp()

conn = self.connections[0]
        self.safe_create_table(
            conn,
            "test", "create table test (data varchar(10))",
        )
        cursor = conn.cursor()
        cursor.execute(
            "insert into test (data) values "
            "('row1'), ('row2'), ('row3'), ('row4'), ('row5')")
        cursor.close()
        self.test_connection = pymysql.connect(**self.databases[0])
        self.addCleanup(self.test_connection.close)

def test_cleanup_rows_unbuffered(self):
        conn = self.test_connection
        cursor = conn.cursor(pymysql.cursors.SSCursor)

cursor.execute("select * from test as t1, test as t2")
        for counter, row in enumerate(cursor):
            if counter > 10:
                break

del cursor
        self.safe_gc_collect()

c2 = conn.cursor()

c2.execute("select 1")
        self.assertEqual(c2.fetchone(), (1,))
        self.assertIsNone(c2.fetchone())

def test_cleanup_rows_buffered(self):
        conn = self.test_connection
        cursor = conn.cursor(pymysql.cursors.Cursor)

cursor.execute("select * from test as t1, test as t2")
        for counter, row in enumerate(cursor):
            if counter > 10:
                break

del cursor
        self.safe_gc_collect()

c2 = conn.cursor()

c2.execute("select 1")

self.assertEqual(
            c2.fetchone(), (1,)
        )
        self.assertIsNone(c2.fetchone())

def test_executemany(self):
        conn = self.test_connection
        cursor = conn.cursor(pymysql.cursors.Cursor)

m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%s, %s)")
        self.assertIsNotNone(m, 'error parse %s')
        self.assertEqual(m.group(3), '', 'group 3 not blank, bug in RE_INSERT_VALUES?')

m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%(id)s, %(name)s)")
        self.assertIsNotNone(m, 'error parse %(name)s')
        self.assertEqual(m.group(3), '', 'group 3 not blank, bug in RE_INSERT_VALUES?')

m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%(id_name)s, %(name)s)")
        self.assertIsNotNone(m, 'error parse %(id_name)s')
        self.assertEqual(m.group(3), '', 'group 3 not blank, bug in RE_INSERT_VALUES?')

m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO TEST (ID, NAME) VALUES (%(id_name)s, %(name)s) ON duplicate update")
        self.assertIsNotNone(m, 'error parse %(id_name)s')
        self.assertEqual(m.group(3), ' ON duplicate update', 'group 3 not ON duplicate update, bug in RE_INSERT_VALUES?')

# https://github.com/PyMySQL/PyMySQL/pull/597
        m = pymysql.cursors.RE_INSERT_VALUES.match("INSERT INTO bloup(foo, bar)VALUES(%s, %s)")
        assert m is not None

# cursor._executed must bee "insert into test (data) values (0),(1),(2),(3),(4),(5),(6),(7),(8),(9)"
        # list args
        data = range(10)
        cursor.executemany("insert into test (data) values (%s)", data)
        self.assertTrue(cursor._executed.endswith(b",(7),(8),(9)"), 'execute many with %s not in one query')

# dict args
        data_dict = [{'data': i} for i in range(10)]
        cursor.executemany("insert into test (data) values (%(data)s)", data_dict)
        self.assertTrue(cursor._executed.endswith(b",(7),(8),(9)"), 'execute many with %(data)s not in one query')

# %% in column set
        cursor.execute("""\
            CREATE TABLE percent_test (
                `A%` INTEGER,
                `B%` INTEGER)""")
        try:
            q = "INSERT INTO percent_test (`A%%`, `B%%`) VALUES (%s, %s)"
            self.assertIsNotNone(pymysql.cursors.RE_INSERT_VALUES.match(q))
            cursor.executemany(q, [(3, 4), (5, 6)])
            self.assertTrue(cursor._executed.endswith(b"(3, 4),(5, 6)"), "executemany with %% not in one query")
        finally:
            cursor.execute("DROP TABLE IF EXISTS percent_test")

X7ROOT File Manager

Editing: test_cursor.py

Upload File

Create Folder