feat(file): report upload failures clearly and verify before storing

src/osw/controller/file/wiki.py

@@ -2,13 +2,84 @@
 import os
 from typing import IO, Any, Dict, List, Optional
 
+import mwclient.errors
+
 from osw.controller.file.base import FileController
 from osw.controller.file.remote import RemoteFileController
 from osw.core import OSW, model
 from osw.utils.wiki import get_namespace, get_title
 from osw.wtsite import WtSite
 
 
+def get_allowed_file_extensions(mw_site) -> Optional[List[str]]:
+    """Queries the file extensions the wiki accepts, None if unavailable"""
+    try:
+        result = mw_site.api(
+            "query", meta="siteinfo", siprop="fileextensions", formatversion=2
+        )
+    except Exception:
+        # only used to enrich an error message, never worth failing over
+        return None
+    extensions = result.get("query", {}).get("fileextensions", [])
+    return [entry["ext"] for entry in extensions if "ext" in entry]
+
+
+def reraise_upload_error(
+    error: mwclient.errors.APIError, mw_site, title: str, suffix: Optional[str]
+) -> None:
+    """Turns a rejected file extension into a readable error, re-raises the rest
+
+    MediaWiki reports a rejected extension as filetype-banned,
+    filetype-banned-type or filetype-badtype, depending on the version.
+    """
+    if "filetype" not in str(getattr(error, "code", "")):
+        raise error
+    allowed = get_allowed_file_extensions(mw_site)
+    hint = (
+        f" Extensions allowed on this wiki: {', '.join(sorted(allowed))}."
+        if allowed
+        else ""
+    )
+    raise ValueError(
+        f"Upload of '{title}' was rejected because the file extension "
+        f"'{suffix}' is not allowed on {mw_site.host}.{hint}"
+    ) from error
+
+
+def assert_upload_success(result: Any, title: str, host: str) -> None:
+    """Raises unless the upload API reports that it stored the file
+
+    mwclient raises only when the response carries an 'error' key. A file that
+    MediaWiki declined for any other reason comes back as a normal return value
+    with a result other than 'Success', so without this check the upload would
+    fail silently.
+    """
+    status = result.get("result") if isinstance(result, dict) else None
+    if status == "Success":
+        return
+    warnings = result.get("warnings") if isinstance(result, dict) else None
+    detail = f" Warnings: {warnings}." if warnings else ""
+    raise ValueError(
+        f"Upload of '{title}' to {host} did not succeed (result: {status}).{detail}"
+    )
+
+
+def store_params_for_upload(
+    se_params: Dict[str, Any], page_existed: bool
+) -> Dict[str, Any]:
+    """Picks the overwrite policy for the entity that accompanies an upload
+
+    An upload creates the file page when it was not there yet. store_entity
+    would then see an existing page and, under the default 'keep existing',
+    leave the metadata unwritten. So the entity has to replace what the upload
+    put there. A page that was already there keeps whatever the caller asked
+    for.
+    """
+    if page_existed:
+        return se_params
+    return {**se_params, "overwrite": "replace remote"}
+
+
 class WikiFileController(model.WikiFile, RemoteFileController):
     """File controller for wiki files"""
 
@@ -131,20 +202,29 @@ def put(self, file: IO, **kwargs: Dict[str, Any]):
         }
         for key in ["entities", "namespace"]:
             se_params.pop(key, None)  # avoid duplicated kwargs
+        # Upload before storing the entity: MediaWiki offers no transaction
+        # across the two writes, so one of them can be left standing. A file
+        # page without metadata is visibly incomplete, while metadata without a
+        # file looks like a valid entity until someone tries to download it.
+        page_existed = self.osw.mw_site.pages[f"{self.namespace}:{self.title}"].exists
+        try:
+            result = self.osw.mw_site.upload(
+                file=file,
+                filename=self.title,
+                # comment="",
+                # description="",
+                ignore=True,
+            )
+        except mwclient.errors.APIError as e:
+            reraise_upload_error(e, self.osw.mw_site, self.title, self.suffix)
+        assert_upload_success(result, self.title, self.osw.mw_site.host)
         self.osw.store_entity(
             OSW.StoreEntityParam(
                 entities=[self.cast(model.WikiFile, **wf_params)],
                 namespace=self.namespace,
-                **se_params,
+                **store_params_for_upload(se_params, page_existed),
             )
         )
-        self.osw.mw_site.upload(
-            file=file,
-            filename=self.title,
-            # comment="",
-            # description="",
-            ignore=True,
-        )
 
     def put_from(self, other: FileController, **kwargs: Dict[str, Any]):
         # if isinstance(file, LocalFileController) and self.suffix is None:

src/osw/core.py

@@ -59,12 +59,14 @@
     get_full_title,
     get_namespace,
     get_title,
