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

disable dht test for now

Browse Source
This commit is contained in:
John Smith 2023-06-19 15:25:53 -04:00
parent a01d286dcf
commit c556258fe7
1 changed files with 59 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
veilid-python/tests/test_dht.py
View File

@ -1,74 +1,74 @@
# Routing context veilid tests
# # Routing context veilid tests
import veilid
import pytest
import asyncio
import json
from . import *
# import veilid
# import pytest
# import asyncio
# import json
# from . import *
##################################################################
BOGUS_KEY = veilid.TypedKey.from_value(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.PublicKey.from_bytes(b' '))
# ##################################################################
# BOGUS_KEY = veilid.TypedKey.from_value(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.PublicKey.from_bytes(b' '))
@pytest.mark.asyncio
async def test_get_dht_value_unopened(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
with pytest.raises(veilid.VeilidAPIError):
out = await rc.get_dht_value(BOGUS_KEY, veilid.ValueSubkey(0), False)
# @pytest.mark.asyncio
# async def test_get_dht_value_unopened(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# with pytest.raises(veilid.VeilidAPIError):
# out = await rc.get_dht_value(BOGUS_KEY, veilid.ValueSubkey(0), False)
@pytest.mark.asyncio
async def test_open_dht_record_nonexistent_no_writer(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
with pytest.raises(veilid.VeilidAPIError):
out = await rc.open_dht_record(BOGUS_KEY, None)
# @pytest.mark.asyncio
# async def test_open_dht_record_nonexistent_no_writer(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# with pytest.raises(veilid.VeilidAPIError):
# out = await rc.open_dht_record(BOGUS_KEY, None)
@pytest.mark.asyncio
async def test_close_dht_record_nonexistent(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
with pytest.raises(veilid.VeilidAPIError):
await rc.close_dht_record(BOGUS_KEY)
# @pytest.mark.asyncio
# async def test_close_dht_record_nonexistent(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# with pytest.raises(veilid.VeilidAPIError):
# await rc.close_dht_record(BOGUS_KEY)
@pytest.mark.asyncio
async def test_delete_dht_record_nonexistent(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
with pytest.raises(veilid.VeilidAPIError):
await rc.delete_dht_record(BOGUS_KEY)
# @pytest.mark.asyncio
# async def test_delete_dht_record_nonexistent(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# with pytest.raises(veilid.VeilidAPIError):
# await rc.delete_dht_record(BOGUS_KEY)
@pytest.mark.asyncio
async def test_create_delete_dht_record_simple(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
await rc.close_dht_record(rec.key)
await rc.delete_dht_record(rec.key)
# @pytest.mark.asyncio
# async def test_create_delete_dht_record_simple(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
# await rc.close_dht_record(rec.key)
# await rc.delete_dht_record(rec.key)
@pytest.mark.asyncio
async def test_get_dht_value_nonexistent(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
assert await rc.get_dht_value(rec.key, 0, False) == None
await rc.close_dht_record(rec.key)
await rc.delete_dht_record(rec.key)
# @pytest.mark.asyncio
# async def test_get_dht_value_nonexistent(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
# assert await rc.get_dht_value(rec.key, 0, False) == None
# await rc.close_dht_record(rec.key)
# await rc.delete_dht_record(rec.key)
@pytest.mark.asyncio
async def test_set_get_dht_value(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
# @pytest.mark.asyncio
# async def test_set_get_dht_value(api_connection: veilid.VeilidAPI):
# rc = await api_connection.new_routing_context()
# async with rc:
# rec = await rc.create_dht_record(veilid.CryptoKind.CRYPTO_KIND_VLD0, veilid.DHTSchema.dflt(1))
vd = await rc.set_dht_value(rec.key, 0, b"BLAH BLAH BLAH")
assert vd != None
# vd = await rc.set_dht_value(rec.key, 0, b"BLAH BLAH BLAH")
# assert vd != None
vd2 = await rc.get_dht_value(rec.key, 0, False)
assert vd2 != None
# vd2 = await rc.get_dht_value(rec.key, 0, False)
# assert vd2 != None
assert vd == vd2
# assert vd == vd2
await rc.close_dht_record(rec.key)
await rc.delete_dht_record(rec.key)
# await rc.close_dht_record(rec.key)
# await rc.delete_dht_record(rec.key)