[IMP] core: introduce mechanism for selection_add cleanup

- Let A and B be two different modules.
- A defines a required Selection field F of model M and B extends
  it through the `selection_add` argument.
- Create records of model M and have some of them have any of the
  options introduced by module B selected for field F.
- Uninstall module B.

The result will be records of Model M with an option for field F that no
longer exists, this makes the registry inconsistent and prone to
crashing (it is sufficient to access the form view of such a record to
trigger a crash).

This commit introduces a mechanism similar to the `ondelete` argument
found in Many2one fields, the argument name is the same but it's
different both in behaviour and in implementation.

The `ondelete` mechanism for Selection fields is enforced for **any**
Selection field with required set to `True`, this means that the
developer is required to set a cleanup behaviour for when their module
is uninstalled. For possible cleanup options, see fields.Selection's
docstring.

As far as implementation goes, everything is implemented in Python
unlike with Many2one fields where the behaviour is delegated to
PostgreSQL.

The `ondelete` setting will be processed during
`ir.model.fields.selection.unlink()` to ensure that the registry is left
in an appropriate state after module uninstall.

odoo/addons/base/models/ir_model.py

@@ -893,7 +893,7 @@ class IrModelFields(models.Model):
             'index': bool(field.index),
             'store': bool(field.store),
             'copied': bool(field.copy),
-            'on_delete': getattr(field, 'ondelete', None),
+            'on_delete': field.ondelete if field.type == 'many2one' else None,
             'related': ".".join(field.related) if field.related else None,
             'readonly': bool(field.readonly),
             'required': bool(field.required),
@@ -1238,16 +1238,7 @@ class IrModelSelection(models.Model):
                               'Please modify them through Python code, '
                               'preferably through a custom addon!'))
 
-        for selection in self:
-            if selection.field_id.store and \
-                    not self.env[selection.field_id.model]._abstract:
-                # replace the value by NULL in the field's corresponding column
-                query = 'UPDATE "{table}" SET "{field}"=NULL WHERE "{field}"=%s'.format(
-                    table=self.env[selection.field_id.model]._table,
-                    field=selection.field_id.name,
-                )
-                self.env.cr.execute(query, [selection.value])
-
+        self._process_ondelete()
         result = super().unlink()
 
         # Reload registry for normal unlink only. For module uninstall, the
@@ -1259,6 +1250,40 @@ class IrModelSelection(models.Model):
 
         return result
 
