[ADD] util.fields: handle base_import.mapping model/fields renames/removal

Implement proper fixing and/or removal of base_import.mapping and records on
model/fields renames/removal:
- renamed fields: update affected base_import.mapping records
- removed fields: remove affected base_import.mapping records
- renamed models: update affected base_import.mapping records `res_model`
- removed models: remove affected base_import.mapping records

This is a follow-up on 5b944f7

closes #100

Signed-off-by: Christophe Simonis (chs) <chs@odoo.com>
Co-authored-by: Alvaro Fuentes <4387571+aj-fuentes@users.noreply.github.com>

Author: pauloamed

src/base/tests/test_util.py

@@ -464,6 +464,72 @@ def test_remove_model(self):
         self.assertFalse(self.cr.fetchall())
 
 
+class TestBaseImportMappings(UnitTestCase):
+    def setUp(self):
+        super().setUp()
+        self.import_mapping = self.env["base_import.mapping"].create(
+            [
+                {"res_model": "res.currency", "column_name": "Column name", "field_name": path}
+                for path in [
+                    "full_name",
+                    "rate_ids/company_id/user_ids/name",
+                    "rate_ids/company_id/user_ids/partner_id/user_ids/name",
+                    "rate_ids/name",
+                ]
+            ]
+        )
+
+        util.flush(self.import_mapping)
+
+    def test_rename_field(self):
+        util.rename_field(self.cr, "res.partner", "user_ids", "renamed_user_ids")
+        util.invalidate(self.import_mapping)
+
+        self.assertEqual(
+            self.import_mapping[2].field_name, "rate_ids/company_id/user_ids/partner_id/renamed_user_ids/name"
+        )
+
+        util.rename_field(self.cr, "res.users", "name", "new_name")
+        util.invalidate(self.import_mapping)
+
+        self.assertEqual(self.import_mapping[1].field_name, "rate_ids/company_id/user_ids/new_name")
+
+    def test_remove_field(self):
+        prev_mappings = self.env["base_import.mapping"].search([])
+
+        util.remove_field(self.cr, "res.currency.rate", "company_id")
+        util.invalidate(self.import_mapping)
+
+        removed_mappings = prev_mappings - self.env["base_import.mapping"].search([])
+        remaining_mappings = self.import_mapping - removed_mappings
+
+        self.assertEqual(len(removed_mappings), 2)
+        self.assertEqual(remaining_mappings[0].field_name, "full_name")
+        self.assertEqual(remaining_mappings[1].field_name, "rate_ids/name")
+
+    def test_rename_model(self):
+        util.rename_model(self.cr, "res.currency", "res.currency2")
+        util.invalidate(self.import_mapping)
+
+        self.assertEqual(self.import_mapping[0].res_model, "res.currency2")
+
+    def test_remove_model(self):
+        prev_mappings = self.env["base_import.mapping"].search([])
+
+        util.remove_model(self.cr, "res.currency.rate")
+        util.invalidate(self.import_mapping)
+
+        removed_mappings = prev_mappings - self.env["base_import.mapping"].search([])
+        remaining_mappings = self.import_mapping - removed_mappings
+
+        self.assertEqual(len(removed_mappings), 3)
+        self.assertEqual(remaining_mappings[0].field_name, "full_name")
+
+        util.remove_model(self.cr, "res.currency")
+        self.cr.execute("SELECT * FROM base_import_mapping WHERE id = %s", [remaining_mappings.id])
+        self.assertFalse(self.cr.fetchall())
+
+
 class TestIterBrowse(UnitTestCase):
     def test_iter_browse_iter(self):
         cr = self.env.cr

src/util/fields.py

@@ -40,13 +40,7 @@ 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,
-    _remove_export_lines,
-    _validate_model,
-    resolve_model_fields_path,
-    table_of_model,
-)
+from .helpers import _dashboard_actions, _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
@@ -63,6 +57,7 @@ def make_index_name(table_name, column_name):
     savepoint,
     table_exists,
 )
+from .records import _remove_import_export_paths
 from .report import add_to_migration_reports, get_anchor_link_to_record
 
 # python3 shims
@@ -208,8 +203,7 @@ def clean_context(context):
             [(fieldname, fieldname + " desc"), model, r"\y{}\y".format(fieldname)],
         )
 
-    # ir.exports.line
-    _remove_export_lines(cr, model, fieldname)
+    _remove_import_export_paths(cr, model, fieldname)
 
     def adapter(leaf, is_or, negated):
         # replace by TRUE_LEAF, unless negated or in a OR operation but not negated
