[FIX] base: properly execute Selection field ondelete actions

With this commit, 2 changes are made to the way that the
fields.Selection.ondelete cleanup action works:

1) We manually delete ir.model.fields.selection **before** the deletion
of ir.model.fields purposefully to **avoid** SQL CASCADE deletes, as we
do not know which selections are deleted in cascade and thus we cannot
perform the corresponding ondelete cleanup action (which is implemented
within the ORM).

2) In some actions, namely 'set default' and 'set null', we write a
"safe" value to the records containing the Selection being deleted,
before this commit this would go through the ORM (records.write()) but
this is problematic if there's a write override for the record's model
that raises an error for the field being written to. With this commit,
we first try to go through the ORM but if there's a failure (because of
the raise in a write override) then we will bypass the ORM and set it
with SQL.

opw-2451126

closes odoo/odoo#67616

X-original-commit: f5c7e861ee3ca0cdeb6c2f06d227ada07f923ed0
Signed-off-by: Adrian Torres (adt) <adt@odoo.com>
This commit is contained in:
Adrian Torres
2021-03-11 06:45:42 +00:00
parent 75243d5a43
commit ab7b5af3e8
4 changed files with 65 additions and 8 deletions
+30 -8
View File
@@ -1370,6 +1370,29 @@ class IrModelSelection(models.Model):
def _process_ondelete(self):
""" Process the 'ondelete' of the given selection values. """
def safe_write(records, fname, value):
if not records:
return
try:
with self.env.cr.savepoint():
records.write({fname: value})
except Exception:
# going through the ORM failed, probably because of an exception
# in an override or possibly a constraint.
_logger.warning(
"Could not fulfill ondelete action for field %s.%s, "
"attempting ORM bypass...", records._name, fname,
)
query = sql.SQL("UPDATE {} SET {}=%s WHERE id IN %s").format(
sql.Identifier(records._table),
sql.Identifier(fname),
)
# if this fails then we're shit out of luck and there's nothing
# we can do except fix on a case-by-case basis
value = field.convert_to_column(value, records)
self.env.cr.execute(query, [value, records._ids])
records.invalidate_cache([fname])
for selection in self:
Model = self.env[selection.field_id.model]
# The field may exist in database but not in registry. In this case
@@ -1392,10 +1415,10 @@ class IrModelSelection(models.Model):
elif callable(ondelete):
ondelete(selection._get_records())
elif ondelete == 'set null':
selection._get_records().write({field.name: False})
safe_write(selection._get_records(), field.name, False)
elif ondelete == 'set default':
value = field.convert_to_write(field.default(Model), Model)
selection._get_records().write({field.name: value})
safe_write(selection._get_records(), field.name, value)
elif ondelete == 'cascade':
selection._get_records().unlink()
else:
@@ -2152,13 +2175,12 @@ class IrModelData(models.Model):
constraints = self.env['ir.model.constraint'].search([('module', 'in', modules.ids)])
constraints._module_data_uninstall()
# Remove fields, selections and relations. Note that the selections of
# removed fields do not require any "data fix", as their corresponding
# column no longer exists. We can therefore completely ignore them. That
# is why selections are removed after fields: most selections are
# deleted on cascade by their corresponding field.
delete(self.env['ir.model.fields'].browse(unique(field_ids)))
# If we delete a selection field, and some of its values have ondelete='cascade',
# we expect the records with that value to be deleted. If we delete the field first,
# the column is dropped and the selection is gone, and thus the records above will not
# be deleted.
delete(self.env['ir.model.fields.selection'].browse(unique(selection_ids)).exists())
delete(self.env['ir.model.fields'].browse(unique(field_ids)))
relations = self.env['ir.model.relation'].search([('module', 'in', modules.ids)])
relations._module_data_uninstall()
@@ -1152,6 +1152,29 @@ class SelectionNonStored(models.Model):
], store=False)
class SelectionRequiredForWriteOverride(models.Model):
_name = 'test_new_api.model_selection_required_for_write_override'
_description = "Model with required selection field for an extension with write override"
my_selection = fields.Selection([
('foo', "Foo"),
('bar', "Bar"),
], required=True, default='foo')
class SelectionRequiredWithWriteOverride(models.Model):
_inherit = 'test_new_api.model_selection_required_for_write_override'
my_selection = fields.Selection(selection_add=[
('divinity', "Divinity: Original Sin 2"),
], ondelete={'divinity': 'set default'})
def write(self, vals):
if 'my_selection' in vals:
raise ValueError("No... no no no")
return super().write(vals)
# Special classes to ensure the correct usage of a shared cache amongst users.
# See the method test_shared_cache_computed_field
class SharedCacheComputeParent(models.Model):
@@ -61,6 +61,7 @@ access_test_new_api_monetary_order_line,access_test_new_api_monetary_order_line,
access_test_new_api_model_selection_base,access_test_new_api_model_selection_base,model_test_new_api_model_selection_base,,1,1,1,1
access_test_new_api_model_selection_required,access_test_new_api_model_selection_required,model_test_new_api_model_selection_required,,1,1,1,1
access_test_new_api_model_selection_non_stored,access_test_new_api_model_selection_non_stored,model_test_new_api_model_selection_non_stored,,1,1,1,1
access_test_new_api_model_selection_required_for_write_override,access_test_new_api_model_selection_required_for_write_override,model_test_new_api_model_selection_required_for_write_override,,1,1,1,1
access_test_new_api_move,access_test_new_api_move,model_test_new_api_move,,1,1,1,1
access_test_new_api_move_line,access_test_new_api_move_line,model_test_new_api_move_line,,1,1,1,1
access_test_new_api_model_shared_cache_compute_parent,access_test_new_api.model_shared_cache_compute_parent,model_test_new_api_model_shared_cache_compute_parent,,1,1,1,1
1 id name model_id:id group_id:id perm_read perm_write perm_create perm_unlink
61 access_test_new_api_model_selection_base access_test_new_api_model_selection_base model_test_new_api_model_selection_base 1 1 1 1
62 access_test_new_api_model_selection_required access_test_new_api_model_selection_required model_test_new_api_model_selection_required 1 1 1 1
63 access_test_new_api_model_selection_non_stored access_test_new_api_model_selection_non_stored model_test_new_api_model_selection_non_stored 1 1 1 1
64 access_test_new_api_model_selection_required_for_write_override access_test_new_api_model_selection_required_for_write_override model_test_new_api_model_selection_required_for_write_override 1 1 1 1
65 access_test_new_api_move access_test_new_api_move model_test_new_api_move 1 1 1 1
66 access_test_new_api_move_line access_test_new_api_move_line model_test_new_api_move_line 1 1 1 1
67 access_test_new_api_model_shared_cache_compute_parent access_test_new_api.model_shared_cache_compute_parent model_test_new_api_model_shared_cache_compute_parent 1 1 1 1
@@ -2928,6 +2928,7 @@ class TestSelectionOndelete(common.TransactionCase):
MODEL_BASE = 'test_new_api.model_selection_base'
MODEL_REQUIRED = 'test_new_api.model_selection_required'
MODEL_NONSTORED = 'test_new_api.model_selection_non_stored'
MODEL_WRITE_OVERRIDE = 'test_new_api.model_selection_required_for_write_override'
def setUp(self):
super().setUp()
@@ -3066,6 +3067,16 @@ class TestSelectionOndelete(common.TransactionCase):
self._unlink_option(self.MODEL_REQUIRED, 'foo')
self.assertEqual(rec.my_selection, 'foo')
@mute_logger('odoo.addons.base.models.ir_model')
def test_write_override_selection(self):
# test that on override to write that raises an error does not prevent the ondelete
# policy from executing and cleaning up what needs to be cleaned up
rec = self.env[self.MODEL_WRITE_OVERRIDE].create({'my_selection': 'divinity'})
self.assertEqual(rec.my_selection, 'divinity')
self._unlink_option(self.MODEL_WRITE_OVERRIDE, 'divinity')
self.assertEqual(rec.my_selection, 'foo')
@common.tagged('selection_ondelete_advanced')
class TestSelectionOndeleteAdvanced(common.TransactionCase):