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_SSCursor.py

import sys

try:
    from pymysql.tests import base
    import pymysql.cursors
except Exception:
    # For local testing from top-level directory, without installing
    sys.path.append('../pymysql')
    from pymysql.tests import base
    import pymysql.cursors

class TestSSCursor(base.PyMySQLTestCase):
    def test_SSCursor(self):
        affected_rows = 18446744073709551615

conn = self.connections[0]
        data = [
            ('America', '', 'America/Jamaica'),
            ('America', '', 'America/Los_Angeles'),
            ('America', '', 'America/Lima'),
            ('America', '', 'America/New_York'),
            ('America', '', 'America/Menominee'),
            ('America', '', 'America/Havana'),
            ('America', '', 'America/El_Salvador'),
            ('America', '', 'America/Costa_Rica'),
            ('America', '', 'America/Denver'),
            ('America', '', 'America/Detroit'),]

try:
            cursor = conn.cursor(pymysql.cursors.SSCursor)

# Create table
            cursor.execute(('CREATE TABLE tz_data ('
                'region VARCHAR(64),'
                'zone VARCHAR(64),'
                'name VARCHAR(64))'))

conn.begin()
            # Test INSERT
            for i in data:
                cursor.execute('INSERT INTO tz_data VALUES (%s, %s, %s)', i)
                self.assertEqual(conn.affected_rows(), 1, 'affected_rows does not match')
            conn.commit()

# Test fetchone()
            iter = 0
            cursor.execute('SELECT * FROM tz_data')
            while True:
                row = cursor.fetchone()
                if row is None:
                    break
                iter += 1

# Test cursor.rowcount
                self.assertEqual(cursor.rowcount, affected_rows,
                    'cursor.rowcount != %s' % (str(affected_rows)))

# Test cursor.rownumber
                self.assertEqual(cursor.rownumber, iter,
                    'cursor.rowcount != %s' % (str(iter)))

# Test row came out the same as it went in
                self.assertEqual((row in data), True,
                    'Row not found in source data')

# Test fetchall
            cursor.execute('SELECT * FROM tz_data')
            self.assertEqual(len(cursor.fetchall()), len(data),
                'fetchall failed. Number of rows does not match')

# Test fetchmany
            cursor.execute('SELECT * FROM tz_data')
            self.assertEqual(len(cursor.fetchmany(2)), 2,
                'fetchmany failed. Number of rows does not match')

# So MySQLdb won't throw "Commands out of sync"
            while True:
                res = cursor.fetchone()
                if res is None:
                    break

# Test update, affected_rows()
            cursor.execute('UPDATE tz_data SET zone = %s', ['Foo'])
            conn.commit()
            self.assertEqual(cursor.rowcount, len(data),
                'Update failed. affected_rows != %s' % (str(len(data))))

# Test executemany
            cursor.executemany('INSERT INTO tz_data VALUES (%s, %s, %s)', data)
            self.assertEqual(cursor.rowcount, len(data),
                'executemany failed. cursor.rowcount != %s' % (str(len(data))))

# Test multiple datasets
            cursor.execute('SELECT 1; SELECT 2; SELECT 3')
            self.assertListEqual(list(cursor), [(1, )])
            self.assertTrue(cursor.nextset())
            self.assertListEqual(list(cursor), [(2, )])
            self.assertTrue(cursor.nextset())
            self.assertListEqual(list(cursor), [(3, )])
            self.assertFalse(cursor.nextset())

finally:
            cursor.execute('DROP TABLE tz_data')
            cursor.close()

__all__ = ["TestSSCursor"]

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

X7ROOT File Manager

Editing: test_SSCursor.py

Upload File

Create Folder