mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-18 06:34:19 +01:00
Changes sk_path to sk_der in Watcher
The Watcher used to receive a secret key file path ion the __init__ to load a secret key for signing. That made testing the Watcher hard, since the file needed to be present. Changing it so the main (pisad) loads the file from disk and passes the data the Watcher on init.
This commit is contained in:
@@ -3,6 +3,7 @@ import pytest
|
||||
import requests
|
||||
from time import sleep
|
||||
from threading import Thread
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
from pisa.api import start_api
|
||||
from pisa.watcher import Watcher
|
||||
@@ -10,7 +11,13 @@ from pisa.tools import bitcoin_cli
|
||||
from pisa import HOST, PORT, c_logger
|
||||
from pisa.conf import MAX_APPOINTMENTS
|
||||
|
||||
from test.unit.conftest import generate_block, generate_blocks, get_random_value_hex, generate_dummy_appointment_data
|
||||
from test.unit.conftest import (
|
||||
generate_block,
|
||||
generate_blocks,
|
||||
get_random_value_hex,
|
||||
generate_dummy_appointment_data,
|
||||
generate_keypair,
|
||||
)
|
||||
|
||||
from common.constants import LOCATOR_LEN_BYTES
|
||||
|
||||
@@ -25,7 +32,13 @@ locator_dispute_tx_map = {}
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def run_api(db_manager):
|
||||
watcher = Watcher(db_manager)
|
||||
sk, pk = generate_keypair()
|
||||
sk_der = sk.private_bytes(
|
||||
encoding=serialization.Encoding.DER,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
watcher = Watcher(db_manager, sk_der)
|
||||
|
||||
api_thread = Thread(target=start_api, args=[watcher])
|
||||
api_thread.daemon = True
|
||||
|
||||
@@ -2,13 +2,20 @@ import pytest
|
||||
from uuid import uuid4
|
||||
from threading import Thread
|
||||
from queue import Queue, Empty
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
from pisa import c_logger
|
||||
from pisa.watcher import Watcher
|
||||
from pisa.responder import Responder
|
||||
from pisa.tools import bitcoin_cli
|
||||
from test.unit.conftest import generate_block, generate_blocks, generate_dummy_appointment, get_random_value_hex
|
||||
from pisa.conf import EXPIRY_DELTA, PISA_SECRET_KEY, MAX_APPOINTMENTS
|
||||
from test.unit.conftest import (
|
||||
generate_block,
|
||||
generate_blocks,
|
||||
generate_dummy_appointment,
|
||||
get_random_value_hex,
|
||||
generate_keypair,
|
||||
)
|
||||
from pisa.conf import EXPIRY_DELTA, MAX_APPOINTMENTS
|
||||
|
||||
from common.tools import check_sha256_hex_format
|
||||
from common.cryptographer import Cryptographer
|
||||
@@ -20,15 +27,18 @@ START_TIME_OFFSET = 1
|
||||
END_TIME_OFFSET = 1
|
||||
TEST_SET_SIZE = 200
|
||||
|
||||
with open(PISA_SECRET_KEY, "rb") as key_file_der:
|
||||
sk_der = key_file_der.read()
|
||||
signing_key = Cryptographer.load_private_key_der(sk_der)
|
||||
public_key = signing_key.public_key()
|
||||
|
||||
signing_key, public_key = generate_keypair()
|
||||
sk_der = signing_key.private_bytes(
|
||||
encoding=serialization.Encoding.DER,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def watcher(db_manager):
|
||||
return Watcher(db_manager)
|
||||
return Watcher(db_manager, sk_der)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
@@ -69,15 +79,6 @@ def test_init(watcher):
|
||||
assert type(watcher.responder) is Responder
|
||||
|
||||
|
||||
def test_init_no_key(db_manager):
|
||||
try:
|
||||
Watcher(db_manager, pisa_sk_file=None)
|
||||
assert False
|
||||
|
||||
except ValueError:
|
||||
assert True
|
||||
|
||||
|
||||
def test_add_appointment(run_bitcoind, watcher):
|
||||
# The watcher automatically fires do_watch and do_subscribe on adding an appointment if it is asleep (initial state)
|
||||
# Avoid this by setting the state to awake.
|
||||
|
||||
Reference in New Issue
Block a user