@@ -969,6 +963,54 @@ def update_field_references(cr, old, new, only_models=None, domain_adapter=None,
     return _update_field_usage_multi(cr, models, old, new, domain_adapter=domain_adapter, skip_inherit=skip_inherit)
 
 
+def _update_impex_renamed_fields_paths(cr, old_field_name, new_field_name, only_models):
+    export_q = cr.mogrify(
+        """
+        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_field_name)],
+    ).decode()
+    impex_data = [(export_q, "ir_exports_line", "name")]
+    if table_exists(cr, "base_import_mapping"):
+        import_q = cr.mogrify(
+            """
+            SELECT id,
+                   res_model,
+                   STRING_TO_ARRAY(field_name, '/')
+              FROM base_import_mapping
+             WHERE field_name ~ %s
+            """,
+            [r"\y{}\y".format(old_field_name)],
+        ).decode()
+        impex_data.append((import_q, "base_import_mapping", "field_name"))
+
+    for query, table, column in impex_data:
+        cr.execute(query)
+        if not cr.rowcount:
+            continue
+        fixed_paths = {}
+        for record_id, related_model, path in cr.fetchall():
+            new_path = [
+                new_field_name
+                if field.field_name == old_field_name and field.field_model in only_models
+                else field.field_name
+                for field in resolve_model_fields_path(cr, related_model, path)
+            ]
+            if len(new_path) == len(path) and new_path != path:
+                fixed_paths[record_id] = "/".join(new_path)
+        if fixed_paths:
+            cr.execute(
+                format_query(cr, "UPDATE {} SET {} = (%s::jsonb)->>(id::text) WHERE id IN %s", table, column),
+                [Json(fixed_paths), tuple(fixed_paths)],
+            )
+
+
 def _update_field_usage_multi(cr, models, old, new, domain_adapter=None, skip_inherit=()):
     assert models
     only_models = None if models == "*" else tuple(models)
@@ -1079,33 +1121,9 @@ 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
+        # ir.exports.line, base_import.mapping # noqa
         if only_models:
-            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)],
-                )
+            _update_impex_renamed_fields_paths(cr, old, new, only_models)
 
         # mail.alias
         if column_exists(cr, "mail_alias", "alias_defaults"):

src/util/helpers.py

@@ -317,27 +317,3 @@ def resolve_model_fields_path(cr, model, path):
         {"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/models.py

@@ -11,7 +11,7 @@
 
 from .const import ENVIRON
 from .fields import IMD_FIELD_PATTERN, remove_field
-from .helpers import _ir_values_value, _remove_export_lines, _validate_model, model_of_table, table_of_model
+from .helpers import _ir_values_value, _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
@@ -31,7 +31,7 @@
 
 # avoid namespace clash
 from .pg import rename_table as pg_rename_table
-from .records import _rm_refs, remove_records, remove_view, replace_record_references_batch
+from .records import _remove_import_export_paths, _rm_refs, remove_records, remove_view, replace_record_references_batch
 from .report import add_to_migration_reports
 
 _logger = logging.getLogger(__name__)
@@ -129,8 +129,7 @@ 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)
+    _remove_import_export_paths(cr, model)
 
     _rm_refs(cr, model)
 

src/util/records.py

@@ -23,7 +23,14 @@
 
 from .const import NEARLYWARN
 from .exceptions import MigrationError
-from .helpers import _get_theme_models, _ir_values_value, _validate_model, model_of_table, table_of_model
+from .helpers import (
+    _get_theme_models,
+    _ir_values_value,
+    _validate_model,
+    model_of_table,
+    resolve_model_fields_path,
+    table_of_model,
+)
 from .indirect_references import indirect_references
 from .inherit import direct_inherit_parents, for_each_inherit
 from .misc import parse_version, version_gte
@@ -474,6 +481,41 @@ def _rm_refs(cr, model, ids=None):
         )
 
 
+def _remove_import_export_paths(cr, model, field=None):
+    export_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:
+        export_q = cr.mogrify(export_q + " WHERE el.name ~ %s ", [r"\y{}\y".format(field)]).decode()
+
+    import_q = """
+        SELECT id,
+               res_model,
+               STRING_TO_ARRAY(field_name, '/')
+          FROM base_import_mapping
+        """
+    if field:
+        import_q = cr.mogrify(import_q + " WHERE field_name ~ %s ", [r"\y{}\y".format(field)]).decode()
+
+    for query, impex_model in [(export_q, "ir.exports.line"), (import_q, "base_import.mapping")]:
+        cr.execute(query)
+        to_rem = [
+            path_id
+            for path_id, related_model, 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, related_model, path)
+            )
+        ]
+        if to_rem:
+            remove_records(cr, impex_model, to_rem)
+
+
 def is_changed(cr, xmlid, interval="1 minute"):
     """
     Return whether a record was changed.