gwp 0.2.1

A standalone, pure Rust gRPC wire protocol for GQL (ISO/IEC 39075)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

"""Shared fixtures for gwp-py tests."""

import socket
import subprocess
import sys
import time
from pathlib import Path

import pytest


def _find_free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _wait_for_port(port: int, timeout: float = 10.0) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.5):
                return
        except OSError:
            time.sleep(0.1)
    raise TimeoutError(f"server did not start on port {port}")


@pytest.fixture(scope="session")
def test_server():
    """Start the gwp-test-server and yield the endpoint address."""
    # Look for the binary in the workspace
    repo_root = Path(__file__).parent.parent.parent
    binary = repo_root / "target" / "release" / "gwp-test-server"
    if sys.platform == "win32":
        binary = binary.with_suffix(".exe")

    if not binary.exists():
        pytest.skip(f"gwp-test-server not found at {binary}")

    port = _find_free_port()
    proc = subprocess.Popen(
        [str(binary), str(port)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    try:
        _wait_for_port(port)
        yield f"localhost:{port}"
    finally:
        proc.terminate()
        proc.wait(timeout=5)