Differential D4865 Diff 17313 swh/lister/phabricator/tests/test_lister.py

Changeset View

Standalone View

swh/lister/phabricator/tests/test_lister.py

# Copyright (C) 2019-2021 The Software Heritage developers		# Copyright (C) 2019-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
from pathlib import Path		from pathlib import Path

import pytest		import pytest
		from requests.exceptions import HTTPError

from swh.lister import USER_AGENT		from swh.lister import USER_AGENT
from swh.lister.phabricator.lister import PhabricatorLister, get_repo_url		from swh.lister.phabricator.lister import PhabricatorLister, get_repo_url


@pytest.fixture		@pytest.fixture
def phabricator_repositories_page1(datadir):		def phabricator_repositories_page1(datadir):
return json.loads(		return json.loads(
▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines	):
expected_nb_origins = len(phabricator_repositories_page1["result"]["data"]) * 2		expected_nb_origins = len(phabricator_repositories_page1["result"]["data"]) * 2

assert stats.pages == 2		assert stats.pages == 2
assert stats.origins == expected_nb_origins		assert stats.origins == expected_nb_origins

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

assert len(scheduler_origins) == expected_nb_origins		assert len(scheduler_origins) == expected_nb_origins


		def test_lister_request_error(
		swh_scheduler, requests_mock, phabricator_repositories_page1,
		):
		FORGE_BASE_URL = "https://forge.softwareheritage.org"

		lister = PhabricatorLister(
		scheduler=swh_scheduler, url=FORGE_BASE_URL, instance="swh", api_token="foo"
		)

		requests_mock.post(
		f"{FORGE_BASE_URL}{lister.API_REPOSITORY_PATH}",
		[
		{"status_code": 200, "json": phabricator_repositories_page1},
		{"status_code": 500, "reason": "Internal Server Error"},
		],
		)

		with pytest.raises(HTTPError):
		lister.run()