mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-17 14:14:22 +01:00
Adds some missing tests
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
[run]
|
[run]
|
||||||
omit =
|
omit =
|
||||||
pisa/pisad.py
|
pisa/pisad.py
|
||||||
|
pisa/logger.py
|
||||||
pisa/sample_conf.py
|
pisa/sample_conf.py
|
||||||
pisa/time_traveler.py
|
pisa/time_traveler.py
|
||||||
pisa/utils/auth_proxy.py
|
pisa/utils/auth_proxy.py
|
||||||
@@ -84,7 +84,7 @@ def get_appointment():
|
|||||||
response.append(job_data)
|
response.append(job_data)
|
||||||
|
|
||||||
if not response:
|
if not response:
|
||||||
response.append({"locator": locator, "status": "not found"})
|
response.append({"locator": locator, "status": "not_found"})
|
||||||
|
|
||||||
response = jsonify(response)
|
response = jsonify(response)
|
||||||
|
|
||||||
|
|||||||
@@ -121,6 +121,7 @@ class Responder:
|
|||||||
Cleaner.delete_completed_jobs(self.jobs, self.tx_job_map, completed_jobs, height)
|
Cleaner.delete_completed_jobs(self.jobs, self.tx_job_map, completed_jobs, height)
|
||||||
self.rebroadcast(txs_to_rebroadcast)
|
self.rebroadcast(txs_to_rebroadcast)
|
||||||
|
|
||||||
|
# NOTCOVERED
|
||||||
else:
|
else:
|
||||||
logger.warning("Reorg found", local_prev_block_hash=prev_block_hash,
|
logger.warning("Reorg found", local_prev_block_hash=prev_block_hash,
|
||||||
remote_prev_block_hash=block.get('previousblockhash'))
|
remote_prev_block_hash=block.get('previousblockhash'))
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import json
|
import json
|
||||||
import pytest
|
import pytest
|
||||||
import requests
|
import requests
|
||||||
|
from os import urandom
|
||||||
from hashlib import sha256
|
from hashlib import sha256
|
||||||
from binascii import unhexlify
|
from binascii import unhexlify
|
||||||
|
|
||||||
@@ -99,6 +100,16 @@ def test_request_appointment(new_appointment):
|
|||||||
assert (all([status == "being_watched" for status in appointment_status]))
|
assert (all([status == "being_watched" for status in appointment_status]))
|
||||||
|
|
||||||
|
|
||||||
|
def test_request_random_appointment():
|
||||||
|
r = requests.get(url=PISA_API + "/get_appointment?locator=" + urandom(32).hex())
|
||||||
|
assert (r.status_code == 200)
|
||||||
|
|
||||||
|
received_appointments = json.loads(r.content)
|
||||||
|
appointment_status = [appointment.pop("status") for appointment in received_appointments]
|
||||||
|
|
||||||
|
assert (all([status == "not_found" for status in appointment_status]))
|
||||||
|
|
||||||
|
|
||||||
def test_add_appointment_multiple_times(new_appointment, n=MULTIPLE_APPOINTMENTS):
|
def test_add_appointment_multiple_times(new_appointment, n=MULTIPLE_APPOINTMENTS):
|
||||||
# Multiple appointments with the same locator should be valid
|
# Multiple appointments with the same locator should be valid
|
||||||
# DISCUSS: #34-store-identical-appointments
|
# DISCUSS: #34-store-identical-appointments
|
||||||
|
|||||||
@@ -43,6 +43,12 @@ def test_get_block(best_block_hash):
|
|||||||
assert block.get('hash') == best_block_hash and 'height' in block and 'previousblockhash' in block and 'tx' in block
|
assert block.get('hash') == best_block_hash and 'height' in block and 'previousblockhash' in block and 'tx' in block
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_random_block():
|
||||||
|
block = BlockProcessor.get_block(urandom(32).hex())
|
||||||
|
|
||||||
|
assert block is None
|
||||||
|
|
||||||
|
|
||||||
def test_get_block_count():
|
def test_get_block_count():
|
||||||
block_count = BlockProcessor.get_block_count()
|
block_count = BlockProcessor.get_block_count()
|
||||||
assert isinstance(block_count, int) and block_count >= 0
|
assert isinstance(block_count, int) and block_count >= 0
|
||||||
@@ -55,6 +61,15 @@ def test_potential_matches(txids, locator_uuid_map):
|
|||||||
assert locator_uuid_map.keys() == potential_matches.keys()
|
assert locator_uuid_map.keys() == potential_matches.keys()
|
||||||
|
|
||||||
|
|
||||||
|
def test_potential_matches_random(locator_uuid_map):
|
||||||
|
txids = [urandom(32).hex() for _ in range(len(locator_uuid_map))]
|
||||||
|
|
||||||
|
potential_matches = BlockProcessor.get_potential_matches(txids, locator_uuid_map)
|
||||||
|
|
||||||
|
# None of the ids should match
|
||||||
|
assert len(potential_matches) == 0
|
||||||
|
|
||||||
|
|
||||||
def test_potential_matches_random_data(locator_uuid_map):
|
def test_potential_matches_random_data(locator_uuid_map):
|
||||||
# The likelihood of finding a potential match with random data should be negligible
|
# The likelihood of finding a potential match with random data should be negligible
|
||||||
txids = [urandom(32).hex() for _ in range(TEST_SET_SIZE)]
|
txids = [urandom(32).hex() for _ in range(TEST_SET_SIZE)]
|
||||||
|
|||||||
@@ -25,7 +25,6 @@ def carrier():
|
|||||||
|
|
||||||
|
|
||||||
def test_send_transaction(run_bitcoind, carrier):
|
def test_send_transaction(run_bitcoind, carrier):
|
||||||
# We are mocking bitcoind and in our simulator txid == tx
|
|
||||||
tx = TX.create_dummy_transaction()
|
tx = TX.create_dummy_transaction()
|
||||||
txid = sha256d(tx)
|
txid = sha256d(tx)
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
from pisa import logging
|
from pisa import logging
|
||||||
from pisa.tools import check_txid_format
|
from pisa.tools import check_txid_format
|
||||||
from pisa.tools import can_connect_to_bitcoind, in_correct_network
|
from pisa.tools import can_connect_to_bitcoind, in_correct_network, bitcoin_cli
|
||||||
|
|
||||||
logging.getLogger().disabled = True
|
logging.getLogger().disabled = True
|
||||||
|
|
||||||
@@ -22,6 +22,15 @@ def test_can_connect_to_bitcoind():
|
|||||||
# assert can_connect_to_bitcoind() is False
|
# assert can_connect_to_bitcoind() is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_bitcoin_cli():
|
||||||
|
try:
|
||||||
|
bitcoin_cli().help()
|
||||||
|
assert True
|
||||||
|
|
||||||
|
except Exception:
|
||||||
|
assert False
|
||||||
|
|
||||||
|
|
||||||
def test_check_txid_format():
|
def test_check_txid_format():
|
||||||
assert(check_txid_format(None) is False)
|
assert(check_txid_format(None) is False)
|
||||||
assert(check_txid_format("") is False)
|
assert(check_txid_format("") is False)
|
||||||
|
|||||||
Reference in New Issue
Block a user