Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/gitlab/tests/test_lister.py
| # Copyright (C) 2017-2021 The Software Heritage developers | # Copyright (C) 2017-2022 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # 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 functools | |||||
| import json | import json | ||||
| import logging | import logging | ||||
| from pathlib import Path | from pathlib import Path | ||||
| from typing import Dict, List | from typing import Dict, List | ||||
| import pytest | import pytest | ||||
| from requests.status_codes import codes | from requests.status_codes import codes | ||||
| from swh.lister import USER_AGENT | from swh.lister import USER_AGENT_TEMPLATE | ||||
| from swh.lister.gitlab.lister import GitLabLister, _parse_id_after | from swh.lister.gitlab.lister import GitLabLister, _parse_id_after | ||||
| from swh.lister.pattern import ListerStats | from swh.lister.pattern import ListerStats | ||||
| from swh.lister.tests.test_utils import assert_sleep_calls | from swh.lister.tests.test_utils import assert_sleep_calls | ||||
| from swh.lister.utils import WAIT_EXP_BASE | from swh.lister.utils import WAIT_EXP_BASE | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| def api_url(instance: str) -> str: | def api_url(instance: str) -> str: | ||||
| return f"https://{instance}/api/v4/" | return f"https://{instance}/api/v4/" | ||||
| def _match_request(request): | def _match_request(request, lister_name="gitlab"): | ||||
| return request.headers.get("User-Agent") == USER_AGENT | return request.headers.get("User-Agent") == USER_AGENT_TEMPLATE % lister_name | ||||
| def test_lister_gitlab(datadir, swh_scheduler, requests_mock): | def test_lister_gitlab(datadir, swh_scheduler, requests_mock): | ||||
| """Gitlab lister supports full listing""" | """Gitlab lister supports full listing""" | ||||
| instance = "gitlab.com" | instance = "gitlab.com" | ||||
| lister = GitLabLister(swh_scheduler, url=api_url(instance), instance=instance) | lister = GitLabLister(swh_scheduler, url=api_url(instance), instance=instance) | ||||
| response = gitlab_page_response(datadir, instance, 1) | response = gitlab_page_response(datadir, instance, 1) | ||||
| Show All 28 Lines | def test_lister_gitlab_heptapod(datadir, swh_scheduler, requests_mock): | ||||
| ) | ) | ||||
| assert lister.LISTER_NAME == name | assert lister.LISTER_NAME == name | ||||
| response = gitlab_page_response(datadir, instance, 1) | response = gitlab_page_response(datadir, instance, 1) | ||||
| requests_mock.get( | requests_mock.get( | ||||
| lister.page_url(), | lister.page_url(), | ||||
| [{"json": response}], | [{"json": response}], | ||||
| additional_matcher=_match_request, | additional_matcher=functools.partial(_match_request, lister_name="heptapod"), | ||||
| ) | ) | ||||
| listed_result = lister.run() | listed_result = lister.run() | ||||
| expected_nb_origins = len(response) | expected_nb_origins = len(response) | ||||
| for entry in response: | for entry in response: | ||||
| assert entry["vcs_type"] in ("hg", "hg_git") | assert entry["vcs_type"] in ("hg", "hg_git") | ||||
| ▲ Show 20 Lines • Show All 276 Lines • Show Last 20 Lines | |||||