added blockio load implementation

This commit is contained in:
Kevin Froman 2021-01-24 23:29:02 +00:00
parent 5ed5b8455a
commit 75234310ca
3 changed files with 23 additions and 6 deletions

View File

@ -3,6 +3,7 @@
Wrap safedb for storing and fetching blocks
"""
from .store import store_block
from .load import load_block, list_blocks_by_type
"""
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by

View File

@ -2,11 +2,12 @@
Get blocks from safedb
"""
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, List
from kasten import Kasten
if TYPE_CHECKING:
from kasten.types import BlockChecksumBytes
from safedb import SafeDB
"""
This program is free software: you can redistribute it and/or modify
@ -23,11 +24,13 @@ You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
def load_block(block: 'BlockChecksumBytes', safe_db: 'SafeDB') -> Kasten:
kasten_generator = get_generator
return Kasten()
#def list_blocks_by_type(block_type: str, safe_db: 'SafeDB'):
return Kasten(
block, safe_db.get(block), generator=None, auto_check_generator=False)
def list_blocks_by_type(
block_type: str, safe_db: 'SafeDB') -> List['BlockChecksumBytes']:
blocks = safe_db.get(f'bl-{block_type}')
return zip(*[iter(blocks)]*64)

View File

@ -16,6 +16,7 @@ setup_default_plugins()
import kasten
from onionrblocks.generators import anonvdf
from onionrblocks import blockcreator
from utils import identifyhome
import safedb
@ -47,6 +48,18 @@ class TestBlockIO(unittest.TestCase):
db.close()
_remove_db(db_file)
def test_list_blocks(self):
db_file = identifyhome.identify_home() + 'test.db'
db = safedb.SafeDB(db_file)
expected_list = []
for i in range(10):
bl = blockcreator.create_anonvdf_block(b'test' + int(i).to_bytes(1, 'big'), 'txt', 60)
blockio.store_block(bl, db)
expected_list.append(bl.id)
l = blockio.list_blocks_by_type('txt', db)
self.assertEqual(len(list(l)), len(expected_list))
for i in l:
self.assertIn(bytes(i), expected_list)
unittest.main()