[18.0][IMP] storage_[file|media|image]: improve replace file wizard

storage_file/__manifest__.py

@@ -22,6 +22,7 @@
         "security/storage_file.xml",
         "data/ir_cron.xml",
         "data/storage_backend.xml",
+        "wizards/replace_file.xml",
         "wizards/swap_backend.xml",
         "data/ir_config_parameter.xml",
     ],

storage_file/models/storage_file.py

@@ -241,7 +241,7 @@ def _compute_extract_filename(self):
             rec.mimetype = mime
 
     def unlink(self):
-        if self._context.get("cleanning_storage_file"):
+        if self.env.context.get("force_delete_storage_file"):
             super().unlink()
         else:
             self.write({"to_delete": True, "active": False})
@@ -259,13 +259,22 @@ def _clean_storage_file(self, batch_size=1000):
         )
         ids = [x[0] for x in self._cr.fetchall()]
         recordset = self.browse(ids[:batch_size])
+        corrupted_ids = []
         for st_file in recordset:
+            if not st_file.relative_path:
+                corrupted_ids.append(st_file.id)
+                continue
             st_file.backend_id.sudo().delete(st_file.relative_path)
-            st_file.with_context(cleanning_storage_file=True).unlink()
+            st_file.with_context(force_delete_storage_file=True).unlink()
             # commit is required since the backend could be an external system
             # therefore, if the record is deleted on the external system
             # we must be sure that the record is also deleted into Odoo
             st_file._cr.commit()
