[19.0][FIX] fs_attachment: don't upload files forced to DB

fs_attachment/models/ir_attachment.py

@@ -237,6 +237,10 @@ def _get_datas_related_values(self, data, mimetype):
                         "db_datas": data,
                     }
                 )
+                # see _is_forced_to_db_and_unreferenced
+                self.env.cr.cache.setdefault(
+                    "fs_attachment_forced_to_db_checksums", set()
+                ).add(values["checksum"])
             else:
                 # Uses the full object storage path; standard Odoo uses a relative path.
                 path = self._get_fs_path(storage, data)
@@ -358,11 +362,40 @@ def _file_read(self, fname, size=None):
     def _file_write(self, bin_data, checksum):
         location = self.env.context.get("storage_location") or self._storage()
         if location in self._get_storage_codes():
+            if self._is_forced_to_db_and_unreferenced(location, bin_data, checksum):
+                return False
             filename = self._storage_file_write(bin_data)
         else:
             filename = super()._file_write(bin_data, checksum)
         return filename
 
+    @api.model
+    def _is_forced_to_db_and_unreferenced(self, location, bin_data, checksum):
+        """Return whether writing the file into the storage can be skipped
+
+        Since Odoo 19, ``create`` and ``_set_attachment_data`` write the file
+        after ``_get_datas_related_values`` for every attachment when the
+        storage is not 'db', including the ones this module forces to the
+        database. Uploading them only leaves an orphan file in the storage
+        and fails when the storage is unreachable.
+
+        The write is skipped only for a content forced to the database in
+        this transaction that no attachment references, so a file shared with
+        an attachment kept in the storage is still written, as are the files
+        written for another purpose (e.g. ``AttachmentFileLikeAdapter``).
+        """
+        if checksum not in self.env.cr.cache.get(
+            "fs_attachment_forced_to_db_checksums", ()
+        ):
+            return False
+        store_fname = f"{location}://{self._get_fs_path(location, bin_data)}"
+        self.flush_model(["store_fname"])
+        self.env.cr.execute(
+            "SELECT 1 FROM ir_attachment WHERE store_fname = %s LIMIT 1",
+            (store_fname,),
+        )
+        return not self.env.cr.fetchone()
+
     @api.model
     def _file_delete(self, fname) -> None:  # pylint: disable=missing-return
         if self._is_file_from_a_storage(fname):

fs_attachment/readme/newsfragments/forced_db_upload.bugfix

@@ -0,0 +1,3 @@
+Attachments forced to the database by `force_db_for_default_attachment_rules`
+are no longer also written into the storage, which left an orphan file and
+failed when the storage was unreachable.

fs_attachment/tests/test_fs_attachment.py

@@ -402,6 +402,68 @@ def test_force_storage_to_fs_ignore_missing_file(self):
         # the file should not have been moved
         self.assertEqual(attachment.store_fname, store_fname, "store_fname not changed")
 
+    def test_force_attachment_in_db_rules_no_file_written(self):
+        self.temp_backend.use_as_default_for_attachments = True
+        self.temp_backend.force_db_for_default_attachment_rules = '{"text/plain": 0}'
+        attachment = self.ir_attachment_model.create(
+            {"name": "test.txt", "raw": b"content"}
+        )
+        attachment.write({"raw": b"new content"})
+        self.env.flush_all()
+        self.assertFalse(attachment.store_fname)
+        self.assertEqual(attachment.db_datas, b"new content")
+        # nothing uploaded to the storage, nothing left for the gc
+        self.assertEqual(os.listdir(self.temp_dir), [])
+        self.assertFalse(
+            self.gc_file_model.search([("fs_storage_code", "=", "tmp_dir")])
+        )
+
+    def test_force_attachment_in_db_rules_storage_unreachable(self):
+        self.temp_backend.use_as_default_for_attachments = True
+        self.temp_backend.force_db_for_default_attachment_rules = '{"text/plain": 0}'
+        with mock.patch.object(
+            type(self.ir_attachment_model),
+            "_storage_file_write",
+            side_effect=MyException("storage unreachable"),
+        ):
+            attachment = self.ir_attachment_model.create(
+                {"name": "test.txt", "raw": b"content"}
+            )
+            self.env.flush_all()
+        self.assertEqual(attachment.db_datas, b"content")
+
+    def test_force_attachment_in_db_rules_shared_content(self):
+        self.temp_backend.use_as_default_for_attachments = True
+        self.temp_backend.force_db_for_default_attachment_rules = '{"text/plain": 0}'
+        in_db = self.ir_attachment_model.create({"name": "a.txt", "raw": b"content"})
+        # same content in the same transaction, not forced to the database:
+        # the file must still be written
+        self.temp_backend.force_db_for_default_attachment_rules = "{}"
+        in_storage = self.ir_attachment_model.create(
+            {"name": "b.txt", "raw": b"content"}
+        )
+        self.env.flush_all()
+        self.assertFalse(in_db.store_fname)
+        self.assertTrue(in_storage.store_fname.startswith("tmp_dir://"))
+        self.assertEqual(os.listdir(self.temp_dir), [in_storage.fs_filename])
+        self.env.invalidate_all()
+        self.assertEqual(in_storage.raw, b"content")
+
+    def test_force_attachment_in_db_rules_content_already_in_storage(self):
+        self.temp_backend.use_as_default_for_attachments = True
+        self.temp_backend.force_db_for_default_attachment_rules = "{}"
+        in_storage = self.ir_attachment_model.create(
+            {"name": "a.txt", "raw": b"content"}
+        )
+        self.env.flush_all()
+        self.temp_backend.force_db_for_default_attachment_rules = '{"text/plain": 0}'
+        in_db = self.ir_attachment_model.create({"name": "b.txt", "raw": b"content"})
+        self.env.flush_all()
+        self.assertFalse(in_db.store_fname)
+        self.assertEqual(os.listdir(self.temp_dir), [in_storage.fs_filename])
+        self.env.invalidate_all()
+        self.assertEqual(in_storage.raw, b"content")
+
     def test_storage_use_filename_obfuscation(self):
         self.temp_backend.base_url = "https://acsone.eu/media"
         self.temp_backend.use_as_default_for_attachments = True