[FIX] util.replace_record_references_batch: handle unique constraints

We already do it for indirect references (res_model/res_id), why not
for the direct ones?

closes #46

Signed-off-by: Christophe Simonis (chs) <[email protected]>

Author: KangOl

src/base/tests/test_util.py

@@ -937,6 +937,19 @@ def test_replace_in_all_jsonb_values(self):
         # also ensure the replace works for multiple embedded quotes
         self.assertEqual(test_partner_title.name, "object.name GONE object.numbercombined")
 
+    def test_replace_record_references_batch__uniqueness(self):
+        c1 = self.env["res.country"].create(
+            {"name": "TEST1", "code": "T1", "state_ids": [(0, 0, {"name": "STATE1", "code": "STATE"})]}
+        )
+        c2 = self.env["res.country"].create(
+            {"name": "TEST2", "code": "T2", "state_ids": [(0, 0, {"name": "STATE2", "code": "STATE"})]}
+        )
+
+        util.replace_record_references_batch(self.env.cr, {c2.id: c1.id}, "res.country")
+        self.env.cr.execute("SELECT count(1) FROM res_country_state WHERE country_id=%s", [c1.id])
+        [count] = self.env.cr.fetchone()
+        self.assertEqual(count, 1)
+
 
 class TestMisc(UnitTestCase):
     @parametrize(

src/util/records.py

@@ -1061,6 +1061,19 @@ def replace_record_references_batch(cr, id_mapping, model_src, model_dst=None, r
                   FROM _upgrade_rrr r
                  WHERE r.old = t.{fk}
             """
+            unique_indexes = _get_unique_indexes_with(cr, table, fk)
+            if unique_indexes:
+                conditions = [""]
+                for _, uniq_cols in unique_indexes:
+                    uniq_cols = set(uniq_cols) - {fk}  # noqa: PLW2901
+                    ands = (
+                        " AND ".join(format_query(cr, "u.{col} = t.{col}", col=col) for col in uniq_cols)
+                        if uniq_cols
+                        else "true"
+                    )
+                    conditions.append("NOT EXISTS(SELECT 1 FROM {table} u WHERE u.{fk} = r.new AND %s)" % ands)
+
+                query += " AND ".join(conditions)
 
             if not column_exists(cr, table, "id"):
                 # seems to be a m2m table. Avoid duplicated entries
@@ -1091,22 +1104,15 @@ def replace_record_references_batch(cr, id_mapping, model_src, model_dst=None, r
                         col2=col2,
                     )
                 )
-                query += """
-                    AND NOT EXISTS(SELECT 1 FROM {table} e WHERE e.{col2} = t.{col2} AND e.{fk} = r.new);
-
-                    DELETE
-                      FROM {table} t
-                     USING _upgrade_rrr r
-                     WHERE t.{fk} = r.old;
-                """
+                query += " AND NOT EXISTS(SELECT 1 FROM {table} e WHERE e.{col2} = t.{col2} AND e.{fk} = r.new)"
 
                 col2_info = target_of(cr, table, col2)  # col2 may not be a FK
                 if col2_info and col2_info[:2] == (model_src_table, "id"):
                     # a m2m on itself, remove the self referencing entries
                     # It only handle 1-level recursions. For multi-level recursions, it should be handled manually.
                     # We can't decide which link to break.
                     # XXX: add a warning?
-                    query += """
+                    query += """;
                         DELETE
                           FROM {table} t
                          USING _upgrade_rrr r
@@ -1117,13 +1123,21 @@ def replace_record_references_batch(cr, id_mapping, model_src, model_dst=None, r
                 cr.execute(format_query(cr, query, table=table, fk=fk, col2=col2))
 
             else:  # it's a model
-                fmt_query = cr.mogrify(format_query(cr, query, table=table, fk=fk)).decode()
+                fmt_query = format_query(cr, query, table=table, fk=fk)
                 parallel_execute(cr, explode_query_range(cr, fmt_query, table=table, alias="t"))
 
                 # track default values to update
                 model = model_of_table(cr, table)
                 fk_def.append((model, fk))
 
+            delete_query = """
+                DELETE
+                  FROM {table} t
+                 USING _upgrade_rrr r
+                 WHERE t.{fk} = r.old
+            """
+            cr.execute(format_query(cr, delete_query, table=table, fk=fk))
+
         if fk_def:
             if table_exists(cr, "ir_values"):
                 column_read, cast_write = _ir_values_value(cr, prefix="v")