liz/veilid
liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

Big refactoring sweep through tests

Browse Source
This commit is contained in:
Teknique 2023-06-17 11:34:09 -07:00
parent b545c5c2be
commit 553ea7a3e0
No known key found for this signature in database
GPG Key ID: 54D9906A294B7FF3
5 changed files with 134 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
veilid-python/tests/__init__.py
View File

@ -1,34 +1,37 @@
from typing import Callable, Awaitable
import os
import pytest
pytest_plugins = ('pytest_asyncio',)
from functools import cache
from typing import AsyncGenerator
import pytest_asyncio
import veilid
from veilid.json_api import _JsonVeilidAPI
pytest_plugins = ("pytest_asyncio",)
##################################################################
VEILID_SERVER = os.getenv("VEILID_SERVER")
if VEILID_SERVER is not None:
vsparts = VEILID_SERVER.split(":")
VEILID_SERVER = vsparts[0]
if len(vsparts) == 2:
VEILID_SERVER_PORT = int(vsparts[1])
else:
VEILID_SERVER_PORT = 5959
else:
VEILID_SERVER = "localhost"
VEILID_SERVER_PORT = 5959
@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_connect_and_run(func: Callable[[veilid.VeilidAPI], Awaitable]):
api = await veilid.json_api_connect(VEILID_SERVER, VEILID_SERVER_PORT, simple_update_callback)
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")
await func(api)
async def simple_update_callback(update: veilid.VeilidUpdate):
print("VeilidUpdate: {}".format(update))
yield api

53
veilid-python/tests/test_basic.py
View File

@ -1,39 +1,46 @@
# Basic veilid tests
import veilid
import socket
import pytest
from . import *
import veilid
from . import api_connection, simple_update_callback
##################################################################
@pytest.mark.asyncio
async def test_connect():
async def func(api: veilid.VeilidAPI):
pass
await simple_connect_and_run(func)
async def test_connect(api_connection):
pass
@pytest.mark.asyncio
async def test_get_node_id():
async def func(api: veilid.VeilidAPI):
# get our own node id
state = await api.get_state()
node_id = state.config.config.network.routing_table.node_id.pop()
await simple_connect_and_run(func)
async def test_get_node_id(api_connection):
state = await api_connection.get_state()
node_ids = state.config.config.network.routing_table.node_id
assert len(node_ids) >= 1
for node_id in node_ids:
assert node_id[4] == ":"
@pytest.mark.asyncio
async def test_fail_connect():
with pytest.raises(Exception):
api = await veilid.json_api_connect("fuahwelifuh32luhwafluehawea", 1, simple_update_callback)
async with api:
pass
with pytest.raises(socket.gaierror) as exc:
await veilid.json_api_connect(
"fuahwelifuh32luhwafluehawea", 1, simple_update_callback
)
assert exc.value.errno == socket.EAI_NONAME
@pytest.mark.asyncio
async def test_version():
async def func(api: veilid.VeilidAPI):
v = await api.veilid_version()
print("veilid_version: {}".format(v.__dict__))
vstr = await api.veilid_version_string()
print("veilid_version_string: {}".format(vstr))
await simple_connect_and_run(func)
async def test_version(api_connection):
v = await api_connection.veilid_version()
print(f"veilid_version: {v.__dict__}")
assert v.__dict__.keys() >= {"_major", "_minor", "_patch"}
vstr = await api_connection.veilid_version_string()
print(f"veilid_version_string: {vstr}")

71
veilid-python/tests/test_crypto.py
View File

@ -1,42 +1,53 @@
# Crypto veilid tests
import veilid
import pytest
from . import *
import veilid
from veilid.api import CryptoSystem
from . import api_connection
##################################################################
@pytest.mark.asyncio
async def test_best_crypto_system():
async def func(api: veilid.VeilidAPI):
bcs = await api.best_crypto_system()
await simple_connect_and_run(func)
@pytest.mark.asyncio
async def test_get_crypto_system():
async def func(api: veilid.VeilidAPI):
cs = await api.get_crypto_system(veilid.CryptoKind.CRYPTO_KIND_VLD0)
# clean up handle early
del cs
await simple_connect_and_run(func)
async def test_best_crypto_system(api_connection):
bcs: CryptoSystem = await api_connection.best_crypto_system()
assert await bcs.default_salt_length() == 16
@pytest.mark.asyncio
async def test_get_crypto_system_invalid():
async def func(api: veilid.VeilidAPI):
with pytest.raises(veilid.VeilidAPIError):
cs = await api.get_crypto_system(veilid.CryptoKind.CRYPTO_KIND_NONE)
await simple_connect_and_run(func)
async def test_get_crypto_system(api_connection):
cs: CryptoSystem = await api_connection.get_crypto_system(
veilid.CryptoKind.CRYPTO_KIND_VLD0
)
assert await cs.default_salt_length() == 16
# clean up handle early
del cs
@pytest.mark.asyncio
async def test_hash_and_verify_password():
async def func(api: veilid.VeilidAPI):
bcs = await api.best_crypto_system()
nonce = await bcs.random_nonce()
salt = nonce.to_bytes()
# Password match
phash = await bcs.hash_password(b"abc123", salt)
assert await bcs.verify_password(b"abc123", phash)
# Password mismatch
phash2 = await bcs.hash_password(b"abc1234", salt)
assert not await bcs.verify_password(b"abc12345", phash)
await simple_connect_and_run(func)
async def test_get_crypto_system_invalid(api_connection):
with pytest.raises(veilid.VeilidAPIErrorInvalidArgument) as exc:
await api_connection.get_crypto_system(veilid.CryptoKind.CRYPTO_KIND_NONE)
assert exc.value.context == "unsupported cryptosystem"
assert exc.value.argument == "kind"
assert exc.value.value == "NONE"
@pytest.mark.asyncio
async def test_hash_and_verify_password(api_connection):
bcs = await api_connection.best_crypto_system()
nonce = await bcs.random_nonce()
salt = nonce.to_bytes()
# Password match
phash = await bcs.hash_password(b"abc123", salt)
assert await bcs.verify_password(b"abc123", phash)
# Password mismatch
phash2 = await bcs.hash_password(b"abc1234", salt)
assert not await bcs.verify_password(b"abc12345", phash)

