Fixed adapters copy-on-write

If a wider context is customised don't affect smaller contexts already
created.

diff --git a/psycopg3/psycopg3/adapt.py b/psycopg3/psycopg3/adapt.py
index 6ba0f798a6ed119a97f776afcafd5fa09b56adf6..5f2773a20dfb458cdf5652cccff62cb69f938c82 100644 (file)
--- a/psycopg3/psycopg3/adapt.py
+++ b/psycopg3/psycopg3/adapt.py
@@ -154,8 +154,10 @@ class AdaptersMap(AdaptContext):
         if template:
             self._dumpers = template._dumpers.copy()
             self._own_dumpers = dict.fromkeys(Format, False)
+            template._own_dumpers = dict.fromkeys(Format, False)
             self._loaders = template._loaders[:]
             self._own_loaders = [False, False]
+            template._own_loaders = [False, False]
             self.types = TypesRegistry(template.types)
         else:
             self._dumpers = {fmt: {} for fmt in Format}

diff --git a/tests/test_adapt.py b/tests/test_adapt.py
index 7e1c05f84d8f2118b5541b3fcfcb1a35b6beb16b..69e05b0747640591c536d74c1951f54b1e3d100c 100644 (file)
--- a/tests/test_adapt.py
+++ b/tests/test_adapt.py
@@ -147,18 +147,48 @@ def test_load_cursor_ctx(conn):
     make_loader("tc").register(TEXT_OID, cur)
     make_bin_loader("bc").register(TEXT_OID, cur)
 
-    r = cur.execute("select 'hello'::text").fetchone()
-    assert r == ("hellotc",)
+    assert cur.execute("select 'hello'::text").fetchone() == ("hellotc",)
     cur.format = pq.Format.BINARY
-    r = cur.execute("select 'hello'::text").fetchone()
-    assert r == ("hellobc",)
+    assert cur.execute("select 'hello'::text").fetchone() == ("hellobc",)
 
     cur = conn.cursor()
-    r = cur.execute("select 'hello'::text").fetchone()
-    assert r == ("hellot",)
+    assert cur.execute("select 'hello'::text").fetchone() == ("hellot",)
     cur.format = pq.Format.BINARY
-    r = cur.execute("select 'hello'::text").fetchone()
-    assert r == ("hellob",)
+    assert cur.execute("select 'hello'::text").fetchone() == ("hellob",)
+
+
+def test_cow_dumpers(conn):
+    make_dumper("t").register(str, conn)
+
+    cur1 = conn.cursor()
+    cur2 = conn.cursor()
+    make_dumper("c2").register(str, cur2)
+
+    r = cur1.execute("select %s::text -- 1", ["hello"]).fetchone()
+    assert r == ("hellot",)
+    r = cur2.execute("select %s::text -- 1", ["hello"]).fetchone()
+    assert r == ("helloc2",)
+
+    make_dumper("t1").register(str, conn)
+    r = cur1.execute("select %s::text -- 2", ["hello"]).fetchone()
+    assert r == ("hellot",)
+    r = cur2.execute("select %s::text -- 2", ["hello"]).fetchone()
+    assert r == ("helloc2",)
+
+
+def test_cow_loaders(conn):
+    make_loader("t").register(TEXT_OID, conn)
+
+    cur1 = conn.cursor()
+    cur2 = conn.cursor()
+    make_loader("c2").register(TEXT_OID, cur2)
+
+    assert cur1.execute("select 'hello'::text").fetchone() == ("hellot",)
+    assert cur2.execute("select 'hello'::text").fetchone() == ("helloc2",)
+
+    make_loader("t1").register(TEXT_OID, conn)
+    assert cur1.execute("select 'hello2'::text").fetchone() == ("hello2t",)
+    assert cur2.execute("select 'hello2'::text").fetchone() == ("hello2c2",)
 
 
 @pytest.mark.parametrize(