veilid/veilid-python/tests/test_table_db.py

128 lines
4.1 KiB
Python
Raw Normal View History

2023-06-18 18:47:39 -04:00
# TableDB veilid tests
import pytest
import veilid
from veilid.api import CryptoSystem
TEST_DB = "__pytest_db"
TEST_NONEXISTENT_DB = "__pytest_nonexistent_db"
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
@pytest.mark.asyncio
async def test_delete_table_db_nonexistent(api_connection: veilid.VeilidAPI):
deleted = await api_connection.delete_table_db(TEST_NONEXISTENT_DB)
assert not deleted
@pytest.mark.asyncio
async def test_open_delete_table_db(api_connection: veilid.VeilidAPI):
# delete test db if it exists
await api_connection.delete_table_db(TEST_DB)
tdb = await api_connection.open_table_db(TEST_DB, 1)
async with tdb:
# delete should fail since it is still open
with pytest.raises(veilid.VeilidAPIErrorGeneric) as exc:
await api_connection.delete_table_db(TEST_DB)
# drop the db
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
# now delete should succeed
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
@pytest.mark.asyncio
async def test_open_twice_table_db(api_connection: veilid.VeilidAPI):
# delete test db if it exists
await api_connection.delete_table_db(TEST_DB)
tdb = await api_connection.open_table_db(TEST_DB, 1)
tdb2 = await api_connection.open_table_db(TEST_DB, 1)
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
# delete should fail because open
with pytest.raises(veilid.VeilidAPIErrorGeneric) as exc:
await api_connection.delete_table_db(TEST_DB)
await tdb.release()
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
# delete should fail because open
with pytest.raises(veilid.VeilidAPIErrorGeneric) as exc:
await api_connection.delete_table_db(TEST_DB)
await tdb2.release()
# delete should now succeed
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted
@pytest.mark.asyncio
async def test_open_twice_table_db_store_load(api_connection: veilid.VeilidAPI):
# delete test db if it exists
await api_connection.delete_table_db(TEST_DB)
tdb = await api_connection.open_table_db(TEST_DB, 1)
async with tdb:
tdb2 = await api_connection.open_table_db(TEST_DB, 1)
async with tdb2:
2023-07-22 13:06:46 -04:00
# store into first db copy
2023-06-18 18:47:39 -04:00
await tdb.store(b"asdf", b"1234")
# load from second db copy
2024-03-16 23:57:46 -04:00
assert await tdb2.load(b"asdf") == b"1234"
2023-06-18 18:47:39 -04:00
# delete should now succeed
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
@pytest.mark.asyncio
async def test_open_twice_table_db_store_delete_load(api_connection: veilid.VeilidAPI):
# delete test db if it exists
await api_connection.delete_table_db(TEST_DB)
tdb = await api_connection.open_table_db(TEST_DB, 1)
async with tdb:
tdb2 = await api_connection.open_table_db(TEST_DB, 1)
async with tdb2:
2023-07-22 13:06:46 -04:00
# store into first db copy
2023-06-18 18:47:39 -04:00
await tdb.store(b"asdf", b"1234")
# delete from second db copy and clean up
await tdb2.delete(b"asdf")
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
# load from first db copy
assert await tdb.load(b"asdf") == None
# delete should now succeed
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted
@pytest.mark.asyncio
async def test_resize_table_db(api_connection: veilid.VeilidAPI):
# delete test db if it exists
await api_connection.delete_table_db(TEST_DB)
tdb = await api_connection.open_table_db(TEST_DB, 1)
async with tdb:
# reopen the db with more columns should fail if it is already open
with pytest.raises(veilid.VeilidAPIErrorGeneric) as exc:
await api_connection.open_table_db(TEST_DB, 2)
2023-07-22 13:06:46 -04:00
2023-06-18 18:47:39 -04:00
tdb2 = await api_connection.open_table_db(TEST_DB, 2)
async with tdb2:
# write something to second column
2023-07-22 13:06:46 -04:00
await tdb2.store(b"qwer", b"5678", col=1)
2023-06-18 18:47:39 -04:00
# reopen the db with fewer columns
tdb = await api_connection.open_table_db(TEST_DB, 1)
async with tdb:
2023-07-22 13:06:46 -04:00
# Should fail access to second column
2023-06-18 18:47:39 -04:00
with pytest.raises(veilid.VeilidAPIErrorGeneric) as exc:
2023-07-22 13:06:46 -04:00
await tdb.load(b"qwer", col=1)
# Should succeed with access to second column
assert await tdb2.load(b"qwer", col=1) == b"5678"
2023-06-18 18:47:39 -04:00
# now delete should succeed
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted