Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/cgit/tests/test_lister.py
| # Copyright (C) 2019-2021 The Software Heritage developers | # Copyright (C) 2019-2021 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 | ||||
| from datetime import datetime, timedelta, timezone | from datetime import datetime, timedelta, timezone | ||||
| import os | |||||
| from typing import List | from typing import List | ||||
| import pytest | import pytest | ||||
| from swh.core.pytest_plugin import requests_mock_datadir_factory | from swh.core.pytest_plugin import requests_mock_datadir_factory | ||||
| from swh.lister import __version__ | from swh.lister import __version__ | ||||
| from swh.lister.cgit.lister import CGitLister, _parse_last_updated_date | from swh.lister.cgit.lister import CGitLister, _parse_last_updated_date | ||||
| from swh.lister.pattern import ListerStats | from swh.lister.pattern import ListerStats | ||||
| ▲ Show 20 Lines • Show All 210 Lines • ▼ Show 20 Lines | ): | ||||
| # test listed repositories | # test listed repositories | ||||
| for listed_origin in scheduler_origins: | for listed_origin in scheduler_origins: | ||||
| assert listed_origin.visit_type == "git" | assert listed_origin.visit_type == "git" | ||||
| assert listed_origin.url.startswith(base_git_url) | assert listed_origin.url.startswith(base_git_url) | ||||
| assert ( | assert ( | ||||
| listed_origin.url.startswith(url) is False | listed_origin.url.startswith(url) is False | ||||
| ), f"url should be mapped to {base_git_url}" | ), f"url should be mapped to {base_git_url}" | ||||
| def test_lister_cgit_get_pages_with_pages_and_retry( | |||||
| requests_mock_datadir, requests_mock, datadir, mocker, swh_scheduler | |||||
| ): | |||||
| url = "https://git.tizen/cgit/" | |||||
| with open(os.path.join(datadir, "https_git.tizen/cgit,ofs=50"), "rb") as page: | |||||
| requests_mock.get( | |||||
| f"{url}?ofs=50", | |||||
| [ | |||||
| {"content": None, "status_code": 429}, | |||||
| {"content": None, "status_code": 429}, | |||||
| {"content": page.read(), "status_code": 200}, | |||||
| ], | |||||
| ) | |||||
| lister_cgit = CGitLister(swh_scheduler, url=url) | |||||
| mocker.patch.object(lister_cgit._get_and_parse.retry, "sleep") | |||||
| repos: List[List[str]] = list(lister_cgit.get_pages()) | |||||
| flattened_repos = sum(repos, []) | |||||
| # we should have 16 repos (listed on 3 pages) | |||||
| assert len(repos) == 3 | |||||
| assert len(flattened_repos) == 16 | |||||