Differential D8289 Diff 29959 swh/lister/gitea/tests/test_lister.py

Changeset View

Standalone View

swh/lister/gitea/tests/test_lister.py

# Copyright (C) 2017-2020 The Software Heritage developers		# Copyright (C) 2017-2020 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
from pathlib import Path		from pathlib import Path
from typing import Dict, List, Tuple		from typing import Dict, List, Tuple

import pytest		import pytest
import requests		import requests

from swh.lister.gitea.lister import GiteaLister, RepoListPage		from swh.lister.gitea.lister import GiteaLister
		from swh.lister.gogs.lister import GogsListerPage
from swh.scheduler.model import ListedOrigin		from swh.scheduler.model import ListedOrigin

TRYGITEA_URL = "https://try.gitea.io/api/v1/"		TRYGITEA_URL = "https://try.gitea.io/api/v1/"
TRYGITEA_P1_URL = TRYGITEA_URL + "repos/search?sort=id&order=asc&limit=3&page=1"		TRYGITEA_P1_URL = TRYGITEA_URL + "repos/search?limit=3&page=1"
TRYGITEA_P2_URL = TRYGITEA_URL + "repos/search?sort=id&order=asc&limit=3&page=2"		TRYGITEA_P2_URL = TRYGITEA_URL + "repos/search?limit=3&page=2"


@pytest.fixture		@pytest.fixture
def trygitea_p1(datadir) -> Tuple[str, Dict[str, str], RepoListPage, List[str]]:		def trygitea_p1(datadir) -> Tuple[str, Dict[str, str], GogsListerPage, List[str]]:
text = Path(datadir, "https_try.gitea.io", "repos_page1").read_text()		text = Path(datadir, "https_try.gitea.io", "repos_page1").read_text()
headers = {		headers = {
"Link": '<{p2}>; rel="next",<{p2}>; rel="last"'.format(p2=TRYGITEA_P2_URL)		"Link": '<{p2}>; rel="next",<{p2}>; rel="last"'.format(p2=TRYGITEA_P2_URL)
}		}
page_result = GiteaLister.results_simplified(json.loads(text))		page_data = json.loads(text)
origin_urls = [r["clone_url"] for r in page_result]		page_result = GogsListerPage(
		repos=GiteaLister.extract_repos(page_data), next_link=TRYGITEA_P2_URL
		)
		origin_urls = [r["clone_url"] for r in page_data["data"]]
return text, headers, page_result, origin_urls		return text, headers, page_result, origin_urls


@pytest.fixture		@pytest.fixture
def trygitea_p2(datadir) -> Tuple[str, Dict[str, str], RepoListPage, List[str]]:		def trygitea_p2(datadir) -> Tuple[str, Dict[str, str], GogsListerPage, List[str]]:
text = Path(datadir, "https_try.gitea.io", "repos_page2").read_text()		text = Path(datadir, "https_try.gitea.io", "repos_page2").read_text()
headers = {		headers = {
"Link": '<{p1}>; rel="prev",<{p1}>; rel="first"'.format(p1=TRYGITEA_P1_URL)		"Link": '<{p1}>; rel="prev",<{p1}>; rel="first"'.format(p1=TRYGITEA_P1_URL)
}		}
page_result = GiteaLister.results_simplified(json.loads(text))		page_data = json.loads(text)
origin_urls = [r["clone_url"] for r in page_result]		page_result = GogsListerPage(
		repos=GiteaLister.extract_repos(page_data), next_link=None
		)
		origin_urls = [r["clone_url"] for r in page_data["data"]]
return text, headers, page_result, origin_urls		return text, headers, page_result, origin_urls


def check_listed_origins(lister_urls: List[str], scheduler_origins: List[ListedOrigin]):		def check_listed_origins(lister_urls: List[str], scheduler_origins: List[ListedOrigin]):
"""Asserts that the two collections have the same origin URLs.		"""Asserts that the two collections have the same origin URLs.

Does not test last_update."""		Does not test last_update."""

Show All 40 Lines	):

calls = [mocker.call(p1_result), mocker.call(p2_result)]		calls = [mocker.call(p1_result), mocker.call(p2_result)]
lister.get_origins_from_page.assert_has_calls(calls)		lister.get_origins_from_page.assert_has_calls(calls)

scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results		scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results

check_listed_origins(p1_origin_urls + p2_origin_urls, scheduler_origins)		check_listed_origins(p1_origin_urls + p2_origin_urls, scheduler_origins)

assert lister.get_state_from_scheduler() is None		lister_state = lister.get_state_from_scheduler()
		assert lister_state.last_seen_next_link == TRYGITEA_P2_URL
		assert lister_state.last_seen_repo_id == p2_result.repos[-1]["id"]


def test_gitea_auth_instance(swh_scheduler, requests_mock, trygitea_p1):		def test_gitea_auth_instance(swh_scheduler, requests_mock, trygitea_p1):
"""Covers token authentication, token from credentials,		"""Covers token authentication, token from credentials,
instance inference from URL."""		instance inference from URL."""

api_token = "teapot"		api_token = "teapot"
instance = "try.gitea.io"		instance = "try.gitea.io"
▲ Show 20 Lines • Show All 47 Lines • Show Last 20 Lines