Onionr/tests/test_blockdb.py

67 lines
1.9 KiB
Python
Raw Permalink Normal View History

2022-02-05 19:25:28 +00:00
#!/usr/bin/env python3
import sys, os
2022-02-07 00:06:34 +00:00
import time
2022-02-05 19:25:28 +00:00
import dbm
sys.path.append(".")
sys.path.append("src/")
import uuid
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
print("Test directory:", TEST_DIR)
os.environ["ONIONR_HOME"] = TEST_DIR
import unittest
from utils import createdirs
createdirs.create_dirs()
import onionrblocks
import blockdb
2022-02-07 00:06:34 +00:00
def _delete_db():
try:
os.remove(blockdb.block_db_path)
except FileNotFoundError:
pass
2022-02-05 19:25:28 +00:00
class TestBlockDB(unittest.TestCase):
def test_store_vdf_block(self):
2022-02-07 00:06:34 +00:00
_delete_db()
2022-02-05 19:25:28 +00:00
bl: Block = onionrblocks.create_anonvdf_block(os.urandom(10), b'bin', 2500)
2022-04-05 06:17:40 +00:00
blockdb.add_block_to_db(bl)
2022-02-05 19:25:28 +00:00
with dbm.open(blockdb.block_db_path, 'r') as b_db:
b_db[bl.id]
def test_get_blocks_by_type(self):
2022-02-07 00:06:34 +00:00
_delete_db()
2022-02-05 19:25:28 +00:00
with dbm.open(blockdb.block_db_path, 'c') as b_db:
bl: Block = onionrblocks.create_anonvdf_block('test', b'txt', 2500)
b_db[bl.id] = bl.raw
2022-02-07 00:06:34 +00:00
bl2: Block = onionrblocks.create_anonvdf_block('test2', b'bin', 2500)
b_db[bl2.id] = bl2.raw
2022-02-05 19:25:28 +00:00
looped = False
2022-02-07 00:06:34 +00:00
for c, block in enumerate(blockdb.get_blocks_by_type('txt')):
2022-02-05 19:25:28 +00:00
looped = True
self.assertEqual(bl.id, block.id)
2022-02-07 00:06:34 +00:00
self.assertTrue(c <= 1)
2022-02-05 19:25:28 +00:00
self.assertTrue(looped)
2022-02-07 00:06:34 +00:00
def test_get_blocks_after_timestamp(self):
_delete_db()
t = round(time.time())
with dbm.open(blockdb.block_db_path, 'c') as b_db:
bl: Block = onionrblocks.create_anonvdf_block('test', b'txt', 2500)
b_db[bl.id] = bl.raw
time.sleep(1)
t2 = round(time.time())
bl2: Block = onionrblocks.create_anonvdf_block('test2', b'bin', 2500)
b_db[bl2.id] = bl2.raw
bls = list(blockdb.get_blocks_after_timestamp(t + 10))
self.assertTrue(len(bls) == 0)
2022-02-05 19:25:28 +00:00
unittest.main()