-    get_uuid,
     is_empty,
     namespace_from_full_title,
     remove_empty,
     title_from_full_title,
 )
+from osw.utils.wiki import (
+    get_uuid as get_uuid_from_osw_id,
+)
 from osw.wiki_tools import SearchParam
 from osw.wtsite import WtPage, WtSite
 
@@ -197,18 +199,21 @@ def get_osw_id(uuid: Union[str, UUID]) -> str:
 
     @staticmethod
     def get_uuid(osw_id: str) -> UUID:
-        """Returns the uuid for a given OSW-ID
+        """Returns the uuid for a given OSW-ID. Kept for backwards compatibility,
+        the implementation lives in osw.utils.wiki.get_uuid()
 
         Parameters
         ----------
         osw_id
-            OSW-ID string, e.g. OSW2ea5b605c91f4e5a95593dff79fdd4a5
+            OSW-ID string, e.g. OSW2ea5b605c91f4e5a95593dff79fdd4a5, with or
+            without file suffixes, e.g.
+            OSW2ea5b605c91f4e5a95593dff79fdd4a5.drawio.png
 
         Returns
         -------
             uuid object, e.g. UUID("2ea5b605-c91f-4e5a-9559-3dff79fdd4a5")
         """
-        return UUID(osw_id.replace("OSW", ""))
+        return get_uuid_from_osw_id(osw_id)
 
     class SortEntitiesResult(OswBaseModel):
         by_name: Dict[str, List[OswBaseModel]]
@@ -1348,7 +1353,7 @@ def validate_entity(cls, entity, values):
             if jsondata is None:  # Guard clause
                 title = title_from_full_title(page.title)
                 try:
-                    uuid_from_title = get_uuid(title)
+                    uuid_from_title = get_uuid_from_osw_id(title)
                 except ValueError:
                     print(
                         f"Error: UUID could not be determined from title: '{title}', "

src/osw/utils/wiki.py

@@ -1,9 +1,18 @@
+import re
 from copy import deepcopy
 from uuid import UUID
 
 # Legacy imports:
 from opensemantic.v1 import get_full_title, get_namespace, get_title  # noqa: F401
 
+OSW_ID_PATTERN = re.compile(
+    r"^(?:OSW)?"  # the prefix, absent when a bare uuid is passed
+    r"([0-9a-f]{8}-?[0-9a-f]{4}-?[0-9a-f]{4}-?[0-9a-f]{4}-?[0-9a-f]{12})"
+    r"(?:\.[\w-]+)*$",  # file suffixes, e.g. '.png' or '.drawio.png'
+    re.IGNORECASE,
+)
+"""Matches an OSW-ID with an optional prefix and any number of file suffixes"""
+
 
 def get_osw_id(uuid: UUID) -> str:
     """Generates a OSW-ID based on the given uuid by prefixing "OSW" and removing
@@ -21,19 +30,33 @@ def get_osw_id(uuid: UUID) -> str:
     return "OSW" + str(uuid).replace("-", "")
 
 
-def get_uuid(osw_id) -> UUID:
-    """Returns the uuid for a given OSW-ID. Duplicate of OSW.get_uuid() from src/sw/core/osw.py
+def get_uuid(osw_id: str) -> UUID:
+    """Returns the uuid for a given OSW-ID. The single implementation, wrapped by
+    OSW.get_uuid() from src/osw/core.py
+
+    A file page keeps its file extension in the title, so the OSW-ID of a file is
+    followed by one or more suffixes that are not part of the uuid. These are
+    ignored, as is the OSW prefix.
 
     Parameters
     ----------
     osw_id
-        OSW-ID string, e.g., OSW2ea5b605c91f4e5a95593dff79fdd4a5
+        OSW-ID string, e.g., OSW2ea5b605c91f4e5a95593dff79fdd4a5, with or without
+        file suffixes, e.g., OSW2ea5b605c91f4e5a95593dff79fdd4a5.drawio.png
 
     Returns
     -------
         uuid object, e.g., UUID("2ea5b605-c91f-4e5a-9559-3dff79fdd4a5")
