From a88f1aab3410cc09a097b37d7d17c644a9b09481 Mon Sep 17 00:00:00 2001 From: Myst <1592048+LeMyst@users.noreply.github.com> Date: Sat, 4 Jul 2026 10:11:00 +0200 Subject: [PATCH 1/3] Add check_constraints() helper to query Wikibase constraint violations Uses the target instance's own wbcheckconstraints API action (from the WikibaseQualityConstraints extension) instead of a hardcoded, Wikidata-specific rule set, so it works against any Wikibase instance with the extension installed. Closes #154 Co-Authored-By: Claude Sonnet 5 --- test/conftest.py | 6 +++++ test/test_wbi_helpers.py | 35 +++++++++++++++++++++++++++- wikibaseintegrator/wbi_helpers.py | 38 +++++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 1 deletion(-) diff --git a/test/conftest.py b/test/conftest.py index 4fc8797e..da813660 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -79,6 +79,7 @@ def __init__(self, mocker: requests_mock_lib.Mocker, base_url: str = 'https://wi self.search_results: list[dict] = [] # wbsearchentities results self.fulltext_results: list[dict] = [] # list=search results self.sparql_bindings: list[dict] = [] # bindings returned by the SPARQL endpoint + self.constraint_results: dict[str, list[dict]] = {} # wbcheckconstraints results, keyed by entity id self.valid_credentials: dict[str, str] = {} # user -> password accepted by (client)login self.login_token = 'aabbccddeeff+\\' self.csrf_token = '0123456789abcdef+\\' @@ -368,6 +369,11 @@ def _action_wbremoveclaims(self, params: dict[str, str]) -> dict: def _action_delete(self, params: dict[str, str]) -> dict: return {'delete': {'title': params.get('title', ''), 'reason': params.get('reason', 'mock deletion'), 'logid': 1}} + def _action_wbcheckconstraints(self, params: dict[str, str]) -> dict: + requested_ids = params['id'].split('|') if 'id' in params else list(self.constraint_results.keys()) + claims = {entity_id: deepcopy(self.constraint_results[entity_id]) for entity_id in requested_ids if entity_id in self.constraint_results} + return {'claims': claims, 'success': 1} + # ---------------------------------------------------------------------- # # SPARQL helpers usable from the tests diff --git a/test/test_wbi_helpers.py b/test/test_wbi_helpers.py index b23db7a9..e4ca3b43 100644 --- a/test/test_wbi_helpers.py +++ b/test/test_wbi_helpers.py @@ -11,7 +11,7 @@ from wikibaseintegrator.wbi_exceptions import AnonymousEditNotAllowedError, MaxRetriesReachedException, ModificationFailed, MWApiError, NonExistentEntityError, SaveFailed from wikibaseintegrator.wbi_helpers import (download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, get_user_agent, lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, remove_claims, - search_entities) + search_entities, check_constraints) class FakeLogin: @@ -254,6 +254,39 @@ def test_search_strict_language_and_limit_parameters(self, wikibase): assert len(results) == 10 +class TestCheckConstraints: + def test_requires_entity_id_or_claim_id(self): + with pytest.raises(ValueError): + check_constraints() + + def test_check_by_entity_id(self, wikibase): + wikibase.constraint_results = { + 'Q582': [{'id': 'Q582$1', 'results': [{'status': 'violation', 'property': 'P17'}]}] + } + + result = check_constraints(entity_id='Q582') + assert wikibase.last_request['action'] == 'wbcheckconstraints' + assert wikibase.last_request['id'] == 'Q582' + assert result['claims']['Q582'][0]['results'][0]['status'] == 'violation' + + def test_check_by_multiple_entity_ids(self, wikibase): + wikibase.constraint_results = { + 'Q582': [{'id': 'Q582$1', 'results': []}], + 'Q1': [{'id': 'Q1$1', 'results': []}], + } + + check_constraints(entity_id=['Q582', 'Q1']) + assert wikibase.last_request['id'] == 'Q582|Q1' + + def test_check_by_claim_id(self, wikibase): + check_constraints(claim_id='Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a') + assert wikibase.last_request['claimid'] == 'Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a' + + def test_check_with_status_filter(self, wikibase): + check_constraints(entity_id='Q582', status=['violation', 'warning']) + assert wikibase.last_request['status'] == 'violation|warning' + + class TestFulltextSearch: def test_fulltext_search(self, wikibase): wikibase.fulltext_results = [{'ns': 0, 'title': 'Q582', 'pageid': 892, 'snippet': 'Villeurbanne'}] diff --git a/wikibaseintegrator/wbi_helpers.py b/wikibaseintegrator/wbi_helpers.py index 52f6a31b..22131f2f 100644 --- a/wikibaseintegrator/wbi_helpers.py +++ b/wikibaseintegrator/wbi_helpers.py @@ -433,6 +433,44 @@ def remove_claims(claim_id: str, summary: str | None = None, baserevid: int | No return mediawiki_api_call_helper(data=params, is_bot=is_bot, **kwargs) +def check_constraints(entity_id: str | list[str] | None = None, claim_id: str | list[str] | None = None, status: list[str] | None = None, allow_anonymous: bool = True, + **kwargs: Any) -> dict: + """ + Check the constraint violations of one or more entities or claims, using the target Wikibase instance's own constraint checker + (the ``wbcheckconstraints`` API module provided by the `WikibaseQualityConstraints + `_ extension). + + This relies entirely on the constraints configured on the Wikibase instance being queried (e.g. on Wikidata itself), rather than + on a hardcoded or Wikidata-specific set of rules, so it works the same way against any Wikibase instance that has the extension + installed. If the extension isn't installed, the underlying API call will fail with a :class:`~wikibaseintegrator.wbi_exceptions.MWApiError`. + + :param entity_id: One or more entity IDs (item, property, lexeme, etc.) whose claims should be checked. + :param claim_id: One or more claim GUIDs to check, instead of checking whole entities. + :param status: Only return results with one of these statuses, e.g. ``['violation']``. By default, the API returns every status. + :param allow_anonymous: Allow anonymous interaction with the MediaWiki API. 'True' by default since this is a read-only action. + :param kwargs: Extra parameters for mediawiki_api_call_helper() + :return: The data returned by the API as a dictionary + """ + if not entity_id and not claim_id: + raise ValueError("You must provide either 'entity_id' or 'claim_id'.") + + params: dict[str, Any] = { + 'action': 'wbcheckconstraints', + 'format': 'json' + } + + if entity_id: + params.update({'id': '|'.join(entity_id) if isinstance(entity_id, list) else entity_id}) + + if claim_id: + params.update({'claimid': '|'.join(claim_id) if isinstance(claim_id, list) else claim_id}) + + if status: + params.update({'status': '|'.join(status)}) + + return mediawiki_api_call_helper(data=params, allow_anonymous=allow_anonymous, **kwargs) + + def search_entities(search_string: str, language: str | None = None, strict_language: bool = False, search_type: str = 'item', max_results: int = 50, dict_result: bool = False, allow_anonymous: bool = True, **kwargs: Any) -> list[dict[str, Any]]: """ From 6c766c5e8a2297161a790d3d89038eb99c9afb57 Mon Sep 17 00:00:00 2001 From: Myst <1592048+LeMyst@users.noreply.github.com> Date: Sat, 4 Jul 2026 10:12:52 +0200 Subject: [PATCH 2/3] Fix isort ordering in test_wbi_helpers.py imports Co-Authored-By: Claude Sonnet 5 --- test/test_wbi_helpers.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/test_wbi_helpers.py b/test/test_wbi_helpers.py index e4ca3b43..42f373b1 100644 --- a/test/test_wbi_helpers.py +++ b/test/test_wbi_helpers.py @@ -9,9 +9,9 @@ from wikibaseintegrator.wbi_config import config as wbi_config from wikibaseintegrator.wbi_exceptions import AnonymousEditNotAllowedError, MaxRetriesReachedException, ModificationFailed, MWApiError, NonExistentEntityError, SaveFailed -from wikibaseintegrator.wbi_helpers import (download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, get_user_agent, - lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, remove_claims, - search_entities, check_constraints) +from wikibaseintegrator.wbi_helpers import (check_constraints, download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, + get_user_agent, lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, + remove_claims, search_entities) class FakeLogin: From 509b9d96bfd032dccc0a625f3977e9ef2db9dd28 Mon Sep 17 00:00:00 2001 From: Myst <1592048+LeMyst@users.noreply.github.com> Date: Sat, 4 Jul 2026 10:35:16 +0200 Subject: [PATCH 3/3] Drop mwoauth dependency, use oauthlib/requests-oauthlib directly The 3-legged OAuth1 handshake (Special:OAuth/initiate, authenticate, token) is reimplemented with requests_oauthlib.OAuth1 and requests, mirroring what mwoauth did internally, removing the extra dependency. Also fixes continue_oauth(), which previously crashed with an AttributeError because mediawiki_api_url/instantiation_time were never initialized after completing the handshake. Fixes #318 Co-Authored-By: Claude Sonnet 5 --- poetry.lock | 45 ++---------------------- pyproject.toml | 1 - test/test_wbi_login.py | 21 +++++++++++ wikibaseintegrator/wbi_login.py | 62 ++++++++++++++++++++++++--------- 4 files changed, 69 insertions(+), 60 deletions(-) diff --git a/poetry.lock b/poetry.lock index b3420926..78560e3b 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1897,27 +1897,6 @@ files = [ [package.dependencies] typing-extensions = {version = "*", markers = "python_version < \"3.11\""} -[[package]] -name = "mwoauth" -version = "0.4.0" -description = "A generic MediaWiki OAuth handshake helper." -optional = false -python-versions = "*" -groups = ["main"] -files = [ - {file = "mwoauth-0.4.0-py3-none-any.whl", hash = "sha256:fed9bc7d6bbabb5f691b918af0ac844e13c9b75d5fa51a898f36d54d798b5fe1"}, - {file = "mwoauth-0.4.0.tar.gz", hash = "sha256:22e3403e748e70146f8eccc1430fe542c9f9c4ff677eff424a52e644f6d8f7c5"}, -] - -[package.dependencies] -oauthlib = "*" -PyJWT = ">=1.0.1" -requests = "*" -requests-oauthlib = "*" - -[package.extras] -flask = ["flask"] - [[package]] name = "mypy" version = "2.1.0" @@ -2421,24 +2400,6 @@ files = [ [package.extras] windows-terminal = ["colorama (>=0.4.6)"] -[[package]] -name = "pyjwt" -version = "2.13.0" -description = "JSON Web Token implementation in Python" -optional = false -python-versions = ">=3.9" -groups = ["main"] -files = [ - {file = "pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728"}, - {file = "pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423"}, -] - -[package.dependencies] -typing_extensions = {version = ">=4.0", markers = "python_version < \"3.11\""} - -[package.extras] -crypto = ["cryptography (>=3.4.0)"] - [[package]] name = "pylint" version = "4.0.6" @@ -3758,12 +3719,12 @@ version = "4.16.0" description = "Backported and Experimental Type Hints for Python 3.9+" optional = false python-versions = ">=3.9" -groups = ["main", "coverage", "dev", "docs", "notebooks"] +groups = ["coverage", "dev", "docs", "notebooks"] files = [ {file = "typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8"}, {file = "typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5"}, ] -markers = {main = "python_version == \"3.10\"", coverage = "python_version == \"3.10\"", docs = "python_version >= \"3.11\" and python_version < \"3.13\""} +markers = {coverage = "python_version == \"3.10\"", docs = "python_version >= \"3.11\" and python_version < \"3.13\""} [[package]] name = "tzdata" @@ -3981,4 +3942,4 @@ files = [ [metadata] lock-version = "2.1" python-versions = "^3.10" -content-hash = "505ee90d158ed8020002f7f485cdc68fa0d5d0f7520267948cf8210d1c5983cb" +content-hash = "3c3710f79630363e3c72c5cde954e2b7833f78db41fc4e10dbd8d9e9ddd90f57" diff --git a/pyproject.toml b/pyproject.toml index 2ac9275f..c53be093 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,7 +41,6 @@ Changelog = "https://github.com/LeMyst/WikibaseIntegrator/releases" [tool.poetry.dependencies] python = "^3.10" backoff = "^2.2.1" -mwoauth = "^0.4.0" oauthlib = "^3.2.2" requests = "^2.32.3" requests-oauthlib = "^2.0.0" diff --git a/test/test_wbi_login.py b/test/test_wbi_login.py index 3e9b0c0f..40b0a866 100644 --- a/test/test_wbi_login.py +++ b/test/test_wbi_login.py @@ -75,6 +75,27 @@ def test_csrf_error_raises_login_error(self, credentials): with pytest.raises(LoginError): wbi_login.OAuth1(consumer_token='wrong', consumer_secret='wrong', access_token='wrong', access_secret='wrong') + def test_three_legged_flow(self, credentials, requests_mock): + # Step 1 (initiate) and step 3 (token) both post to Special:OAuth on the index endpoint. + requests_mock.post(credentials.mediawiki_index_url, [ + {'text': 'oauth_token=request-key&oauth_token_secret=request-secret'}, + {'text': 'oauth_token=access-key&oauth_token_secret=access-secret'}, + ]) + + login = wbi_login.OAuth1(consumer_token='consumer-token', consumer_secret='consumer-secret') + assert 'oauth_token=request-key' in login.redirect + assert login.request_token_key == 'request-key' + + login.continue_oauth('https://example.org/callback?oauth_token=request-key&oauth_verifier=some-verifier') + assert login.get_edit_token() == credentials.csrf_token + + def test_three_legged_flow_wrong_request_token(self, credentials, requests_mock): + requests_mock.post(credentials.mediawiki_index_url, text='oauth_token=request-key&oauth_token_secret=request-secret') + + login = wbi_login.OAuth1(consumer_token='consumer-token', consumer_secret='consumer-secret') + with pytest.raises(LoginError): + login.continue_oauth('https://example.org/callback?oauth_token=unexpected-key&oauth_verifier=some-verifier') + class TestOAuth2: def test_successful_flow(self, credentials, requests_mock): diff --git a/wikibaseintegrator/wbi_login.py b/wikibaseintegrator/wbi_login.py index 1a109f3f..0fbc6ffc 100644 --- a/wikibaseintegrator/wbi_login.py +++ b/wikibaseintegrator/wbi_login.py @@ -5,11 +5,13 @@ import time import webbrowser from typing import Any, cast +from urllib.parse import parse_qs, urlencode -from mwoauth import ConsumerToken, Handshaker, OAuthException +import requests from oauthlib.oauth2 import BackendApplicationClient, InvalidClientError from requests import Session from requests.cookies import RequestsCookieJar +from requests_oauthlib import OAuth1 as OAuth1Auth from requests_oauthlib import OAuth1Session, OAuth2Session from wikibaseintegrator.wbi_backoff import wbi_backoff @@ -176,19 +178,35 @@ def __init__(self, consumer_token: str | None = None, consumer_secret: str | Non super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url) else: # Oauth procedure, based on https://www.mediawiki.org/wiki/OAuth/For_Developers - # Construct a "consumer" from the key/secret provided by MediaWiki - self.oauth1_consumer_token = ConsumerToken(consumer_token, consumer_secret) + self.consumer_token = consumer_token + self.consumer_secret = consumer_secret + self.mediawiki_index_url = mediawiki_index_url + self.mediawiki_api_url = str(mediawiki_api_url or config['MEDIAWIKI_API_URL']) + self.token_renew_period = token_renew_period + self.user_agent = user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None) - # Construct handshaker with wiki URI and consumer - self.handshaker = Handshaker(mw_uri=mediawiki_index_url, consumer_token=self.oauth1_consumer_token, callback=callback_url, - user_agent=get_user_agent(user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None))) + # Step 1: Initiate -- ask MediaWiki for a temporary key/secret for the user + auth = OAuth1Auth(consumer_token, client_secret=consumer_secret, callback_uri=callback_url) + response = requests.post(url=mediawiki_index_url, params={'title': "Special:OAuth/initiate"}, auth=auth, headers={'User-Agent': get_user_agent(self.user_agent)}, timeout=config['TIMEOUT']) + + request_token = self._parse_token_response(response.text) + self.request_token_key = request_token['oauth_token'] + self.request_token_secret = request_token['oauth_token_secret'] - # Step 1: Initialize -- ask MediaWiki for a temp key/secret for user # redirect -> authorization -> callback url - try: - self.redirect, self.request_token = self.handshaker.initiate(callback=callback_url) - except OAuthException as err: - raise LoginError(err) from err + params = {'title': "Special:OAuth/authenticate", 'oauth_token': self.request_token_key, 'oauth_consumer_key': consumer_token} + self.redirect = mediawiki_index_url + '?' + urlencode(params) + + @staticmethod + def _parse_token_response(content: str) -> dict[str, str]: + if content.startswith("Error: "): + raise LoginError(content[len("Error: "):]) + + credentials = parse_qs(content) + if not credentials or 'oauth_token' not in credentials or 'oauth_token_secret' not in credentials: + raise LoginError(f"MediaWiki response lacks token information: {content!r}") + + return {'oauth_token': credentials['oauth_token'][0], 'oauth_token_secret': credentials['oauth_token_secret'][0]} def continue_oauth(self, oauth_callback_data: str | None = None) -> None: """ @@ -205,17 +223,27 @@ def continue_oauth(self, oauth_callback_data: str | None = None) -> None: # input the url from redirect after authorization response_qs = oauth_callback_data.split('?')[-1] + callback_data = parse_qs(response_qs) + + if not callback_data or 'oauth_token' not in callback_data or 'oauth_verifier' not in callback_data: + raise LoginError(f"Query string lacks token information: {callback_data!r}") + + request_token_key = callback_data['oauth_token'][0] + verifier = callback_data['oauth_verifier'][0] + + if self.request_token_key != request_token_key: + raise LoginError(f"Unexpected request token key {request_token_key!r}, expected {self.request_token_key!r}.") # Step 3: Complete -- obtain authorized key/secret for "resource owner" - access_token = self.handshaker.complete(self.request_token, response_qs) + auth = OAuth1Auth(self.consumer_token, client_secret=self.consumer_secret, resource_owner_key=self.request_token_key, resource_owner_secret=self.request_token_secret, verifier=verifier) + response = requests.post(url=self.mediawiki_index_url, params={'title': "Special:OAuth/token"}, auth=auth, headers={'User-Agent': get_user_agent(self.user_agent)}, timeout=config['TIMEOUT']) - if self.oauth1_consumer_token is None: - raise ValueError("oauth1_consumer_token can't be None") + access_token = self._parse_token_response(response.text) # input the access token to return a csrf (edit) token - self.session = OAuth1Session(client_key=self.oauth1_consumer_token.key, client_secret=self.oauth1_consumer_token.secret, resource_owner_key=access_token.key, - resource_owner_secret=access_token.secret) - self.generate_edit_credentials() + session = OAuth1Session(client_key=self.consumer_token, client_secret=self.consumer_secret, resource_owner_key=access_token['oauth_token'], + resource_owner_secret=access_token['oauth_token_secret']) + super().__init__(session=session, token_renew_period=self.token_renew_period, user_agent=self.user_agent, mediawiki_api_url=self.mediawiki_api_url) class Login(_Login):