Modify form: make loading content from file upload work

See the comments in #1974

Author: roland-ruedenauer

src/moin/items/__init__.py

@@ -1591,10 +1591,22 @@ def do_modify(self, item_may=None):
 
             form = self.ModifyForm.from_request(request)
             meta, data, contenttype_guessed, comment = form._dump(self)
-            if contenttype_guessed:
-                m = re.search("charset=(.+?)($|;)", contenttype_guessed)
-                if m:
-                    data = str(data, m.group(1))
+            if data is not None:
+                if not isinstance(data, (str, bytes)):
+                    data = data.read()
+                if isinstance(data, bytes):
+                    encoding = "utf-8"
+                    if contenttype_guessed:
+                        if m := re.search("charset=(.+?)($|;)", contenttype_guessed):
+                            encoding = m.group(1)
+                    try:
+                        data = str(data, encoding)
+                    except ValueError:
+                        flash(_("Invalid data content received. Expecting text content."), "error")
+                        data = ""
+                if not isinstance(data, str):
+                    abort(422, description="invalid content")
+
             state = dict(fqname=self.fqname, itemid=meta.get(ITEMID), meta=meta)
             if form.validate(state):
                 if request.values.get("preview"):
@@ -1609,6 +1621,9 @@ def do_modify(self, item_may=None):
                     else:  # TODO: make preview button inactive for empty items, see #1539
                         flash(_("No preview available for empty items."), "error")
                     close_file(old_item.rev.data)
+                    # update content form text data if data originated from a file upload
+                    if data and form["content_form"]["data_file"]:
+                        form["content_form"]["data_text"] = data
                 else:
                     # user clicked OK/Save button, check for conflicts,
                     if "charset" in self.contenttype:

src/moin/utils/edit_locking.py

@@ -30,6 +30,8 @@
 To stress-test edit locking, use the Locust-based tests in /contrib/loadtesting/.
 """
 
+from __future__ import annotations
+
 import os
 import time
 import sqlite3
@@ -145,7 +147,7 @@ def get_user_name(self):
             user_name = request.remote_user or request.remote_addr
         return user_name
 
-    def put_draft(self, data_in, overwrite=True):
+    def put_draft(self, data_in: str | None, overwrite: bool = True) -> None:
         """
         Only 1 item draft is saved per user. Most recent item draft overlays prior item draft.
 
@@ -159,7 +161,7 @@ def put_draft(self, data_in, overwrite=True):
         """
         rev_id = self.rev_id
         draft_rev_number = self.rev_number
-        draft, data = self.get_draft()
+        draft, _ = self.get_draft()
         if draft:
             # draft may be of this item or remnant of a prior abandoned edit
             u_name, i_id, i_name, rev_number, save_time, i_rev_id = draft
@@ -170,22 +172,24 @@ def put_draft(self, data_in, overwrite=True):
                 draft_rev_number = rev_number  # XXX per line 1074 in __init__
                 rev_id = i_rev_id
                 self.draft_name = self.make_draft_name(rev_id)
-            self.cursor.execute("""DELETE FROM editdraft WHERE user_name = ? """, (self.user_name,))
+            self.cursor.execute("""DELETE FROM editdraft WHERE user_name = ?""", (self.user_name,))
+
         if data_in:
-            data_in = data_in.encode(self.coding)
+            data = data_in.encode(self.coding)
             with open(self.draft_name, "wb") as f:
-                f.write(data_in)
+                f.write(data)
             save_time = int(time.time())
         else:
             save_time = 0  # indicates user is editing item but has not done a preview, no draft has been saved
+
         self.cursor.execute(
             """INSERT INTO editdraft(user_name, item_id, item_name, rev_number, save_time, rev_id)
                            VALUES(?,?,?,?,?,?)""",
             (self.user_name, self.item_id, self.item_name, draft_rev_number, save_time, rev_id),
         )
         self.conn.commit()
 
-    def get_draft(self):
+    def get_draft(self) -> tuple[sqlite3.Row, str] | tuple[sqlite3.Row, None] | tuple[None, None]:
         """
         Return None, None if no draft available; else tuple of row fields, textarea data or None.
 
@@ -205,8 +209,8 @@ def get_draft(self):
                     try:
                         with open(self.draft_name, "rb") as f:
                             data = f.read()
-                        data = data.decode(self.coding)
-                        return draft, data
+                        content = data.decode(self.coding)
+                        return draft, content
                     except OSError:
                         logging.error(f"User {u_name} failed to load draft for: {i_name}")
                         return draft, None