+
+    Raises
+    ------
+    ValueError
+        If no OSW-ID can be read from the given string.
     """
-    return UUID(osw_id.replace("OSW", ""))
+    match = OSW_ID_PATTERN.match(osw_id)
+    if match is None:
+        raise ValueError(f"No OSW-ID could be read from '{osw_id}'")
+    return UUID(match.group(1))
 
 
 def namespace_from_full_title(full_title: str) -> str:

tests/test_wiki_file_upload_errors.py

@@ -0,0 +1,140 @@
+"""Unit tests for upload error reporting in osw.controller.file.wiki.
+
+Regression guard for #51: a file rejected because of its extension must produce
+an error that names the extension, instead of a bare MediaWiki API code.
+
+Also guards the silent-failure path: mwclient raises only on an 'error' key, so
+an upload MediaWiki declined by other means must be caught by inspecting the
+returned result.
+"""
+
+import mwclient.errors
+import pytest
+
+from osw.controller.file.wiki import (
+    assert_upload_success,
+    get_allowed_file_extensions,
+    reraise_upload_error,
+    store_params_for_upload,
+)
+
+
+class _FakeSite:
+    host = "wiki.example.org"
+
+    def __init__(self, extensions=None, fail=False):
+        self._extensions = extensions
+        self._fail = fail
+
+    def api(self, *args, **kwargs):
+        if self._fail:
+            raise RuntimeError("siteinfo unavailable")
+        return {"query": {"fileextensions": [{"ext": e} for e in self._extensions]}}
+
+
+def _api_error(code):
+    return mwclient.errors.APIError(code, "info", {})
+
+
+@pytest.mark.parametrize(
+    "code", ["filetype-banned", "filetype-banned-type", "filetype-badtype"]
+)
+def test_rejected_extension_names_the_extension(code):
+    site = _FakeSite(extensions=["png", "pdf"])
+
+    with pytest.raises(ValueError) as exc_info:
+        reraise_upload_error(_api_error(code), site, "OSW123.exe", ".exe")
+
+    message = str(exc_info.value)
+    assert ".exe" in message
+    assert "OSW123.exe" in message
+    assert "wiki.example.org" in message
+    assert "pdf, png" in message  # allowed extensions, sorted
+
+
+def test_original_error_is_chained():
+    site = _FakeSite(extensions=["png"])
+    original = _api_error("filetype-banned")
+
+    with pytest.raises(ValueError) as exc_info:
+        reraise_upload_error(original, site, "OSW123.exe", ".exe")
+
+    assert exc_info.value.__cause__ is original
+
+
+def test_unrelated_api_error_is_reraised_unchanged():
+    site = _FakeSite(extensions=["png"])
+    original = _api_error("readapidenied")
+
+    with pytest.raises(mwclient.errors.APIError) as exc_info:
+        reraise_upload_error(original, site, "OSW123.png", ".png")
+
+    assert exc_info.value is original
+
+
+def test_message_omits_the_hint_when_siteinfo_fails():
+    """A failing siteinfo lookup must not mask the upload error."""
+    site = _FakeSite(fail=True)
+
+    with pytest.raises(ValueError) as exc_info:
+        reraise_upload_error(_api_error("filetype-banned"), site, "a.exe", ".exe")
+
+    assert "allowed on this wiki" not in str(exc_info.value)
+    assert ".exe" in str(exc_info.value)
+
+
+def test_get_allowed_file_extensions_returns_none_on_failure():
+    assert get_allowed_file_extensions(_FakeSite(fail=True)) is None
+
+
+def test_get_allowed_file_extensions_reads_siteinfo():
+    site = _FakeSite(extensions=["png", "jpg"])
+
+    assert get_allowed_file_extensions(site) == ["png", "jpg"]
+
+
+def test_successful_upload_passes():
+    assert (
+        assert_upload_success({"result": "Success"}, "a.png", "wiki.example.org")
+        is None
+    )
+
+
+@pytest.mark.parametrize("result", [{}, None, {"result": "Poll"}])
+def test_upload_without_success_raises(result):
+    """Anything but a Success result means the file did not arrive."""
+    with pytest.raises(ValueError) as exc_info:
+        assert_upload_success(result, "a.png", "wiki.example.org")
+
+    assert "a.png" in str(exc_info.value)
+    assert "wiki.example.org" in str(exc_info.value)
+
+
+def test_a_page_the_upload_created_gets_its_metadata_written():
+    """Otherwise store_entity keeps the empty page the upload just left behind."""
+    assert store_params_for_upload({}, page_existed=False) == {
+        "overwrite": "replace remote"
+    }
+
+
+def test_an_existing_page_keeps_the_callers_overwrite_policy():
+    se_params = {"overwrite": "keep existing", "edit_comment": "hi"}
+
+    assert store_params_for_upload(se_params, page_existed=True) == se_params
+
+
+def test_store_params_are_not_mutated():
+    se_params = {"edit_comment": "hi"}
+
+    store_params_for_upload(se_params, page_existed=False)
+
+    assert se_params == {"edit_comment": "hi"}
+
+
+def test_warned_upload_reports_the_warnings():
+    result = {"result": "Warning", "warnings": {"badfilename": "a_png"}}
+
+    with pytest.raises(ValueError) as exc_info:
+        assert_upload_success(result, "a.png", "wiki.example.org")
+
+    assert "badfilename" in str(exc_info.value)