+    def _process_ondelete(self):
+        """ Process the 'ondelete' of the given selection values. """
+        for selection in self:
+            Model = self.env[selection.field_id.model]
+            field = Model._fields[selection.field_id.name]
+            if not field.store or Model._abstract:
+                continue
+
+            ondelete = (field.ondelete or {}).get(selection.value) or 'set null'
+            if callable(ondelete):
+                ondelete(selection._get_records())
+            elif ondelete == 'set null':
+                selection._get_records().write({field.name: False})
+            elif ondelete == 'set default':
+                value = field.convert_to_write(field.default(Model), Model)
+                selection._get_records().write({field.name: value})
+            elif ondelete == 'cascade':
+                selection._get_records().unlink()
+            else:
+                # this shouldn't happen... simply a sanity check
+                raise ValueError(_(
+                    "The ondelete policy %r is not valid for field %r"
+                ) % (ondelete, selection))
+
+    def _get_records(self):
+        """ Return the records having 'self' as a value. """
+        self.ensure_one()
+        Model = self.env[self.field_id.model]
+        query = 'SELECT id FROM "{table}" WHERE "{field}"=%s'.format(
+            table=Model._table, field=self.field_id.name,
+        )
+        self.env.cr.execute(query, [self.value])
+        return Model.browse(r[0] for r in self.env.cr.fetchall())
+
 
 class IrModelConstraint(models.Model):
     """

odoo/addons/test_new_api/models/test_new_api.py

@@ -894,3 +894,100 @@ class StateMixin(models.AbstractModel):
         ('confirmed', 'Confirmed'),
         ('done', 'Done'),
     ])
+
+
+class SelectionBase(models.Model):
+    _name = 'test_new_api.model_selection_base'
+    _description = "Model with a base selection field"
+
+    my_selection = fields.Selection([
+        ('foo', "Foo"),
+        ('bar', "Bar"),
+    ])
+
+
+class SelectionBaseNullExplicit(models.Model):
+    _inherit = 'test_new_api.model_selection_base'
+    _description = "Model with a selection field extension with ondelete null"
+
+    my_selection = fields.Selection(selection_add=[
+        ('quux', "Quux"),
+    ], ondelete={'quux': 'set null'})
+
+
+class SelectionBaseNullImplicit(models.Model):
+    _inherit = 'test_new_api.model_selection_base'
+    _description = "Model with a selection field extension without ondelete"
+
+    my_selection = fields.Selection(selection_add=[
+        ('ham', "Ham"),
+    ])
+
+
+class SelectionRequired(models.Model):
+    _name = 'test_new_api.model_selection_required'
+    _description = "Model with a required selection field"
+
+    active = fields.Boolean(default=True)
+    my_selection = fields.Selection([
+        ('foo', "Foo"),
+        ('bar', "Bar"),
+    ], required=True, default='foo')
+
+
+class SelectionRequiredDefault(models.Model):
+    _inherit = 'test_new_api.model_selection_required'
+    _description = "Model with a selection field extension with ondelete default"
+
+    my_selection = fields.Selection(selection_add=[
+        ('baz', "Baz"),
+    ], ondelete={'baz': 'set default'})
+
+
+class SelectionRequiredCascade(models.Model):
+    _inherit = 'test_new_api.model_selection_required'
+    _description = "Model with a selection field extension with ondelete cascade"
+
+    my_selection = fields.Selection(selection_add=[
+        ('eggs', "Eggs"),
+    ], ondelete={'eggs': 'cascade'})
+
+
+class SelectionRequiredLiteral(models.Model):
+    _inherit = 'test_new_api.model_selection_required'
+    _description = "Model with a selection field extension with ondelete set <option>"
+
+    my_selection = fields.Selection(selection_add=[
+        ('bacon', "Bacon"),
+    ], ondelete={'bacon': lambda r: r.write({'my_selection': 'bar'})})
+
+
+class SelectionRequiredMultiple(models.Model):
+    _inherit = 'test_new_api.model_selection_required'
+    _description = "Model with a selection field extension with multiple ondelete policies"
+
+    my_selection = fields.Selection(selection_add=[
+        ('pikachu', "Pikachu"),
+        ('eevee', "Eevee"),
+    ], ondelete={'pikachu': 'set default', 'eevee': lambda r: r.write({'my_selection': 'bar'})})
+
+
+class SelectionRequiredCallback(models.Model):
+    _inherit = 'test_new_api.model_selection_required'
+    _description = "Model with a selection field extension with ondelete callback"
+
+    my_selection = fields.Selection(selection_add=[
+        ('knickers', "Oh la la"),
+    ], ondelete={
+        'knickers': lambda recs: recs.write({'active': False, 'my_selection': 'foo'}),
+    })
+
+
+class SelectionNonStored(models.Model):
+    _name = 'test_new_api.model_selection_non_stored'
+    _description = "Model with non-stored selection field"
+
+    my_selection = fields.Selection([
+        ('foo', "Foo"),
+        ('bar', "Bar"),
+    ], store=False)

odoo/addons/test_new_api/security/ir.model.access.csv

@@ -52,3 +52,6 @@ access_test_new_api_model_private_address_onchange,access_test_new_api_model_pri
 access_test_new_api_req_m2o_transient,access_test_new_api_req_m2o_transient,model_test_new_api_req_m2o_transient,,0,0,0,0
 access_test_new_api_country,access_test_new_api_country,model_test_new_api_country,base.group_user,1,1,1,1
 access_test_new_api_city,access_test_new_api_city,model_test_new_api_city,base.group_user,1,1,1,1
+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

odoo/addons/test_new_api/tests/test_new_fields.py

@@ -11,7 +11,7 @@ import io
 from PIL import Image
 import psycopg2
 
-from odoo import fields
+from odoo import models, fields
 from odoo.addons.base.tests.common import TransactionCaseWithUserDemo
 from odoo.exceptions import AccessError, UserError, ValidationError
 from odoo.tests import common
@@ -2462,3 +2462,205 @@ class TestSelectionDeleteUpdate(common.TransactionCase):
             ('field_id.name', '=', 'state'),
             ('value', '=', 'confirmed'),
         ], limit=1).unlink()
+
+
+@common.tagged('selection_ondelete_base')
+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'
+
+    def setUp(self):
+        super().setUp()
+        # enable unlinking ir.model.fields.selection
+        self.patch(self.registry, 'ready', False)
+
+    def _unlink_option(self, model, option):
+        self.env['ir.model.fields.selection'].search([
+            ('field_id.model', '=', model),
+            ('field_id.name', '=', 'my_selection'),
+            ('value', '=', option),
+        ], limit=1).unlink()
+
+    def test_ondelete_default(self):
+        # create some records, one of which having the extended selection option
+        rec1 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'foo'})
+        rec2 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'bar'})
+        rec3 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'baz'})
+
+        # test that all values are correct before the removal of the value
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'baz')
+
+        # unlink the extended option (simulates a module uninstall)
+        self._unlink_option(self.MODEL_REQUIRED, 'baz')
+
+        # verify that the ondelete policy has succesfully been applied
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'foo')   # reset to default
+
+    def test_ondelete_base_null_explicit(self):
+        rec1 = self.env[self.MODEL_BASE].create({'my_selection': 'foo'})
+        rec2 = self.env[self.MODEL_BASE].create({'my_selection': 'bar'})
+        rec3 = self.env[self.MODEL_BASE].create({'my_selection': 'quux'})
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'quux')
+
+        self._unlink_option(self.MODEL_BASE, 'quux')
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertFalse(rec3.my_selection)
+
+    def test_ondelete_base_null_implicit(self):
+        rec1 = self.env[self.MODEL_BASE].create({'my_selection': 'foo'})
+        rec2 = self.env[self.MODEL_BASE].create({'my_selection': 'bar'})
+        rec3 = self.env[self.MODEL_BASE].create({'my_selection': 'ham'})
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'ham')
+
+        self._unlink_option(self.MODEL_BASE, 'ham')
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertFalse(rec3.my_selection)
+
+    def test_ondelete_cascade(self):
+        rec1 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'foo'})
+        rec2 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'bar'})
+        rec3 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'eggs'})
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'eggs')
+
+        self._unlink_option(self.MODEL_REQUIRED, 'eggs')
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertFalse(rec3.exists())
+
+    def test_ondelete_literal(self):
+        rec1 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'foo'})
+        rec2 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'bar'})
+        rec3 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'bacon'})
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'bacon')
+
+        self._unlink_option(self.MODEL_REQUIRED, 'bacon')
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'bar')
+
+    def test_ondelete_multiple_explicit(self):
+        rec1 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'foo'})
+        rec2 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'eevee'})
+        rec3 = self.env[self.MODEL_REQUIRED].create({'my_selection': 'pikachu'})
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'eevee')
+        self.assertEqual(rec3.my_selection, 'pikachu')
+
+        self._unlink_option(self.MODEL_REQUIRED, 'eevee')
+        self._unlink_option(self.MODEL_REQUIRED, 'pikachu')
+
+        self.assertEqual(rec1.my_selection, 'foo')
+        self.assertEqual(rec2.my_selection, 'bar')
+        self.assertEqual(rec3.my_selection, 'foo')
+
+    def test_ondelete_callback(self):
+        rec = self.env[self.MODEL_REQUIRED].create({'my_selection': 'knickers'})
+
+        self.assertEqual(rec.my_selection, 'knickers')
+
+        self._unlink_option(self.MODEL_REQUIRED, 'knickers')
+
+        self.assertEqual(rec.my_selection, 'foo')
+        self.assertFalse(rec.active)
+
+    def test_non_stored_selection(self):
+        rec = self.env[self.MODEL_NONSTORED].create({})
+        rec.my_selection = 'foo'
+
+        self.assertEqual(rec.my_selection, 'foo')
+
+        self._unlink_option(self.MODEL_NONSTORED, 'foo')
+
+        self.assertFalse(rec.my_selection)
+
+
+@common.tagged('selection_ondelete_advanced')
+class TestSelectionOndeleteAdvanced(common.TransactionCase):
+
+    MODEL_BASE = 'test_new_api.model_selection_base'
+    MODEL_REQUIRED = 'test_new_api.model_selection_required'
+
+    def setUp(self):
+        super().setUp()
+        # necessary cleanup for resetting changes in the registry
+        for model_name in (self.MODEL_BASE, self.MODEL_REQUIRED):
+            Model = self.registry[model_name]
+            self.addCleanup(setattr, Model, '__bases__', Model.__bases__)
+        self.addCleanup(self.registry.model_cache.clear)
+
+    def test_ondelete_unexisting_policy(self):
+        class Foo(models.Model):
+            _inherit = self.MODEL_REQUIRED
+
+            my_selection = fields.Selection(selection_add=[
+                ('random', "Random stuff"),
+            ], ondelete={'random': 'poop'})
+
+        Foo._build_model(self.registry, self.env.cr)
+
+        with self.assertRaises(ValueError):
+            self.registry.setup_models(self.env.cr)
+
+    def test_ondelete_default_no_default(self):
+        class Foo(models.Model):
+            _inherit = self.MODEL_BASE
+
+            my_selection = fields.Selection(selection_add=[
+                ('corona', "Corona beers suck"),
+            ], ondelete={'corona': 'set default'})
+
+        Foo._build_model(self.registry, self.env.cr)
+
+        with self.assertRaises(AssertionError):
+            self.registry.setup_models(self.env.cr)
+
+    def test_ondelete_required_null_explicit(self):
+        class Foo(models.Model):
+            _inherit = self.MODEL_REQUIRED
+
+            my_selection = fields.Selection(selection_add=[
+                ('brap', "Brap"),
+            ], ondelete={'brap': 'set null'})
+
+        Foo._build_model(self.registry, self.env.cr)
+
+        with self.assertRaises(ValueError):
+            self.registry.setup_models(self.env.cr)
+
+    def test_ondelete_required_null_implicit(self):
+        class Foo(models.Model):
+            _inherit = self.MODEL_REQUIRED
+
+            my_selection = fields.Selection(selection_add=[
+                ('boing', "Boyoyoyoing"),
+            ])
+
+        Foo._build_model(self.registry, self.env.cr)
+
+        with self.assertRaises(ValueError):
+            self.registry.setup_models(self.env.cr)

odoo/addons/test_new_api/tests/test_schema.py

@@ -3,13 +3,24 @@ from odoo.models import MetaModel
 from odoo.tests import common
 
 
+def get_model_name(cls):
+    name = cls._name
+    if not name:
+        [name] = cls._inherit if isinstance(cls._inherit, list) else [cls._inherit]
+    assert isinstance(name, str)
+    return name
+
+
 class TestReflection(common.TransactionCase):
     """ Test the reflection into 'ir.model', 'ir.model.fields', etc. """
 
     def test_models_fields(self):
         """ check that all models and fields are reflected as expected. """
         # retrieve the models defined in this module, and check them
-        model_names = {cls._name for cls in MetaModel.module_to_models['test_new_api']}
+        model_names = {
+            get_model_name(cls)
+            for cls in MetaModel.module_to_models['test_new_api']
+        }
         ir_models = self.env['ir.model'].search([('model', 'in', list(model_names))])
         self.assertEqual(len(ir_models), len(model_names))
         for ir_model in ir_models:

odoo/fields.py

@@ -2134,6 +2134,24 @@ class Selection(Field):
             > result = [('a', 'A'), ('c', 'C'), ('b', 'B')]
     :type selection_add: list(tuple(str,str))
 
+    :param ondelete: provides a fallback mechanism for any overridden
+        field with a selection_add. It is a dict that maps every option
+        from the selection_add to a fallback action.
+
+        This fallback action will be applied to all records whose
+        selection_add option maps to it.
+
+        The actions can be any of the following:
+            - 'set null' -- the default, all records with this option
+              will have their selection value set to False.
+            - 'cascade' -- all records with this option will be
+              deleted along with the option itself.
+            - 'set default' -- all records with this option will be
+              set to the default of the field definition
+            - <callable> -- a callable whose first and only argument will be
+              the set of records containing the specified Selection option,
+              for custom processing
+
     The attribute ``selection`` is mandatory except in the case of
     ``related`` or extended fields.
     """
