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

andreabak · andreabak · commit 045f760fb65f · 2024-06-14T15:49:13.000+02:00
Implement proper fixing and/or removal of ir.exports and
ir.exports.line records on model/fields renames/removal:
- renamed fields: update affected ir.exports.line records
- removed fields: remove affected ir.exports.line records
- renamed models: update affected ir.exports records `resource`
- removed models: remove affected ir.exports.line / ir.exports records

Some of these cases were already handled but have been improved.

Specifically:
- for renamed fields, previously was done by doing a simple string
  replacement, which is not good enough because export lines can
  reference fields paths, and these in turn *might* contain paths to
  multiple fields with the same name on different models, only some of
  which are affected. Now the fields path get properly resolved for
  renaming, only the affected fields path parts are updated.
- for removed fields, previously no action was taken, leaving a broken
  export template and UI traceback. Now the affected export lines are
  removed, using fields paths resolution.
- for renamed and removed models, this was already handled by the
  indirect_references mechanism, but now export lines for the removed
  models are checked with the same mechanism for fields above.

Additionally, unit tests have been added to make sure these cases are
properly handled by the code.
diff --git a/src/base/tests/test_util.py b/src/base/tests/test_util.py
@@ -390,6 +390,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
diff --git a/src/util/fields.py b/src/util/fields.py
@@ -16,7 +16,7 @@
 
 import psycopg2
 from psycopg2 import sql
-from psycopg2.extras import Json
+from psycopg2.extras import Json, execute_values
 
 try:
     from odoo import release
@@ -40,7 +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, _validate_model, 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
@@ -79,6 +79,126 @@ def make_index_name(table_name, column_name):
 )
 
 
