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..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) +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: @@ -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]]: """