Python pymysql 模块，Warning() 实例源码

我们从Python开源项目中，提取了以下29个代码示例，用于说明如何使用pymysql.Warning()。

项目：watchmen 作者：lycclsltt | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：watchmen 作者：lycclsltt | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：touch-pay-client 作者：HackPucBemobi | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：touch-pay-client 作者：HackPucBemobi | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：bawk 作者：jttwnsnd | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：bawk 作者：jttwnsnd | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：Flask-NvRay-Blog 作者：rui7157 | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                self.assertTrue("Incorrect integer value" in str(w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：Flask-NvRay-Blog 作者：rui7157 | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                self.assertTrue("Incorrect integer value" in str(w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：rekall-agent-server 作者：rekall-innovations | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：rekall-agent-server 作者：rekall-innovations | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：deb-python-pymysql 作者：openstack | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：deb-python-pymysql 作者：openstack | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：ServerlessCrawler-VancouverRealState 作者：MarcelloLins | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：ServerlessCrawler-VancouverRealState 作者：MarcelloLins | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：QualquerMerdaAPI 作者：tiagovizoto | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：QualquerMerdaAPI 作者：tiagovizoto | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：metrics 作者：Jeremy-Friedman | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                self.assertTrue("Incorrect integer value" in str(w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：metrics 作者：Jeremy-Friedman | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                self.assertTrue("Incorrect integer value" in str(w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：gardenbot 作者：GoestaO | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：gardenbot 作者：GoestaO | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：prophessor 作者：paulhauner | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                self.assertTrue("Incorrect integer value" in str(w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：blog_flask 作者：momantai | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：blog_flask 作者：momantai | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：flask 作者：bobohope | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：flask 作者：bobohope | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：power-pi 作者：Knapsacks | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：power-pi 作者：Knapsacks | 项目源码 | 文件源码

def test_issue_491(self):
        """ Test warning propagation """
        conn = pymysql.connect(charset="utf8", **self.databases[0])

        with warnings.catch_warnings():
            # Ignore all warnings other than pymysql generated ones
            warnings.simplefilter("ignore")
            warnings.simplefilter("error", category=pymysql.Warning)

            # verify for both buffered and unbuffered cursor types
            for cursor_class in (cursors.Cursor, cursors.SSCursor):
                c = conn.cursor(cursor_class)
                try:
                    c.execute("SELECT CAST('124b' AS SIGNED)")
                    c.fetchall()
                except pymysql.Warning as e:
                    # Warnings should have errorcode and string message, just like exceptions
                    self.assertEqual(len(e.args), 2)
                    self.assertEqual(e.args[0], 1292)
                    self.assertTrue(isinstance(e.args[1], text_type))
                else:
                    self.fail("Should raise Warning")
                finally:
                    c.close()

项目：power-pi 作者：Knapsacks | 项目源码 | 文件源码

def test_load_warnings(self):
        """Test load local infile produces the appropriate warnings"""
        conn = self.connections[0]
        c = conn.cursor()
        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'data',
                                'load_local_warn_data.txt')
        try:
            with warnings.catch_warnings(record=True) as w:
                warnings.simplefilter('always')
                c.execute(
                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
                )
                self.assertEqual(w[0].category, Warning)
                expected_message = "Incorrect integer value"
                if expected_message not in str(w[-1].message):
                    self.fail("%r not in %r" % (expected_message, w[-1].message))
        finally:
            c.execute("DROP TABLE test_load_local")
            c.close()

项目：graphscale 作者：schrockn | 项目源码 | 文件源码

def disable_pymysql_warnings() -> Iterator:
    filterwarnings('ignore', category=pymysql.Warning)  # type: ignore
    yield
    resetwarnings()