diff --git a/src/osw/controller/file/wiki.py b/src/osw/controller/file/wiki.py index 8a03229..de6b2a7 100644 --- a/src/osw/controller/file/wiki.py +++ b/src/osw/controller/file/wiki.py @@ -2,6 +2,8 @@ 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 @@ -9,6 +11,75 @@ 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: diff --git a/src/osw/core.py b/src/osw/core.py index fd37d31..6f485f2 100644 --- a/src/osw/core.py +++ b/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}', " diff --git a/src/osw/utils/wiki.py b/src/osw/utils/wiki.py index 054c3f7..9140f92 100644 --- a/src/osw/utils/wiki.py +++ b/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: diff --git a/tests/test_wiki_file_upload_errors.py b/tests/test_wiki_file_upload_errors.py new file mode 100644 index 0000000..211cf1b --- /dev/null +++ b/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) diff --git a/tests/utils/utils_test.py b/tests/utils/utils_test.py index c9519c4..8241b02 100644 --- a/tests/utils/utils_test.py +++ b/tests/utils/utils_test.py @@ -1,6 +1,9 @@ import uuid +import pytest + import osw.model.entity as model +from osw.core import OSW from osw.utils.regex import count_match_groups from osw.utils.strings import camel_case, pascal_case from osw.utils.wiki import ( @@ -28,6 +31,49 @@ def test_get_uuid(): assert get_uuid(osw_id) == uuid_ +@pytest.mark.parametrize( + "suffix", ["", ".png", ".drawio.png", ".tar.gz", ".jpg-2", ".a.b.c.d"] +) +def test_get_uuid_ignores_file_suffixes(suffix): + """File pages carry their extension in the title, ahead of the OSW-ID.""" + uuid_ = uuid.uuid4() + osw_id = f"OSW{str(uuid_).replace('-', '')}{suffix}" + + assert get_uuid(osw_id) == uuid_ + + +@pytest.mark.parametrize("prefix", ["", "OSW", "osw"]) +def test_get_uuid_accepts_a_bare_uuid(prefix): + uuid_ = uuid.uuid4() + + assert get_uuid(f"{prefix}{str(uuid_).replace('-', '')}") == uuid_ + assert get_uuid(f"{prefix}{uuid_}") == uuid_ # dashed + + +@pytest.mark.parametrize( + "osw_id", + [ + "", + "OSW", + "Category:OSW2ea5b605c91f4e5a95593dff79fdd4a5", # full title, not an ID + "OSW2ea5b605c91f4e5a95593dff79fdd4a", # one character short + "OSW2ea5b605c91f4e5a95593dff79fdd4a5x", + "OSW2ea5b605c91f4e5a95593dff79fdd4a5.", # empty suffix + "OSWzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz", + ], +) +def test_get_uuid_rejects_what_is_not_an_osw_id(osw_id): + with pytest.raises(ValueError): + get_uuid(osw_id) + + +def test_osw_get_uuid_delegates(): + """OSW.get_uuid is a wrapper, so it must handle suffixes just the same.""" + osw_id = "OSW2ea5b605c91f4e5a95593dff79fdd4a5.drawio.png" + + assert OSW.get_uuid(osw_id) == get_uuid(osw_id) + + def test_get_entity_namespace(): class DummyClass(model.Entity): pass