+def _get_resolved_ir_exports(cr, models=None, fields=None):
+    """
+    Return a list of ir.exports.line records which models or fields match the given arguments.
+
+    Export lines can reference nested models through relationship field "paths"
+    (e.g. "partner_id/country_id/name"), therefore these needs to be resolved properly.
+
+    Only one of ``models`` or ``fields`` arguments should be provided.
+
+    :param list[str] models: a list of model names to match in exports
+    :param list[(str, str)] fields: a list of (model, field) tuples to match in exports
+    :return: the resolved field paths parts for each matched export line id
+    :rtype: dict[int, list[FieldsPathPart]]
+
+    :meta private: exclude from online docs
+    """
+    assert bool(models) ^ bool(fields), "One of models or fields must be given, and not both."
+
+    # Get the model fields paths for exports.
+    # When matching fields we can already broadly filter on field names (will be double-checked later).
+    # When matching models we can't exclude anything because we don't know intermediate models.
+    where = ""
+    params = {}
+    if fields:
+        fields = {(model, fields) for model, fields in fields}  # noqa: C416  # make sure set[tuple]
+        where = "WHERE el.name ~ ANY(%(field_names)s)"
+        params["field_names"] = [f[1] for f in fields]
+    cr.execute(
+        """
+        SELECT el.id, e.resource AS model, string_to_array(el.name, '/') AS path
+          FROM ir_exports e
+          JOIN ir_exports_line el ON e.id = el.export_id
+        {where}
+        """.format(where=where),
+        params,
+    )
+    paths_to_line_ids = {}
+    for line_id, model, path in cr.fetchall():
+        paths_to_line_ids.setdefault((model, tuple(path)), set()).add(line_id)
+
+    # Resolve intermediate models for all model fields paths, filter only matching paths parts
+    matching_paths_parts = {}
+    for model, path in paths_to_line_ids:
+        resolved_paths = resolve_model_fields_path(cr, model, path)
+        if fields:
+            matching_parts = [p for p in resolved_paths if (p.field_model, p.field_name) in fields]
+        else:
+            matching_parts = [p for p in resolved_paths if p.field_model in models]
+        if not matching_parts:
+            continue
+        matching_paths_parts[(model, path)] = matching_parts
+
+    # Return the matched parts for each export line id
+    result = {}
+    for (model, path), matching_parts in matching_paths_parts.items():
+        line_ids = paths_to_line_ids.get((model, path))
+        if not line_ids:
+            continue  # wut?
+        for line_id in line_ids:
+            result.setdefault(line_id, []).extend(matching_parts)
+    return result
+
+
+def rename_ir_exports_fields(cr, models_fields_map):
+    """
+    Rename fields references in ir.exports.line records.
+
+    :param dict[str, dict[str, str]] models_fields_map: a dict of models to the fields rename dict,
+        like: `{"model.name": {"old_field": "new_field", ...}, ...}`
+
+    :meta private: exclude from online docs
+    """
+    matching_exports = _get_resolved_ir_exports(
+        cr,
+        fields=[(model, field) for model, fields_map in models_fields_map.items() for field in fields_map],
+    )
+    if not matching_exports:
+        return
+    _logger.debug("Renaming %d export template lines with renamed fields", len(matching_exports))
+    fixed_lines_paths = {}
+    for line_id, resolved_paths in matching_exports.items():
+        for path_part in resolved_paths:
+            assert path_part.field_model in models_fields_map
+            fields_map = models_fields_map[path_part.field_model]
+            assert path_part.field_name in fields_map
+            assert path_part.path[path_part.part_index - 1] == path_part.field_name
+            new_field_name = fields_map[path_part.field_name]
+            fixed_path = fixed_lines_paths.get(line_id, list(path_part.path))
+            fixed_path[path_part.part_index - 1] = new_field_name
+            fixed_lines_paths[line_id] = fixed_path
+    execute_values(
+        cr,
+        """
+        UPDATE ir_exports_line el
+           SET name = v.name
+          FROM (VALUES %s) AS v(id, name)
+         WHERE el.id = v.id
+        """,
+        [(k, "/".join(v)) for k, v in fixed_lines_paths.items()],
+    )
+
+
+def remove_ir_exports_lines(cr, models=None, fields=None):
+    """
+    Delete ir.exports.line records that reference models or fields that are/will be removed.
+
+    Only one of ``models`` or ``fields`` arguments should be provided.
+
+    :param list[str] models: a list of model names to match in exports
+    :param list[(str, str)] fields: a list of (model, field) tuples to match in exports
+
+    :meta private: exclude from online docs
+    """
+    matching_exports = _get_resolved_ir_exports(cr, models=models, fields=fields)
+    if not matching_exports:
+        return
+    _logger.debug("Deleting %d export template lines with removed models/fields", len(matching_exports))
+    cr.execute("DELETE FROM ir_exports_line WHERE id IN %s", [tuple(matching_exports.keys())])
+
+
 def ensure_m2o_func_field_data(cr, src_table, column, dst_table):
     """
     Fix broken m2o relations.
@@ -202,6 +322,9 @@ def clean_context(context):
             [(fieldname, fieldname + " desc"), model, r"\y{}\y".format(fieldname)],
         )
 
+    # ir.exports
+    remove_ir_exports_lines(cr, fields=[(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:
@@ -1070,22 +1193,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
-        q = """
-            UPDATE ir_exports_line l
-               SET name = regexp_replace(l.name, %(old)s, %(new)s, 'g')
-        """
+        # ir.exports
         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)
+            rename_ir_exports_fields(cr, {model: {old: new} for model in only_models})
 
         # mail.alias
         if column_exists(cr, "mail_alias", "alias_defaults"):
diff --git a/src/util/indirect_references.py b/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),
diff --git a/src/util/models.py b/src/util/models.py
@@ -10,7 +10,7 @@
 import re
 
 from .const import ENVIRON
-from .fields import IMD_FIELD_PATTERN, remove_field
+from .fields import IMD_FIELD_PATTERN, remove_field, remove_ir_exports_lines
 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
@@ -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_ir_exports_lines(cr, models=[model])
+
     _rm_refs(cr, model)
 
     cr.execute(
