Differential D6286 Diff 22781 swh/lister/gitlab/tests/test_lister.py

Changeset View

Standalone View

swh/lister/gitlab/tests/test_lister.py

# Copyright (C) 2017-2021 The Software Heritage developers		# Copyright (C) 2017-2021 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 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
from swh.lister.gitlab.lister import IGNORED_DVCS, 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:
Show All 40 Lines	def test_lister_gitlab_heptapod(datadir, swh_scheduler, requests_mock):
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)

requests_mock.get(		requests_mock.get(
lister.page_url(), [{"json": response}], additional_matcher=_match_request,		lister.page_url(), [{"json": response}], additional_matcher=_match_request,
)		)

listed_result = lister.run()		listed_result = lister.run()
expected_nb_origins = 0		expected_nb_origins = len(response)

for entry in response:		for entry in response:
if entry["vcs_type"] in IGNORED_DVCS:		assert entry["vcs_type"] in ("hg", "hg_git")
		anlambertUnsubmitted Done Inline Actions I would rather check the visit_type is hg in the listed origins here. anlambert: I would rather check the visit_type is hg in the listed origins here.
		ardumontAuthorUnsubmitted Done Inline Actions It's both hg_git and hg. Hency why i added this and the check below becomes clearer. But i can think of something else then ;) ardumont: It's both hg_git and hg. Hency why i added this and the check below becomes clearer. But i can…
		anlambertUnsubmitted Not Done Inline Actions I was thinking about that check instead: scheduler_origins = lister.scheduler.get_listed_origins( lister.lister_obj.id ).results assert all([origin.visit_type == "hg" for origin in scheduler_origins]) anlambert: I was thinking about that check instead: ```lang=python scheduler_origins = lister.scheduler.
		ardumontAuthorUnsubmitted Done Inline Actions That test is already done below line 86 in the original code though. Here, i'm merely pointing out that the input does indeed have more vcs-type than usual. And then that it's mapped to hg (with the existing check). Isn't it enough? ardumont: That test is already done below line 86 in the original code though. Here, i'm merely pointing…
		anlambertUnsubmitted Not Done Inline Actions Ah right, it was hidden by Phabricator UI, I think we are good then. anlambert: Ah right, it was hidden by Phabricator UI, I think we are good then.
continue
expected_nb_origins += 1

assert listed_result == ListerStats(pages=1, origins=expected_nb_origins)		assert listed_result == ListerStats(pages=1, origins=expected_nb_origins)

scheduler_origins = lister.scheduler.get_listed_origins(		scheduler_origins = lister.scheduler.get_listed_origins(
lister.lister_obj.id		lister.lister_obj.id
).results		).results
assert len(scheduler_origins) == expected_nb_origins		assert len(scheduler_origins) == expected_nb_origins

for listed_origin in scheduler_origins:		for listed_origin in scheduler_origins:
assert listed_origin.visit_type == "hg"		assert listed_origin.visit_type == "hg"
		ardumontAuthorUnsubmitted Done Inline Actions Here is the test already done as mentioned above. ardumont: Here is the test already done as mentioned above.
assert listed_origin.url.startswith(f"https://{instance}")		assert listed_origin.url.startswith(f"https://{instance}")
assert listed_origin.last_update is not None		assert listed_origin.last_update is not None


def gitlab_page_response(datadir, instance: str, id_after: int) -> List[Dict]:		def gitlab_page_response(datadir, instance: str, id_after: int) -> List[Dict]:
"""Return list of repositories (out of test dataset)"""		"""Return list of repositories (out of test dataset)"""
datapath = Path(datadir, f"https_{instance}", f"api_response_page{id_after}.json")		datapath = Path(datadir, f"https_{instance}", f"api_response_page{id_after}.json")
return json.loads(datapath.read_text()) if datapath.exists else []		return json.loads(datapath.read_text()) if datapath.exists else []
▲ Show 20 Lines • Show All 253 Lines • Show Last 20 Lines