Differential D2069 Diff 6987 swh/graph/tests/conftest.py

Changeset View

Standalone View

swh/graph/tests/conftest.py

	import multiprocessing			import multiprocessing
	import pytest			import pytest
	from pathlib import Path			from pathlib import Path

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

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

	SWH_GRAPH_ROOT = Path(__file__).parents[3]			SWH_GRAPH_ROOT = Path(__file__).parents[3]
	TEST_GRAPH_PATH = SWH_GRAPH_ROOT / 'tests/dataset/output/example'			TEST_GRAPH_PATH = SWH_GRAPH_ROOT / 'tests/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):
	backend = Backend(graph_path=str(TEST_GRAPH_PATH))			backend = Backend(graph_path=str(TEST_GRAPH_PATH))
	with backend:			with backend:
	with loop_context() as loop:			with loop_context() as loop:
	app = make_app(backend=backend)			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()


	@pytest.fixture(scope="module")			@pytest.fixture(scope="module")
	def graph_client():			def graph_client():
	queue = multiprocessing.Queue()			queue = multiprocessing.Queue()
	server = GraphServerProcess(queue)			server = GraphServerProcess(queue)
	server.start()			server.start()
	url = queue.get()			url = queue.get()
	yield RemoteGraphClient(str(url))			yield RemoteGraphClient(str(url))
	server.terminate()			server.terminate()


				@pytest.fixture(scope="module")
				def graph():
				with graph_load(str(TEST_GRAPH_PATH)) as g:
				yield g