Differential D7908 Diff 28548 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

	from .utils import get_query_response			import pytest

				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 = """
	{			{
	origin(url: "http://example.com/forge1/") {			origin(url: "http://example.com/non-existing") {
	url			url
	}			}
	}			}
	"""			"""
	data, errors = get_query_response(client, query_str)			assert_missing_object(client, query_str, "origin")
	assert data["origin"] is None
	assert len(errors) == 1
	assert errors[0]["message"] == "Requested object is not available"


	def test_get(client):			@pytest.mark.parametrize("origin", get_origins())
	query_str = """			def test_get(client, storage, origin):
	{			query_str = f"""
	origin(url: "http://example.com/forge1") {			{{
				origin(url: "{origin.url}") {{
	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
	}			}}
	}			}}
	}			}}
	}			}}
	"""			"""
	data, _ = get_query_response(client, query_str)
				ardumontUnsubmitted Done Inline Actions same remark... etc... ardumont: same remark... etc...
	origin = data["origin"]			response, _ = get_query_response(client, query_str)
	assert origin["url"] == "http://example.com/forge1"			data_origin = response["origin"]
	assert len(origin["visits"]["nodes"]) == 2			storage_origin = storage.origin_get([origin.url])[0]
	assert origin["latestVisit"]["visitId"] == 2			visits_and_statuses = storage.origin_visit_get_with_statuses(origin.url).results
	assert len(origin["snapshots"]["nodes"]) == 1			assert data_origin["url"] == storage_origin.url
				assert data_origin["id"] == storage_origin.id.hex()
				assert len(data_origin["visits"]["nodes"]) == len(visits_and_statuses)
				assert data_origin["latestVisit"]["visitId"] == visits_and_statuses[-1].visit.visit
				snapshots = storage.origin_snapshot_get_all(origin.url)
				assert len(data_origin["snapshots"]["nodes"]) == len(snapshots)