mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-17 14:14:22 +01:00
Fixes memory leak bug regarding test db
This commit is contained in:
@@ -7,7 +7,6 @@ from threading import Thread
|
|||||||
from hashlib import sha256
|
from hashlib import sha256
|
||||||
from binascii import unhexlify
|
from binascii import unhexlify
|
||||||
|
|
||||||
from pisa.conf import DB_PATH
|
|
||||||
from apps.cli.blob import Blob
|
from apps.cli.blob import Blob
|
||||||
from pisa.api import start_api
|
from pisa.api import start_api
|
||||||
from pisa.responder import Job
|
from pisa.responder import Job
|
||||||
@@ -31,8 +30,7 @@ def run_bitcoind():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="session")
|
@pytest.fixture(scope="session")
|
||||||
def run_api():
|
def run_api(db_manager):
|
||||||
db_manager = DBManager(DB_PATH)
|
|
||||||
watcher = Watcher(db_manager)
|
watcher = Watcher(db_manager)
|
||||||
|
|
||||||
api_thread = Thread(target=start_api, args=[watcher])
|
api_thread = Thread(target=start_api, args=[watcher])
|
||||||
@@ -48,7 +46,7 @@ def prng_seed():
|
|||||||
random.seed(0)
|
random.seed(0)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="module")
|
@pytest.fixture(scope="session")
|
||||||
def db_manager():
|
def db_manager():
|
||||||
manager = DBManager("test_db")
|
manager = DBManager("test_db")
|
||||||
yield manager
|
yield manager
|
||||||
|
|||||||
@@ -80,8 +80,18 @@ def test_load_appointments_db(db_manager):
|
|||||||
assert set(values) == set(local_appointments.values()) and (len(values) == len(local_appointments))
|
assert set(values) == set(local_appointments.values()) and (len(values) == len(local_appointments))
|
||||||
|
|
||||||
|
|
||||||
def test_get_last_known_block(db_manager):
|
def test_get_last_known_block():
|
||||||
|
db_path = "empty_db"
|
||||||
|
|
||||||
|
# First we check if the db exists, and if so we delete it
|
||||||
|
if os.path.isdir(db_path):
|
||||||
|
shutil.rmtree(db_path)
|
||||||
|
|
||||||
|
# Check that the db can be created if it does not exist
|
||||||
|
db_manager = open_create_db(db_path)
|
||||||
|
|
||||||
# Trying to get any last block for either the watcher or the responder should return None for an empty db
|
# Trying to get any last block for either the watcher or the responder should return None for an empty db
|
||||||
|
|
||||||
for key in [WATCHER_LAST_BLOCK_KEY, RESPONDER_LAST_BLOCK_KEY]:
|
for key in [WATCHER_LAST_BLOCK_KEY, RESPONDER_LAST_BLOCK_KEY]:
|
||||||
assert db_manager.get_last_known_block(key) is None
|
assert db_manager.get_last_known_block(key) is None
|
||||||
|
|
||||||
@@ -91,6 +101,9 @@ def test_get_last_known_block(db_manager):
|
|||||||
db_manager.db.put(key.encode("utf-8"), block_hash.encode("utf-8"))
|
db_manager.db.put(key.encode("utf-8"), block_hash.encode("utf-8"))
|
||||||
assert db_manager.get_last_known_block(key) == block_hash
|
assert db_manager.get_last_known_block(key) == block_hash
|
||||||
|
|
||||||
|
# Removing test db
|
||||||
|
shutil.rmtree(db_path)
|
||||||
|
|
||||||
|
|
||||||
def test_create_entry(db_manager):
|
def test_create_entry(db_manager):
|
||||||
key = get_random_value_hex(32)
|
key = get_random_value_hex(32)
|
||||||
|
|||||||
Reference in New Issue
Block a user