test: fix failure in test_connection_warn_close

Create all the connections beforehand otherwise some may reuse the same
address, resulting in a duplicated warning message, which would be
omitted and fail the test.

diff --git a/tests/test_connection.py b/tests/test_connection.py
index ad3a262e0dbb60a55abe9caa909d5caaf32ea70e..240bba0f96fc772e5fe48195120b203a86468a3a 100644 (file)
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -17,7 +17,7 @@ from psycopg import pq
 from psycopg.rows import tuple_row
 from psycopg.conninfo import conninfo_to_dict, timeout_from_conninfo
 
-from .acompat import is_async, skip_async, skip_sync, sleep
+from .acompat import skip_async, skip_sync, sleep
 from .fix_crdb import crdb_anydb
 from .test_adapt import make_bin_dumper, make_dumper
 from ._test_cursor import my_row_factory
@@ -182,43 +182,40 @@ def test_cursor_closed(conn):
 
 
 @pytest.mark.slow
-@pytest.mark.xfail(
-    pq.__impl__ in ("c", "binary")
-    and sys.version_info[:2] == (3, 12)
-    and (not is_async(__name__)),
-    reason="Something with Exceptions, C, Python 3.12",
-)
 def test_connection_warn_close(conn_cls, dsn, recwarn, gc_collect):
-    conn = conn_cls.connect(dsn)
-    conn.close()
-    del conn
+    # First create all the connections to test, to avoid some to reuse the same
+    # address, resulting in an omitted warning.
+    conn1 = conn_cls.connect(dsn)
+    conn2 = conn_cls.connect(dsn)
+    conn3 = conn_cls.connect(dsn)
+    conn4 = conn_cls.connect(dsn)
+
+    with conn_cls.connect(dsn) as conn5:
+        pass
+    del conn5
     assert not recwarn, [str(w.message) for w in recwarn.list]
 
-    conn = conn_cls.connect(dsn)
-    del conn
+    conn1.close()
+    del conn1
+    assert not recwarn, [str(w.message) for w in recwarn.list]
+
+    del conn2
     gc_collect()
     assert conn_cls.__name__ in str(recwarn.pop(ResourceWarning).message)
 
-    conn = conn_cls.connect(dsn)
-    conn.execute("select 1")
-    del conn
+    conn3.execute("select 1")
+    del conn3
     gc_collect()
     assert conn_cls.__name__ in str(recwarn.pop(ResourceWarning).message)
 
-    conn = conn_cls.connect(dsn)
     try:
-        conn.execute("select wat")
+        conn4.execute("select wat")
     except psycopg.ProgrammingError:
         pass
-    del conn
+    del conn4
     gc_collect()
     assert conn_cls.__name__ in str(recwarn.pop(ResourceWarning).message)
 
-    with conn_cls.connect(dsn) as conn:
-        pass
-    del conn
-    assert not recwarn, [str(w.message) for w in recwarn.list]
-
 
 @pytest.mark.usefixtures("testctx")
 def test_context_commit(conn_cls, conn, dsn):

diff --git a/tests/test_connection_async.py b/tests/test_connection_async.py
index 7eb5a27e30ab2fc892766f5ee6642b28249786ce..5583821fae2a1aae2dbe5a95e046bbf7e338bcf2 100644 (file)
--- a/tests/test_connection_async.py
+++ b/tests/test_connection_async.py
@@ -14,7 +14,7 @@ from psycopg import pq
 from psycopg.rows import tuple_row
 from psycopg.conninfo import conninfo_to_dict, timeout_from_conninfo
 
-from .acompat import asleep, is_async, skip_async, skip_sync
+from .acompat import asleep, skip_async, skip_sync
 from .fix_crdb import crdb_anydb
 from .test_adapt import make_bin_dumper, make_dumper
 from ._test_cursor import my_row_factory
@@ -177,43 +177,40 @@ async def test_cursor_closed(aconn):
 # TODO: the INERROR started failing in the C implementation in Python 3.12a7
 # compiled with Cython-3.0.0b3, not before.
 @pytest.mark.slow
-@pytest.mark.xfail(
-    pq.__impl__ in ("c", "binary")
-    and sys.version_info[:2] == (3, 12)
-    and not is_async(__name__),
-    reason="Something with Exceptions, C, Python 3.12",
-)
 async def test_connection_warn_close(aconn_cls, dsn, recwarn, gc_collect):
-    conn = await aconn_cls.connect(dsn)
-    await conn.close()
-    del conn
+    # First create all the connections to test, to avoid some to reuse the same
+    # address, resulting in an omitted warning.
+    conn1 = await aconn_cls.connect(dsn)
+    conn2 = await aconn_cls.connect(dsn)
+    conn3 = await aconn_cls.connect(dsn)
+    conn4 = await aconn_cls.connect(dsn)
+
+    async with await aconn_cls.connect(dsn) as conn5:
+        pass
+    del conn5
     assert not recwarn, [str(w.message) for w in recwarn.list]
 
-    conn = await aconn_cls.connect(dsn)
-    del conn
+    await conn1.close()
+    del conn1
+    assert not recwarn, [str(w.message) for w in recwarn.list]
+
+    del conn2
     gc_collect()
     assert aconn_cls.__name__ in str(recwarn.pop(ResourceWarning).message)
 
-    conn = await aconn_cls.connect(dsn)
-    await conn.execute("select 1")
-    del conn
+    await conn3.execute("select 1")
+    del conn3
     gc_collect()
     assert aconn_cls.__name__ in str(recwarn.pop(ResourceWarning).message)
 
-    conn = await aconn_cls.connect(dsn)
     try:
-        await conn.execute("select wat")
+        await conn4.execute("select wat")
     except psycopg.ProgrammingError:
         pass
-    del conn
+    del conn4
     gc_collect()
     assert aconn_cls.__name__ in str(recwarn.pop(ResourceWarning).message)
 
-    async with await aconn_cls.connect(dsn) as conn:
-        pass
-    del conn
-    assert not recwarn, [str(w.message) for w in recwarn.list]
-
 
 @pytest.mark.usefixtures("testctx")
 async def test_context_commit(aconn_cls, aconn, dsn):