Add test for ignored exception when .close() fails

This commit is contained in:
Jon Dufresne 2020-02-11 17:55:37 -08:00
parent d5110e5191
commit e5322a71bb

View File

@ -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"<cursor object at 0x[0-9a-fA-F]+; closed: 0>$",
re.M,
),
)
self.assertIn("\n__main__.MyException: \n", output)
self.assertRegex(
output,
re.compile(
r"ResourceWarning: unclosed cursor "
r"<cursor object at 0x[0-9a-fA-F]+; closed: 0> "
r"for connection "
r"<connection object at 0x[0-9a-fA-F]+; dsn: '.*', closed: 0>$",
re.M,
),
)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)