[IMP] lift_constraint: allow to cascade dropping constraints

openupgradelib/openupgrade.py

@@ -2802,11 +2802,13 @@ def is_module_installed(cr, module):
     return bool(cr.fetchone())
 
 
-def lift_constraints(cr, table, column):
+def lift_constraints(cr, table, column, cascade=False):
     """Lift all constraints on column in table.
     Typically, you use this in a pre-migrate script where you adapt references
     for many2one fields with changed target objects.
-    If everything went right, the constraints will be recreated"""
+    Set cascade=True if other constraints depend on the one you want to delete,
+    which ie is the case for primary keys.
+    If everything went right, the constraints will be recreated."""
     cr.execute(
         "select relname, array_agg(conname) from "
         "(select t1.relname, c.conname "
@@ -2830,8 +2832,16 @@ def lift_constraints(cr, table, column):
     )
     for table, constraints in cr.fetchall():
         cr.execute(
-            "alter table %s drop constraint %s",
-            (AsIs(table), AsIs(", drop constraint ".join(constraints))),
+            "alter table %s drop constraint if exists %s",
+            (
+                AsIs(table),
+                AsIs(
+                    ", drop constraint if exists ".join(
+                        (constraint if not cascade else (constraint + " cascade"))
+                        for constraint in constraints
+                    )
+                ),
+            ),
         )
 
 

tests/test_openupgradelib.py

@@ -1,6 +1,8 @@
 import os
 import unittest
 
+import psycopg2
+
 import odoo
 
 # needs to be imported after Odoo initialized, done in test setup
@@ -92,6 +94,33 @@ def test_rename_field_references(self):
         openupgrade.openupgrade_tools.invalidate_cache(self.env, flush=True)
         self.assertEqual(test_filter.domain, "[('renamed_name', '=', 'test')]")
 
+    def test_lift_constraints(self):
+        self.env.cr.execute("SAVEPOINT test")
+        with self.assertRaises(psycopg2.errors.DependentObjectsStillExist):
+            openupgrade.lift_constraints(
+                self.env.cr,
+                "res_partner",
+                "id",
+            )
+        self.env.cr.execute("ROLLBACK TO SAVEPOINT test")
+        self.env.cr.execute("SAVEPOINT test")
+        admin_partner = self.env.ref("base.user_admin").partner_id
+        with self.assertRaises(psycopg2.errors.ForeignKeyViolation):
+            self.env.cr.execute(
+                "DELETE FROM res_partner WHERE id=%s", (admin_partner.id,)
+            )
+        self.env.cr.execute("ROLLBACK TO SAVEPOINT test")
+        self.env.cr.execute("SAVEPOINT test")
+        openupgrade.lift_constraints(
+            self.env.cr,
+            "res_partner",
+            "id",
+            cascade=True,
+        )
+        self.env.cr.execute("DELETE FROM res_partner WHERE id=%s", (admin_partner.id,))
+        self.assertFalse(admin_partner.exists())
+        self.env.cr.execute("ROLLBACK TO SAVEPOINT test")
+
     def tearDown(self):
         super().tearDown()
         self.cr.close()