[ADD] util.fields: handle ir.exports model/fields renames/removal

src/base/tests/test_util.py

@@ -19,7 +19,7 @@
 
 from odoo.addons.base.maintenance.migrations import util
 from odoo.addons.base.maintenance.migrations.testing import UnitTestCase, parametrize
-from odoo.addons.base.maintenance.migrations.util.domains import _adapt_one_domain
+from odoo.addons.base.maintenance.migrations.util.domains import _adapt_one_domain, _model_of_path
 from odoo.addons.base.maintenance.migrations.util.exceptions import MigrationError
 
 
@@ -42,6 +42,24 @@ def test_adapt_renamed_field(self):
 
         self.assertEqual(match_domain, new_domain)
 
+    @parametrize(
+        [
+            ("res.currency", [], "res.currency"),
+            ("res.currency", ["rate_ids"], "res.currency.rate"),
+            ("res.currency", ("rate_ids", "company_id"), "res.company"),
+            ("res.currency", ["rate_ids", "company_id", "user_ids"], "res.users"),
+            ("res.currency", ("rate_ids", "company_id", "user_ids", "partner_id"), "res.partner"),
+            ("res.users", ["partner_id"], "res.partner"),
+            ("res.users", ["nonexistent_field"], None),
+            ("res.users", ("partner_id", "active"), None),
+            ("res.users", ("partner_id", "active", "name"), None),
+            ("res.users", ("partner_id", "removed_field"), None),
+        ]
+    )
+    def test_model_of_path(self, model, path, expected):
+        cr = self.env.cr
+        self.assertEqual(_model_of_path(cr, model, path), expected)
+
     def test_change_no_leaf(self):
         # testing plan: updata path of a domain where the last element is not changed
 
@@ -389,6 +407,63 @@ def test_remove_field(self, domain, expected):
         self.assertEqual(altered_domain, expected)
 
 
+class TestIrExports(UnitTestCase):
+    def setUp(self):
+        super().setUp()
+        self.export = self.env["ir.exports"].create(
+            [
+                {
+                    "name": "Test currency export",
+                    "resource": "res.currency",
+                    "export_fields": [
+                        (0, 0, {"name": "full_name"}),
+                        (0, 0, {"name": "rate_ids/company_id/user_ids/name"}),
+                        (0, 0, {"name": "rate_ids/company_id/user_ids/partner_id/user_ids/name"}),
+                        (0, 0, {"name": "rate_ids/name"}),
+                    ],
+                }
+            ]
+        )
+        util.flush(self.export)
+
+    def _invalidate(self):
+        util.invalidate(self.export.export_fields)
+        util.invalidate(self.export)
+
+    def test_rename_field(self):
+        util.rename_field(self.cr, "res.partner", "user_ids", "renamed_user_ids")
+        self._invalidate()
+        self.assertEqual(
+            self.export.export_fields[2].name, "rate_ids/company_id/user_ids/partner_id/renamed_user_ids/name"
+        )
+
+        util.rename_field(self.cr, "res.users", "name", "new_name")
+        self._invalidate()
+        self.assertEqual(self.export.export_fields[1].name, "rate_ids/company_id/user_ids/new_name")
+
+    def test_remove_field(self):
+        util.remove_field(self.cr, "res.currency.rate", "company_id")
+        self._invalidate()
+        self.assertEqual(len(self.export.export_fields), 2)
+        self.assertEqual(self.export.export_fields[0].name, "full_name")
+        self.assertEqual(self.export.export_fields[1].name, "rate_ids/name")
+
+    def test_rename_model(self):
+        util.rename_model(self.cr, "res.currency", "res.currency2")
+        self._invalidate()
+        self.assertEqual(self.export.resource, "res.currency2")
+
+    def test_remove_model(self):
+        util.remove_model(self.cr, "res.currency.rate")
+        self._invalidate()
+        self.assertEqual(len(self.export.export_fields), 1)
+        self.assertEqual(self.export.export_fields[0].name, "full_name")
+
+        util.remove_model(self.cr, "res.currency")
+        self.cr.execute("SELECT * FROM ir_exports WHERE id = %s", [self.export.id])
+        self.assertFalse(self.cr.fetchall())
+
+
 class TestIterBrowse(UnitTestCase):
     def test_iter_browse_iter(self):
         cr = self.env.cr
