Mirroristas/planetmint
2
0
Fork 0
mirror of https://github.com/planetmint/planetmint.git synced 2025-11-27 07:48:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

test_queries.py -> fixed

Browse Source
This commit is contained in:
andrei 2022-05-05 16:19:14 +03:00
parent 41a65520d7
commit b1e1928702
2 changed files with 18 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
planetmint/backend/tarantool/query.py
View File

@ -99,9 +99,6 @@ def get_transactions(connection, transactions_ids: list):
def store_metadatas(connection, metadata: list): def store_metadatas(connection, metadata: list):
space = connection.space("meta_data") space = connection.space("meta_data")
for meta in metadata: for meta in metadata:
data = meta["data"] if not "metadata" in meta else meta["metadata"]
if data:
space.insert((meta["id"], meta["data"] if not "metadata" in meta else meta["metadata"])) space.insert((meta["id"], meta["data"] if not "metadata" in meta else meta["metadata"]))
@ -120,25 +117,24 @@ def get_metadata(connection, transaction_ids: list):
# asset: {"id": "asset_id"} # asset: {"id": "asset_id"}
# asset: {"data": any} -> insert (tx_id, asset["data"]). # asset: {"data": any} -> insert (tx_id, asset["data"]).
# def store_asset(connection, asset: dict, tx_id=None): # def store_asset(connection, asset: dict, tx_id=None):
def store_asset(connection, asset: dict): def store_asset(connection, asset):
space = connection.space("assets") space = connection.space("assets")
# print(f"DATA store asset: {asset}")
try: try:
if isinstance(asset, dict):
space.insert((asset, asset["id"], asset["id"]))
elif isinstance(asset, tuple):
space.insert(asset) space.insert(asset)
# if tx_id is not None: else:
# space.insert((asset, tx_id, tx_id)) raise Exception(f"Unkown asset format for this query :: {asset} ::")
# else: except Exception as err: # TODO Add Raise For Duplicate
# space.insert((asset, str(asset["id"]), str(asset["id"]))) # TODO Review this function print(err)
except: # TODO Add Raise For Duplicate
print("DUPLICATE ERROR")
@register_query(TarantoolDB) @register_query(TarantoolDB)
def store_assets(connection, assets: list): def store_assets(connection, assets: list):
space = connection.space("assets")
for asset in assets: for asset in assets:
try: try:
space.insert(asset) store_asset(asset=asset)
except Exception as ex: # TODO Raise ERROR for Duplicate except Exception as ex: # TODO Raise ERROR for Duplicate
print(f"EXCEPTION : {ex} ") print(f"EXCEPTION : {ex} ")
@ -156,8 +152,8 @@ def get_assets(connection, assets_ids: list) -> list:
_returned_data = [] _returned_data = []
for _id in list(set(assets_ids)): for _id in list(set(assets_ids)):
asset = get_asset(connection, _id) asset = get_asset(connection, _id)
_returned_data.append(tuple(asset)) _returned_data.append(asset)
return sorted(_returned_data, key=lambda k: k[0]["id"], reverse=False) return sorted(_returned_data, key=lambda k: k["id"], reverse=False)
@register_query(TarantoolDB) @register_query(TarantoolDB)

10
tests/backend/tarantool/test_queries.py
View File

@ -51,11 +51,11 @@ def test_write_assets(db_conn):
# conn = Connection().get_connection() # conn = Connection().get_connection()
conn = db_conn.get_connection() conn = db_conn.get_connection()
assets = [ assets = [
({'id': '1', 'data': '1'}, '1', '1'), {'id': '1', 'data': '1'},
({'id': '2', 'data': '2'}, '2', '2'), {'id': '2', 'data': '2'},
({'id': '3', 'data': '3'}, '3', '3'), {'id': '3', 'data': '3'},
# Duplicated id. Should not be written to the database # Duplicated id. Should not be written to the database
({'id': '1', 'data': '1'}, '1', '1'), {'id': '1', 'data': '1'}
] ]
# write the assets # write the assets
@ -63,7 +63,7 @@ def test_write_assets(db_conn):
query.store_asset(connection=conn, asset=asset) query.store_asset(connection=conn, asset=asset)
# check that 3 assets were written to the database # check that 3 assets were written to the database
documents = query.get_assets(assets_ids=[asset[2] for asset in assets], connection=conn) documents = query.get_assets(assets_ids=[asset["id"] for asset in assets], connection=conn)
assert len(documents) == 3 assert len(documents) == 3
assert list(documents) == assets[:-1] assert list(documents) == assets[:-1]