Differential D8629 Diff 31151 swh/graphql/tests/functional/test_origin_node.py

Changeset View

Standalone View

swh/graphql/tests/functional/test_origin_node.py

	# Copyright (C) 2022 The Software Heritage developers			# Copyright (C) 2022 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 pytest			import pytest

				from . import utils
	from ..data import get_origins			from ..data import get_origins
	from .utils import assert_missing_object, get_query_response


	def test_invalid_get(client):			def test_invalid_get(client):
	query_str = """			query_str = """
	{			query getOrigin {
	origin(url: "http://example.com/non-existing") {			origin(url: "http://example.com/non-existing") {
	url			url
	}			}
	}			}
	"""			"""
	assert_missing_object(client, query_str, "origin")			utils.assert_missing_object(client, query_str, "origin")


	@pytest.mark.parametrize("origin", get_origins())			@pytest.mark.parametrize("origin", get_origins())
	def test_get(client, storage, origin):			def test_get(client, storage, origin):
	query_str = (			query_str = """
	"""			query getOrigin($url: String!) {
	{			origin(url: $url) {
	origin(url: "%s") {
	url			url
	id			id
	visits(first: 10) {			visits(first: 10) {
	nodes {			nodes {
	id			id
	}			}
	}			}
	latestVisit {			latestVisit {
	visitId			visitId
	}			}
	snapshots(first: 2) {			snapshots(first: 2) {
	nodes {			nodes {
	id			id
	}			}
	}			}
	}			}
	}			}
	"""			"""
	% origin.url
	)

	response, _ = get_query_response(client, query_str)			response, _ = utils.get_query_response(client, query_str, url=origin.url)
	data_origin = response["origin"]			data_origin = response["origin"]
	storage_origin = storage.origin_get([origin.url])[0]			storage_origin = storage.origin_get([origin.url])[0]
	visits_and_statuses = storage.origin_visit_get_with_statuses(origin.url).results			visits_and_statuses = storage.origin_visit_get_with_statuses(origin.url).results
	assert data_origin["url"] == storage_origin.url			assert data_origin["url"] == storage_origin.url
	assert data_origin["id"] == storage_origin.id.hex()			assert data_origin["id"] == storage_origin.id.hex()
	assert len(data_origin["visits"]["nodes"]) == len(visits_and_statuses)			assert len(data_origin["visits"]["nodes"]) == len(visits_and_statuses)
	assert data_origin["latestVisit"]["visitId"] == visits_and_statuses[-1].visit.visit			assert data_origin["latestVisit"]["visitId"] == visits_and_statuses[-1].visit.visit
	snapshots = storage.origin_snapshot_get_all(origin.url)			snapshots = storage.origin_snapshot_get_all(origin.url)
	assert len(data_origin["snapshots"]["nodes"]) == len(snapshots)			assert len(data_origin["snapshots"]["nodes"]) == len(snapshots)


	def test_latest_visit_type_filter(client):			def test_latest_visit_type_filter(client):
	query_str = """			query_str = """
	{			query getOrigin($url: String!, $visitType: String!) {
	origin(url: "%s") {			origin(url: $url) {
	latestVisit(visitType: "%s") {			latestVisit(visitType: $visitType) {
	visitId			visitId
	}			}
	}			}
	}			}
	"""			"""
	data, _ = get_query_response(client, query_str % (get_origins()[0].url, "git"))			data, _ = utils.get_query_response(
				client, query_str, url=get_origins()[0].url, visitType="git"
				)
	assert data["origin"] == {"latestVisit": {"visitId": 3}}			assert data["origin"] == {"latestVisit": {"visitId": 3}}

	data, _ = get_query_response(client, query_str % (get_origins()[0].url, "hg"))			data, _ = utils.get_query_response(
				client, query_str, url=get_origins()[0].url, visitType="hg"
				)
	assert data["origin"] == {"latestVisit": None}			assert data["origin"] == {"latestVisit": None}


	def test_latest_visit_require_snapshot_filter(client):			def test_latest_visit_require_snapshot_filter(client):
	query_str = """			query_str = """
	{			query getOrigin($url: String!, $requireSnapshot: Boolean!) {
	origin(url: "%s") {			origin(url: $url) {
	latestVisit(requireSnapshot: %s) {			latestVisit(requireSnapshot: $requireSnapshot) {
	visitId			visitId
	}			}
	}			}
	}			}
	"""			"""
	data, _ = get_query_response(client, query_str % (get_origins()[1].url, "true"))			data, _ = utils.get_query_response(
				client,
				query_str,
				url=get_origins()[1].url,
				requireSnapshot=True,
				)
	assert data["origin"] == {"latestVisit": {"visitId": 2}}			assert data["origin"] == {"latestVisit": {"visitId": 2}}


	def test_latest_visit_allowed_statuses_filter(client):			def test_latest_visit_allowed_statuses_filter(client):
	query_str = """			query_str = """
	{			query getOrigin($url: String!, $allowedStatuses: [VisitStatusState!]!) {
	origin(url: "%s") {			origin(url: $url) {
	latestVisit(allowedStatuses: [partial]) {			latestVisit(allowedStatuses: $allowedStatuses) {
	visitId			visitId
	statuses {			statuses {
	nodes {			nodes {
	status			status
	}			}
	}			}
	}			}
	}			}
	}			}
	"""			"""
	data, _ = get_query_response(client, query_str % (get_origins()[1].url))			data, _ = utils.get_query_response(
				client,
				query_str,
				url=get_origins()[1].url,
				allowedStatuses=["partial"],
				)
	assert data["origin"] == {			assert data["origin"] == {
	"latestVisit": {"statuses": {"nodes": [{"status": "partial"}]}, "visitId": 2}			"latestVisit": {"statuses": {"nodes": [{"status": "partial"}]}, "visitId": 2}
	}			}