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
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], ...]
This diff is collapsed.
from typing import Any, TYPE_CHECKING, Optional, cast
if TYPE_CHECKING:
import mwclient.page
class MwClientError(RuntimeError): class MwClientError(RuntimeError):
"""Base class for all mwclient errors."""
pass pass
class MediaWikiVersionError(MwClientError): class MediaWikiVersionError(MwClientError):
"""The version of MediaWiki is not supported."""
pass pass
class APIDisabledError(MwClientError): class APIDisabledError(MwClientError):
"""The API is disabled on the wiki."""
pass pass
class MaximumRetriesExceeded(MwClientError): class MaximumRetriesExceeded(MwClientError):
"""The maximum number of retries for a request has been exceeded."""
pass pass
class APIError(MwClientError): 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.code = code
self.info = info self.info = info
super(APIError, self).__init__(code, info, kwargs) super().__init__(code, info, kwargs)
class UserNotFound(APIError): class UserNotFound(APIError):
...@@ -31,38 +48,80 @@ class UserCreateError(APIError): ...@@ -31,38 +48,80 @@ class UserCreateError(APIError):
class InsufficientPermission(MwClientError): class InsufficientPermission(MwClientError):
"""Raised when the user does not have sufficient permissions to perform an
action."""
pass pass
class UserBlocked(InsufficientPermission): class UserBlocked(InsufficientPermission):
"""Raised when attempting to perform an action while blocked."""
pass pass
class EditError(MwClientError): class EditError(MwClientError):
"""Base class for errors related to editing pages."""
pass pass
class ProtectedPageError(EditError, InsufficientPermission): class ProtectedPageError(EditError, InsufficientPermission):
"""Raised when attempting to edit a protected page.
def __init__(self, page, code=None, info=None):
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.page = page
self.code = code self.code = code
self.info = info self.info = info
def __str__(self): def __str__(self) -> str:
if self.info is not None: if self.info is not None:
return self.info return self.info
return 'You do not have the "edit" right.' return 'You do not have the "edit" right.'
class FileExists(EditError): 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): class LoginError(MwClientError):
super(LoginError, self).__init__( """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, site,
{'result': code, 'reason': info} # For backwards-compability {'result': code, 'reason': info} # For backwards-compability
) )
...@@ -70,43 +129,54 @@ class LoginError(MwClientError): ...@@ -70,43 +129,54 @@ class LoginError(MwClientError):
self.code = code self.code = code
self.info = info self.info = info
def __str__(self): def __str__(self) -> str:
return self.info return self.info
class OAuthAuthorizationError(LoginError): 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 pass
class AssertUserFailedError(MwClientError): class AssertUserFailedError(MwClientError):
"""Raised when the user assertion fails."""
def __init__(self): def __init__(self) -> None:
super(AssertUserFailedError, self).__init__(( super().__init__(
'By default, mwclient protects you from accidentally editing ' 'By default, mwclient protects you from accidentally editing '
'without being logged in. If you actually want to edit without ' 'without being logged in. If you actually want to edit without '
'logging in, you can set force_login on the Site object to False.' 'logging in, you can set force_login on the Site object to False.'
)) )
def __str__(self): def __str__(self) -> str:
return self.args[0] return cast(str, self.args[0])
class EmailError(MwClientError): class EmailError(MwClientError):
"""Base class for email errors."""
pass pass
class NoSpecifiedEmail(EmailError): class NoSpecifiedEmail(EmailError):
pass """Raised when trying to email a user who has not specified an email"""
class NoWriteApi(MwClientError):
pass pass
class InvalidResponse(MwClientError): 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): def __init__(self, response_text: Optional[str] = None) -> None:
super(InvalidResponse, self).__init__(( super().__init__((
'Did not get a valid JSON response from the server. Check that ' 'Did not get a valid JSON response from the server. Check that '
'you used the correct hostname. If you did, the server might ' 'you used the correct hostname. If you did, the server might '
'be wrongly configured or experiencing temporary problems.'), 'be wrongly configured or experiencing temporary problems.'),
...@@ -114,9 +184,10 @@ class InvalidResponse(MwClientError): ...@@ -114,9 +184,10 @@ class InvalidResponse(MwClientError):
) )
self.response_text = response_text self.response_text = response_text
def __str__(self): def __str__(self) -> str:
return self.args[0] return cast(str, self.args[0])
class InvalidPageTitle(MwClientError): class InvalidPageTitle(MwClientError):
"""Raised when an invalid page title is used."""
pass pass
import io
from typing import Optional, Mapping, Any, overload
import mwclient.listing import mwclient.listing
import mwclient.page import mwclient.page
from mwclient._types import Namespace
from mwclient.util import handle_limit
class Image(mwclient.page.Page): 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): def __init__(
super(Image, self).__init__( self,
site: 'mwclient.client.Site',
name: str,
info: Optional[Mapping[str, Any]] = None
) -> None:
super().__init__(
site, name, info, extra_properties={ site, name, info, extra_properties={
'imageinfo': ( 'imageinfo': (
('iiprop', ('iiprop',
...@@ -16,7 +38,7 @@ class Image(mwclient.page.Page): ...@@ -16,7 +38,7 @@ class Image(mwclient.page.Page):
self.imagerepository = self._info.get('imagerepository', '') self.imagerepository = self._info.get('imagerepository', '')
self.imageinfo = self._info.get('imageinfo', ({}, ))[0] self.imageinfo = self._info.get('imageinfo', ({}, ))[0]
def imagehistory(self): def imagehistory(self) -> 'mwclient.listing.PageProperty':
""" """
Get file revision info for the given file. Get file revision info for the given file.
...@@ -27,8 +49,16 @@ class Image(mwclient.page.Page): ...@@ -27,8 +49,16 @@ class Image(mwclient.page.Page):
iiprop='timestamp|user|comment|url|size|sha1|metadata|mime|archivename' iiprop='timestamp|user|comment|url|size|sha1|metadata|mime|archivename'
) )
def imageusage(self, namespace=None, filterredir='all', redirect=False, def imageusage(
limit=None, generator=True): 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. List pages that use the given file.
...@@ -38,21 +68,53 @@ class Image(mwclient.page.Page): ...@@ -38,21 +68,53 @@ class Image(mwclient.page.Page):
kwargs = dict(mwclient.listing.List.generate_kwargs( kwargs = dict(mwclient.listing.List.generate_kwargs(
prefix, title=self.name, namespace=namespace, filterredir=filterredir prefix, title=self.name, namespace=namespace, filterredir=filterredir
)) ))
(max_items, api_chunk_size) = handle_limit(limit, max_items, api_chunk_size)
if redirect: if redirect:
kwargs['%sredirect' % prefix] = '1' kwargs[f'{prefix}redirect'] = '1'
return mwclient.listing.List.get_list(generator)( 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. List duplicates of the current file.
API doc: https://www.mediawiki.org/wiki/API:Duplicatefiles 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 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 directly to the stream. Otherwise the file content will be stored in memory
...@@ -64,19 +126,16 @@ class Image(mwclient.page.Page): ...@@ -64,19 +126,16 @@ class Image(mwclient.page.Page):
... image.download(fd) ... image.download(fd)
Args: Args:
destination (file object): Destination file destination: Destination file
""" """
url = self.imageinfo['url'] url = self.imageinfo['url']
if destination is not None: if destination is not None:
res = self.site.connection.get(url, stream=True) res = self.site.connection.get(url, stream=True)
for chunk in res.iter_content(1024): for chunk in res.iter_content(1024):
destination.write(chunk) destination.write(chunk)
return None
else: else:
return self.site.connection.get(url).content return self.site.connection.get(url).content
def __repr__(self): def __repr__(self) -> str:
return "<%s object '%s' for %s>" % ( return f"<{self.__class__.__name__} object '{self.name}' for {self.site}>"
self.__class__.__name__,
self.name.encode('utf-8'),
self.site
)
from mwclient.util import parse_timestamp from typing import ( # noqa: F401
import mwclient.page Optional, Tuple, Any, Union, Iterator, Mapping, Iterable, Type, Dict
)
import mwclient.image import mwclient.image
import mwclient.page
from mwclient._types import Namespace
from mwclient.util import parse_timestamp, handle_limit
class List: class List:
...@@ -9,11 +14,27 @@ class List: ...@@ -9,11 +14,27 @@ class List:
This is a class providing lazy iteration. This means that the This is a class providing lazy iteration. This means that the
content is loaded in chunks as long as the response hints at content is loaded in chunks as long as the response hints at
continuing content. 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, def __init__(
limit=None, return_values=None, max_items=None, self,
*args, **kwargs): 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 # NOTE: Fix limit
self.site = site self.site = site
self.list_name = list_name self.list_name = list_name
...@@ -23,23 +44,28 @@ class List: ...@@ -23,23 +44,28 @@ class List:
kwargs.update(args) kwargs.update(args)
self.args = kwargs self.args = kwargs
if limit is None: (max_items, api_chunk_size) = handle_limit(limit, max_items, api_chunk_size)
limit = site.api_limit
self.args[self.prefix + 'limit'] = str(limit) # 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.count = 0
self.max_items = max_items self.max_items = max_items
self._iter = iter(range(0)) self._iter = iter(range(0)) # type: Iterator[Any]
self.last = False self.last = False
self.result_member = list_name self.result_member = list_name
self.return_values = return_values self.return_values = return_values
def __iter__(self): def __iter__(self) -> 'List':
return self return self
def __next__(self): def __next__(self) -> Any:
if self.max_items is not None: if self.max_items is not None:
if self.count >= self.max_items: if self.count >= self.max_items:
raise StopIteration raise StopIteration
...@@ -64,12 +90,12 @@ class List: ...@@ -64,12 +90,12 @@ class List:
if isinstance(self, GeneratorList): if isinstance(self, GeneratorList):
return item return item
if type(self.return_values) is tuple: 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: if self.return_values is not None:
return item[self.return_values] return item[self.return_values]
return item return item
def load_chunk(self): def load_chunk(self) -> None:
"""Query a new chunk of data """Query a new chunk of data
If the query is empty, `raise StopIteration`. If the query is empty, `raise StopIteration`.
...@@ -77,11 +103,8 @@ class List: ...@@ -77,11 +103,8 @@ class List:
Else, update the iterator accordingly. Else, update the iterator accordingly.
If 'continue' is in the response, it is added to `self.args` If 'continue' is in the response, it is added to `self.args`
(new style continuation, added in MediaWiki 1.21). (new style continuation, added in MediaWiki 1.21, default
since MediaWiki 1.26).
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).
Else, set `self.last` to True. Else, set `self.last` to True.
""" """
...@@ -102,14 +125,10 @@ class List: ...@@ -102,14 +125,10 @@ class List:
# New style continuation, added in MediaWiki 1.21 # New style continuation, added in MediaWiki 1.21
self.args.update(data['continue']) 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: else:
self.last = True 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`.""" """Set `self._iter` to the API response `data`."""
if self.result_member not in data['query']: if self.result_member not in data['query']:
self._iter = iter(range(0)) self._iter = iter(range(0))
...@@ -118,35 +137,33 @@ class List: ...@@ -118,35 +137,33 @@ class List:
else: else:
self._iter = iter(data['query'][self.result_member].values()) self._iter = iter(data['query'][self.result_member].values())
def __repr__(self): def __repr__(self) -> str:
return "<%s object '%s' for %s>" % ( return f"<{self.__class__.__name__} object '{self.list_name}' for {self.site}>"
self.__class__.__name__,
self.list_name,
self.site
)
@staticmethod @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) kwargs.update(args)
for key, value in kwargs.items(): for key, value in kwargs.items():
if value is not None and value is not False: if value is not None and value is not False:
yield _prefix + key, value yield _prefix + key, value
@staticmethod @staticmethod
def get_prefix(prefix, generator=False): def get_prefix(prefix: str, generator: bool = False) -> str:
return ('g' if generator else '') + prefix return ('g' if generator else '') + prefix
@staticmethod @staticmethod
def get_list(generator=False): def get_list(generator: bool = False) -> Union[Type['GeneratorList'], Type['List']]:
return GeneratorList if generator else List return GeneratorList if generator else List
class NestedList(List): class NestedList(List):
def __init__(self, nested_param, *args, **kwargs): def __init__(self, nested_param: str, *args: Any, **kwargs: Any) -> None:
super(NestedList, self).__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self.nested_param = nested_param 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]) self._iter = iter(data['query'][self.result_member][self.nested_param])
...@@ -158,9 +175,17 @@ class GeneratorList(List): ...@@ -158,9 +175,17 @@ class GeneratorList(List):
this subclass turns the data into Page, Image or Category objects. this subclass turns the data into Page, Image or Category objects.
""" """
def __init__(self, site, list_name, prefix, *args, **kwargs): def __init__(
super(GeneratorList, self).__init__(site, list_name, prefix, self,
*args, **kwargs) 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'] self.args['g' + self.prefix + 'limit'] = self.args[self.prefix + 'limit']
del self.args[self.prefix + 'limit'] del self.args[self.prefix + 'limit']
...@@ -173,40 +198,64 @@ class GeneratorList(List): ...@@ -173,40 +198,64 @@ class GeneratorList(List):
self.page_class = mwclient.page.Page self.page_class = mwclient.page.Page
def __next__(self): def __next__(self) -> Union['mwclient.page.Page', 'mwclient.image.Image', 'Category']:
info = super(GeneratorList, self).__next__() info = super().__next__()
if info['ns'] == 14: if info['ns'] == 14:
return Category(self.site, '', info) return Category(self.site, '', info)
if info['ns'] == 6: if info['ns'] == 6:
return mwclient.image.Image(self.site, '', info) return mwclient.image.Image(self.site, '', info)
return mwclient.page.Page(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 # Put this here so that the constructor does not fail
# on uninitialized sites # on uninitialized sites
self.args['iiprop'] = 'timestamp|user|comment|url|size|sha1|metadata|archivename' 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): 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) mwclient.page.Page.__init__(self, site, name, info)
kwargs = {} kwargs = {} # type: Dict[str, Any]
kwargs['gcmtitle'] = self.name kwargs['gcmtitle'] = self.name
if namespace: if namespace:
kwargs['gcmnamespace'] = namespace kwargs['gcmnamespace'] = namespace
GeneratorList.__init__(self, site, 'categorymembers', 'cm', **kwargs) GeneratorList.__init__(self, site, 'categorymembers', 'cm', **kwargs)
def __repr__(self): def __repr__(self) -> str:
return "<%s object '%s' for %s>" % ( return f"<{self.__class__.__name__} object '{self.name}' for {self.site}>"
self.__class__.__name__,
self.name.encode('utf-8'), def members(
self.site self,
) prop: str = 'ids|title',
namespace: Optional[Namespace] = None,
def members(self, prop='ids|title', namespace=None, sort='sortkey', sort: str = 'sortkey',
dir='asc', start=None, end=None, generator=True): dir: str = 'asc',
start: Optional[str] = None,
end: Optional[str] = None,
generator: bool = True
) -> 'List':
prefix = self.get_prefix('cm', generator) prefix = self.get_prefix('cm', generator)
kwargs = dict(self.generate_kwargs(prefix, prop=prop, namespace=namespace, kwargs = dict(self.generate_kwargs(prefix, prop=prop, namespace=namespace,
sort=sort, dir=dir, start=start, end=end, sort=sort, dir=dir, start=start, end=end,
...@@ -216,8 +265,15 @@ class Category(mwclient.page.Page, GeneratorList): ...@@ -216,8 +265,15 @@ class Category(mwclient.page.Page, GeneratorList):
class PageList(GeneratorList): class PageList(GeneratorList):
def __init__(self, site, prefix=None, start=None, namespace=0, redirects='all', def __init__(
end=None): 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 self.namespace = namespace
kwargs = {} kwargs = {}
...@@ -228,36 +284,44 @@ class PageList(GeneratorList): ...@@ -228,36 +284,44 @@ class PageList(GeneratorList):
if end: if end:
kwargs['gapto'] = end kwargs['gapto'] = end
super(PageList, self).__init__(site, 'allpages', 'ap', super().__init__(site, 'allpages', 'ap', gapnamespace=str(namespace),
gapnamespace=str(namespace), gapfilterredir=redirects, **kwargs)
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) 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. """Return the page of name `name` as an object.
If self.namespace is not zero, use {namespace}:{name} as the If self.namespace is not zero, use {namespace}:{name} as the
page name, otherwise guess the namespace from the name using page name, otherwise guess the namespace from the name using
`self.guess_namespace`. `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: Returns:
One of Category, Image or Page (default), according to namespace. One of Category, Image or Page (default), according to namespace.
""" """
if self.namespace != 0: if self.namespace != 0:
full_page_name = u"{namespace}:{name}".format( full_page_name = f"{self.site.namespaces[self.namespace]}:{name}" \
namespace=self.site.namespaces[self.namespace], # type: Union[str, int, 'mwclient.page.Page']
name=name,
)
namespace = self.namespace namespace = self.namespace
else: else:
full_page_name = name full_page_name = name
try: if isinstance(name, str):
namespace = self.guess_namespace(name) namespace = self.guess_namespace(name)
except AttributeError: else:
# raised when `namespace` doesn't have a `startswith` attribute
namespace = 0 namespace = 0
cls = { cls = {
...@@ -265,16 +329,16 @@ class PageList(GeneratorList): ...@@ -265,16 +329,16 @@ class PageList(GeneratorList):
6: mwclient.image.Image, 6: mwclient.image.Image,
}.get(namespace, mwclient.page.Page) }.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 """Guess the namespace from name
If name starts with any of the site's namespaces' names or If name starts with any of the site's namespaces' names or
default_namespaces, use that. Else, return zero. default_namespaces, use that. Else, return zero.
Args: Args:
name (str): The pagename as a string (having `.startswith`) name: The name of the page.
Returns: Returns:
The id of the guessed namespace or zero. The id of the guessed namespace or zero.
...@@ -282,26 +346,34 @@ class PageList(GeneratorList): ...@@ -282,26 +346,34 @@ class PageList(GeneratorList):
for ns in self.site.namespaces: for ns in self.site.namespaces:
if ns == 0: if ns == 0:
continue continue
namespace = '%s:' % self.site.namespaces[ns].replace(' ', '_') namespace = f'{self.site.namespaces[ns].replace(" ", "_")}:'
if name.startswith(namespace): if name.startswith(namespace):
return ns 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 return 0
class PageProperty(List): class PageProperty(List):
def __init__(self, page, prop, prefix, *args, **kwargs): def __init__(
super(PageProperty, self).__init__(page.site, prop, prefix, self,
titles=page.name, page: 'mwclient.page.Page',
*args, **kwargs) 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.page = page
self.generator = 'prop' self.generator = 'prop'
def set_iter(self, data): def set_iter(self, data: Mapping[str, Any]) -> None:
for page in data['query']['pages'].values(): for page in data['query']['pages'].values():
if page['title'] == self.page.name: if page['title'] == self.page.name:
self._iter = iter(page.get(self.list_name, ())) self._iter = iter(page.get(self.list_name, ()))
...@@ -311,16 +383,21 @@ class PageProperty(List): ...@@ -311,16 +383,21 @@ class PageProperty(List):
class PagePropertyGenerator(GeneratorList): class PagePropertyGenerator(GeneratorList):
def __init__(self, page, prop, prefix, *args, **kwargs): def __init__(
super(PagePropertyGenerator, self).__init__(page.site, prop, prefix, self,
titles=page.name, page: 'mwclient.page.Page',
*args, **kwargs) 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 self.page = page
class RevisionsIterator(PageProperty): class RevisionsIterator(PageProperty):
def load_chunk(self): def load_chunk(self) -> None:
if 'rvstartid' in self.args and 'rvstart' in self.args: if 'rvstartid' in self.args and 'rvstart' in self.args:
del self.args['rvstart'] del self.args['rvstart']
return super(RevisionsIterator, self).load_chunk() return super().load_chunk()
This diff is collapsed.
import time
import logging import logging
import time
from typing import Callable, Optional, Any
from mwclient.errors import MaximumRetriesExceeded from mwclient.errors import MaximumRetriesExceeded
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
...@@ -17,24 +19,30 @@ class Sleepers: ...@@ -17,24 +19,30 @@ class Sleepers:
using the `make` method. using the `make` method.
>>> sleeper = sleepers.make() >>> sleeper = sleepers.make()
Args: Args:
max_retries (int): The maximum number of retries to perform. max_retries: The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry. retry_timeout: The time to sleep for each past retry.
callback (Callable[[int, Any], None]): A callable to be called on each retry. callback: A callable to be called on each retry.
Attributes: Attributes:
max_retries (int): The maximum number of retries to perform. max_retries: The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry. retry_timeout: The time to sleep for each past retry.
callback (callable): A callable to be called on each 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.max_retries = max_retries
self.retry_timeout = retry_timeout self.retry_timeout = retry_timeout
self.callback = callback self.callback = callback
def make(self, args=None): def make(self, args: Optional[Any] = None) -> 'Sleeper':
""" """
Creates a new `Sleeper` object. Creates a new `Sleeper` object.
Args: Args:
args (Any): Arguments to be passed to the `callback` callable. args: Arguments to be passed to the `callback` callable.
Returns: Returns:
Sleeper: A `Sleeper` object. Sleeper: A `Sleeper` object.
""" """
...@@ -48,25 +56,32 @@ class Sleeper: ...@@ -48,25 +56,32 @@ class Sleeper:
and a `MaximumRetriesExceeded` is raised. The sleeper object should be discarded and a `MaximumRetriesExceeded` is raised. The sleeper object should be discarded
once the operation is successful. once the operation is successful.
Args: Args:
args (Any): Arguments to be passed to the `callback` callable. args: Arguments to be passed to the `callback` callable.
max_retries (int): The maximum number of retries to perform. max_retries: The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry. retry_timeout: The time to sleep for each past retry.
callback (callable, None]): A callable to be called on each retry. callback: A callable to be called on each retry.
Attributes: Attributes:
args (Any): Arguments to be passed to the `callback` callable. args: Arguments to be passed to the `callback` callable.
retries (int): The number of retries that have been performed. retries: The number of retries that have been performed.
max_retries (int): The maximum number of retries to perform. max_retries: The maximum number of retries to perform.
retry_timeout (int): The time to sleep for each past retry. retry_timeout: The time to sleep for each past retry.
callback (callable): A callable to be called on each 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.args = args
self.retries = 0 self.retries = 0
self.max_retries = max_retries self.max_retries = max_retries
self.retry_timeout = retry_timeout self.retry_timeout = retry_timeout
self.callback = callback 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 Sleeps for a minimum of `min_time` seconds. The actual sleeping time will increase
with the number of retries. with the number of retries.
......
import time import time
import io 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. """Parses a string containing a timestamp.
Args: Args:
t (str): A string containing a timestamp. t: A string containing a timestamp.
Returns: Returns:
time.struct_time: A timestamp. time.struct_time: A timestamp.
...@@ -16,9 +18,38 @@ def parse_timestamp(t): ...@@ -16,9 +18,38 @@ def parse_timestamp(t):
return time.strptime(t, '%Y-%m-%dT%H:%M:%SZ') 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: while True:
data = stream.read(chunk_size) data = stream.read(chunk_size)
if not data: if not data:
break break
yield io.BytesIO(data) 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)
[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] [build-system]
requires = ["setuptools>=40.6.0", "wheel"] requires = ["hatchling"]
build-backend = "setuptools.build_meta" build-backend = "hatchling.build"
[tool.pytest.ini_options] [tool.hatch.version]
addopts = "--cov mwclient test" path = "mwclient/client.py"
[tool.hatch.build.targets.sdist]
exclude = [
"/.flake8",
"/.gitattributes",
"/.github",
"/.readthedocs.yaml",
]
[tool.bumpversion] [tool.bumpversion]
current_version = "0.10.1" current_version = "0.11.0"
commit = true commit = true
tag = true tag = true
[[tool.bumpversion.files]]
filename = "setup.py"
search = "version='{current_version}'"
replace = "version='{new_version}'"
[[tool.bumpversion.files]] [[tool.bumpversion.files]]
filename = "mwclient/client.py" filename = "mwclient/client.py"
search = "__version__ = '{current_version}'"
replace = "__version__ = '{new_version}'"
[[tool.bumpversion.files]] [[tool.bumpversion.files]]
filename = "README.md" 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
#!/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
)
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"
from copy import deepcopy
from datetime import date
import json import json
from io import BytesIO
import logging import logging
import sys import sys
import time import time
import unittest import unittest
import unittest.mock as mock 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 pytest
import requests import requests
import responses import responses
from requests_oauthlib import OAuth1 from requests_oauthlib import OAuth1
import mwclient
if __name__ == "__main__": if __name__ == "__main__":
print() print()
print("Note: Running in stand-alone mode. Consult the README") print("Note: Running in stand-alone mode. Consult the README")
...@@ -26,24 +26,61 @@ logging.basicConfig(level=logging.DEBUG) ...@@ -26,24 +26,61 @@ logging.basicConfig(level=logging.DEBUG)
class TestCase(unittest.TestCase): class TestCase(unittest.TestCase):
def metaResponse(self, **kwargs): def metaResponse(self, version='1.24wmf17', rights=None):
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}}}' if rights is None:
tpl = tpl % {'version': kwargs.get('version', '1.24wmf17'), rights = [
'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"])) "createaccount", "read", "edit", "createpage", "createtalk",
} "editmyusercss", "editmyuserjs", "viewmywatchlist",
"editmywatchlist", "viewmyprivateinfo", "editmyprivateinfo",
res = json.loads(tpl) "editmyoptions", "centralauth-merge", "abusefilter-view",
if kwargs.get('writeapi', True): "abusefilter-log", "translate", "vipsscaler-test", "upload"
res['query']['general']['writeapi'] = '' ]
# @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): def metaResponseAsJson(self, **kwargs):
return json.dumps(self.metaResponse(**kwargs)) return json.dumps(self.metaResponse(**kwargs))
def httpShouldReturn(self, body=None, callback=None, scheme='https', host='test.wikipedia.org', path='/w/', def httpShouldReturn(self, body=None, callback=None, scheme='https', host='test.wikipedia.org', path='/w/',
script='api', headers=None, status=200, method='GET'): 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 mock = responses.GET if method == 'GET' else responses.POST
if body is None: if body is None:
responses.add_callback(mock, url, callback=callback) responses.add_callback(mock, url, callback=callback)
...@@ -90,7 +127,15 @@ class TestClient(TestCase): ...@@ -90,7 +127,15 @@ class TestClient(TestCase):
def testVersion(self): def testVersion(self):
# The version specified in setup.py should equal the one specified in client.py # 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__ assert version == mwclient.__version__
...@@ -120,8 +165,8 @@ class TestClient(TestCase): ...@@ -120,8 +165,8 @@ class TestClient(TestCase):
site = mwclient.Site('test.wikipedia.org') site = mwclient.Site('test.wikipedia.org')
assert len(responses.calls) == 2 assert len(responses.calls) == 2
assert 'retry-after' in responses.calls[0].response.headers assert 'retry-after' in responses.calls[0].response.headers # type: ignore
assert 'retry-after' not in responses.calls[1].response.headers assert 'retry-after' not in responses.calls[1].response.headers # type: ignore
@responses.activate @responses.activate
def test_http_error(self): def test_http_error(self):
...@@ -169,6 +214,7 @@ class TestClient(TestCase): ...@@ -169,6 +214,7 @@ class TestClient(TestCase):
site = mwclient.Site('test.wikipedia.org') 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 'action=query' in responses.calls[0].request.url
assert 'meta=siteinfo%7Cuserinfo' in responses.calls[0].request.url assert 'meta=siteinfo%7Cuserinfo' in responses.calls[0].request.url
...@@ -196,7 +242,8 @@ class TestClient(TestCase): ...@@ -196,7 +242,8 @@ class TestClient(TestCase):
self.httpShouldReturn(self.metaResponseAsJson()) self.httpShouldReturn(self.metaResponseAsJson())
with pytest.raises(RuntimeError): 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 @responses.activate
def test_oauth(self): def test_oauth(self):
...@@ -353,9 +400,9 @@ class TestClient(TestCase): ...@@ -353,9 +400,9 @@ class TestClient(TestCase):
} }
}), method='GET') }), 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 @responses.activate
def test_smw_response_v2(self): def test_smw_response_v2(self):
...@@ -395,9 +442,9 @@ class TestClient(TestCase): ...@@ -395,9 +442,9 @@ class TestClient(TestCase):
} }
}), method='GET') }), 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 @responses.activate
def test_repr(self): def test_repr(self):
...@@ -469,6 +516,18 @@ class TestClient(TestCase): ...@@ -469,6 +516,18 @@ class TestClient(TestCase):
site.raw_api("query", "GET", retry_on_error=False) site.raw_api("query", "GET", retry_on_error=False)
assert timesleep.call_count == 25 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): class TestLogin(TestCase):
@mock.patch('mwclient.client.Site.site_init') @mock.patch('mwclient.client.Site.site_init')
...@@ -553,7 +612,7 @@ class TestLogin(TestCase): ...@@ -553,7 +612,7 @@ class TestLogin(TestCase):
# this would be done by site_init usually, but we're mocking it # this would be done by site_init usually, but we're mocking it
site.version = (1, 28, 0) site.version = (1, 28, 0)
success = site.clientlogin(username='myusername', password='mypassword') 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 call_args = raw_api.call_args_list
...@@ -602,7 +661,7 @@ class TestLogin(TestCase): ...@@ -602,7 +661,7 @@ class TestLogin(TestCase):
'clientlogin', 'POST', 'clientlogin', 'POST',
username='myusername', username='myusername',
password='mypassword', password='mypassword',
loginreturnurl='%s://%s' % (site.scheme, site.host), loginreturnurl=f'{site.scheme}://{site.host}',
logintoken=login_token logintoken=login_token
) )
...@@ -629,7 +688,7 @@ class TestLogin(TestCase): ...@@ -629,7 +688,7 @@ class TestLogin(TestCase):
# this would be done by site_init usually, but we're mocking it # this would be done by site_init usually, but we're mocking it
site.version = (1, 28, 0) site.version = (1, 28, 0)
success = site.clientlogin(username='myusername', password='mypassword') 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 call_args = raw_api.call_args_list
...@@ -684,6 +743,47 @@ class TestClientApiMethods(TestCase): ...@@ -684,6 +743,47 @@ class TestClientApiMethods(TestCase):
assert revisions[1]['revid'] == 689816909 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): class TestClientUploadArgs(TestCase):
def setUp(self): def setUp(self):
...@@ -734,7 +834,7 @@ class TestClientUploadArgs(TestCase): ...@@ -734,7 +834,7 @@ class TestClientUploadArgs(TestCase):
# Test that methods are called, and arguments sent as expected # Test that methods are called, and arguments sent as expected
self.configure() 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 args, kwargs = self.raw_call.call_args
data = args[1] data = args[1]
...@@ -750,19 +850,46 @@ class TestClientUploadArgs(TestCase): ...@@ -750,19 +850,46 @@ class TestClientUploadArgs(TestCase):
self.configure() self.configure()
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.site.upload(file=StringIO('test')) self.site.upload(file=BytesIO(b'test'))
def test_upload_ambigitious_args(self): def test_upload_ambigitious_args(self):
self.configure() self.configure()
with pytest.raises(TypeError): 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): def test_upload_missing_upload_permission(self):
self.configure(rights=['read']) self.configure(rights=['read'])
with pytest.raises(mwclient.errors.InsufficientPermission): 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): class TestClientGetTokens(TestCase):
...@@ -1420,6 +1547,74 @@ class TestUser(TestCase): ...@@ -1420,6 +1547,74 @@ class TestUser(TestCase):
assert real_call_kwargs == mock_call_kwargs assert real_call_kwargs == mock_call_kwargs
assert mock_call.args == call_args[2].args 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__': if __name__ == '__main__':
unittest.main() unittest.main()
import unittest import unittest
import pytest import pytest
import logging
import requests
import responses
import json
import mwclient 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 import unittest.mock as mock
...@@ -22,7 +20,54 @@ class TestList(unittest.TestCase): ...@@ -22,7 +20,54 @@ class TestList(unittest.TestCase):
def setUp(self): def setUp(self):
pass 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: if ns is None:
ns = [0, 0, 0] ns = [0, 0, 0]
mock_site.get.side_effect = [ mock_site.get.side_effect = [
...@@ -64,19 +109,65 @@ class TestList(unittest.TestCase): ...@@ -64,19 +109,65 @@ class TestList(unittest.TestCase):
# Test that the list fetches all three responses # Test that the list fetches all three responses
# and yields dicts when return_values not set # and yields dicts when return_values not set
lst = List(mock_site, 'allpages', 'ap', limit=2) lst = List(mock_site, 'allpages', 'ap', api_chunk_size=2)
self.setupDummyResponses(mock_site, 'allpages') 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] vals = [x for x in lst]
assert len(vals) == 3 assert len(vals) == 3
assert type(vals[0]) == dict 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') @mock.patch('mwclient.client.Site')
def test_list_with_str_return_value(self, mock_site): def test_list_with_str_return_value(self, mock_site):
# Test that the List yields strings when return_values is string # Test that the List yields strings when return_values is string
lst = List(mock_site, 'allpages', 'ap', limit=2, return_values='title') 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] vals = [x for x in lst]
assert len(vals) == 3 assert len(vals) == 3
...@@ -88,18 +179,116 @@ class TestList(unittest.TestCase): ...@@ -88,18 +179,116 @@ class TestList(unittest.TestCase):
lst = List(mock_site, 'allpages', 'ap', limit=2, lst = List(mock_site, 'allpages', 'ap', limit=2,
return_values=('title', 'ns')) return_values=('title', 'ns'))
self.setupDummyResponses(mock_site, 'allpages') self.setupDummyResponsesTwo(mock_site, 'allpages')
vals = [x for x in lst] vals = [x for x in lst]
assert len(vals) == 3 assert len(vals) == 3
assert type(vals[0]) == tuple 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') @mock.patch('mwclient.client.Site')
def test_generator_list(self, mock_site): def test_generator_list(self, mock_site):
# Test that the GeneratorList yields Page objects # Test that the GeneratorList yields Page objects
mock_site.api_limit = 500
lst = GeneratorList(mock_site, 'pages', 'p') 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] vals = [x for x in lst]
assert len(vals) == 3 assert len(vals) == 3
...@@ -107,5 +296,172 @@ class TestList(unittest.TestCase): ...@@ -107,5 +296,172 @@ class TestList(unittest.TestCase):
assert type(vals[1]) == mwclient.image.Image assert type(vals[1]) == mwclient.image.Image
assert type(vals[2]) == mwclient.listing.Category 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__': if __name__ == '__main__':
unittest.main() unittest.main()
This diff is collapsed.
[tox] [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] [gh-actions]
python = python =
...@@ -8,21 +9,32 @@ python = ...@@ -8,21 +9,32 @@ python =
3.8: py38 3.8: py38
3.9: py39 3.9: py39
3.10: py310 3.10: py310
3.11: py311, flake 3.11: py311
3.12: py312 3.12: py312
3.13: py313 3.13: py313, flake, integration, mypy
3.14: py314
[testenv] [testenv]
deps = extras = testing
pytest commands = pytest -v --cov mwclient test
pytest-cov
responses
setuptools
mock
commands = py.test -v --cov mwclient test
[testenv:flake] [testenv:flake]
deps = deps =
flake8 flake8
commands = commands =
flake8 mwclient flake8 mwclient
[testenv:integration]
deps =
pytest
commands = pytest test/integration.py -v
[testenv:mypy]
deps =
mypy
pytest
responses
types-requests
types-setuptools
commands =
mypy