Changeset View
Changeset View
Standalone 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.server.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' | ||||
Show All 18 Lines | |||||
@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 |