@@ -2142,6 +2160,7 @@ class Selection(Field):
     _slots = {
         'selection': None,              # [(value, string), ...], function or method name
         'validate': True,               # whether validating upon write
+        'ondelete': None,               # {value: policy} (what to do when value is deleted)
     }
 
     def __init__(self, selection=Default, string=Default, **kwargs):
@@ -2175,15 +2194,14 @@ class Selection(Field):
                     _logger.warning("%s: selection attribute will be ignored as the field is related", self)
                 selection = field.args['selection']
                 if isinstance(selection, list):
-                    if (
-                        values is not None
-                        and values != [kv[0] for kv in selection]
-                    ):
+                    if values is not None and values != [kv[0] for kv in selection]:
                         _logger.warning("%s: selection=%r overrides existing selection; use selection_add instead", self, selection)
                     values = [kv[0] for kv in selection]
                     labels.update(selection)
+                    self.ondelete = {}
                 else:
                     self.selection = selection
+                    self.ondelete = None
 
             if 'selection_add' in field.args:
                 if self.related:
@@ -2193,8 +2211,41 @@ class Selection(Field):
                     "%s: selection_add=%r must be a list" % (self, selection_add)
                 assert values is not None, \
                     "%s: selection_add=%r on non-list selection %r" % (self, selection_add, self.selection)
