mirror of
https://gitlab.com/veilid/veilid.git
synced 2024-11-22 08:56:58 -06:00
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
import appdirs
|
|
import errno
|
|
import os
|
|
import socket
|
|
import sys
|
|
import re
|
|
from collections.abc import Callable
|
|
from functools import cache
|
|
|
|
from veilid.json_api import _JsonVeilidAPI
|
|
|
|
import veilid
|
|
|
|
ERRNO_PATTERN = re.compile(r"errno (\d+)", re.IGNORECASE)
|
|
|
|
|
|
class VeilidTestConnectionError(Exception):
|
|
"""The test client could not connect to the veilid-server."""
|
|
|
|
pass
|
|
|
|
|
|
@cache
|
|
def server_info() -> tuple[str, int]:
|
|
"""Return the hostname and port of the test server."""
|
|
VEILID_SERVER_NETWORK = os.getenv("VEILID_SERVER_NETWORK")
|
|
if VEILID_SERVER_NETWORK is None:
|
|
return "localhost", 5959
|
|
|
|
hostname, *rest = VEILID_SERVER_NETWORK.split(":")
|
|
if rest:
|
|
return hostname, int(rest[0])
|
|
return hostname, 5959
|
|
|
|
def ipc_path_exists(path: str) -> bool:
|
|
"""Determine if an IPC socket exists in a platform independent way."""
|
|
if os.name == 'nt':
|
|
if not path.upper().startswith("\\\\.\\PIPE\\"):
|
|
return False
|
|
return path[9:] in os.listdir("\\\\.\\PIPE")
|
|
else:
|
|
return os.path.exists(path)
|
|
|
|
@cache
|
|
def ipc_info() -> str:
|
|
"""Return the path of the ipc socket of the test server."""
|
|
VEILID_SERVER_IPC = os.getenv("VEILID_SERVER_IPC")
|
|
if VEILID_SERVER_IPC is not None:
|
|
return VEILID_SERVER_IPC
|
|
|
|
if os.name == 'nt':
|
|
return '\\\\.\\PIPE\\veilid-server\\0'
|
|
|
|
ipc_0_path = "/var/db/veilid-server/ipc/0"
|
|
if os.path.exists(ipc_0_path):
|
|
return ipc_0_path
|
|
|
|
# hack to deal with rust's 'directories' crate case-inconsistency
|
|
if sys.platform.startswith('darwin'):
|
|
data_dir = appdirs.user_data_dir("org.Veilid.Veilid")
|
|
else:
|
|
data_dir = appdirs.user_data_dir("veilid","veilid")
|
|
ipc_0_path = os.path.join(data_dir, "ipc", "0")
|
|
return ipc_0_path
|
|
|
|
|
|
async def api_connector(callback: Callable) -> _JsonVeilidAPI:
|
|
"""Return an API connection if possible.
|
|
|
|
If the connection fails due to an inability to connect to the
|
|
server's socket, raise an easy-to-catch VeilidTestConnectionError.
|
|
"""
|
|
|
|
ipc_path = ipc_info()
|
|
hostname, port = server_info()
|
|
|
|
try:
|
|
if ipc_path_exists(ipc_path):
|
|
return await veilid.json_api_connect_ipc(ipc_path, callback)
|
|
else:
|
|
return await veilid.json_api_connect(hostname, port, callback)
|
|
except OSError as exc:
|
|
# This is a little goofy. The underlying Python library handles
|
|
# connection errors in 2 ways, depending on how many connections
|
|
# it attempted to make:
|
|
#
|
|
# - If it only tried to connect to one IP address socket, the
|
|
# library propagates the one single OSError it got.
|
|
#
|
|
# - If it tried to connect to multiple sockets, perhaps because
|
|
# the hostname resolved to several addresses (e.g. "localhost"
|
|
# => 127.0.0.1 and ::1), then the library raises one exception
|
|
# with all the failure exception strings joined together.
|
|
|
|
# If errno is set, it's the first kind of exception. Check that
|
|
# it's the code we expected.
|
|
if exc.errno is not None:
|
|
if exc.errno == errno.ECONNREFUSED:
|
|
raise VeilidTestConnectionError
|
|
raise
|
|
|
|
# If not, use a regular expression to find all the errno values
|
|
# in the combined error string. Check that all of them have the
|
|
# code we're looking for.
|
|
errnos = ERRNO_PATTERN.findall(str(exc))
|
|
if all(int(err) == errno.ECONNREFUSED for err in errnos):
|
|
raise VeilidTestConnectionError
|
|
|
|
raise
|