[FIX] util/records: check for new duplicates

`replace_record_references_batch` tries to ensure there are no
duplicates when replacing records. It fails when the duplicate is being
introduced as a result of two ids mapping to the same target id _and_ no
original conflicting record exists.

closes #243

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

Author: aj-fuentes

src/base/tests/test_util.py

@@ -389,6 +389,62 @@ def test_adapt_domain_view(self):
         self.assertIn("courriel", view_search_2.arch)
 
 
+@unittest.skipUnless(
+    util.version_gte("13.0"), "This test is incompatible with old style odoo.addons.base.maintenance.migrations.util"
+)
+class TestReplaceReferences(UnitTestCase):
+    def setUp(self):
+        super().setUp()
+        self.env.cr.execute(
+            """
+            CREATE TABLE dummy_model(
+                             id serial PRIMARY KEY,
+                             res_id int,
+                             res_model varchar,
+                             extra varchar,
+                             CONSTRAINT uniq_constr UNIQUE(res_id, res_model, extra)
+                         );
+
+            INSERT INTO dummy_model(res_model, res_id, extra)
+                 VALUES -- the target is there with same res_id
+                        ('res.users', 1, 'x'),
+                        ('res.partner', 1, 'x'),
+
+                        -- two with same target and the target is there
+                        ('res.users', 2, 'x'),
+                        ('res.users', 3, 'x'),
+                        ('res.partner', 2, 'x'),
+
+                        -- two with same target and the target is not there
+                        ('res.users', 4, 'x'),
+                        ('res.users', 5, 'x'),
+
+                        -- target is there different res_id
+                        ('res.users', 6, 'x'),
+                        ('res.partner', 4, 'x')
+            """
+        )
+
+    def _ir_dummy(self, cr, bound_only=True):
+        yield util.IndirectReference("dummy_model", "res_model", "res_id")
+
+    def test_replace_record_references_batch__full_unique(self):
+        cr = self.env.cr
+        mapping = {1: 1, 2: 2, 3: 2, 4: 3, 5: 3, 6: 4}
+        with mock.patch("odoo.upgrade.util.records.indirect_references", self._ir_dummy):
+            util.replace_record_references_batch(cr, mapping, "res.users", "res.partner")
+
+        cr.execute("SELECT res_model, res_id, extra FROM dummy_model ORDER BY res_id, res_model")
+        data = cr.fetchall()
+        expected = [
+            ("res.partner", 1, "x"),
+            ("res.partner", 2, "x"),
+            ("res.partner", 3, "x"),
+            ("res.partner", 4, "x"),
+        ]
+        self.assertEqual(data, expected)
+
+
 class TestRemoveFieldDomains(UnitTestCase):
     @parametrize(
         [

src/util/records.py

@@ -1689,6 +1689,7 @@ def replace_record_references_batch(cr, id_mapping, model_src, model_dst=None, r
                SET {upd}
                    {jmap_expr_upd}
               FROM _upgrade_rrr
+              {{extra_join}}
              WHERE {whr}
                AND _upgrade_rrr.old = t.{res_id}
             """,
@@ -1705,19 +1706,43 @@ def replace_record_references_batch(cr, id_mapping, model_src, model_dst=None, r
         if ir.res_model_id:
             unique_indexes += _get_unique_indexes_with(cr, ir.table, ir.res_id, ir.res_model_id)
         if unique_indexes:
+            query = format_query(
+                cr,
+                query,
+                extra_join=format_query(
+                    cr,
+                    """
+         LEFT JOIN _upgrade_rrr AS _upgrade_rrr2
+                ON _upgrade_rrr.new = _upgrade_rrr2.new
+               AND _upgrade_rrr.old < _upgrade_rrr2.old
+         LEFT JOIN {table} t2
+                ON _upgrade_rrr2.old = t2.{res_id}
+                    """,
+                    table=ir.table,
+                    res_id=ir.res_id,
+                ),
+            )
             conditions = []
             for _, uniq_cols in unique_indexes:
                 uniq_cols = set(uniq_cols) - {ir.res_id, ir.res_model, ir.res_model_id}  # noqa: PLW2901
                 conditions.append(
                     format_query(
                         cr,
-                        "NOT EXISTS(SELECT 1 FROM {table} WHERE {res_model_whr} AND {jmap_expr} AND {where_clause})",
+                        """
+                         -- there is no target already present within the same constraint
+                        NOT EXISTS(SELECT 1 FROM {table} WHERE {res_model_whr} AND {jmap_expr} AND {where_clause})
+                         -- there is no other entry with the same target within the same constraint
+                        AND NOT (t2 IS NOT NULL AND {where_clause2})
+                        """,
                         table=ir.table,
                         res_model_whr=res_model_whr,
                         jmap_expr=jmap_expr,
                         where_clause=SQLStr(" AND ".join(format_query(cr, "{0}=t.{0}", col) for col in uniq_cols))
                         if uniq_cols
                         else SQLStr("True"),
+                        where_clause2=SQLStr(" AND ".join(format_query(cr, "t2.{0}=t.{0}", col) for col in uniq_cols))
+                        if uniq_cols
+                        else SQLStr("True"),
                     )
                 )
             query = format_query(
@@ -1734,7 +1759,7 @@ def replace_record_references_batch(cr, id_mapping, model_src, model_dst=None, r
             )
             cr.execute(query, locals())
         else:
-            fmt_query = cr.mogrify(query.format(**locals()), locals()).decode()
+            fmt_query = cr.mogrify(format_query(cr, query, extra_join=SQLStr("")), locals()).decode()
             parallel_execute(cr, explode_query_range(cr, fmt_query, table=ir.table, alias="t"))
 
     # reference fields