Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-12-25 07:19:26 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix python test

Browse Source
This commit is contained in:
Christien Rioux 2024-03-06 16:10:01 -05:00
parent 78e84f6573
commit 6d7c62fb6b
1 changed files with 115 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -69,21 +69,27 @@ async def test_routing_context_app_message_loopback():
async with rc: async with rc:
# make a new local private route # make a new local private route
prl, blob = await api.new_private_route() prl, blob = await api.new_private_route()
try:
# import it as a remote route as well so we can send to it
prr = await api.import_remote_private_route(blob)
try:
# send an app message to our own private route
message = b"abcd1234"
await rc.app_message(prr, message)
# import it as a remote route as well so we can send to it # we should get the same message back
prr = await api.import_remote_private_route(blob) update: veilid.VeilidUpdate = await asyncio.wait_for(
app_message_queue.get(), timeout=10
)
# send an app message to our own private route assert isinstance(update.detail, veilid.VeilidAppMessage)
message = b"abcd1234" assert update.detail.message == message
await rc.app_message(prr, message) finally:
# release imported private route
# we should get the same message back await api.release_private_route(prr)
update: veilid.VeilidUpdate = await asyncio.wait_for( finally:
app_message_queue.get(), timeout=10 # release local private route
) await api.release_private_route(prl)
assert isinstance(update.detail, veilid.VeilidAppMessage)
assert update.detail.message == message
@pytest.mark.asyncio @pytest.mark.asyncio
@ -109,29 +115,36 @@ async def test_routing_context_app_call_loopback():
async with rc: async with rc:
# make a new local private route # make a new local private route
prl, blob = await api.new_private_route() prl, blob = await api.new_private_route()
try:
# import it as a remote route as well so we can send to it
prr = await api.import_remote_private_route(blob)
try:
# import it as a remote route as well so we can send to it # send an app message to our own private route
prr = await api.import_remote_private_route(blob) request = b"abcd1234"
app_call_task = asyncio.create_task(rc.app_call(prr, request), name="app call task")
# send an app message to our own private route # we should get the same request back
request = b"abcd1234" update: veilid.VeilidUpdate = await asyncio.wait_for(app_call_queue.get(), timeout=10)
app_call_task = asyncio.create_task(rc.app_call(prr, request), name="app call task") appcall = update.detail
# we should get the same request back assert isinstance(appcall, veilid.VeilidAppCall)
update: veilid.VeilidUpdate = await asyncio.wait_for(app_call_queue.get(), timeout=10) assert appcall.message == request
appcall = update.detail
assert isinstance(appcall, veilid.VeilidAppCall) # now we reply to the request
assert appcall.message == request reply = b"qwer5678"
await api.app_call_reply(appcall.call_id, reply)
# now we reply to the request # now we should get the reply from the call
reply = b"qwer5678" result = await app_call_task
await api.app_call_reply(appcall.call_id, reply) assert result == reply
# now we should get the reply from the call
result = await app_call_task
assert result == reply
finally:
# release imported private route
await api.release_private_route(prr)
finally:
# release local private route
await api.release_private_route(prl)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_routing_context_app_message_loopback_big_packets(): async def test_routing_context_app_message_loopback_big_packets():
@ -162,28 +175,35 @@ async def test_routing_context_app_message_loopback_big_packets():
async with rc: async with rc:
# make a new local private route # make a new local private route
prl, blob = await api.new_private_route() prl, blob = await api.new_private_route()
try:
# import it as a remote route as well so we can send to it
prr = await api.import_remote_private_route(blob)
try:
# do this test 5 times
for _ in range(5):
# send a random sized random app message to our own private route
message = random.randbytes(random.randint(0, 32768))
await rc.app_message(prr, message)
# import it as a remote route as well so we can send to it sent_messages.add(message)
prr = await api.import_remote_private_route(blob)
# do this test 100 times # we should get the same messages back
for _ in range(100): print(len(sent_messages))
# send a random sized random app message to our own private route for n in range(len(sent_messages)):
message = random.randbytes(random.randint(0, 32768)) print(n)
await rc.app_message(prr, message) update: veilid.VeilidUpdate = await asyncio.wait_for(
app_message_queue.get(), timeout=10
)
assert isinstance(update.detail, veilid.VeilidAppMessage)
sent_messages.add(message) assert update.detail.message in sent_messages
finally:
# release imported private route
await api.release_private_route(prr)
finally:
# release local private route
await api.release_private_route(prl)
# we should get the same messages back
print(len(sent_messages))
for n in range(len(sent_messages)):
print(n)
update: veilid.VeilidUpdate = await asyncio.wait_for(
app_message_queue.get(), timeout=10
)
assert isinstance(update.detail, veilid.VeilidAppMessage)
assert update.detail.message in sent_messages
@pytest.mark.asyncio @pytest.mark.asyncio
@ -216,30 +236,36 @@ async def test_routing_context_app_call_loopback_big_packets():
await api.debug("purge routes") await api.debug("purge routes")
app_call_task = asyncio.create_task(app_call_queue_task_handler(api), name="app call task") app_call_task = asyncio.create_task(app_call_queue_task_handler(api), name="app call task")
try:
# make a routing context that uses a safety route
rc = await (await api.new_routing_context()).with_sequencing(
veilid.Sequencing.ENSURE_ORDERED
)
async with rc:
# make a new local private route
prl, blob = await api.new_custom_private_route(
[veilid.CryptoKind.CRYPTO_KIND_VLD0],
veilid.Stability.RELIABLE,
veilid.Sequencing.ENSURE_ORDERED)
try:
# import it as a remote route as well so we can send to it
prr = await api.import_remote_private_route(blob)
try:
# do this test 5 times
for _ in range(5):
# send a random sized random app message to our own private route
message = random.randbytes(random.randint(0, 32768))
out_message = await rc.app_call(prr, message)
# make a routing context that uses a safety route assert message == out_message
rc = await (await api.new_routing_context()).with_sequencing( finally:
veilid.Sequencing.ENSURE_ORDERED # release imported private route
) await api.release_private_route(prr)
async with rc: finally:
# make a new local private route # release local private route
prl, blob = await api.new_custom_private_route( await api.release_private_route(prl)
[veilid.CryptoKind.CRYPTO_KIND_VLD0], finally:
veilid.Stability.RELIABLE, app_call_task.cancel()
veilid.Sequencing.ENSURE_ORDERED)
# import it as a remote route as well so we can send to it
prr = await api.import_remote_private_route(blob)
# do this test 10 times
for _ in range(10):
# send a random sized random app message to our own private route
message = random.randbytes(random.randint(0, 32768))
out_message = await rc.app_call(prr, message)
assert message == out_message
app_call_task.cancel()
@pytest.mark.skipif(os.getenv("NOSKIP") != "1", reason="unneeded test, only for performance check") @pytest.mark.skipif(os.getenv("NOSKIP") != "1", reason="unneeded test, only for performance check")
@ -266,16 +292,22 @@ async def test_routing_context_app_message_loopback_bandwidth():
async with rc: async with rc:
# make a new local private route # make a new local private route
prl, blob = await api.new_private_route() prl, blob = await api.new_private_route()
try:
# import it as a remote route as well so we can send to it
prr = await api.import_remote_private_route(blob)
try:
# do this test 1000 times
message = random.randbytes(16384)
for _ in range(10000):
# send a random sized random app message to our own private route
await rc.app_message(prr, message)
# import it as a remote route as well so we can send to it # we should get the same number of messages back (not storing all that data)
prr = await api.import_remote_private_route(blob) for _ in range(10000):
await asyncio.wait_for(app_message_queue.get(), timeout=10)
# do this test 1000 times finally:
message = random.randbytes(16384) # release imported private route
for _ in range(10000): await api.release_private_route(prr)
# send a random sized random app message to our own private route finally:
await rc.app_message(prr, message) # release local private route
await api.release_private_route(prl)
# we should get the same number of messages back (not storing all that data)
for _ in range(10000):
await asyncio.wait_for(app_message_queue.get(), timeout=10)