@@ -670,6 +745,25 @@ def test_model_table_convertion(self):
             self.assertEqual(table, self.env[model]._table)
             self.assertEqual(util.model_of_table(cr, table), model)
 
+    def test_resolve_model_fields_path(self):
+        cr = self.env.cr
+
+        # test with provided paths
+        model, path = "res.currency", ["rate_ids", "company_id", "user_ids", "partner_id"]
+        expected_result = [
+            util.FieldsPathPart("res.currency", "rate_ids", "res.currency.rate"),
+            util.FieldsPathPart("res.currency.rate", "company_id", "res.company"),
+            util.FieldsPathPart("res.company", "user_ids", "res.users"),
+            util.FieldsPathPart("res.users", "partner_id", "res.partner"),
+        ]
+        result = util.resolve_model_fields_path(cr, model, path)
+        self.assertEqual(result, expected_result)
+
+        model, path = "res.users", ("partner_id", "removed_field", "user_id")
+        expected_result = [util.FieldsPathPart("res.users", "partner_id", "res.partner")]
+        result = util.resolve_model_fields_path(cr, model, path)
+        self.assertEqual(result, expected_result)
+
 
 @unittest.skipIf(
     util.version_gte("saas~17.1"),

src/util/domains.py

@@ -33,7 +33,7 @@
     from openerp.tools.safe_eval import safe_eval
 
 from .const import NEARLYWARN
-from .helpers import _dashboard_actions, _validate_model
+from .helpers import _dashboard_actions, _validate_model, resolve_model_fields_path
 from .inherit import for_each_inherit
 from .misc import SelfPrintEvalContext
 from .pg import column_exists, get_value_or_en_translation, table_exists
@@ -160,21 +160,13 @@ def _get_domain_fields(cr):
 
 
 def _model_of_path(cr, model, path):
-    for field in path:
-        cr.execute(
-            """
-            SELECT relation
-              FROM ir_model_fields
-             WHERE model = %s
-               AND name = %s
-        """,
-            [model, field],
-        )
-        if not cr.rowcount:
-            return None
-        [model] = cr.fetchone()
-
-    return model
+    if not path:
+        return model
+    path = tuple(path)
+    resolved_parts = resolve_model_fields_path(cr, model, path)
+    if len(resolved_parts) == len(path):
+        return resolved_parts[-1].relation_model
+    return None
 
 
 def _valid_path_to(cr, path, from_, to):

src/util/fields.py

@@ -40,7 +40,13 @@ def make_index_name(table_name, column_name):
 from .const import ENVIRON
 from .domains import _adapt_one_domain, _replace_path, _valid_path_to, adapt_domains
 from .exceptions import SleepyDeveloperError
-from .helpers import _dashboard_actions, _validate_model, table_of_model
+from .helpers import (
+    _dashboard_actions,
+    _remove_export_lines,
+    _validate_model,
+    resolve_model_fields_path,
+    table_of_model,
+)
 from .inherit import for_each_inherit
 from .misc import SelfPrintEvalContext, log_progress, version_gte
 from .orm import env, invalidate
@@ -202,6 +208,9 @@ def clean_context(context):
             [(fieldname, fieldname + " desc"), model, r"\y{}\y".format(fieldname)],
         )
 
+    # ir.exports.line
+    _remove_export_lines(cr, model, fieldname)
+
     def adapter(leaf, is_or, negated):
         # replace by TRUE_LEAF, unless negated or in a OR operation but not negated
         if is_or ^ negated:
