Changeset View
Changeset View
Standalone 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 multiprocessing | import multiprocessing | ||||
| from pathlib import Path | from pathlib import Path | ||||
| import subprocess | import subprocess | ||||
| 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.client import RemoteGraphClient | from swh.graph.http_client import RemoteGraphClient | ||||
| from swh.graph.naive_client import NaiveClient | from swh.graph.http_naive_client import NaiveClient | ||||
| SWH_GRAPH_TESTS_ROOT = Path(__file__).parents[0] | SWH_GRAPH_TESTS_ROOT = Path(__file__).parents[0] | ||||
| TEST_GRAPH_PATH = SWH_GRAPH_TESTS_ROOT / "dataset/compressed/example" | TEST_GRAPH_PATH = SWH_GRAPH_TESTS_ROOT / "dataset/compressed/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 | # Lazy import to allow debian packaging | ||||
| from swh.graph.backend import Backend | from swh.graph.http_server import make_app | ||||
| from swh.graph.server.app import make_app | |||||
| try: | try: | ||||
| backend = Backend(graph_path=str(TEST_GRAPH_PATH)) | config = {"graph": {"path": TEST_GRAPH_PATH}} | ||||
| with loop_context() as loop: | with loop_context() as loop: | ||||
| app = make_app(backend=backend, debug=True) | app = make_app(config=config, 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) | ||||
| Show All 30 Lines | |||||