Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • w/shared/mwclient
1 result
Show changes
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 2449 additions and 713 deletions
mwclient/_types.py 0 → 100644
View file @ 15eb10cc
from http.cookiejar import CookieJar
from typing import Union, Tuple, Mapping
Cookies = Union[Mapping[str, str], CookieJar]
Namespace = Union[str, int]
VersionTuple = Tuple[Union[int, str], ...]
mwclient/client.py
View file @ 15eb10cc
This diff is collapsed.
mwclient/errors.py
View file @ 15eb10cc
from typing import Any, TYPE_CHECKING, Optional, cast
if TYPE_CHECKING:
import mwclient.page
class MwClientError(RuntimeError):
"""Base class for all mwclient errors."""
pass
class MediaWikiVersionError(MwClientError):
"""The version of MediaWiki is not supported."""
pass
class APIDisabledError(MwClientError):
"""The API is disabled on the wiki."""
pass
class MaximumRetriesExceeded(MwClientError):
"""The maximum number of retries for a request has been exceeded."""
pass
class APIError(MwClientError):
"""Base class for errors returned by the MediaWiki API.
def __init__(self, code, info, kwargs):
Attributes:
code (Optional[str]): The error code returned by the API.
info (str): The error message returned by the API.
kwargs (Optional[Any]): Additional information.
"""
def __init__(self, code: Optional[str], info: str, kwargs: Optional[Any]) -> None:
self.code = code
self.info = info
super(APIError, self).__init__(code, info, kwargs)
super().__init__(code, info, kwargs)
class UserNotFound(APIError):
......@@ -31,38 +48,80 @@ class UserCreateError(APIError):
class InsufficientPermission(MwClientError):
"""Raised when the user does not have sufficient permissions to perform an
action."""
pass
class UserBlocked(InsufficientPermission):
"""Raised when attempting to perform an action while blocked."""
pass
class EditError(MwClientError):
"""Base class for errors related to editing pages."""
pass
class ProtectedPageError(EditError, InsufficientPermission):
def __init__(self, page, code=None, info=None):
"""Raised when attempting to edit a protected page.
Attributes:
page (mwclient.page.Page): The page for which the edit attempt was made.
code (Optional[str]): The error code returned by the API.
info (Optional[str]): The error message returned by the API.
"""
def __init__(
self,
page: 'mwclient.page.Page',
code: Optional[str] = None,
info: Optional[str] = None
) -> None:
self.page = page
self.code = code
self.info = info
def __str__(self):
def __str__(self) -> str:
if self.info is not None:
return self.info
return 'You do not have the "edit" right.'
class FileExists(EditError):
pass
"""
Raised when trying to upload a file that already exists.
See also: https://www.mediawiki.org/wiki/API:Upload#Upload_warnings
Attributes:
file_name (str): The name of the file that already exists.
"""
def __init__(self, file_name: str) -> None:
self.file_name = file_name
def __str__(self) -> str:
return (
f'The file "{self.file_name}" already exists. '
f'Set ignore=True to overwrite it.'
)
class LoginError(MwClientError):
def __init__(self, site, code, info):
super(LoginError, self).__init__(
class LoginError(MwClientError):
"""Base class for login errors.
Attributes:
site (mwclient.site.Site): The site object on which the login attempt
was made.
code (str): The error code returned by the API.
info (str): The error message returned by the API.
"""
def __init__(
self, site: 'mwclient.client.Site', code: Optional[str], info: str
) -> None:
super().__init__(
site,
{'result': code, 'reason': info} # For backwards-compability
)
......@@ -70,43 +129,54 @@ class LoginError(MwClientError):
self.code = code
self.info = info
def __str__(self):
def __str__(self) -> str:
return self.info
class OAuthAuthorizationError(LoginError):
"""Raised when OAuth authorization fails.
Attributes:
site (mwclient.site.Site): The site object on which the login attempt
was made.
code (str): The error code returned by the API.
info (str): The error message returned by the API.
"""
pass
class AssertUserFailedError(MwClientError):
def __init__(self):
super(AssertUserFailedError, self).__init__((
"""Raised when the user assertion fails."""
def __init__(self) -> None:
super().__init__(
'By default, mwclient protects you from accidentally editing '
'without being logged in. If you actually want to edit without '
'logging in, you can set force_login on the Site object to False.'
))
)
def __str__(self):
return self.args[0]
def __str__(self) -> str:
return cast(str, self.args[0])
class EmailError(MwClientError):
"""Base class for email errors."""
pass
class NoSpecifiedEmail(EmailError):
pass
class NoWriteApi(MwClientError):
"""Raised when trying to email a user who has not specified an email"""
pass
class InvalidResponse(MwClientError):
"""Raised when the server returns an invalid JSON response.
Attributes:
response_text (str): The response text from the server.
"""
def __init__(self, response_text=None):
super(InvalidResponse, self).__init__((
def __init__(self, response_text: Optional[str] = None) -> None:
super().__init__((
'Did not get a valid JSON response from the server. Check that '
'you used the correct hostname. If you did, the server might '
'be wrongly configured or experiencing temporary problems.'),
......@@ -114,9 +184,10 @@ class InvalidResponse(MwClientError):
)
self.response_text = response_text
def __str__(self):
return self.args[0]
def __str__(self) -> str:
return cast(str, self.args[0])
class InvalidPageTitle(MwClientError):
"""Raised when an invalid page title is used."""
pass
mwclient/image.py
View file @ 15eb10cc
import io
from typing import Optional, Mapping, Any, overload
import mwclient.listing
import mwclient.page
from mwclient._types import Namespace
from mwclient.util import handle_limit
class Image(mwclient.page.Page):
"""
Represents an image on a MediaWiki wiki represented by a
:class:`~mwclient.client.Site` object.
Args:
site (mwclient.client.Site): The site object this page belongs to.
name (Union[str, int, Page]): The title of the page, the page ID, or
another :class:`Page` object to copy.
info (Optional[dict]): Page info, if already fetched, e.g., when
iterating over a list of pages. If not provided, the page info
will be fetched from the API.
"""
def __init__(self, site, name, info=None):
super(Image, self).__init__(
def __init__(
self,
site: 'mwclient.client.Site',
name: str,
info: Optional[Mapping[str, Any]] = None
) -> None:
super().__init__(
site, name, info, extra_properties={
'imageinfo': (
('iiprop',
......@@ -16,7 +38,7 @@ class Image(mwclient.page.Page):
self.imagerepository = self._info.get('imagerepository', '')
self.imageinfo = self._info.get('imageinfo', ({}, ))[0]
def imagehistory(self):
def imagehistory(self) -> 'mwclient.listing.PageProperty':
"""
Get file revision info for the given file.
......@@ -27,8 +49,16 @@ class Image(mwclient.page.Page):
iiprop='timestamp|user|comment|url|size|sha1|metadata|mime|archivename'
)
def imageusage(self, namespace=None, filterredir='all', redirect=False,
limit=None, generator=True):
def imageusage(
self,
namespace: Optional[Namespace] = None,
filterredir: str = 'all',
redirect: bool = False,
limit: Optional[int] = None,
generator: bool = True,
max_items: Optional[int] = None,
api_chunk_size: Optional[int] = None
) -> 'mwclient.listing.List':
"""
List pages that use the given file.
......@@ -38,21 +68,53 @@ class Image(mwclient.page.Page):
kwargs = dict(mwclient.listing.List.generate_kwargs(
prefix, title=self.name, namespace=namespace, filterredir=filterredir
))
(max_items, api_chunk_size) = handle_limit(limit, max_items, api_chunk_size)
if redirect:
kwargs['%sredirect' % prefix] = '1'
kwargs[f'{prefix}redirect'] = '1'
return mwclient.listing.List.get_list(generator)(
self.site, 'imageusage', 'iu', limit=limit, return_values='title', **kwargs
self.site,
'imageusage',
'iu',
max_items=max_items,
api_chunk_size=api_chunk_size,
return_values='title',
**kwargs
)
def duplicatefiles(self, limit=None):
def duplicatefiles(
self,
limit: Optional[int] = None,
max_items: Optional[int] = None,
api_chunk_size: Optional[int] = None
) -> 'mwclient.listing.PageProperty':
"""
List duplicates of the current file.
API doc: https://www.mediawiki.org/wiki/API:Duplicatefiles
limit sets a hard cap on the total number of results, it does
not only specify the API chunk size.
"""
return mwclient.listing.PageProperty(self, 'duplicatefiles', 'df', dflimit=limit)
(max_items, api_chunk_size) = handle_limit(limit, max_items, api_chunk_size)
return mwclient.listing.PageProperty(
self,
'duplicatefiles',
'df',
max_items=max_items,
api_chunk_size=api_chunk_size
)
@overload
def download(self) -> bytes:
...
def download(self, destination=None):
@overload
def download(self, destination: io.BufferedWriter) -> None:
...
def download(
self, destination: Optional[io.BufferedWriter] = None
) -> Optional[bytes]:
"""
Download the file. If `destination` is given, the file will be written
directly to the stream. Otherwise the file content will be stored in memory
......@@ -64,19 +126,16 @@ class Image(mwclient.page.Page):
... image.download(fd)
Args:
destination (file object): Destination file
destination: Destination file
"""
url = self.imageinfo['url']
if destination is not None:
res = self.site.connection.get(url, stream=True)
for chunk in res.iter_content(1024):
destination.write(chunk)
return None
else:
return self.site.connection.get(url).content
def __repr__(self):
return "<%s object '%s' for %s>" % (
self.__class__.__name__,
self.name.encode('utf-8'),
self.site
)
def __repr__(self) -> str:
return f"<{self.__class__.__name__} object '{self.name}' for {self.site}>"
mwclient/listing.py
View file @ 15eb10cc
from mwclient.util import parse_timestamp
import mwclient.page
from typing import ( # noqa: F401
Optional, Tuple, Any, Union, Iterator, Mapping, Iterable, Type, Dict
)
import mwclient.image
import mwclient.page
from mwclient._types import Namespace
from mwclient.util import parse_timestamp, handle_limit
class List:
......@@ -9,11 +14,27 @@ class List:
This is a class providing lazy iteration. This means that the
content is loaded in chunks as long as the response hints at
continuing content.
max_items limits the total number of items that will be yielded
by this iterator. api_chunk_size sets the number of items that
will be requested from the wiki per API call (this iterator itself
always yields one item at a time). limit does the same as
api_chunk_size for backward compatibility, but is deprecated due
to its misleading name.
"""
def __init__(self, site, list_name, prefix,
limit=None, return_values=None, max_items=None,
*args, **kwargs):
def __init__(
self,
site: 'mwclient.client.Site',
list_name: str,
prefix: str,
limit: Optional[int] = None,
return_values: Union[str, Tuple[str, ...], None] = None,
max_items: Optional[int] = None,
api_chunk_size: Optional[int] = None,
*args: Tuple[str, Any],
**kwargs: Any
) -> None:
# NOTE: Fix limit
self.site = site
self.list_name = list_name
......@@ -23,23 +44,28 @@ class List:
kwargs.update(args)
self.args = kwargs
if limit is None:
limit = site.api_limit
self.args[self.prefix + 'limit'] = str(limit)
(max_items, api_chunk_size) = handle_limit(limit, max_items, api_chunk_size)
# for efficiency, if max_items is set and api_chunk_size is not,
# set the chunk size to max_items so we don't retrieve
# unneeded extra items (so long as it's below API limit)
api_limit = site.api_limit
api_chunk_size = api_chunk_size or min(max_items or api_limit, api_limit)
self.args[self.prefix + 'limit'] = str(api_chunk_size)
self.count = 0
self.max_items = max_items
self._iter = iter(range(0))
self._iter = iter(range(0)) # type: Iterator[Any]
self.last = False
self.result_member = list_name
self.return_values = return_values
def __iter__(self):
def __iter__(self) -> 'List':
return self
def __next__(self):
def __next__(self) -> Any:
if self.max_items is not None:
if self.count >= self.max_items:
raise StopIteration
......@@ -64,12 +90,12 @@ class List:
if isinstance(self, GeneratorList):
return item
if type(self.return_values) is tuple:
return tuple((item[i] for i in self.return_values))
return tuple(item[i] for i in self.return_values)
if self.return_values is not None:
return item[self.return_values]
return item
def load_chunk(self):
def load_chunk(self) -> None:
"""Query a new chunk of data
If the query is empty, `raise StopIteration`.
......@@ -77,11 +103,8 @@ class List:
Else, update the iterator accordingly.
If 'continue' is in the response, it is added to `self.args`
(new style continuation, added in MediaWiki 1.21).
If not, but 'query-continue' is in the response, query its
item called `self.list_name` and add this to `self.args` (old
style continuation).
(new style continuation, added in MediaWiki 1.21, default
since MediaWiki 1.26).
Else, set `self.last` to True.
"""
......@@ -102,14 +125,10 @@ class List:
# New style continuation, added in MediaWiki 1.21
self.args.update(data['continue'])
elif self.list_name in data.get('query-continue', ()):
# Old style continuation
self.args.update(data['query-continue'][self.list_name])
else:
self.last = True
def set_iter(self, data):
def set_iter(self, data: Mapping[str, Any]) -> None:
"""Set `self._iter` to the API response `data`."""
if self.result_member not in data['query']:
self._iter = iter(range(0))
......@@ -118,35 +137,33 @@ class List:
else:
self._iter = iter(data['query'][self.result_member].values())
def __repr__(self):
return "<%s object '%s' for %s>" % (
self.__class__.__name__,
self.list_name,
self.site
)
def __repr__(self) -> str:
return f"<{self.__class__.__name__} object '{self.list_name}' for {self.site}>"
@staticmethod
def generate_kwargs(_prefix, *args, **kwargs):
def generate_kwargs(
_prefix: str, *args: Tuple[str, Any], **kwargs: Any
) -> Iterable[Tuple[str, Any]]:
kwargs.update(args)
for key, value in kwargs.items():
if value is not None and value is not False:
yield _prefix + key, value
@staticmethod
def get_prefix(prefix, generator=False):
def get_prefix(prefix: str, generator: bool = False) -> str:
return ('g' if generator else '') + prefix
@staticmethod
def get_list(generator=False):
def get_list(generator: bool = False) -> Union[Type['GeneratorList'], Type['List']]:
return GeneratorList if generator else List
class NestedList(List):
def __init__(self, nested_param, *args, **kwargs):
super(NestedList, self).__init__(*args, **kwargs)
def __init__(self, nested_param: str, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.nested_param = nested_param
def set_iter(self, data):
def set_iter(self, data: Mapping[str, Any]) -> None:
self._iter = iter(data['query'][self.result_member][self.nested_param])
......@@ -158,9 +175,17 @@ class GeneratorList(List):
this subclass turns the data into Page, Image or Category objects.
"""
def __init__(self, site, list_name, prefix, *args, **kwargs):
super(GeneratorList, self).__init__(site, list_name, prefix,
*args, **kwargs)
def __init__(
self,
site: 'mwclient.client.Site',
list_name: str,
prefix: str,
*args: Tuple[str, Any],
**kwargs: Any
) -> None:
super().__init__(
site, list_name, prefix, *args, **kwargs # type: ignore[arg-type]
)
self.args['g' + self.prefix + 'limit'] = self.args[self.prefix + 'limit']
del self.args[self.prefix + 'limit']
......@@ -173,40 +198,64 @@ class GeneratorList(List):
self.page_class = mwclient.page.Page
def __next__(self):
info = super(GeneratorList, self).__next__()
def __next__(self) -> Union['mwclient.page.Page', 'mwclient.image.Image', 'Category']:
info = super().__next__()
if info['ns'] == 14:
return Category(self.site, '', info)
if info['ns'] == 6:
return mwclient.image.Image(self.site, '', info)
return mwclient.page.Page(self.site, '', info)
def load_chunk(self):
def load_chunk(self) -> None:
# Put this here so that the constructor does not fail
# on uninitialized sites
self.args['iiprop'] = 'timestamp|user|comment|url|size|sha1|metadata|archivename'
return super(GeneratorList, self).load_chunk()
return super().load_chunk()
class Category(mwclient.page.Page, GeneratorList):
"""
Represents a category on a MediaWiki wiki represented by a
:class:`~mwclient.client.Site` object.
Args:
site (mwclient.client.Site): The site object this page belongs to.
name (Union[str, int, Page]): The title of the page, the page ID, or
another :class:`Page` object to copy.
info (Optional[dict]): Page info, if already fetched, e.g., when
iterating over a list of pages. If not provided, the page info
will be fetched from the API.
namespace (Union[int, str, None]): The namespace of the category
members to list.
"""
def __init__(self, site, name, info=None, namespace=None):
def __init__(
self,
site: 'mwclient.client.Site',
name: str,
info: Optional[Mapping[str, Any]] = None,
namespace: Optional[Namespace] = None
) -> None:
mwclient.page.Page.__init__(self, site, name, info)
kwargs = {}
kwargs = {} # type: Dict[str, Any]
kwargs['gcmtitle'] = self.name
if namespace:
kwargs['gcmnamespace'] = namespace
GeneratorList.__init__(self, site, 'categorymembers', 'cm', **kwargs)
def __repr__(self):
return "<%s object '%s' for %s>" % (
self.__class__.__name__,
self.name.encode('utf-8'),
self.site
)
def members(self, prop='ids|title', namespace=None, sort='sortkey',
dir='asc', start=None, end=None, generator=True):
def __repr__(self) -> str:
return f"<{self.__class__.__name__} object '{self.name}' for {self.site}>"
def members(
self,
prop: str = 'ids|title',
namespace: Optional[Namespace] = None,
sort: str = 'sortkey',
dir: str = 'asc',
start: Optional[str] = None,
end: Optional[str] = None,
generator: bool = True
) -> 'List':
prefix = self.get_prefix('cm', generator)
kwargs = dict(self.generate_kwargs(prefix, prop=prop, namespace=namespace,
sort=sort, dir=dir, start=start, end=end,
......@@ -216,8 +265,15 @@ class Category(mwclient.page.Page, GeneratorList):
class PageList(GeneratorList):
def __init__(self, site, prefix=None, start=None, namespace=0, redirects='all',
end=None):
def __init__(
self,
site: 'mwclient.client.Site',
prefix: Optional[str] = None,
start: Optional[str] = None,
namespace: int = 0,
redirects: str = 'all',
end: Optional[str] = None
):
self.namespace = namespace
kwargs = {}
......@@ -228,36 +284,44 @@ class PageList(GeneratorList):
if end:
kwargs['gapto'] = end
super(PageList, self).__init__(site, 'allpages', 'ap',
gapnamespace=str(namespace),
gapfilterredir=redirects,
**kwargs)
super().__init__(site, 'allpages', 'ap', gapnamespace=str(namespace),
gapfilterredir=redirects, **kwargs)
def __getitem__(self, name):
def __getitem__(
self, name: Union[str, int, 'mwclient.page.Page']
) -> Union['mwclient.page.Page', 'mwclient.image.Image', 'Category']:
return self.get(name, None)
def get(self, name, info=()):
def get(
self,
name: Union[str, int, 'mwclient.page.Page'],
info: Optional[Mapping[str, Any]] = None
) -> Union['mwclient.page.Page', 'mwclient.image.Image', 'Category']:
"""Return the page of name `name` as an object.
If self.namespace is not zero, use {namespace}:{name} as the
page name, otherwise guess the namespace from the name using
`self.guess_namespace`.
Args:
name: The name of the page as a string, the page ID as an int, or
another :class:`Page` object.
info: Page info, if already fetched, e.g., when iterating over a
list of pages. If not provided, the page info will be fetched
from the API.
Returns:
One of Category, Image or Page (default), according to namespace.
"""
if self.namespace != 0:
full_page_name = u"{namespace}:{name}".format(
namespace=self.site.namespaces[self.namespace],
name=name,
)
full_page_name = f"{self.site.namespaces[self.namespace]}:{name}" \
# type: Union[str, int, 'mwclient.page.Page']
namespace = self.namespace
else:
full_page_name = name
try:
if isinstance(name, str):
namespace = self.guess_namespace(name)
except AttributeError:
# raised when `namespace` doesn't have a `startswith` attribute
else:
namespace = 0
cls = {
......@@ -265,16 +329,16 @@ class PageList(GeneratorList):
6: mwclient.image.Image,
}.get(namespace, mwclient.page.Page)
return cls(self.site, full_page_name, info)
return cls(self.site, full_page_name, info) # type: ignore[no-any-return]
def guess_namespace(self, name):
def guess_namespace(self, name: str) -> int:
"""Guess the namespace from name
If name starts with any of the site's namespaces' names or
default_namespaces, use that. Else, return zero.
Args:
name (str): The pagename as a string (having `.startswith`)
name: The name of the page.
Returns:
The id of the guessed namespace or zero.
......@@ -282,26 +346,34 @@ class PageList(GeneratorList):
for ns in self.site.namespaces:
if ns == 0:
continue
namespace = '%s:' % self.site.namespaces[ns].replace(' ', '_')
namespace = f'{self.site.namespaces[ns].replace(" ", "_")}:'
if name.startswith(namespace):
return ns
elif ns in self.site.default_namespaces:
namespace = '%s:' % self.site.default_namespaces[ns].replace(' ', '_')
if name.startswith(namespace):
return ns
return 0
class PageProperty(List):
def __init__(self, page, prop, prefix, *args, **kwargs):
super(PageProperty, self).__init__(page.site, prop, prefix,
titles=page.name,
*args, **kwargs)
def __init__(
self,
page: 'mwclient.page.Page',
prop: str,
prefix: str,
*args: Tuple[str, Any],
**kwargs: Any
) -> None:
super().__init__(
page.site,
prop,
prefix,
titles=page.name,
*args, # type: ignore[arg-type]
**kwargs,
)
self.page = page
self.generator = 'prop'
def set_iter(self, data):
def set_iter(self, data: Mapping[str, Any]) -> None:
for page in data['query']['pages'].values():
if page['title'] == self.page.name:
self._iter = iter(page.get(self.list_name, ()))
......@@ -311,16 +383,21 @@ class PageProperty(List):
class PagePropertyGenerator(GeneratorList):
def __init__(self, page, prop, prefix, *args, **kwargs):
super(PagePropertyGenerator, self).__init__(page.site, prop, prefix,
titles=page.name,
*args, **kwargs)
def __init__(
self,
page: 'mwclient.page.Page',
prop: str,
prefix: str,
*args: Tuple[str, Any],
**kwargs: Any
) -> None:
super().__init__(page.site, prop, prefix, titles=page.name, *args, **kwargs)
self.page = page
class RevisionsIterator(PageProperty):
def load_chunk(self):
def load_chunk(self) -> None:
if 'rvstartid' in self.args and 'rvstart' in self.args:
del self.args['rvstart']
return super(RevisionsIterator, self).load_chunk()
return super().load_chunk()
mwclient/page.py
View file @ 15eb10cc
This diff is collapsed.
mwclient/sleep.py
View file @ 15eb10cc
import time
import logging
import time
from typing import Callable, Optional, Any
from mwclient.errors import MaximumRetriesExceeded
log = logging.getLogger(__name__)
......@@ -17,24 +19,30 @@ class Sleepers:
using the `make` method.
>>> sleeper = sleepers.make()
Args:
max_retries (int): The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry.
callback (Callable[[int, Any], None]): A callable to be called on each retry.
max_retries: The maximum number of retries to perform.
retry_timeout: The time to sleep for each past retry.
callback: A callable to be called on each retry.
Attributes:
max_retries (int): The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry.
callback (callable): A callable to be called on each retry.
max_retries: The maximum number of retries to perform.
retry_timeout: The time to sleep for each past retry.
callback: A callable to be called on each retry.
"""
def __init__(self, max_retries, retry_timeout, callback=lambda *x: None):
def __init__(
self,
max_retries: int,
retry_timeout: int,
callback: Callable[['Sleeper', int, Optional[Any]], Any] = lambda *x: None
) -> None:
self.max_retries = max_retries
self.retry_timeout = retry_timeout
self.callback = callback
def make(self, args=None):
def make(self, args: Optional[Any] = None) -> 'Sleeper':
"""
Creates a new `Sleeper` object.
Args:
args (Any): Arguments to be passed to the `callback` callable.
args: Arguments to be passed to the `callback` callable.
Returns:
Sleeper: A `Sleeper` object.
"""
......@@ -48,25 +56,32 @@ class Sleeper:
and a `MaximumRetriesExceeded` is raised. The sleeper object should be discarded
once the operation is successful.
Args:
args (Any): Arguments to be passed to the `callback` callable.
max_retries (int): The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry.
callback (callable, None]): A callable to be called on each retry.
args: Arguments to be passed to the `callback` callable.
max_retries: The maximum number of retries to perform.
retry_timeout: The time to sleep for each past retry.
callback: A callable to be called on each retry.
Attributes:
args (Any): Arguments to be passed to the `callback` callable.
retries (int): The number of retries that have been performed.
max_retries (int): The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry.
callback (callable): A callable to be called on each retry.
args: Arguments to be passed to the `callback` callable.
retries: The number of retries that have been performed.
max_retries: The maximum number of retries to perform.
retry_timeout: The time to sleep for each past retry.
callback: A callable to be called on each retry.
"""
def __init__(self, args, max_retries, retry_timeout, callback):
def __init__(
self,
args: Any,
max_retries: int,
retry_timeout: int,
callback: Callable[['Sleeper', int, Optional[Any]], Any]
) -> None:
self.args = args
self.retries = 0
self.max_retries = max_retries
self.retry_timeout = retry_timeout
self.callback = callback
def sleep(self, min_time=0):
def sleep(self, min_time: int = 0) -> None:
"""
Sleeps for a minimum of `min_time` seconds. The actual sleeping time will increase
with the number of retries.
......
mwclient/util.py
View file @ 15eb10cc
import time
import io
from typing import Optional, Iterable, Tuple, BinaryIO
import warnings
def parse_timestamp(t):
def parse_timestamp(t: Optional[str]) -> time.struct_time:
"""Parses a string containing a timestamp.
Args:
t (str): A string containing a timestamp.
t: A string containing a timestamp.
Returns:
time.struct_time: A timestamp.
......@@ -16,9 +18,38 @@ def parse_timestamp(t):
return time.strptime(t, '%Y-%m-%dT%H:%M:%SZ')
def read_in_chunks(stream, chunk_size):
def read_in_chunks(stream: BinaryIO, chunk_size: int) -> Iterable[io.BytesIO]:
while True:
data = stream.read(chunk_size)
if not data:
break
yield io.BytesIO(data)
def handle_limit(
limit: Optional[int], max_items: Optional[int], api_chunk_size: Optional[int]
) -> Tuple[Optional[int], Optional[int]]:
"""
Consistently handles 'limit', 'api_chunk_size' and 'max_items' -
https://github.com/mwclient/mwclient/issues/259 . In version 0.11,
'api_chunk_size' was introduced as a better name for 'limit', but
we still accept 'limit' with a deprecation warning. 'max_items'
does what 'limit' sounds like it should.
"""
if limit:
if api_chunk_size:
warnings.warn(
"limit and api_chunk_size both specified, this is not supported! limit "
"is deprecated, will use value of api_chunk_size",
DeprecationWarning
)
else:
warnings.warn(
"limit is deprecated as its name and purpose are confusing. use "
"api_chunk_size to set the number of items retrieved from the API at "
"once, and/or max_items to limit the total number of items that will be "
"yielded",
DeprecationWarning
)
api_chunk_size = limit
return (max_items, api_chunk_size)
pyproject.toml
View file @ 15eb10cc
[project]
name = "mwclient"
dynamic = ["version"]
description = "MediaWiki API client"
readme = "README.md"
requires-python = ">=3.6"
authors = [
{ name = "Bryan Tong Minh", email = "bryan.tongminh@gmail.com" },
]
keywords = ["mediawiki", "wikipedia"]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Internet :: WWW/HTTP",
"Topic :: Software Development :: Libraries :: Python Modules",
]
dependencies = [
"requests",
"requests-oauthlib",
]
[project.optional-dependencies]
docs = [
"sphinx",
"sphinx-rtd-theme",
]
testing = [
"pytest",
"pytest-cov",
"responses>=0.3.0",
"responses!=0.6.0",
"setuptools; python_version < '3.8'",
]
[project.urls]
Documentation = "https://mwclient.readthedocs.io/"
Repository = "https://github.com/mwclient/mwclient"
Issues = "https://github.com/mwclient/mwclient/issues"
Changelog = "https://github.com/mwclient/mwclient/releases"
[build-system]
requires = ["setuptools>=40.6.0", "wheel"]
build-backend = "setuptools.build_meta"
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.pytest.ini_options]
addopts = "--cov mwclient test"
[tool.hatch.version]
path = "mwclient/client.py"
[tool.hatch.build.targets.sdist]
exclude = [
"/.flake8",
"/.gitattributes",
"/.github",
"/.readthedocs.yaml",
]
[tool.bumpversion]
current_version = "0.10.1"
current_version = "0.11.0"
commit = true
tag = true
[[tool.bumpversion.files]]
filename = "setup.py"
search = "version='{current_version}'"
replace = "version='{new_version}'"
[[tool.bumpversion.files]]
filename = "mwclient/client.py"
search = "__version__ = '{current_version}'"
replace = "__version__ = '{new_version}'"
[[tool.bumpversion.files]]
filename = "README.md"
[tool.mypy]
packages = ["mwclient", "test"]
strict = true
warn_unreachable = true
[[tool.mypy.overrides]]
module = "test.*"
disallow_untyped_calls = false
disallow_untyped_defs = false
[[tool.mypy.overrides]]
module = "requests_oauthlib"
ignore_missing_imports = true
setup.py deleted 100644 → 0
View file @ 6c38cf50
#!/usr/bin/env python
import os
import sys
from setuptools import setup
here = os.path.abspath(os.path.dirname(__file__))
README = open(os.path.join(here, 'README.md')).read()
needs_pytest = set(['pytest', 'test', 'ptr']).intersection(sys.argv)
pytest_runner = ['pytest-runner'] if needs_pytest else []
setup(name='mwclient',
# See https://mwclient.readthedocs.io/en/latest/development/#making-a-release
# for how to update this field and release a new version.
version='0.10.1',
description='MediaWiki API client',
long_description=README,
long_description_content_type='text/markdown',
classifiers=[
'Programming Language :: Python',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
'Programming Language :: Python :: 3.11',
'Programming Language :: Python :: 3.12',
],
keywords='mediawiki wikipedia',
author='Bryan Tong Minh',
author_email='bryan.tongminh@gmail.com',
url='https://github.com/mwclient/mwclient',
license='MIT',
packages=['mwclient'],
install_requires=['requests-oauthlib'],
setup_requires=pytest_runner,
tests_require=['pytest', 'pytest-cov',
'responses>=0.3.0', 'responses!=0.6.0', 'setuptools'],
zip_safe=True
)
test/integration.py 0 → 100644
View file @ 15eb10cc
import http.client
import shutil
import subprocess
import time
from typing import Optional
from urllib.error import URLError, HTTPError
from urllib.request import urlopen
import pytest
import mwclient
if shutil.which("podman"):
_CONTAINER_RUNTIME = "podman"
elif shutil.which("docker"):
_CONTAINER_RUNTIME = "docker"
else:
raise RuntimeError("Neither podman nor docker is installed")
@pytest.fixture(
scope="class",
params=[("latest", "5002"), ("legacy", "5003"), ("lts", "5004")],
ids=["latest", "legacy", "lts"]
)
def site(request):
"""
Run a mediawiki container for the duration of the class, yield
a Site instance for it, then clean it up on exit. This is
parametrized so we get three containers and run the tests three
times for three mediawiki releases. We use podman because it's
much easier to use rootless than docker.
"""
(tag, port) = request.param
container = f"mwclient-{tag}"
# create the container, using upstream's official image. see
# https://hub.docker.com/_/mediawiki
args = [_CONTAINER_RUNTIME, "run", "--name", container, "-p", f"{port}:80",
"-d", f"docker.io/library/mediawiki:{tag}"]
subprocess.run(args)
# configure the wiki far enough that we can use the API. if you
# use this interactively the CSS doesn't work, I don't know why,
# don't think it really matters
args = [_CONTAINER_RUNTIME, "exec", container, "runuser", "-u", "www-data", "--",
"php", "/var/www/html/maintenance/install.php", "--server",
f"http://localhost:{port}", "--dbtype", "sqlite", "--pass", "weakpassword",
"--dbpath", "/var/www/data", "mwclient-test-wiki", "root"]
subprocess.run(args)
# create a regular user
args = [_CONTAINER_RUNTIME, "exec", container, "runuser", "-u", "www-data", "--",
"php", "/var/www/html/maintenance/createAndPromote.php", "testuser",
"weakpassword"]
subprocess.run(args)
# create an admin user
args = [_CONTAINER_RUNTIME, "exec", container, "runuser", "-u", "www-data", "--",
"php", "/var/www/html/maintenance/createAndPromote.php", "sysop",
"weakpassword", "--bureaucrat", "--sysop", "--interface-admin"]
subprocess.run(args)
# create a bot user
args = [_CONTAINER_RUNTIME, "exec", container, "runuser", "-u", "www-data", "--",
"php", "/var/www/html/maintenance/createAndPromote.php", "testbot",
"weakpassword", "--bot"]
subprocess.run(args)
# disable anonymous editing (we can't use redirection via podman
# exec for some reason, so we use sed)
args = [_CONTAINER_RUNTIME, "exec", container, "runuser", "-u", "www-data", "--",
"sed", "-i", r"$ a\$wgGroupPermissions['*']['edit'] = false;",
"/var/www/html/LocalSettings.php"]
subprocess.run(args)
# allow editing by users
args = [_CONTAINER_RUNTIME, "exec", container, "runuser", "-u", "www-data", "--",
"sed", "-i", r"$ a\$wgGroupPermissions['user']['edit'] = true;",
"/var/www/html/LocalSettings.php"]
subprocess.run(args)
# block until the server is actually running, up to 30 seconds
start = int(time.time())
resp: Optional[http.client.HTTPResponse] = None
while not resp:
try:
resp = urlopen(f"http://localhost:{port}")
except (ValueError, URLError, HTTPError) as err:
if int(time.time()) - start > 30:
print("Waited more than 30 seconds for server to start!")
raise err
else:
time.sleep(0.1)
# set up mwclient.site instance and yield it
yield mwclient.Site(f"localhost:{port}", path="/", scheme="http", force_login=False)
# -t=0 just hard stops it immediately, saves time
args = [_CONTAINER_RUNTIME, "stop", "-t=0", container]
subprocess.run(args)
args = [_CONTAINER_RUNTIME, "rm", container]
subprocess.run(args)
class TestAnonymous:
def test_page_load(self, site):
"""Test we can read a page from the sites."""
pg = site.pages["Main_Page"]
text = pg.text()
assert text.startswith("<strong>MediaWiki has been installed")
def test_page_create(self, site):
"""Test we get expected error if we try to create a page."""
pg = site.pages["Anonymous New Page"]
with pytest.raises(mwclient.errors.ProtectedPageError):
pg.edit("Hi I'm a new page", "create new page")
def test_expandtemplates(self, site):
"""Test we can expand templates."""
wikitext = site.expandtemplates("{{CURRENTYEAR}}")
assert wikitext == str(time.localtime().tm_year)
def test_expandtemplates_with_parsetree(self, site):
"""Test we can expand templates and get the parse tree."""
wikitext, parsetree = site.expandtemplates("{{CURRENTYEAR}}", generatexml=True)
assert wikitext == str(time.localtime().tm_year)
assert parsetree == "<root><template><title>CURRENTYEAR</title></template></root>"
class TestLogin:
def test_login_wrong_password(self, site):
"""Test we raise correct error for login() with wrong password."""
assert not site.logged_in
with pytest.raises(mwclient.errors.LoginError):
site.login(username="testuser", password="thisiswrong")
assert not site.logged_in
def test_login(self, site):
"""
Test we can log in to the sites with login() and do authed
stuff.
"""
site.login(username="testuser", password="weakpassword")
assert site.logged_in
# test we can create a page
pg = site.pages["Authed New Page"]
pg.edit("Hi I'm a new page", "create new page")
# we have to reinit because of Page.exists
# https://github.com/mwclient/mwclient/issues/354
pg = site.pages["Authed New Page"]
assert pg.text() == "Hi I'm a new page"
# test we can move it
ret = pg.move("Authed Moved Page")
pg = site.pages["Authed Moved Page"]
assert pg.text() == "Hi I'm a new page"
def test_page_delete(self, site):
"""Test we can login, create and delete a page as sysop."""
site.login(username="sysop", password="weakpassword")
pg = site.pages["Sysop New Page"]
pg.edit("Hi I'm a new page", "create new page")
pg = site.pages["Sysop New Page"]
assert pg.text() == "Hi I'm a new page"
assert pg.exists == True
pg.delete()
pg = site.pages["Sysop New Page"]
assert pg.text() == ""
assert pg.exists == False
class TestClientLogin:
def test_clientlogin_wrong_password(self, site):
"""Test we raise correct error for clientlogin() with wrong password."""
with pytest.raises(mwclient.errors.LoginError):
site.clientlogin(username="testuser", password="thisiswrong")
assert not site.logged_in
def test_clientlogin(self, site):
"""
Test we can log in to the site with clientlogin() and
create a page.
"""
site.clientlogin(username="testuser", password="weakpassword")
assert site.logged_in
pg = site.pages["Anonymous New Page"]
pg.edit("Hi I'm a new page", "create new page")
pg = site.pages["Anonymous New Page"]
assert pg.text() == "Hi I'm a new page"
test/test_client.py
View file @ 15eb10cc
from copy import deepcopy
from datetime import date
import json
from io import BytesIO
import logging
import sys
import time
import unittest
import unittest.mock as mock
from copy import deepcopy
from datetime import date
from io import StringIO
import mwclient
import pkg_resources # part of setuptools
import pytest
import requests
import responses
from requests_oauthlib import OAuth1
import mwclient
if __name__ == "__main__":
print()
print("Note: Running in stand-alone mode. Consult the README")
......@@ -26,24 +26,61 @@ logging.basicConfig(level=logging.DEBUG)
class TestCase(unittest.TestCase):
def metaResponse(self, **kwargs):
tpl = '{"query":{"general":{"generator":"MediaWiki %(version)s"},"namespaces":{"-1":{"*":"Special","canonical":"Special","case":"first-letter","id":-1},"-2":{"*":"Media","canonical":"Media","case":"first-letter","id":-2},"0":{"*":"","case":"first-letter","content":"","id":0},"1":{"*":"Talk","canonical":"Talk","case":"first-letter","id":1,"subpages":""},"10":{"*":"Template","canonical":"Template","case":"first-letter","id":10,"subpages":""},"100":{"*":"Test namespace 1","canonical":"Test namespace 1","case":"first-letter","id":100,"subpages":""},"101":{"*":"Test namespace 1 talk","canonical":"Test namespace 1 talk","case":"first-letter","id":101,"subpages":""},"102":{"*":"Test namespace 2","canonical":"Test namespace 2","case":"first-letter","id":102,"subpages":""},"103":{"*":"Test namespace 2 talk","canonical":"Test namespace 2 talk","case":"first-letter","id":103,"subpages":""},"11":{"*":"Template talk","canonical":"Template talk","case":"first-letter","id":11,"subpages":""},"1198":{"*":"Translations","canonical":"Translations","case":"first-letter","id":1198,"subpages":""},"1199":{"*":"Translations talk","canonical":"Translations talk","case":"first-letter","id":1199,"subpages":""},"12":{"*":"Help","canonical":"Help","case":"first-letter","id":12,"subpages":""},"13":{"*":"Help talk","canonical":"Help talk","case":"first-letter","id":13,"subpages":""},"14":{"*":"Category","canonical":"Category","case":"first-letter","id":14},"15":{"*":"Category talk","canonical":"Category talk","case":"first-letter","id":15,"subpages":""},"2":{"*":"User","canonical":"User","case":"first-letter","id":2,"subpages":""},"2500":{"*":"VisualEditor","canonical":"VisualEditor","case":"first-letter","id":2500},"2501":{"*":"VisualEditor talk","canonical":"VisualEditor talk","case":"first-letter","id":2501},"2600":{"*":"Topic","canonical":"Topic","case":"first-letter","defaultcontentmodel":"flow-board","id":2600},"3":{"*":"User talk","canonical":"User talk","case":"first-letter","id":3,"subpages":""},"4":{"*":"Wikipedia","canonical":"Project","case":"first-letter","id":4,"subpages":""},"460":{"*":"Campaign","canonical":"Campaign","case":"case-sensitive","defaultcontentmodel":"Campaign","id":460},"461":{"*":"Campaign talk","canonical":"Campaign talk","case":"case-sensitive","id":461},"5":{"*":"Wikipedia talk","canonical":"Project talk","case":"first-letter","id":5,"subpages":""},"6":{"*":"File","canonical":"File","case":"first-letter","id":6},"7":{"*":"File talk","canonical":"File talk","case":"first-letter","id":7,"subpages":""},"710":{"*":"TimedText","canonical":"TimedText","case":"first-letter","id":710},"711":{"*":"TimedText talk","canonical":"TimedText talk","case":"first-letter","id":711},"8":{"*":"MediaWiki","canonical":"MediaWiki","case":"first-letter","id":8,"subpages":""},"828":{"*":"Module","canonical":"Module","case":"first-letter","id":828,"subpages":""},"829":{"*":"Module talk","canonical":"Module talk","case":"first-letter","id":829,"subpages":""},"866":{"*":"CNBanner","canonical":"CNBanner","case":"first-letter","id":866},"867":{"*":"CNBanner talk","canonical":"CNBanner talk","case":"first-letter","id":867,"subpages":""},"9":{"*":"MediaWiki talk","canonical":"MediaWiki talk","case":"first-letter","id":9,"subpages":""},"90":{"*":"Thread","canonical":"Thread","case":"first-letter","id":90},"91":{"*":"Thread talk","canonical":"Thread talk","case":"first-letter","id":91},"92":{"*":"Summary","canonical":"Summary","case":"first-letter","id":92},"93":{"*":"Summary talk","canonical":"Summary talk","case":"first-letter","id":93}},"userinfo":{"anon":"","groups":["*"],"id":0,"name":"127.0.0.1","rights": %(rights)s}}}'
tpl = tpl % {'version': kwargs.get('version', '1.24wmf17'),
'rights': json.dumps(kwargs.get('rights', ["createaccount", "read", "edit", "createpage", "createtalk", "writeapi", "editmyusercss", "editmyuserjs", "viewmywatchlist", "editmywatchlist", "viewmyprivateinfo", "editmyprivateinfo", "editmyoptions", "centralauth-merge", "abusefilter-view", "abusefilter-log", "translate", "vipsscaler-test", "upload"]))
}
res = json.loads(tpl)
if kwargs.get('writeapi', True):
res['query']['general']['writeapi'] = ''
def metaResponse(self, version='1.24wmf17', rights=None):
if rights is None:
rights = [
"createaccount", "read", "edit", "createpage", "createtalk",
"editmyusercss", "editmyuserjs", "viewmywatchlist",
"editmywatchlist", "viewmyprivateinfo", "editmyprivateinfo",
"editmyoptions", "centralauth-merge", "abusefilter-view",
"abusefilter-log", "translate", "vipsscaler-test", "upload"
]
# @formatter:off
namespaces = {
-2: {"id": -2, "*": "Media", "canonical": "Media", "case": "first-letter"},
-1: {"id": -1, "*": "Special", "canonical": "Special", "case": "first-letter"},
0: {"id": 0, "*": "", "case": "first-letter", "content": ""},
1: {"id": 1, "*": "Talk", "canonical": "Talk", "case": "first-letter", "subpages": ""},
2: {"id": 2, "*": "User", "canonical": "User", "case": "first-letter", "subpages": ""},
3: {"id": 3, "*": "User talk", "canonical": "User talk", "case": "first-letter", "subpages": ""},
4: {"id": 4, "*": "Wikipedia", "canonical": "Project", "case": "first-letter", "subpages": ""},
5: {"id": 5, "*": "Wikipedia talk", "canonical": "Project talk", "case": "first-letter", "subpages": ""},
6: {"id": 6, "*": "File", "canonical": "File", "case": "first-letter"},
7: {"id": 7, "*": "File talk", "canonical": "File talk", "case": "first-letter", "subpages": ""},
8: {"id": 8, "*": "MediaWiki", "canonical": "MediaWiki", "case": "first-letter", "subpages": ""},
9: {"id": 9, "*": "MediaWiki talk", "canonical": "MediaWiki talk", "case": "first-letter", "subpages": ""},
10: {"id": 10, "*": "Template", "canonical": "Template", "case": "first-letter", "subpages": ""},
11: {"id": 11, "*": "Template talk", "canonical": "Template talk", "case": "first-letter", "subpages": ""},
12: {"id": 12, "*": "Help", "canonical": "Help", "case": "first-letter", "subpages": ""},
13: {"id": 13, "*": "Help talk", "canonical": "Help talk", "case": "first-letter", "subpages": ""},
14: {"id": 14, "*": "Category", "canonical": "Category", "case": "first-letter"},
15: {"id": 15, "*": "Category talk", "canonical": "Category talk", "case": "first-letter", "subpages": ""},
}
# @formatter:on
return res
return {
"query": {
"general": {
"generator": f"MediaWiki {version}"
},
"namespaces": namespaces,
"userinfo": {
"anon": "",
"groups": ["*"],
"id": 0,
"name": "127.0.0.1",
"rights": rights
}
}
}
def metaResponseAsJson(self, **kwargs):
return json.dumps(self.metaResponse(**kwargs))
def httpShouldReturn(self, body=None, callback=None, scheme='https', host='test.wikipedia.org', path='/w/',
script='api', headers=None, status=200, method='GET'):
url = '{scheme}://{host}{path}{script}.php'.format(scheme=scheme, host=host, path=path, script=script)
url = f'{scheme}://{host}{path}{script}.php'
mock = responses.GET if method == 'GET' else responses.POST
if body is None:
responses.add_callback(mock, url, callback=callback)
......@@ -90,7 +127,15 @@ class TestClient(TestCase):
def testVersion(self):
# The version specified in setup.py should equal the one specified in client.py
version = pkg_resources.require("mwclient")[0].version
if sys.version_info >= (3, 8):
import importlib.metadata
version = importlib.metadata.version("mwclient")
else:
import pkg_resources # part of setuptools
version = pkg_resources.require("mwclient")[0].version
assert version == mwclient.__version__
......@@ -120,8 +165,8 @@ class TestClient(TestCase):
site = mwclient.Site('test.wikipedia.org')
assert len(responses.calls) == 2
assert 'retry-after' in responses.calls[0].response.headers
assert 'retry-after' not in responses.calls[1].response.headers
assert 'retry-after' in responses.calls[0].response.headers # type: ignore
assert 'retry-after' not in responses.calls[1].response.headers # type: ignore
@responses.activate
def test_http_error(self):
......@@ -169,6 +214,7 @@ class TestClient(TestCase):
site = mwclient.Site('test.wikipedia.org')
assert responses.calls[0].request.url is not None
assert 'action=query' in responses.calls[0].request.url
assert 'meta=siteinfo%7Cuserinfo' in responses.calls[0].request.url
......@@ -196,7 +242,8 @@ class TestClient(TestCase):
self.httpShouldReturn(self.metaResponseAsJson())
with pytest.raises(RuntimeError):
site = mwclient.Site('test.wikipedia.org', httpauth=1)
site = mwclient.Site('test.wikipedia.org',
httpauth=1) # type: ignore[arg-type]
@responses.activate
def test_oauth(self):
......@@ -353,9 +400,9 @@ class TestClient(TestCase):
}
}), method='GET')
answers = set(result['fulltext'] for result in site.ask('test'))
answers = {result['fulltext'] for result in site.ask('test')}
assert answers == set(('Serendipitet', 'Indeks (bibliotekfag)'))
assert answers == {'Serendipitet', 'Indeks (bibliotekfag)'}
@responses.activate
def test_smw_response_v2(self):
......@@ -395,9 +442,9 @@ class TestClient(TestCase):
}
}), method='GET')
answers = set(result['fulltext'] for result in site.ask('test'))
answers = {result['fulltext'] for result in site.ask('test')}
assert answers == set(('Serendipitet', 'Indeks (bibliotekfag)'))
assert answers == {'Serendipitet', 'Indeks (bibliotekfag)'}
@responses.activate
def test_repr(self):
......@@ -469,6 +516,18 @@ class TestClient(TestCase):
site.raw_api("query", "GET", retry_on_error=False)
assert timesleep.call_count == 25
@responses.activate
def test_connection_options(self):
self.httpShouldReturn(self.metaResponseAsJson())
args = {"timeout": 60, "stream": False}
site = mwclient.Site('test.wikipedia.org', connection_options=args)
assert site.requests == args
with pytest.warns(DeprecationWarning):
site = mwclient.Site('test.wikipedia.org', reqs=args)
assert site.requests == args
with pytest.raises(ValueError):
site = mwclient.Site('test.wikipedia.org', reqs=args, connection_options=args)
class TestLogin(TestCase):
@mock.patch('mwclient.client.Site.site_init')
......@@ -553,7 +612,7 @@ class TestLogin(TestCase):
# this would be done by site_init usually, but we're mocking it
site.version = (1, 28, 0)
success = site.clientlogin(username='myusername', password='mypassword')
url = '%s://%s' % (site.scheme, site.host)
url = f'{site.scheme}://{site.host}'
call_args = raw_api.call_args_list
......@@ -602,7 +661,7 @@ class TestLogin(TestCase):
'clientlogin', 'POST',
username='myusername',
password='mypassword',
loginreturnurl='%s://%s' % (site.scheme, site.host),
loginreturnurl=f'{site.scheme}://{site.host}',
logintoken=login_token
)
......@@ -629,7 +688,7 @@ class TestLogin(TestCase):
# this would be done by site_init usually, but we're mocking it
site.version = (1, 28, 0)
success = site.clientlogin(username='myusername', password='mypassword')
url = '%s://%s' % (site.scheme, site.host)
url = f'{site.scheme}://{site.host}'
call_args = raw_api.call_args_list
......@@ -684,6 +743,47 @@ class TestClientApiMethods(TestCase):
assert revisions[1]['revid'] == 689816909
class TestVersionTupleFromGenerator:
@pytest.mark.parametrize('version, expected', [
('MediaWiki 1.24', (1, 24)),
('MediaWiki 1.24.0', (1, 24, 0)),
('MediaWiki 1.24.0-wmf.1', (1, 24, 0, 'wmf', 1)),
('MediaWiki 1.24.1alpha', (1, 24, 1, 'alpha')),
('MediaWiki 1.24.1alpha1', (1, 24, 1, 'alpha', 1)),
('MediaWiki 1.24.1-rc.3', (1, 24, 1, 'rc', 3)),
])
def test_version_tuple_from_generator(self, version, expected):
assert mwclient.Site.version_tuple_from_generator(version) == expected
def test_version_tuple_from_generator_empty(self):
with pytest.raises(mwclient.errors.MediaWikiVersionError):
mwclient.Site.version_tuple_from_generator('')
def test_version_tuple_from_generator_invalid_prefix(self):
with pytest.raises(mwclient.errors.MediaWikiVersionError):
mwclient.Site.version_tuple_from_generator('Foo 1.24.1')
def test_version_tuple_from_generator_no_minor(self):
with pytest.raises(mwclient.errors.MediaWikiVersionError):
mwclient.Site.version_tuple_from_generator('MediaWiki 1')
def test_version_tuple_from_generator_major_is_not_number(self):
with pytest.raises(mwclient.errors.MediaWikiVersionError):
mwclient.Site.version_tuple_from_generator('MediaWiki foo.24.1')
def test_version_tuple_from_generator_minor_is_not_number(self):
with pytest.raises(mwclient.errors.MediaWikiVersionError):
mwclient.Site.version_tuple_from_generator('MediaWiki 1.foo.1')
def test_version_tuple_from_generator_major_and_minor_are_not_numbers(self):
with pytest.raises(mwclient.errors.MediaWikiVersionError):
mwclient.Site.version_tuple_from_generator('MediaWiki foo.bar.1')
def test_version_tuple_from_generator_patch_is_not_number(self):
assert mwclient.Site.version_tuple_from_generator('MediaWiki 1.24.foo') == (1, 24, 'foo')
class TestClientUploadArgs(TestCase):
def setUp(self):
......@@ -734,7 +834,7 @@ class TestClientUploadArgs(TestCase):
# Test that methods are called, and arguments sent as expected
self.configure()
self.site.upload(file=StringIO('test'), filename=self.vars['fname'], comment=self.vars['comment'])
self.site.upload(file=BytesIO(b'test'), filename=self.vars['fname'], comment=self.vars['comment'])
args, kwargs = self.raw_call.call_args
data = args[1]
......@@ -750,19 +850,46 @@ class TestClientUploadArgs(TestCase):
self.configure()
with pytest.raises(TypeError):
self.site.upload(file=StringIO('test'))
self.site.upload(file=BytesIO(b'test'))
def test_upload_ambigitious_args(self):
self.configure()
with pytest.raises(TypeError):
self.site.upload(filename='Test', file=StringIO('test'), filekey='abc')
self.site.upload(filename='Test', file=BytesIO(b'test'), filekey='abc')
def test_upload_missing_upload_permission(self):
self.configure(rights=['read'])
with pytest.raises(mwclient.errors.InsufficientPermission):
self.site.upload(filename='Test', file=StringIO('test'))
self.site.upload(filename='Test', file=BytesIO(b'test'))
def test_upload_file_exists(self):
self.configure()
self.raw_call.side_effect = [
self.makePageResponse(title='File:Test.jpg', imagerepository='local',
imageinfo=[{
"comment": "",
"height": 1440,
"metadata": [],
"sha1": "69a764a9cf8307ea4130831a0aa0b9b7f9585726",
"size": 123,
"timestamp": "2013-12-22T07:11:07Z",
"user": "TestUser",
"width": 2160
}]),
json.dumps({'query': {'tokens': {'csrftoken': self.vars['token']}}}),
json.dumps({
'upload': {'result': 'Warning',
'warnings': {'duplicate': ['Test.jpg'],
'exists': 'Test.jpg'},
'filekey': '1apyzwruya84.da2cdk.1.jpg',
'sessionkey': '1apyzwruya84.da2cdk.1.jpg'}
})
]
with pytest.raises(mwclient.errors.FileExists):
self.site.upload(file=BytesIO(b'test'), filename='Test.jpg', ignore=False)
class TestClientGetTokens(TestCase):
......@@ -1420,6 +1547,74 @@ class TestUser(TestCase):
assert real_call_kwargs == mock_call_kwargs
assert mock_call.args == call_args[2].args
class TestClientExpandtemplates(TestCase):
def setUp(self):
self.raw_call = mock.patch('mwclient.client.Site.raw_call').start()
def configure(self, version='1.24'):
self.raw_call.return_value = self.metaResponseAsJson(version=version)
self.site = mwclient.Site('test.wikipedia.org')
def tearDown(self):
mock.patch.stopall()
def test_expandtemplates_1_13(self):
self.configure('1.16')
self.raw_call.return_value = json.dumps({
'expandtemplates': {
'*': '2024'
}
})
wikitext = self.site.expandtemplates('{{CURRENTYEAR}}')
assert wikitext == '2024'
def test_expandtemplates_1_13_generatexml(self):
self.configure('1.16')
self.raw_call.return_value = json.dumps({
'parsetree': {
'*': '<root><template><title>CURRENTYEAR</title></template></root>'
},
'expandtemplates': {
'*': '2024'
}
})
expanded = self.site.expandtemplates('{{CURRENTYEAR}}', generatexml=True)
assert isinstance(expanded, tuple)
assert expanded[0] == '2024'
assert expanded[1] == '<root><template><title>CURRENTYEAR</title></template></root>'
def test_expandtemplates_1_24(self):
self.configure('1.24')
self.raw_call.return_value = json.dumps({
'expandtemplates': {
'wikitext': '2024'
}
})
wikitext = self.site.expandtemplates('{{CURRENTYEAR}}')
assert wikitext == '2024'
def test_expandtemplates_1_24_generatexml(self):
self.configure('1.24')
self.raw_call.return_value = json.dumps({
'expandtemplates': {
'parsetree': '<root><template><title>CURRENTYEAR</title></template></root>',
'wikitext': '2024'
}
})
expanded = self.site.expandtemplates('{{CURRENTYEAR}}', generatexml=True)
assert isinstance(expanded, tuple)
assert expanded[0] == '2024'
assert expanded[1] == '<root><template><title>CURRENTYEAR</title></template></root>'
if __name__ == '__main__':
unittest.main()
test/test_listing.py
View file @ 15eb10cc
import unittest
import pytest
import logging
import requests
import responses
import json
import mwclient
from mwclient.listing import List, GeneratorList
from mwclient.listing import List, NestedList, GeneratorList
from mwclient.listing import Category, PageList, RevisionsIterator
from mwclient.page import Page
import unittest.mock as mock
......@@ -22,7 +20,54 @@ class TestList(unittest.TestCase):
def setUp(self):
pass
def setupDummyResponses(self, mock_site, result_member, ns=None):
def setupDummyResponsesOne(self, mock_site, result_member, ns=None):
if ns is None:
ns = [0, 0, 0]
mock_site.get.side_effect = [
{
'continue': {
'apcontinue': 'Kre_Mbaye',
'continue': '-||'
},
'query': {
result_member: [
{
"pageid": 19839654,
"ns": ns[0],
"title": "Kre'fey",
},
]
}
},
{
'continue': {
'apcontinue': 'Kre_Blip',
'continue': '-||'
},
'query': {
result_member: [
{
"pageid": 19839654,
"ns": ns[1],
"title": "Kre-O",
}
]
}
},
{
'query': {
result_member: [
{
"pageid": 30955295,
"ns": ns[2],
"title": "Kre-O Transformers",
}
]
}
},
]
def setupDummyResponsesTwo(self, mock_site, result_member, ns=None):
if ns is None:
ns = [0, 0, 0]
mock_site.get.side_effect = [
......@@ -64,19 +109,65 @@ class TestList(unittest.TestCase):
# Test that the list fetches all three responses
# and yields dicts when return_values not set
lst = List(mock_site, 'allpages', 'ap', limit=2)
self.setupDummyResponses(mock_site, 'allpages')
lst = List(mock_site, 'allpages', 'ap', api_chunk_size=2)
self.setupDummyResponsesTwo(mock_site, 'allpages')
vals = [x for x in lst]
assert len(vals) == 3
assert type(vals[0]) == dict
assert lst.args["aplimit"] == "2"
assert mock_site.get.call_count == 2
@mock.patch('mwclient.client.Site')
def test_list_limit_deprecated(self, mock_site):
# Test that the limit arg acts as api_chunk_size but generates
# DeprecationWarning
with pytest.deprecated_call():
lst = List(mock_site, 'allpages', 'ap', limit=2)
self.setupDummyResponsesTwo(mock_site, 'allpages')
vals = [x for x in lst]
assert len(vals) == 3
assert type(vals[0]) == dict
assert lst.args["aplimit"] == "2"
assert mock_site.get.call_count == 2
@mock.patch('mwclient.client.Site')
def test_list_max_items(self, mock_site):
# Test that max_items properly caps the list
# iterations
mock_site.api_limit = 500
lst = List(mock_site, 'allpages', 'ap', max_items=2)
self.setupDummyResponsesTwo(mock_site, 'allpages')
vals = [x for x in lst]
assert len(vals) == 2
assert type(vals[0]) == dict
assert lst.args["aplimit"] == "2"
assert mock_site.get.call_count == 1
@mock.patch('mwclient.client.Site')
def test_list_max_items_continuation(self, mock_site):
# Test that max_items and api_chunk_size work together
mock_site.api_limit = 500
lst = List(mock_site, 'allpages', 'ap', max_items=2, api_chunk_size=1)
self.setupDummyResponsesOne(mock_site, 'allpages')
vals = [x for x in lst]
assert len(vals) == 2
assert type(vals[0]) == dict
assert lst.args["aplimit"] == "1"
assert mock_site.get.call_count == 2
@mock.patch('mwclient.client.Site')
def test_list_with_str_return_value(self, mock_site):
# Test that the List yields strings when return_values is string
lst = List(mock_site, 'allpages', 'ap', limit=2, return_values='title')
self.setupDummyResponses(mock_site, 'allpages')
self.setupDummyResponsesTwo(mock_site, 'allpages')
vals = [x for x in lst]
assert len(vals) == 3
......@@ -88,18 +179,116 @@ class TestList(unittest.TestCase):
lst = List(mock_site, 'allpages', 'ap', limit=2,
return_values=('title', 'ns'))
self.setupDummyResponses(mock_site, 'allpages')
self.setupDummyResponsesTwo(mock_site, 'allpages')
vals = [x for x in lst]
assert len(vals) == 3
assert type(vals[0]) == tuple
@mock.patch('mwclient.client.Site')
def test_list_empty(self, mock_site):
# Test that we handle an empty response from get correctly
# (stop iterating)
lst = List(mock_site, 'allpages', 'ap', limit=2,
return_values=('title', 'ns'))
mock_site.get.side_effect = [{}]
vals = [x for x in lst]
assert len(vals) == 0
@mock.patch('mwclient.client.Site')
def test_list_invalid(self, mock_site):
# Test that we handle the response for a list that doesn't
# exist correctly (set an empty iterator, then stop
# iterating)
mock_site.api_limit = 500
lst = List(mock_site, 'allpagess', 'ap')
mock_site.get.side_effect = [
{
'batchcomplete': '',
'warnings': {
'main': {'*': 'Unrecognized parameter: aplimit.'},
'query': {'*': 'Unrecognized value for parameter "list": allpagess'}
},
'query': {
'userinfo': {
'id': 0,
'name': 'DEAD:BEEF:CAFE',
'anon': ''
}
}
}
]
vals = [x for x in lst]
assert len(vals) == 0
@mock.patch('mwclient.client.Site')
def test_list_repr(self, mock_site):
# Test __repr__ of a List is as expected
mock_site.__str__.return_value = "some wiki"
lst = List(mock_site, 'allpages', 'ap', limit=2,
return_values=('title', 'ns'))
assert repr(lst) == "<List object 'allpages' for some wiki>"
@mock.patch('mwclient.client.Site')
def test_get_list(self, mock_site):
# Test get_list behaves as expected
lst = List.get_list()(mock_site, 'allpages', 'ap', limit=2,
return_values=('title', 'ns'))
genlst = List.get_list(True)(mock_site, 'allpages', 'ap', limit=2,
return_values=('title', 'ns'))
assert isinstance(lst, List)
assert not isinstance(lst, GeneratorList)
assert isinstance(genlst, GeneratorList)
@mock.patch('mwclient.client.Site')
def test_nested_list(self, mock_site):
# Test NestedList class works as expected
mock_site.api_limit = 500
nested = NestedList('entries', mock_site, 'checkuserlog', 'cul')
mock_site.get.side_effect = [
# this is made-up because I do not have permissions on any
# wiki with this extension installed and the extension doc
# does not show a sample API response
{
'query': {
'checkuserlog': {
'entries': [
{
'user': 'Dreamyjazz',
'action': 'users',
'ip': '172.18.0.1',
'message': 'suspected sockpuppet',
'time': 1662328680
},
{
'user': 'Dreamyjazz',
'action': 'ip',
'targetuser': 'JohnDoe124',
'message': 'suspected sockpuppet',
'time': 1662328380
},
]
}
}
}
]
vals = [x for x in nested]
assert len(vals) == 2
assert vals[0]['action'] == 'users'
assert vals[1]['action'] == 'ip'
@mock.patch('mwclient.client.Site')
def test_generator_list(self, mock_site):
# Test that the GeneratorList yields Page objects
mock_site.api_limit = 500
lst = GeneratorList(mock_site, 'pages', 'p')
self.setupDummyResponses(mock_site, 'pages', ns=[0, 6, 14])
self.setupDummyResponsesTwo(mock_site, 'pages', ns=[0, 6, 14])
vals = [x for x in lst]
assert len(vals) == 3
......@@ -107,5 +296,172 @@ class TestList(unittest.TestCase):
assert type(vals[1]) == mwclient.image.Image
assert type(vals[2]) == mwclient.listing.Category
@mock.patch('mwclient.client.Site')
def test_category(self, mock_site):
# Test that Category works as expected
mock_site.__str__.return_value = "some wiki"
mock_site.api_limit = 500
# first response is for Page.__init__ as Category inherits
# from both Page and GeneratorList, second response is for
# the Category treated as an iterator with the namespace
# filter applied, third response is for the Category.members()
# call without a namespace filter
mock_site.get.side_effect = [
{
'query': {
'pages': {
'54565': {
'pageid': 54565, 'ns': 14, 'title': 'Category:Furry things'
}
}
}
},
{
'query': {
'pages': {
'36245': {
'pageid': 36245,
'ns': 118,
'title': 'Draft:Cat'
},
'36275': {
'pageid': 36275,
'ns': 118,
'title': 'Draft:Dog'
}
}
}
},
{
'query': {
'pages': {
'36245': {
'pageid': 36245,
'ns': 118,
'title': 'Draft:Cat'
},
'36275': {
'pageid': 36275,
'ns': 118,
'title': 'Draft:Dog'
},
'36295': {
'pageid': 36295,
'ns': 0,
'title': 'Hamster'
}
}
}
},
]
cat = Category(mock_site, 'Category:Furry things', namespace=118)
assert repr(cat) == "<Category object 'Category:Furry things' for some wiki>"
assert cat.args['gcmnamespace'] == 118
vals = [x for x in cat]
assert len(vals) == 2
assert vals[0].name == "Draft:Cat"
newcat = cat.members()
assert 'gcmnamespace' not in newcat.args
vals = [x for x in newcat]
assert len(vals) == 3
assert vals[2].name == "Hamster"
@mock.patch('mwclient.client.Site')
def test_pagelist(self, mock_site):
# Test that PageList works as expected
mock_site.__str__.return_value = "some wiki"
mock_site.api_limit = 500
mock_site.namespaces = {0: "", 6: "Image", 14: "Category"}
mock_site.get.return_value = {
'query': {
'pages': {
'8052484': {
'pageid': 8052484, 'ns': 0, 'title': 'Impossible'
}
}
}
}
pl = PageList(mock_site, start="Herring", end="Marmalade")
assert pl.args["gapfrom"] == "Herring"
assert pl.args["gapto"] == "Marmalade"
pg = pl["Impossible"]
assert isinstance(pg, Page)
assert mock_site.get.call_args[0] == ("query",)
assert mock_site.get.call_args[1]["titles"] == "Impossible"
# covers the catch of AttributeError in get()
pg = pl[8052484]
assert isinstance(pg, Page)
assert mock_site.get.call_args[0] == ("query",)
assert mock_site.get.call_args[1]["pageids"] == 8052484
pg = pl["Category:Spreads"]
assert mock_site.get.call_args[1]["titles"] == "Category:Spreads"
assert isinstance(pg, Category)
pl = PageList(mock_site, prefix="Ham")
assert pl.args["gapprefix"] == "Ham"
@mock.patch('mwclient.client.Site')
def test_revisions_iterator(self, mock_site):
# Test RevisionsIterator, including covering a line of
# PageProperty.set_iter
mock_site.api_limit = 500
mock_site.get.return_value = {
'query': {
'pages': {
'8052484': {
'pageid': 8052484,
'ns': 0,
'title': 'Impossible',
'revisions': [
{
"revid": 5000,
"parentid": 4999,
"user": "Bob",
"comment": "an edit"
},
{
"revid": 4999,
"parentid": 4998,
"user": "Alice",
"comment": "an earlier edit"
}
]
}
}
}
}
page = mock.MagicMock()
page.site = mock_site
page.name = "Impossible"
rvi = RevisionsIterator(
page, "revisions", "rv", rvstartid=5, rvstart="2001-01-15T14:56:00Z"
)
assert "rvstart" in rvi.args and "rvstartid" in rvi.args
vals = [x for x in rvi]
assert "rvstart" not in rvi.args and "rvstartid" in rvi.args
assert len(vals) == 2
assert vals[0]["comment"] == "an edit"
assert vals[1]["comment"] == "an earlier edit"
# now test the StopIteration line in PageProperty.set_iter
# by mocking a return value for a different page
mock_site.get.return_value = {
'query': {
'pages': {
'8052485': {
'pageid': '8052485',
'ns': 0,
'title': 'Impractical'
}
}
}
}
rvi = RevisionsIterator(
page, "revisions", "rv", rvstartid=5, rvstart="2001-01-15T14:56:00Z"
)
vals = [x for x in rvi]
assert len(vals) == 0
if __name__ == '__main__':
unittest.main()
test/test_page.py
View file @ 15eb10cc
This diff is collapsed.
tox.ini
View file @ 15eb10cc
[tox]
envlist = py35,py36,py37,py38,py39,py310,py311,py312,py313,flake
envlist = py36,py37,py38,py39,py310,py311,py312,py313,py314,flake,mypy
isolated_build = true
[gh-actions]
python =
......@@ -8,21 +9,32 @@ python =
3.8: py38
3.9: py39
3.10: py310
3.11: py311, flake
3.11: py311
3.12: py312
3.13: py313
3.13: py313, flake, integration, mypy
3.14: py314
[testenv]
deps =
pytest
pytest-cov
responses
setuptools
mock
commands = py.test -v --cov mwclient test
extras = testing
commands = pytest -v --cov mwclient test
[testenv:flake]
deps =
flake8
commands =
flake8 mwclient
[testenv:integration]
deps =
pytest
commands = pytest test/integration.py -v
[testenv:mypy]
deps =
mypy
pytest
responses
types-requests
types-setuptools
commands =
mypy