Changeset View
Changeset View
Standalone View
Standalone View
swh/graph/tests/conftest.py
| # Copyright (C) 2019-2021 The Software Heritage developers | |||||
| # See the AUTHORS file at the top-level directory of this distribution | |||||
| # License: GNU General Public License version 3, or any later version | |||||
| # See top-level LICENSE file for more information | |||||
| 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.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.graph import load as graph_load | ||||
| from swh.graph.naive_client import NaiveClient | |||||
| from swh.graph.server.app import make_app | 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): | ||||
| Show All 10 Lines | def run(self): | ||||
| 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) | ||||
| @pytest.fixture(scope="module") | @pytest.fixture(scope="module", params=["remote", "naive"]) | ||||
| def graph_client(): | def graph_client(request): | ||||
seirl: This is a bit subjective, but i would probably prefer a fixture for `graph_remote_client`, a… | |||||
vlorentzAuthorUnsubmitted Done Inline ActionsI don't think I can do that without graph_client always starting a WebGraph process, even if we use a naive client vlorentz: I don't think I can do that without `graph_client` always starting a WebGraph process, even if… | |||||
| 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: | |||||
| with open(SWH_GRAPH_TESTS_ROOT / "dataset/example.nodes.csv") as fd: | |||||
| nodes = [node for (node,) in csv.reader(fd, delimiter=" ")] | |||||
| with open(SWH_GRAPH_TESTS_ROOT / "dataset/example.edges.csv") as fd: | |||||
| edges = list(csv.reader(fd, delimiter=" ")) | |||||
| yield NaiveClient(nodes=nodes, edges=edges) | |||||
| @pytest.fixture(scope="module") | @pytest.fixture(scope="module") | ||||
| def graph(): | def graph(): | ||||
| with graph_load(str(TEST_GRAPH_PATH)) as g: | with graph_load(str(TEST_GRAPH_PATH)) as g: | ||||
| yield g | yield g | ||||
This is a bit subjective, but i would probably prefer a fixture for graph_remote_client, a fixture for graph_naive_client, and finally a fixture graph_client that does the dispatch.