From beea7f6c35806baf29e949f25bff78b55557067e Mon Sep 17 00:00:00 2001 From: Inada Naoki Date: Thu, 17 Jan 2019 20:54:21 +0900 Subject: [PATCH 1/2] Remove auto show warnings --- pymysql/cursors.py | 31 ------------------------------- 1 file changed, 31 deletions(-) diff --git a/pymysql/cursors.py b/pymysql/cursors.py index a6d645d4..b3a690e6 100644 --- a/pymysql/cursors.py +++ b/pymysql/cursors.py @@ -2,7 +2,6 @@ from __future__ import print_function, absolute_import from functools import partial import re -import warnings from ._compat import range_type, text_type, PY2 from . import err @@ -35,8 +34,6 @@ class Cursor(object): #: Default value of max_allowed_packet is 1048576. max_stmt_length = 1024000 - _defer_warnings = False - def __init__(self, connection): self.connection = connection self.description = None @@ -46,7 +43,6 @@ def __init__(self, connection): self._executed = None self._result = None self._rows = None - self._warnings_handled = False def close(self): """ @@ -90,9 +86,6 @@ def _nextset(self, unbuffered=False): """Get the next query set""" conn = self._get_db() current_result = self._result - # for unbuffered queries warnings are only available once whole result has been read - if unbuffered: - self._show_warnings() if current_result is None or current_result is not conn._result: return None if not current_result.has_next: @@ -347,26 +340,6 @@ def _do_get_result(self): self.description = result.description self.lastrowid = result.insert_id self._rows = result.rows - self._warnings_handled = False - - if not self._defer_warnings: - self._show_warnings() - - def _show_warnings(self): - if self._warnings_handled: - return - self._warnings_handled = True - if self._result and (self._result.has_next or not self._result.warning_count): - return - ws = self._get_db().show_warnings() - if ws is None: - return - for w in ws: - msg = w[-1] - if PY2: - if isinstance(msg, unicode): - msg = msg.encode('utf-8', 'replace') - warnings.warn(err.Warning(*w[1:3]), stacklevel=4) def __iter__(self): return iter(self.fetchone, None) @@ -427,8 +400,6 @@ class SSCursor(Cursor): possible to scroll backwards, as only the current row is held in memory. """ - _defer_warnings = True - def _conv_row(self, row): return row @@ -468,7 +439,6 @@ def fetchone(self): self._check_executed() row = self.read_next() if row is None: - self._show_warnings() return None self.rownumber += 1 return row @@ -502,7 +472,6 @@ def fetchmany(self, size=None): for i in range_type(size): row = self.read_next() if row is None: - self._show_warnings() break rows.append(row) self.rownumber += 1 From 66a6c4da817a28a5536015b190d071239972fd45 Mon Sep 17 00:00:00 2001 From: Inada Naoki Date: Thu, 17 Jan 2019 21:29:56 +0900 Subject: [PATCH 2/2] Remove tests for auto show warnings --- pymysql/tests/test_basic.py | 12 ------------ pymysql/tests/test_issues.py | 25 ------------------------- pymysql/tests/test_load_local.py | 24 ------------------------ 3 files changed, 61 deletions(-) diff --git a/pymysql/tests/test_basic.py b/pymysql/tests/test_basic.py index c2d53904..38c8cb64 100644 --- a/pymysql/tests/test_basic.py +++ b/pymysql/tests/test_basic.py @@ -2,7 +2,6 @@ import datetime import json import time -import warnings import pytest @@ -378,14 +377,3 @@ def test_issue_288(self): age = values(age)""")) cursor.execute('commit') self._verify_records(data) - - def test_warnings(self): - con = self.connect() - cur = con.cursor() - with warnings.catch_warnings(record=True) as ws: - warnings.simplefilter("always") - cur.execute("drop table if exists no_exists_table") - self.assertEqual(len(ws), 1) - self.assertEqual(ws[0].category, pymysql.Warning) - if u"no_exists_table" not in str(ws[0].message): - self.fail("'no_exists_table' not in %s" % (str(ws[0].message),)) diff --git a/pymysql/tests/test_issues.py b/pymysql/tests/test_issues.py index 05ecf286..3775f314 100644 --- a/pymysql/tests/test_issues.py +++ b/pymysql/tests/test_issues.py @@ -485,28 +485,3 @@ def test_issue_363(self): # don't assert the exact internal binary value, as it could # vary across implementations self.assertTrue(isinstance(row[0], bytes)) - - 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() diff --git a/pymysql/tests/test_load_local.py b/pymysql/tests/test_load_local.py index eafa6e19..30186e3a 100644 --- a/pymysql/tests/test_load_local.py +++ b/pymysql/tests/test_load_local.py @@ -2,7 +2,6 @@ from pymysql.tests import base import os -import warnings __all__ = ["TestLoadLocal"] @@ -64,29 +63,6 @@ def test_unbuffered_load_file(self): c = conn.cursor() c.execute("DROP TABLE test_load_local") - def test_load_warnings(self): - """Test load local infile produces the appropriate warnings""" - conn = self.connect() - 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() - if __name__ == "__main__": import unittest