Differential D6167 Diff 22343 swh/graph/tests/conftest.py

Changeset View

Standalone View

swh/graph/tests/conftest.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 csv			import csv
	import multiprocessing			import multiprocessing
	from pathlib import Path			from pathlib import Path

	from aiohttp.test_utils import TestClient, TestServer, loop_context			from aiohttp.test_utils import TestClient, TestServer, loop_context
	import pytest			import pytest

	from swh.graph.backend import Backend
	from swh.graph.client import RemoteGraphClient			from swh.graph.client import RemoteGraphClient
	from swh.graph.graph import load as graph_load
	from swh.graph.naive_client import NaiveClient			from swh.graph.naive_client import NaiveClient
	from swh.graph.server.app import make_app

	SWH_GRAPH_TESTS_ROOT = Path(__file__).parents[0]			SWH_GRAPH_TESTS_ROOT = Path(__file__).parents[0]
	TEST_GRAPH_PATH = SWH_GRAPH_TESTS_ROOT / "dataset/output/example"			TEST_GRAPH_PATH = SWH_GRAPH_TESTS_ROOT / "dataset/output/example"


	class GraphServerProcess(multiprocessing.Process):			class GraphServerProcess(multiprocessing.Process):
	def __init__(self, q, args, *kwargs):			def __init__(self, q, args, *kwargs):
	self.q = q			self.q = q
	super().__init__(args, *kwargs)			super().__init__(args, *kwargs)

	def run(self):			def run(self):
				# Lazy import to allow debian packaging
				from swh.graph.backend import Backend
				from swh.graph.server.app import make_app

	try:			try:
	backend = Backend(graph_path=str(TEST_GRAPH_PATH))			backend = Backend(graph_path=str(TEST_GRAPH_PATH))
	with loop_context() as loop:			with loop_context() as loop:
	app = make_app(backend=backend, debug=True)			app = make_app(backend=backend, debug=True)
	client = TestClient(TestServer(app), loop=loop)			client = TestClient(TestServer(app), loop=loop)
	loop.run_until_complete(client.start_server())			loop.run_until_complete(client.start_server())
	url = client.make_url("/graph/")			url = client.make_url("/graph/")
	self.q.put(url)			self.q.put(url)
	loop.run_forever()			loop.run_forever()
	except Exception as e:			except Exception as e:
	self.q.put(e)			self.q.put(e)
				ardumontAuthorUnsubmitted Done Inline Actions the coverage report here lies, it's run locally (i know because i use some pdb stuff). ardumont: the coverage report here lies, it's run locally (i know because i use some pdb stuff).


	@pytest.fixture(scope="module", params=["remote", "naive"])			@pytest.fixture(scope="module", params=["remote", "naive"])
	def graph_client(request):			def graph_client(request):
	if request.param == "remote":			if request.param == "remote":
	queue = multiprocessing.Queue()			queue = multiprocessing.Queue()
	server = GraphServerProcess(queue)			server = GraphServerProcess(queue)
	server.start()			server.start()
	res = queue.get()			res = queue.get()
	if isinstance(res, Exception):			if isinstance(res, Exception):
	raise res			raise res
	yield RemoteGraphClient(str(res))			yield RemoteGraphClient(str(res))
	server.terminate()			server.terminate()
	else:			else:
	with open(SWH_GRAPH_TESTS_ROOT / "dataset/example.nodes.csv") as fd:			with open(SWH_GRAPH_TESTS_ROOT / "dataset/example.nodes.csv") as fd:
	nodes = [node for (node,) in csv.reader(fd, delimiter=" ")]			nodes = [node for (node,) in csv.reader(fd, delimiter=" ")]
	with open(SWH_GRAPH_TESTS_ROOT / "dataset/example.edges.csv") as fd:			with open(SWH_GRAPH_TESTS_ROOT / "dataset/example.edges.csv") as fd:
	edges = list(csv.reader(fd, delimiter=" "))			edges = list(csv.reader(fd, delimiter=" "))
	yield NaiveClient(nodes=nodes, edges=edges)			yield NaiveClient(nodes=nodes, edges=edges)


	@pytest.fixture(scope="module")			@pytest.fixture(scope="module")
	def graph():			def graph():
				# Lazy import to allow debian packaging
				from swh.graph.graph import load as graph_load

	with graph_load(str(TEST_GRAPH_PATH)) as g:			with graph_load(str(TEST_GRAPH_PATH)) as g:
	yield g			yield g