Differential D5875 Diff 21287 swh/web/tests/api/views/test_origin.py

Changeset View

Standalone View

swh/web/tests/api/views/test_origin.py

# Copyright (C) 2015-2020 The Software Heritage developers		# Copyright (C) 2015-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 Affero General Public License version 3, or any later version		# License: GNU Affero 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 timedelta		from datetime import timedelta
import json		import json

from hypothesis import given		from hypothesis import given
import pytest		import pytest
from requests.utils import parse_header_links

from swh.indexer.storage.model import OriginIntrinsicMetadataRow		from swh.indexer.storage.model import OriginIntrinsicMetadataRow
from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes
from swh.model.model import Origin, OriginVisit, OriginVisitStatus		from swh.model.model import Origin, OriginVisit, OriginVisitStatus
from swh.storage.exc import StorageAPIError, StorageDBError		from swh.storage.exc import StorageAPIError, StorageDBError
from swh.storage.utils import now		from swh.storage.utils import now
from swh.web.api.utils import enrich_origin, enrich_origin_visit		from swh.web.api.utils import enrich_origin, enrich_origin_visit
from swh.web.common.exc import BadInputExc		from swh.web.common.exc import BadInputExc
from swh.web.common.origin_visits import get_origin_visits		from swh.web.common.origin_visits import get_origin_visits
from swh.web.common.utils import reverse		from swh.web.common.utils import reverse
		from swh.web.tests.api.views.utils import scroll_results
from swh.web.tests.data import (		from swh.web.tests.data import (
INDEXER_TOOL,		INDEXER_TOOL,
ORIGIN_MASTER_REVISION,		ORIGIN_MASTER_REVISION,
ORIGIN_METADATA_KEY,		ORIGIN_METADATA_KEY,
ORIGIN_METADATA_VALUE,		ORIGIN_METADATA_VALUE,
)		)
from swh.web.tests.strategies import new_origin, new_snapshots, origin, visit_dates		from swh.web.tests.strategies import new_origin, new_snapshots, origin, visit_dates
from swh.web.tests.utils import check_api_get_responses		from swh.web.tests.utils import check_api_get_responses


def _scroll_results(api_client, url):
"""Iterates through pages of results, and returns them all."""
results = []

while True:
rv = check_api_get_responses(api_client, url, status_code=200)

results.extend(rv.data)

if "Link" in rv:
for link in parse_header_links(rv["Link"]):
if link["rel"] == "next":
# Found link to next page of results
url = link["url"]
break
else:
# No link with 'rel=next'
break
else:
# No Link header
break

return results


def test_api_lookup_origin_visits_raise_error(api_client, mocker):		def test_api_lookup_origin_visits_raise_error(api_client, mocker):
mock_get_origin_visits = mocker.patch("swh.web.api.views.origin.get_origin_visits")		mock_get_origin_visits = mocker.patch("swh.web.api.views.origin.get_origin_visits")
err_msg = "voluntary error to check the bad request middleware."		err_msg = "voluntary error to check the bad request middleware."

mock_get_origin_visits.side_effect = BadInputExc(err_msg)		mock_get_origin_visits.side_effect = BadInputExc(err_msg)

url = reverse("api-1-origin-visits", url_args={"origin_url": "http://foo"})		url = reverse("api-1-origin-visits", url_args={"origin_url": "http://foo"})
rv = check_api_get_responses(api_client, url, status_code=400)		rv = check_api_get_responses(api_client, url, status_code=400)
▲ Show 20 Lines • Show All 319 Lines • ▼ Show 20 Lines
@pytest.mark.parametrize("origin_count", [1, 2, 10, 100])		@pytest.mark.parametrize("origin_count", [1, 2, 10, 100])
def test_api_origins_scroll(api_client, archive_data, origin_count):		def test_api_origins_scroll(api_client, archive_data, origin_count):
page_result = archive_data.origin_list(limit=10000)		page_result = archive_data.origin_list(limit=10000)
origins = page_result.results		origins = page_result.results
origin_urls = {origin.url for origin in origins}		origin_urls = {origin.url for origin in origins}

url = reverse("api-1-origins", query_params={"origin_count": origin_count})		url = reverse("api-1-origins", query_params={"origin_count": origin_count})

results = _scroll_results(api_client, url)		results = scroll_results(api_client, url)

assert len(results) == len(origins)		assert len(results) == len(origins)
assert {origin["url"] for origin in results} == origin_urls		assert {origin["url"] for origin in results} == origin_urls


@given(origin())		@given(origin())
def test_api_origin_by_url(api_client, archive_data, origin):		def test_api_origin_by_url(api_client, archive_data, origin):
origin_url = origin["url"]		origin_url = origin["url"]
▲ Show 20 Lines • Show All 160 Lines • ▼ Show 20 Lines	def test_api_origin_search_scroll(api_client, archive_data, mocker, limit, backend):
}		}

url = reverse(		url = reverse(
"api-1-origin-search",		"api-1-origin-search",
url_args={"url_pattern": "github.com"},		url_args={"url_pattern": "github.com"},
query_params={"limit": limit},		query_params={"limit": limit},
)		)

results = _scroll_results(api_client, url)		results = scroll_results(api_client, url)

assert {origin["url"] for origin in results} == expected_origins		assert {origin["url"] for origin in results} == expected_origins


@pytest.mark.parametrize("backend", ["swh-search", "swh-storage"])		@pytest.mark.parametrize("backend", ["swh-search", "swh-storage"])
def test_api_origin_search_limit(api_client, archive_data, tests_data, mocker, backend):		def test_api_origin_search_limit(api_client, archive_data, tests_data, mocker, backend):
if backend == "swh-search":		if backend == "swh-search":
tests_data["search"].origin_update(		tests_data["search"].origin_update(
▲ Show 20 Lines • Show All 129 Lines • Show Last 20 Lines