+
+                ondelete = field.args.get('ondelete') or {}
+                new_values = [kv[0] for kv in selection_add if kv[0] not in values]
+                for key in new_values:
+                    ondelete.setdefault(key, 'set null')
+                if self.required and new_values and 'set null' in ondelete.values():
+                    raise ValueError(
+                        "%r: required selection fields must define an ondelete policy that "
+                        "implements the proper cleanup of the corresponding records upon "
+                        "module uninstallation. Please use one or more of the following "
+                        "policies: 'set default' (if the field has a default defined), 'cascade', "
+                        "or a single-argument callable where the argument is the recordset "
+                        "containing the specified option." % self
+                    )
+
+                # check ondelete values
+                for key, val in ondelete.items():
+                    if callable(val) or val in ('set null', 'cascade'):
+                        continue
+                    if val == 'set default':
+                        assert self.default is not None, (
+                            "%r: ondelete policy of type 'set default' is invalid for this field "
+                            "as it does not define a default! Either define one in the base "
+                            "field, or change the chosen ondelete policy" % self
+                        )
+                        continue
+                    raise ValueError(
+                        "%r: ondelete policy %r for selection value %r is not a valid ondelete "
+                        "policy, please choose one of 'set null', 'set default', 'cascade' or "
+                        "a callable" % (self, val, key)
+                    )
+
                 values = merge_sequences(values, [kv[0] for kv in selection_add])
                 labels.update(kv for kv in selection_add if len(kv) == 2)
+                self.ondelete.update(ondelete)
 
         if values is not None:
             self.selection = [(value, labels[value]) for value in values]