46
veilid-python/tests/test_routing_context.py
View File

@ -1,34 +1,37 @@
# Routing context veilid tests
import veilid
import pytest
import asyncio
import json
from . import *
import pytest
import veilid
from . import api_connection, server_info
##################################################################
@pytest.mark.asyncio
async def test_routing_contexts():
async def func(api: veilid.VeilidAPI):
rc = await api.new_routing_context()
rcp = await rc.with_privacy()
rcps = await rcp.with_sequencing(veilid.Sequencing.ENSURE_ORDERED)
rcpsr = await rcps.with_custom_privacy(veilid.Stability.RELIABLE)
await simple_connect_and_run(func)
async def test_routing_contexts(api_connection):
rc = await api_connection.new_routing_context()
rcp = await rc.with_privacy()
rcps = await rcp.with_sequencing(veilid.Sequencing.ENSURE_ORDERED)
await rcps.with_custom_privacy(veilid.Stability.RELIABLE)
@pytest.mark.asyncio
async def test_routing_context_app_message_loopback():
app_message_queue = asyncio.Queue()
async def app_message_queue_update_callback(update: veilid.VeilidUpdate):
if update.kind == veilid.VeilidUpdateKind.APP_MESSAGE:
await app_message_queue.put(update)
api = await veilid.json_api_connect(VEILID_SERVER, VEILID_SERVER_PORT, app_message_queue_update_callback)
hostname, port = server_info()
api = await veilid.json_api_connect(
hostname, port, app_message_queue_update_callback
)
async with api:
# purge routes to ensure we start fresh
await api.debug("purge routes")
@ -46,23 +49,24 @@ async def test_routing_context_app_message_loopback():
await rc.app_message(prr, message)
# we should get the same message back
update: veilid.VeilidUpdate = await asyncio.wait_for(app_message_queue.get(), timeout=10)
update: veilid.VeilidUpdate = await asyncio.wait_for(
app_message_queue.get(), timeout=10
)
appmsg: veilid.VeilidAppMessage = update.detail
assert appmsg.message == message
@pytest.mark.asyncio
async def test_routing_context_app_call_loopback():
app_call_queue = asyncio.Queue()
async def app_call_queue_update_callback(update: veilid.VeilidUpdate):
if update.kind == veilid.VeilidUpdateKind.APP_CALL:
await app_call_queue.put(update)
api = await veilid.json_api_connect(VEILID_SERVER, VEILID_SERVER_PORT, app_call_queue_update_callback)
hostname, port = server_info()
api = await veilid.json_api_connect(hostname, port, app_call_queue_update_callback)
async with api:
# purge routes to ensure we start fresh
await api.debug("purge routes")
@ -77,10 +81,14 @@ async def test_routing_context_app_call_loopback():
# send an app message to our own private route
request = b"abcd1234"
app_call_task = asyncio.create_task(rc.app_call(prr, request), name = "app call task")
app_call_task = asyncio.create_task(
rc.app_call(prr, request), name="app call task"
)
# we should get the same request back
update: veilid.VeilidUpdate = await asyncio.wait_for(app_call_queue.get(), timeout=10)
update: veilid.VeilidUpdate = await asyncio.wait_for(
app_call_queue.get(), timeout=10
)
appcall: veilid.VeilidAppCall = update.detail
assert appcall.message == request

2
veilid-python/veilid/json_api.py
View File

@ -1072,5 +1072,5 @@ class _JsonCryptoSystem(CryptoSystem):
async def json_api_connect(
host: str, port: int, update_callback: Callable[[VeilidUpdate], Awaitable]
) -> VeilidAPI:
) -> _JsonVeilidAPI:
return await _JsonVeilidAPI.connect(host, port, update_callback)