tests/__init__ => conftest
This commit is contained in:
parent
553ea7a3e0
commit
f1aad14099
@ -1,37 +0,0 @@
|
||||
import os
|
||||
from functools import cache
|
||||
from typing import AsyncGenerator
|
||||
|
||||
import pytest_asyncio
|
||||
import veilid
|
||||
from veilid.json_api import _JsonVeilidAPI
|
||||
|
||||
pytest_plugins = ("pytest_asyncio",)
|
||||
|
||||
|
||||
@cache
|
||||
def server_info() -> tuple[str, int]:
|
||||
"""Return the hostname and port of the test server."""
|
||||
VEILID_SERVER = os.getenv("VEILID_SERVER")
|
||||
if VEILID_SERVER is None:
|
||||
return "localhost", 5959
|
||||
|
||||
hostname, *rest = VEILID_SERVER.split(":")
|
||||
if rest:
|
||||
return hostname, int(rest[0])
|
||||
return hostname, 5959
|
||||
|
||||
|
||||
async def simple_update_callback(update: veilid.VeilidUpdate):
|
||||
print(f"VeilidUpdate: {update}")
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def api_connection() -> AsyncGenerator[_JsonVeilidAPI, None]:
|
||||
hostname, port = server_info()
|
||||
api = await veilid.json_api_connect(hostname, port, simple_update_callback)
|
||||
async with api:
|
||||
# purge routes to ensure we start fresh
|
||||
await api.debug("purge routes")
|
||||
|
||||
yield api
|
37
veilid-python/tests/conftest.py
Normal file
37
veilid-python/tests/conftest.py
Normal file
@ -0,0 +1,37 @@
|
||||
import os
|
||||
from functools import cache
|
||||
from typing import AsyncGenerator
|
||||
|
||||
import pytest_asyncio
|
||||
import veilid
|
||||
from veilid.json_api import _JsonVeilidAPI
|
||||
|
||||
pytest_plugins = ("pytest_asyncio",)
|
||||
|
||||
|
||||
@cache
|
||||
def server_info() -> tuple[str, int]:
|
||||
"""Return the hostname and port of the test server."""
|
||||
VEILID_SERVER = os.getenv("VEILID_SERVER")
|
||||
if VEILID_SERVER is None:
|
||||
return "localhost", 5959
|
||||
|
||||
hostname, *rest = VEILID_SERVER.split(":")
|
||||
if rest:
|
||||
return hostname, int(rest[0])
|
||||
return hostname, 5959
|
||||
|
||||
|
||||
async def simple_update_callback(update: veilid.VeilidUpdate):
|
||||
print(f"VeilidUpdate: {update}")
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def api_connection() -> AsyncGenerator[_JsonVeilidAPI, None]:
|
||||
hostname, port = server_info()
|
||||
api = await veilid.json_api_connect(hostname, port, simple_update_callback)
|
||||
async with api:
|
||||
# purge routes to ensure we start fresh
|
||||
await api.debug("purge routes")
|
||||
|
||||
yield api
|
@ -5,9 +5,7 @@ import socket
|
||||
import pytest
|
||||
import veilid
|
||||
|
||||
from . import api_connection, simple_update_callback
|
||||
|
||||
##################################################################
|
||||
from .conftest import simple_update_callback
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
@ -4,10 +4,6 @@ import pytest
|
||||
import veilid
|
||||
from veilid.api import CryptoSystem
|
||||
|
||||
from . import api_connection
|
||||
|
||||
##################################################################
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_best_crypto_system(api_connection):
|
||||
|
@ -1,12 +1,11 @@
|
||||
# Routing context veilid tests
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
import pytest
|
||||
import veilid
|
||||
|
||||
from . import api_connection, server_info
|
||||
from .conftest import server_info
|
||||
|
||||
##################################################################
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user