Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_fetch_group_id_not_found(self):
"""Test whether an error is thrown when the group id is not found"""
setup_http_server(empty_mbox=True)
client = GroupsioClient('beta/api', self.tmp_path, 'jsmith@example.com', 'aaaaa', verify=False)
with self.assertRaises(BackendError):
client.fetch()
def test_invalid_auth(self):
"""Test whether it fails when the authentication goes wrong"""
# Set up a mock HTTP server
httpretty.register_uri(httpretty.GET,
BUGZILLA_LOGIN_URL,
body="401 Client Error: Authorization Required",
status=401)
with self.assertRaises(BackendError):
_ = BugzillaRESTClient(BUGZILLA_SERVER_URL,
user='jsmith@example.com',
password='1234')
self.assertEqual(backend.origin, GOOGLE_SEARCH_URL)
self.assertEqual(backend.tag, GOOGLE_SEARCH_URL)
backend = GoogleHits(['bitergia', 'grimoirelab'], tag='')
self.assertEqual(backend.keywords, ['bitergia', 'grimoirelab'])
self.assertEqual(backend.origin, GOOGLE_SEARCH_URL)
self.assertEqual(backend.tag, GOOGLE_SEARCH_URL)
backend = GoogleHits(['bitergia', 'grimoirelab'], tag='', max_retries=1, sleep_time=100)
self.assertEqual(backend.keywords, ['bitergia', 'grimoirelab'])
self.assertEqual(backend.origin, GOOGLE_SEARCH_URL)
self.assertEqual(backend.tag, GOOGLE_SEARCH_URL)
self.assertEqual(backend.max_retries, 1)
self.assertEqual(backend.sleep_time, 100)
with self.assertRaises(BackendError):
_ = GoogleHits([''], tag='')
def test_missing_credentials(self):
"""Test whether an exception is thrown when `api_username` and `api_token` aren't defined together"""
with self.assertRaises(BackendError):
_ = Discourse(DISCOURSE_SERVER_URL, api_token='1234')
with self.assertRaises(BackendError):
_ = Discourse(DISCOURSE_SERVER_URL, api_username='user')
def test_invalid_auth(self):
"""Test whether it fails when the authentication goes wrong"""
# Set up a mock HTTP server
httpretty.register_uri(httpretty.POST,
BUGZILLA_LOGIN_URL,
body="",
status=200)
with self.assertRaises(BackendError):
_ = BugzillaClient(BUGZILLA_SERVER_URL,
user='jsmith@example.com',
password='1234')
def test_missing_credentials(self):
"""Test whether an exception is thrown when `api_username` and `api_token` aren't defined together"""
with self.assertRaises(BackendError):
_ = Discourse(DISCOURSE_SERVER_URL, api_token='1234')
with self.assertRaises(BackendError):
_ = Discourse(DISCOURSE_SERVER_URL, api_username='user')
def test_unknown_version(self):
"""Test whether an exception is thrown when the gerrit version is unknown"""
mock_check_ouput_version_unknown.side_effect = mock_check_ouput_version_unknown
client = GerritClient(GERRIT_REPO, GERRIT_USER)
with self.assertRaises(BackendError):
_ = client.version
def __init_issue_extra_fields(self, issue):
"""Add fields to an issue"""
issue['notes_data'] = []
issue['award_emoji_data'] = []
def __init_merge_extra_fields(self, merge):
"""Add fields to a merge requests"""
merge['notes_data'] = []
merge['award_emoji_data'] = []
merge['versions_data'] = []
class _OutdatedMRsList(BackendError):
"""Exception raised when the list of MRs is outdated."""
message = "MRs list is outdated; you should fetch a new one."
class GitLabClient(HttpClient, RateLimitHandler):
"""Client for retieving information from GitLab API
:param owner: GitLab owner
:param repository: GitLab owner's repository
:param token: GitLab auth token to access the API
:param is_oauth_token: True if the token is OAuth (default False)
:param base_url: GitLab URL in enterprise edition case;
when no value is set the backend will be fetch the data
from the GitLab public site.
:param sleep_for_rate: sleep until rate limit is reset
def next_retrieve_group_item(self, last_item=None, entry=None):
"""Return the item to start from in next reviews group."""
next_item = None
gerrit_version = self.version
if gerrit_version[0] == 2 and gerrit_version[1] > 9:
if last_item is None:
next_item = 0
else:
next_item = last_item
elif gerrit_version[0] == 2 and gerrit_version[1] == 9:
# https://groups.google.com/forum/#!topic/repo-discuss/yQgRR5hlS3E
cause = "Gerrit 2.9.0 does not support pagination"
raise BackendError(cause=cause)
else:
if entry is not None:
next_item = entry['sortKey']
return next_item
logger.warning("Revisions not found in %s [page id: %s], page skipped",
page['title'], page['pageid'])
continue
yield page_reviews
npages += 1
logger.info("Total number of pages: %i, skipped %i", tpages, tpages - npages)
logger.info("Looking for pages at url '%s'", self.url)
# from_date can not be older than MAX_RECENT_DAYS days ago
if from_date:
if (datetime_utcnow() - from_date).days >= MAX_RECENT_DAYS:
cause = "Can't get incremental pages older than %i days." % MAX_RECENT_DAYS
cause += " Do a complete analysis without from_date for older changes."
raise BackendError(cause=cause)
namespaces_contents = self.__get_namespaces_contents()
if not from_date:
return fetch_all_pages(namespaces_contents)
else:
return fetch_incremental_changes(namespaces_contents)