+        if corrupted_ids:
+            _logger.error(
+                "Skipping corrupted files with no relative_path IDS=%s",
+                ",".join(map(str, corrupted_ids)),
+            )
         done = len(recordset)
         self.env["ir.cron"]._notify_progress(
             done=done, remaining=0 if done <= batch_size else len(ids) - done

storage_file/tests/test_storage_file.py

@@ -227,6 +227,24 @@ def test_unlink(self):
         self.assertEqual(len(files), 0)
         self.assertNotIn(relative_path, backend.list_files())
 
+    def test_clean_storage_file_skips_corrupted(self):
+        # Do not commit during the test
+        self.cr.commit = lambda: True
+        stfile = self._create_storage_file()
+        corrupted = self.env["storage.file"].create(
+            {"name": "corrupted.txt", "backend_id": stfile.backend_id.id}
+        )
+        self.assertFalse(corrupted.relative_path)
+        (stfile | corrupted).unlink()
+        with self.assertLogs(
+            "odoo.addons.storage_file.models.storage_file", level="ERROR"
+        ) as logs:
+            self.env["storage.file"]._clean_storage_file()
+        self.assertIn(str(corrupted.id), logs.output[0])
+        # The valid file is cleaned, the corrupted one is kept
+        self.assertFalse(stfile.exists())
+        self.assertTrue(corrupted.exists())
+
     def test_public_access1(self):
         """
         Test the public access (when is_public on the backend).

storage_file/wizards/replace_file.py

@@ -1,7 +1,7 @@
 # Copyright 2023 Camptocamp SA
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
 
-from odoo import fields, models
+from odoo import api, fields, models
 
 
 class StorageFileReplace(models.TransientModel):
@@ -11,6 +11,32 @@ class StorageFileReplace(models.TransientModel):
     file_id = fields.Many2one("storage.file")
     data = fields.Binary()
     file_name = fields.Char()
+    keep_history = fields.Boolean(
+        default=True,
+        help="Keep an archived copy of the record holding the replaced file.",
+    )
+    # Support for relations is up to the modules defining them
+    keep_relations = fields.Boolean(
+        default=True,
+        help="Link the archived copy to the same records.",
+    )
+    keep_relations_visible = fields.Boolean(
+        compute="_compute_keep_relations_visible",
+    )
+
+    @api.depends("keep_history")
+    def _compute_keep_relations_visible(self):
+        for rec in self:
+            rec.keep_relations_visible = (
+                rec.keep_history and rec._has_replaced_relations()
+            )
+
+    def _has_replaced_relations(self):
+        """Tell if the record owning the replaced file has relations to keep.
+
+        Hook for the modules defining relations to it.
+        """
+        return False
 
     def _get_file_from_data(self):
         file_model = self.env["storage.file"].sudo()
@@ -25,3 +51,61 @@ def _prepare_file_values(self):
 
     def confirm(self):
         return
+
+    def _replace_file(self, owner):
+        """Make ``owner`` point to a new file built from the wizard data.
+
+        Without ``keep_history``, the replaced file is dropped: nothing
+        refers to it anymore.
+
+        :param owner: record delegating to 'storage.file' via ``file_id``
+        :return: the archived copy of ``owner`` if ``keep_history`` is set
+        """
+        old_file = owner.file_id
+        archived = self.keep_history and self._copy_replaced(owner)
+        # Creating the file with its data already stores it on the backend
+        owner.file_id = self._get_file_from_data()
+        if archived:
+            # 'active' is the one of the file, shared with ``owner``
+            # until the replacement: archive it only now.
+            archived.active = False
+        else:
+            self._drop_replaced_file(old_file)
+        return archived
+
+    def _drop_replaced_file(self, old_file):
+        """Drop the replaced file, not owned by any record anymore."""
+        old_file = old_file.sudo()
+        # Forcing the deletion skips the cleanup cron, which also removes
+        # the file from the backend: do it here.
+        if old_file.relative_path:
+            old_file.backend_id.delete(old_file.relative_path)
+        old_file.with_context(force_delete_storage_file=True).unlink()
+
+    def _copy_replaced(self, owner):
+        """Copy ``owner``, keeping the file being replaced.
+
+        The copy is returned active: it is archived by ``_replace_file``
+        once ``owner`` points to the new file.
+        """
+        archived = owner.sudo().copy(self._copy_replaced_prepare_values(owner))
+        if self.keep_relations:
+            self._copy_replaced_relations(owner, archived)
+        return archived
+
+    def _copy_replaced_relations(self, owner, archived):
+        """Link ``archived`` to the same records as ``owner``.
+
+        Hook for the modules defining relations to ``owner``.
+        """
+        return
+
+    def _copy_replaced_prepare_values(self, owner):
+        old_file = owner.file_id
+        return {
+            # Reuse the file instead of duplicating it
+            "file_id": old_file.id,
+            # Otherwise the creation sets the default backend of the owner
+            # and writes it on the reused file, which moves it.
+            "backend_id": old_file.backend_id.id,
+        }

storage_file/wizards/replace_file.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<!-- Copyright 2026 Camptocamp SA
+     License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -->
+<odoo>
+    <record id="storage_file_replace_view_form" model="ir.ui.view">
+        <field name="name">storage.file.replace.form</field>
+        <field name="model">storage.file.replace</field>
+        <field name="arch" type="xml">
+            <form string="Replace File">
+                <sheet>
+                    <field name="file_id" invisible="1" />
+                    <group string="File" name="file">
+                        <field name="data" filename="file_name" />
+                    </group>
+                    <group string="Name" name="name">
+                        <field name="file_name" />
+                    </group>
+                    <group name="options">
+                        <field name="keep_history" />
+                        <field name="keep_relations_visible" invisible="1" />
+                        <field
+                            name="keep_relations"
+                            invisible="not keep_relations_visible"
+                        />
+                    </group>
+                    <footer>
+                        <button name="confirm" string="Confirm" type="object" />
+                        <button special="cancel" string="Cancel" class="btn-default" />
+                    </footer>
+                </sheet>
+            </form>
+        </field>
+    </record>
+</odoo>

storage_file_swap_backend_queue/tests/test_swap_backend_queue.py

@@ -111,7 +111,7 @@ def test_job_handles_missing_record(self):
         """Deleted records between enqueue and execution are reported."""
         stfile = self._create_storage_file(data=b"payload")
         file_id = stfile.id
-        stfile.with_context(cleanning_storage_file=True).unlink()
+        stfile.with_context(force_delete_storage_file=True).unlink()
         records = self.env["storage.file"].browse(file_id)
         result = records._swap_backend_job(self.backend_b.id)
         self.assertIn("no longer exists", result)

storage_image/tests/test_storage_replace_file.py

@@ -58,3 +58,48 @@ def test_wizard_change_file(self):
         self.assertEqual(image.file_id.name, self.filename_1)
         self.assertEqual(image.file_id.data, self.filedata_1)
         self.assertEqual(image.file_id.file_type, "image")
+
+    def test_wizard_change_file_keep_history(self):
+        image = self._create_storage_image(self.filename_1, self.filedata_1)
+        old_file = image.file_id
+        wiz_form = Form(
+            self.env["storage.file.replace"].with_context(
+                **{"active_model": "storage.image", "active_id": image.id}
+            ),
+            view="storage_image.storage_file_replace_view_form",
+        )
+        wiz_form.file_name = self.filename_2
+        wiz_form.data = self.filedata_2
+        wiz_form.keep_history = True
+        wiz_form.save().confirm()
+        # The image holds the new file and stays active
+        self.assertTrue(image.active)
+        self.assertEqual(image.file_id.data, self.filedata_2)
+        # An archived copy holds the old file, untouched
+        archived_image = self.env["storage.image"].search(
+            [("file_id", "=", old_file.id), ("active", "=", False)]
+        )
+        self.assertEqual(len(archived_image), 1)
+        self.assertEqual(archived_image.data, self.filedata_1)
+        self.assertEqual(archived_image.backend_id, old_file.backend_id)
+        self.assertFalse(old_file.to_delete)
+
+    def test_wizard_change_file_no_history(self):
+        image = self._create_storage_image(self.filename_1, self.filedata_1)
+        old_file = image.file_id
+        backend, relative_path = old_file.backend_id, old_file.relative_path
+        wiz_form = Form(
+            self.env["storage.file.replace"].with_context(
+                **{"active_model": "storage.image", "active_id": image.id}
+            ),
+            view="storage_image.storage_file_replace_view_form",
+        )
+        wiz_form.file_name = self.filename_2
+        wiz_form.data = self.filedata_2
+        wiz_form.keep_history = False
+        wiz_form.save().confirm()
+        self.assertTrue(image.active)
+        self.assertEqual(image.file_id.data, self.filedata_2)
+        # Nothing refers to the old file anymore: it is deleted
+        self.assertFalse(old_file.exists())
+        self.assertNotIn(relative_path, backend.list_files())

storage_image/wizards/replace_file.py

@@ -34,5 +34,5 @@ def _prepare_file_values(self):
     def confirm(self):
         res = super().confirm()
         if self.image_id and self.data:
-            self.image_id.file_id = self._get_file_from_data()
+            self._replace_file(self.image_id)
         return res

storage_image/wizards/replace_file.xml

@@ -3,31 +3,21 @@
      License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -->
 <odoo>
     <record id="storage_file_replace_view_form" model="ir.ui.view">
-        <field name="name">storage.file.replace.form</field>
         <field name="model">storage.file.replace</field>
+        <field name="inherit_id" ref="storage_file.storage_file_replace_view_form" />
+        <field name="mode">primary</field>
         <field name="arch" type="xml">
-            <form string="Replace File" version="7.0">
-                <sheet>
-                    <field name="file_id" invisible="1" />
-                    <field name="image_id" invisible="1" />
-                    <group string="Image" name="image">
-                        <field
-                            name="data"
-                            widget="image"
-                            filename="file_name"
-                            nolabel="1"
-                            options="{'size':(128,128)}"
-                        />
-                    </group>
-                    <group string="Name" name="name">
-                        <field name="file_name" />
-                    </group>
-                    <footer>
-                        <button name="confirm" string="Confirm" type="object" />
-                        <button special="cancel" string="Cancel" class="btn-default" />
-                    </footer>
-                </sheet>
-            </form>
+            <field name="file_id" position="after">
+                <field name="image_id" invisible="1" />
+            </field>
+            <group name="file" position="attributes">
+                <attribute name="string">Image</attribute>
+            </group>
+            <field name="data" position="attributes">
+                <attribute name="widget">image</attribute>
+                <attribute name="nolabel">1</attribute>
+                <attribute name="options">{'size': (128, 128)}</attribute>
+            </field>
         </field>
     </record>
 

storage_image_product/__init__.py

@@ -1 +1,2 @@
 from . import models
+from . import wizards

storage_image_product/tests/__init__.py

@@ -1 +1,2 @@
 from . import test_product_image_relation
+from . import test_replace_file

storage_image_product/tests/test_replace_file.py

@@ -0,0 +1,77 @@
+# Copyright 2026 Camptocamp SA
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+from odoo.tests import Form
+
+from .common import ProductImageCaseMixin, TransactionComponentCase
+
+
+class TestReplaceFile(TransactionComponentCase, ProductImageCaseMixin):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.image = cls._create_storage_image("logo-image.jpg")
+        cls.template = cls.env["product.template"].create({"name": "Image product"})
+        cls.category = cls.env["product.category"].create({"name": "Image categ"})
+        cls.product_relation = cls.env["product.image.relation"].create(
+            {"image_id": cls.image.id, "product_tmpl_id": cls.template.id}
+        )
+        cls.category_relation = cls.env["category.image.relation"].create(
+            {"image_id": cls.image.id, "category_id": cls.category.id}
+        )
+
+    def _replace_image_file(self, keep_history, keep_relations=False):
+        wiz_form = Form(
+            self.env["storage.file.replace"].with_context(
+                active_model="storage.image", active_id=self.image.id
+            ),
+            view="storage_image.storage_file_replace_view_form",
+        )
+        wiz_form.file_name = "white-image.jpg"
+        wiz_form.data = self._get_file_content("white-image.jpg")
+        wiz_form.keep_history = keep_history
+        if keep_history:
+            wiz_form.keep_relations = keep_relations
+        wiz_form.save().confirm()
+
+    def _relations(self, model):
+        images = self.image | self._archived_image()
+        return (
+            self.env[model]
+            .with_context(active_test=False)
+            .search([("image_id", "in", images.ids)])
+        )
+
+    def _archived_image(self):
+        return self.env["storage.image"].search(
+            [("file_id", "=", self.old_file.id), ("active", "=", False)]
+        )
+
+    def test_replace_keep_history_and_relations(self):
+        self.old_file = self.image.file_id
+        self._replace_image_file(keep_history=True, keep_relations=True)
+        archived_image = self._archived_image()
+        self.assertEqual(len(archived_image), 1)
+        self.assertTrue(self.image.active)
+        self.assertNotEqual(self.image.file_id, self.old_file)
+        for model, current in (
+            ("product.image.relation", self.product_relation),
+            ("category.image.relation", self.category_relation),
+        ):
+            archived_relation = self._relations(model) - current
+            self.assertEqual(archived_relation.image_id, archived_image)
+            self.assertFalse(archived_relation.active)
+        # The relations of the archived image are hidden on the product
+        self.assertEqual(self.template.image_ids, self.product_relation)
+        self.assertEqual(self.template.main_image_id, self.image)
+
+    def test_replace_keep_history_only(self):
+        self.old_file = self.image.file_id
+        self._replace_image_file(keep_history=True)
+        self.assertEqual(len(self._archived_image()), 1)
+        self.assertEqual(
+            self._relations("product.image.relation"), self.product_relation
+        )
+        self.assertEqual(
+            self._relations("category.image.relation"), self.category_relation
+        )