[FIX] core: complex flushing when searching on one2many fields

Consider two models A and B, where
 - model A has a many2one_reference field 'res_id' with model field 'res_model';
 - model B has an auto-join one2many field 'stuff_ids' to A using field 'res_id';
 - the field 'res_model' is not flushed on some record.

      model     | A                 | B
     -----------+-------------------+-------------------
      memory    | res_model = B     |
     -----------+-------------------+-------------------
      database  | res_model = NULL  | id = 42
                | res_id = 42       |
                | foo = 'bar'       |

Now, perform a search on model B that should return record with id=42 by
matching some condition on the unflushed record in model A, like:

    B.search([('stuff_ids.foo', '=', 'bar')])

Before this patch, the search method would not flush the field
'res_model', which causes the method to return incorrect results.  This
patch fixes the issue by ensuring that searches on one2many fields flush
all the fields on which the one2many field depends.

The issue was discovered while working on task 2735672.

closes odoo/odoo#96115

Signed-off-by: Raphael Collet <rco@odoo.com>
This commit is contained in:
Raphael Collet
2022-07-16 17:05:45 +02:00
parent f3590959b0
commit 519d8fe492
3 changed files with 50 additions and 29 deletions
@@ -70,7 +70,7 @@ class TestCrmPerformance(CrmPerformanceCase):
country_be = self.env.ref('base.be')
lang_be = self.env['res.lang']._lang_get('fr_BE')
with freeze_time(self.reference_now), self.assertQueryCount(user_sales_leads=177): # com 166
with freeze_time(self.reference_now), self.assertQueryCount(user_sales_leads=178): # com 167
self.env.cr._now = self.reference_now # force create_date to check schedulers
with Form(self.env['crm.lead']) as lead_form:
lead_form.country_id = country_be
@@ -3247,6 +3247,23 @@ class TestMany2oneReference(common.TransactionCase):
foo = m.browse(1 if not ids[0] else (ids[0] + 1))
self.assertTrue(foo.unlink())
def test_search_inverse_one2many_autojoin(self):
record = self.env['test_new_api.inverse_m2o_ref'].create({})
# the one2many field 'model_ids' should be auto_join=True
self.patch(type(record).model_ids, 'auto_join', True)
# create a reference to record
reference = self.env['test_new_api.model_many2one_reference'].create({'res_id': record.id})
reference.res_model = record._name
# the model field 'res_model' is not in database yet
self.assertTrue(self.env.cache.has_dirty_fields(reference, [type(reference).res_model]))
# searching on the one2many should flush the field 'res_model'
records = record.search([('model_ids.create_date', '!=', False)])
self.assertIn(record, records)
@common.tagged('selection_abstract')
class TestSelectionDeleteUpdate(common.TransactionCase):
+32 -28
View File
@@ -4796,43 +4796,47 @@ class BaseModel(metaclass=MetaModel):
to_flush = defaultdict(set) # {model_name: field_names}
if fields:
to_flush[self._name].update(fields)
# also take into account the fields in the record rules
domain = list(domain) + (self.env['ir.rule']._compute_domain(self._name, 'read') or [])
for arg in domain:
if isinstance(arg, str):
continue
if not isinstance(arg[0], str):
continue
model_name = self._name
for fname in arg[0].split('.'):
field = self.env[model_name]._fields.get(fname)
def collect_from_domain(model, domain):
for arg in domain:
if isinstance(arg, str):
continue
if not isinstance(arg[0], str):
continue
comodel = collect_from_path(model, arg[0])
if arg[1] in ('child_of', 'parent_of') and comodel._parent_store:
# hierarchy operators need the parent field
collect_from_path(comodel, comodel._parent_name)
def collect_from_path(model, path):
# path is a dot-separated sequence of field names
for fname in path.split('.'):
field = model._fields.get(fname)
if not field:
break
to_flush[model_name].add(fname)
to_flush[model._name].add(fname)
if field.type == 'one2many' and field.inverse_name:
to_flush[field.comodel_name].add(field.inverse_name)
field_domain = field.get_domain_list(model)
if field_domain:
collect_from_domain(self.env[field.comodel_name], field_domain)
# DLE P111: `test_message_process_email_partner_find`
# Search on res.users with email_normalized in domain
# must trigger the recompute and flush of res.partner.email_normalized
if field.related_field:
model = self
if field.related:
# DLE P129: `test_transit_multi_companies`
# `self.env['stock.picking'].search([('product_id', '=', product.id)])`
# Should flush `stock.move.picking_ids` as `product_id` on `stock.picking` is defined as:
# `product_id = fields.Many2one('product.product', 'Product', related='move_lines.product_id', readonly=False)`
for f in field.related.split('.'):
rfield = model._fields.get(f)
if rfield:
to_flush[model._name].add(f)
if rfield.type in ('many2one', 'one2many', 'many2many'):
model = self.env[rfield.comodel_name]
if rfield.type == 'one2many' and rfield.inverse_name:
to_flush[rfield.comodel_name].add(rfield.inverse_name)
if field.comodel_name:
model_name = field.comodel_name
# hierarchy operators need the parent field
if arg[1] in ('child_of', 'parent_of'):
model = self.env[model_name]
if model._parent_store:
to_flush[model_name].add(model._parent_name)
collect_from_path(model, field.related)
if field.relational:
model = self.env[field.comodel_name]
# return the model found by traversing all fields (used in collect_from_domain)
return model
# also take into account the fields in the record rules
domain = list(domain) + (self.env['ir.rule']._compute_domain(self._name, 'read') or [])
collect_from_domain(self, domain)
# flush the order fields
order_spec = order or self._order