@@ -1071,21 +1080,32 @@ def _update_field_usage_multi(cr, models, old, new, domain_adapter=None, skip_in
         cr.execute(q.format(col_prefix=col_prefix), p)
 
         # ir.exports.line
-        q = """
-            UPDATE ir_exports_line l
-               SET name = regexp_replace(l.name, %(old)s, %(new)s, 'g')
-        """
         if only_models:
-            q += """
-              FROM ir_exports e
-             WHERE e.id = l.export_id
-               AND e.resource IN %(models)s
-               AND
-            """
-        else:
-            q += "WHERE "
-        q += "l.name ~ %(old)s"
-        cr.execute(q, p)
+            cr.execute(
+                """
+                SELECT el.id,
+                       e.resource,
+                       STRING_TO_ARRAY(el.name, '/')
+                  FROM ir_exports_line el
+                  JOIN ir_exports e
+                    ON el.export_id = e.id
+                 WHERE el.name ~ %s
+                """,
+                [r"\y{}\y".format(old)],
+            )
+            fixed_lines_paths = {}
+            for line_id, line_model, line_path in cr.fetchall():
+                new_path = [
+                    new if x.field_name == old and x.field_model in only_models else x.field_name
+                    for x in resolve_model_fields_path(cr, line_model, line_path)
+                ]
+                if len(new_path) == len(line_path) and new_path != line_path:
+                    fixed_lines_paths[line_id] = "/".join(new_path)
+            if fixed_lines_paths:
+                cr.execute(
+                    "UPDATE ir_exports_line SET name = (%s::jsonb)->>(id::text) WHERE id IN %s",
+                    [Json(fixed_lines_paths), tuple(fixed_lines_paths)],
+                )
 
         # mail.alias
         if column_exists(cr, "mail_alias", "alias_defaults"):

src/util/helpers.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import logging
 import os
+from collections import namedtuple
 
 import lxml
 
@@ -214,3 +215,127 @@ def _get_theme_models():
         "theme.website.menu": "website.menu",
         "theme.ir.attachment": "ir.attachment",
     }
