diff --git a/tests/test_warnings.py b/tests/test_warnings.py index 3e09c595..cf003de1 100644 --- a/tests/test_warnings.py +++ b/tests/test_warnings.py @@ -1,3 +1,6 @@ +import re +import subprocess +import sys import unittest import warnings @@ -54,6 +57,56 @@ class WarningsTest(unittest.TestCase): # No warning as no cursor was instantiated. self.assertEquals(cm, []) + @skip_before_python(3) + def test_broken_close(self): + script = """ +import psycopg2 + +class MyException(Exception): + pass + +class MyCurs(psycopg2.extensions.cursor): + def close(self): + raise MyException + +def f(): + conn = psycopg2.connect(%(dsn)r) + try: + conn.cursor(cursor_factory=MyCurs, scrollable=True) + finally: + conn.close() + +f() +""" % {"dsn": dsn} + p = subprocess.Popen( + [sys.executable, "-Walways", "-c", script], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + output, _ = p.communicate() + output = output.decode() + # Normalize line endings. + output = "\n".join(output.splitlines()) + self.assertRegex( + output, + re.compile( + r"^Exception ignored in: " + r"$", + re.M, + ), + ) + self.assertIn("\n__main__.MyException: \n", output) + self.assertRegex( + output, + re.compile( + r"ResourceWarning: unclosed cursor " + r" " + r"for connection " + r"$", + re.M, + ), + ) + def test_suite(): return unittest.TestLoader().loadTestsFromName(__name__)