Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/core/tests/test_lister.py
# Copyright (C) 2019 the Software Heritage developers | # Copyright (C) 2019 the Software Heritage developers | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import abc | import abc | ||||
import datetime | import datetime | ||||
import time | import time | ||||
from unittest import TestCase | from unittest import TestCase | ||||
from unittest.mock import Mock, patch | from unittest.mock import Mock, patch | ||||
import requests_mock | import requests_mock | ||||
from sqlalchemy import create_engine | from sqlalchemy import create_engine | ||||
from typing import Any, Callable, Optional, Pattern, Type, Union | |||||
from swh.lister.core.abstractattribute import AbstractAttribute | from swh.lister.core.abstractattribute import AbstractAttribute | ||||
from swh.lister.tests.test_utils import init_db | from swh.lister.tests.test_utils import init_db | ||||
def noop(*args, **kwargs): | def noop(*args, **kwargs): | ||||
pass | pass | ||||
class HttpListerTesterBase(abc.ABC): | class HttpListerTesterBase(abc.ABC): | ||||
"""Testing base class for listers. | """Testing base class for listers. | ||||
This contains methods for both :class:`HttpSimpleListerTester` and | This contains methods for both :class:`HttpSimpleListerTester` and | ||||
:class:`HttpListerTester`. | :class:`HttpListerTester`. | ||||
See :class:`swh.lister.gitlab.tests.test_lister` for an example of how | See :class:`swh.lister.gitlab.tests.test_lister` for an example of how | ||||
to customize for a specific listing service. | to customize for a specific listing service. | ||||
""" | """ | ||||
Lister = AbstractAttribute('The lister class to test') | Lister = AbstractAttribute( | ||||
lister_subdir = AbstractAttribute('bitbucket, github, etc.') | 'Lister class to test') # type: Union[AbstractAttribute, Type[Any]] | ||||
good_api_response_file = AbstractAttribute('Example good response body') | lister_subdir = AbstractAttribute( | ||||
'bitbucket, github, etc.') # type: Union[AbstractAttribute, str] | |||||
good_api_response_file = AbstractAttribute( | |||||
'Example good response body') # type: Union[AbstractAttribute, str] | |||||
LISTER_NAME = 'fake-lister' | LISTER_NAME = 'fake-lister' | ||||
# May need to override this if the headers are used for something | # May need to override this if the headers are used for something | ||||
def response_headers(self, request): | def response_headers(self, request): | ||||
return {} | return {} | ||||
# May need to override this if the server uses non-standard rate limiting | # May need to override this if the server uses non-standard rate limiting | ||||
# method. | # method. | ||||
▲ Show 20 Lines • Show All 110 Lines • ▼ Show 20 Lines | class HttpListerTester(HttpListerTesterBase, abc.ABC): | ||||
"""Base testing class for subclass of | """Base testing class for subclass of | ||||
:class:`swh.lister.core.indexing_lister.IndexingHttpLister` | :class:`swh.lister.core.indexing_lister.IndexingHttpLister` | ||||
See :class:`swh.lister.github.tests.test_gh_lister` for an example of how | See :class:`swh.lister.github.tests.test_gh_lister` for an example of how | ||||
to customize for a specific listing service. | to customize for a specific listing service. | ||||
""" | """ | ||||
last_index = AbstractAttribute('Last index in good_api_response') | last_index = AbstractAttribute( | ||||
first_index = AbstractAttribute('First index in good_api_response') | 'Last index ' | ||||
bad_api_response_file = AbstractAttribute('Example bad response body') | 'in good_api_response') # type: Union[AbstractAttribute, int] | ||||
entries_per_page = AbstractAttribute('Number of results in good response') | first_index = AbstractAttribute( | ||||
test_re = AbstractAttribute('Compiled regex matching the server url. Must' | 'First index in ' | ||||
' capture the index value.') | ' good_api_response') # type: Union[AbstractAttribute, Optional[int]] | ||||
convert_type = str | bad_api_response_file = AbstractAttribute( | ||||
'Example bad response body') # type: Union[AbstractAttribute, str] | |||||
entries_per_page = AbstractAttribute( | |||||
'Number of results in ' | |||||
'good response') # type: Union[AbstractAttribute, int] | |||||
test_re = AbstractAttribute( | |||||
'Compiled regex matching the server url. Must capture the ' | |||||
'index value.') # type: Union[AbstractAttribute, Pattern] | |||||
convert_type = str # type: Callable[..., Any] | |||||
"""static method used to convert the "request_index" to its right type (for | """static method used to convert the "request_index" to its right type (for | ||||
indexing listers for example, this is in accordance with the model's | indexing listers for example, this is in accordance with the model's | ||||
"indexable" column). | "indexable" column). | ||||
""" | """ | ||||
def mock_response(self, request, context): | def mock_response(self, request, context): | ||||
self.fl.reset_backoff() | self.fl.reset_backoff() | ||||
self.rate_limit = 1 | self.rate_limit = 1 | ||||
▲ Show 20 Lines • Show All 163 Lines • ▼ Show 20 Lines | |||||
class HttpSimpleListerTester(HttpListerTesterBase, abc.ABC): | class HttpSimpleListerTester(HttpListerTesterBase, abc.ABC): | ||||
"""Base testing class for subclass of | """Base testing class for subclass of | ||||
:class:`swh.lister.core.simple)_lister.SimpleLister` | :class:`swh.lister.core.simple)_lister.SimpleLister` | ||||
See :class:`swh.lister.pypi.tests.test_lister` for an example of how | See :class:`swh.lister.pypi.tests.test_lister` for an example of how | ||||
to customize for a specific listing service. | to customize for a specific listing service. | ||||
""" | """ | ||||
entries = AbstractAttribute('Number of results in good response') | entries = AbstractAttribute( | ||||
PAGE = AbstractAttribute("The server api's unique page to retrieve and " | 'Number of results ' | ||||
"parse for information") | 'in good response') # type: Union[AbstractAttribute, int] | ||||
PAGE = AbstractAttribute( | |||||
"URL of the server api's unique page to retrieve and " | |||||
"parse for information") # type: Union[AbstractAttribute, str] | |||||
def get_fl(self, override_config=None): | def get_fl(self, override_config=None): | ||||
"""Retrieve an instance of fake lister (fl). | """Retrieve an instance of fake lister (fl). | ||||
""" | """ | ||||
if override_config or self.fl is None: | if override_config or self.fl is None: | ||||
self.fl = self.Lister( | self.fl = self.Lister( | ||||
override_config=override_config) | override_config=override_config) | ||||
▲ Show 20 Lines • Show All 56 Lines • Show Last 20 Lines |