+
+
+FieldsPathPart = namedtuple("FieldsPathPart", "field_model field_name relation_model")
+FieldsPathPart.__doc__ = """
+Encapsulate information about a field within a fields path.
+
+:param str field_model: model of the field
+:param str field_name: name of the field
+:param str relation_model: target model of the field, if relational, otherwise ``None``
+"""
+for _f in FieldsPathPart._fields:
+    getattr(FieldsPathPart, _f).__doc__ = None
+
+
+def resolve_model_fields_path(cr, model, path):
+    """
+    Resolve model fields paths.
+
+    This function returns a list of :class:`~odoo.upgrade.util.helpers.FieldsPathPart`
+    where each item describes a field in ``path`` (in the same order). The returned list
+    could be shorter than the original ``path`` due to a missing field or model, or
+    because there is a non-relational field in the path. The only non-relational field
+    allowed in a fields path is the last one, in which case the returned list has the same
+    length as the input ``path``.
+
+    .. example::
+
+       .. code-block:: python
+
+          >>> util.resolve_model_fields_path(cr, "res.partner", "user_ids.partner_id.title".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users'),
+          FieldsPathPart(field_model='res.users', field_name='partner_id', relation_model='res.partner'),
+          FieldsPathPart(field_model='res.partner', field_name='title', relation_model='res.partner.title')]
+
+       Last field is not relational:
+
+       .. code-block:: python
+
+          >>> resolve_model_fields_path(cr, "res.partner", "user_ids.active".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users'),
+          FieldsPathPart(field_model='res.users', field_name='active', relation_model=None)]
+
+       The path is wrong, it uses a non-relational field:
+
+       .. code-block:: python
+
+          >>> resolve_model_fields_path(cr, "res.partner", "user_ids.active.name".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users'),
+          FieldsPathPart(field_model='res.users', field_name='active', relation_model=None)]
+
+       The path is broken, it uses a non-existing field:
+
+       .. code-block:: python
+
+          >>> resolve_model_fields_path(cr, "res.partner", "user_ids.non_existing_id.active".split("."))
+          [FieldsPathPart(field_model='res.partner', field_name='user_ids', relation_model='res.users')]
+
+    :param str model: starting model of the fields path
+    :param typing.Sequence[str] path: fields path
+    :return: resolved fields path parts
+    :rtype: list(:class:`~odoo.upgrade.util.helpers.FieldsPathPart`)
+    """
+    path = list(path)
+    cr.execute(
+        """
+        WITH RECURSIVE resolved_fields_path AS (
+            -- non-recursive term
+               SELECT imf.model AS field_model,
+                      imf.name AS field_name,
+                      imf.relation AS relation_model,
+                      p.path AS path,
+                      1 AS part_index
+                 FROM (VALUES (%(model)s, %(path)s)) p(model, path)
+                 JOIN ir_model_fields imf
+                   ON imf.model = p.model
+                  AND imf.name = p.path[1]
+
+            UNION ALL
+
+            -- recursive term
+               SELECT rimf.model AS field_model,
+                      rimf.name AS field_name,
+                      rimf.relation AS relation_model,
+                      rfp.path AS path,
+                      rfp.part_index + 1 AS part_index
+                 FROM resolved_fields_path rfp
+                 JOIN ir_model_fields rimf
+                   ON rimf.model = rfp.relation_model
+                  AND rimf.name = rfp.path[rfp.part_index + 1]
+                WHERE cardinality(rfp.path) > rfp.part_index
+        )
+        SELECT field_model,
+               field_name,
+               relation_model
+          FROM resolved_fields_path
+         ORDER BY part_index
+        """,
+        {"model": model, "path": list(path)},
+    )
+    return [FieldsPathPart(**row) for row in cr.dictfetchall()]
+
+
+def _remove_export_lines(cr, model, field=None):
+    q = """
+    SELECT el.id,
+           e.resource,
+           STRING_TO_ARRAY(el.name, '/')
+      FROM ir_exports_line el
+      JOIN ir_exports e
+        ON el.export_id = e.id
+    """
+    if field:
+        q = cr.mogrify(q + " WHERE el.name ~ %s ", [r"\y{}\y".format(field)]).decode()
+    cr.execute(q)
+    to_rem = [
+        line_id
+        for line_id, line_model, line_path in cr.fetchall()
+        if any(
+            x.field_model == model and (field is None or x.field_name == field)
+            for x in resolve_model_fields_path(cr, line_model, line_path)
+        )
+    ]
+    if to_rem:
+        cr.execute("DELETE FROM ir_exports_line WHERE id IN %s", [tuple(to_rem)])

src/util/indirect_references.py

@@ -39,7 +39,7 @@ def indirect_references(cr, bound_only=False):
         IR("ir_model_fields", "relation", None),  # destination of a relation field
         IR("ir_model_data", "model", "res_id"),
         IR("ir_filters", "model_id", None, set_unknown=True),  # YUCK!, not an id
-        IR("ir_exports", "resource", None, set_unknown=True),
+        IR("ir_exports", "resource", None),
         IR("ir_ui_view", "model", None, set_unknown=True),
         IR("ir_values", "model", "res_id"),
         IR("wkf_transition", "trigger_model", None),

src/util/models.py

@@ -11,7 +11,7 @@
 
 from .const import ENVIRON
 from .fields import IMD_FIELD_PATTERN, remove_field
-from .helpers import _ir_values_value, _validate_model, model_of_table, table_of_model
+from .helpers import _ir_values_value, _remove_export_lines, _validate_model, model_of_table, table_of_model
 from .indirect_references import indirect_references
 from .inherit import for_each_inherit, inherit_parents
 from .misc import _cached, chunks, log_progress
@@ -129,6 +129,9 @@ def remove_model(cr, model, drop_table=True, ignore_m2m=()):
                 cr.execute(query, args + (tuple(ids),))
                 notify = notify or bool(cr.rowcount)
 
+    # for ir.exports.line we have to take care of "nested" references in fields "paths"
+    _remove_export_lines(cr, model)
+
     _rm_refs(cr, model)
 
     cr.execute(