mirror of
https://github.com/yzernik/squeaknode.git
synced 2026-08-13 12:33:25 +02:00
50af9bb905
* Don't check for existing sent offer when getting requested sent offer. * Check if received offer is already saved before requesting from peer.
323 lines
13 KiB
Python
323 lines
13 KiB
Python
# MIT License
|
|
#
|
|
# Copyright (c) 2020 Jonathan Zernik
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
# of this software and associated documentation files (the "Software"), to deal
|
|
# in the Software without restriction, including without limitation the rights
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
# furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in all
|
|
# copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
|
# SOFTWARE.
|
|
import mock
|
|
import pytest
|
|
|
|
from squeaknode.core.lightning_address import LightningAddressHostPort
|
|
from squeaknode.core.squeak_core import SqueakCore
|
|
from squeaknode.db.squeak_db import SqueakDb
|
|
from squeaknode.node.squeak_store import SqueakStore
|
|
|
|
|
|
@pytest.fixture
|
|
def squeak_db():
|
|
return mock.Mock(spec=SqueakDb)
|
|
|
|
|
|
@pytest.fixture
|
|
def squeak_core():
|
|
return mock.Mock(spec=SqueakCore)
|
|
|
|
|
|
@pytest.fixture
|
|
def lightning_host_port():
|
|
return LightningAddressHostPort(host="my_lightning_host", port=8765)
|
|
|
|
|
|
@pytest.fixture
|
|
def price_msat():
|
|
return 777
|
|
|
|
|
|
@pytest.fixture
|
|
def max_squeaks():
|
|
return 55
|
|
|
|
|
|
@pytest.fixture
|
|
def max_squeaks_per_public_key_per_block():
|
|
return 10
|
|
|
|
|
|
@pytest.fixture
|
|
def squeak_retention_s():
|
|
return 360000
|
|
|
|
|
|
@pytest.fixture
|
|
def received_offer_retention_s():
|
|
return 3600
|
|
|
|
|
|
@pytest.fixture
|
|
def sent_offer_retention_s():
|
|
return 7200
|
|
|
|
|
|
@pytest.fixture
|
|
def inserted_signing_profile_id(squeak_db, signing_profile):
|
|
yield squeak_db.insert_profile(signing_profile)
|
|
|
|
|
|
@pytest.fixture
|
|
def squeak_store(
|
|
squeak_db,
|
|
squeak_core,
|
|
max_squeaks,
|
|
max_squeaks_per_public_key_per_block,
|
|
squeak_retention_s,
|
|
received_offer_retention_s,
|
|
sent_offer_retention_s,
|
|
):
|
|
return SqueakStore(
|
|
squeak_db,
|
|
squeak_core,
|
|
max_squeaks,
|
|
max_squeaks_per_public_key_per_block,
|
|
squeak_retention_s,
|
|
received_offer_retention_s,
|
|
sent_offer_retention_s,
|
|
)
|
|
|
|
|
|
def test_save_squeak(squeak_store, squeak_db, squeak_core, block_header, squeak, squeak_hash):
|
|
with mock.patch.object(squeak_db, 'get_number_of_squeaks', autospec=True) as mock_get_number_of_squeaks, \
|
|
mock.patch.object(squeak_db, 'number_of_squeaks_with_public_key_with_block_height', autospec=True) as mock_number_of_squeaks_with_public_key_with_block_height, \
|
|
mock.patch.object(squeak_db, 'insert_squeak', autospec=True) as mock_insert_squeak, \
|
|
mock.patch.object(squeak_store.new_squeak_listener, 'handle_new_item', autospec=True) as mock_handle_new_squeak, \
|
|
mock.patch.object(squeak_core, 'get_block_header', autospec=True) as mock_get_block_header:
|
|
mock_get_number_of_squeaks.return_value = 0
|
|
mock_number_of_squeaks_with_public_key_with_block_height.return_value = 0
|
|
mock_get_block_header.return_value = block_header
|
|
mock_insert_squeak.return_value = squeak_hash
|
|
squeak_store.save_squeak(squeak)
|
|
|
|
mock_insert_squeak.assert_called_once_with(squeak, block_header)
|
|
mock_handle_new_squeak.assert_called_once_with(squeak)
|
|
|
|
|
|
def test_save_squeak_above_max(squeak_store, squeak_db, squeak_core, block_header, squeak, squeak_hash, max_squeaks):
|
|
with mock.patch.object(squeak_db, 'get_number_of_squeaks', autospec=True) as mock_get_number_of_squeaks, \
|
|
mock.patch.object(squeak_db, 'number_of_squeaks_with_public_key_with_block_height', autospec=True) as mock_number_of_squeaks_with_public_key_with_block_height, \
|
|
mock.patch.object(squeak_db, 'insert_squeak', autospec=True) as mock_insert_squeak, \
|
|
mock.patch.object(squeak_store.new_squeak_listener, 'handle_new_item', autospec=True) as mock_handle_new_squeak, \
|
|
mock.patch.object(squeak_core, 'get_block_header', autospec=True) as mock_get_block_header:
|
|
mock_get_number_of_squeaks.return_value = max_squeaks + 1
|
|
mock_number_of_squeaks_with_public_key_with_block_height.return_value = 0
|
|
mock_get_block_header.return_value = block_header
|
|
mock_insert_squeak.return_value = squeak_hash
|
|
|
|
with pytest.raises(Exception):
|
|
squeak_store.save_squeak(squeak)
|
|
|
|
assert mock_insert_squeak.call_count == 0
|
|
assert mock_handle_new_squeak.call_count == 0
|
|
|
|
|
|
def test_save_squeak_above_max_per_pubkey(squeak_store, squeak_db, squeak_core, block_header, squeak, squeak_hash, max_squeaks_per_public_key_per_block):
|
|
with mock.patch.object(squeak_db, 'get_number_of_squeaks', autospec=True) as mock_get_number_of_squeaks, \
|
|
mock.patch.object(squeak_db, 'number_of_squeaks_with_public_key_with_block_height', autospec=True) as mock_number_of_squeaks_with_public_key_with_block_height, \
|
|
mock.patch.object(squeak_db, 'insert_squeak', autospec=True) as mock_insert_squeak, \
|
|
mock.patch.object(squeak_store.new_squeak_listener, 'handle_new_item', autospec=True) as mock_handle_new_squeak, \
|
|
mock.patch.object(squeak_core, 'get_block_header', autospec=True) as mock_get_block_header:
|
|
mock_get_number_of_squeaks.return_value = 0
|
|
mock_number_of_squeaks_with_public_key_with_block_height.return_value = max_squeaks_per_public_key_per_block + 1
|
|
mock_get_block_header.return_value = block_header
|
|
mock_insert_squeak.return_value = squeak_hash
|
|
|
|
with pytest.raises(Exception):
|
|
squeak_store.save_squeak(squeak)
|
|
|
|
assert mock_insert_squeak.call_count == 0
|
|
assert mock_handle_new_squeak.call_count == 0
|
|
|
|
|
|
def test_save_secret_key(squeak_store, squeak_db, squeak_core, squeak, squeak_hash, secret_key):
|
|
with mock.patch.object(squeak_db, 'get_squeak', autospec=True) as mock_get_squeak, \
|
|
mock.patch.object(squeak_db, 'set_squeak_secret_key', autospec=True) as mock_set_squeak_secret_key, \
|
|
mock.patch.object(squeak_store, 'unlock_squeak', autospec=True) as mock_unlock_squeak, \
|
|
mock.patch.object(squeak_store.new_secret_key_listener, 'handle_new_item', autospec=True) as mock_handle_new_secret_key:
|
|
mock_get_squeak.return_value = squeak
|
|
squeak_store.save_secret_key(squeak_hash, secret_key)
|
|
|
|
mock_set_squeak_secret_key.assert_called_once_with(
|
|
squeak_hash, secret_key)
|
|
mock_handle_new_secret_key.assert_called_once_with(squeak)
|
|
mock_unlock_squeak.assert_called_once_with(squeak_hash)
|
|
|
|
|
|
def test_unlock_squeak(squeak_store, squeak_db, squeak_core, squeak, squeak_hash, secret_key, squeak_content):
|
|
with mock.patch.object(squeak_db, 'get_squeak', autospec=True) as mock_get_squeak, \
|
|
mock.patch.object(squeak_db, 'get_squeak_secret_key', autospec=True) as mock_get_squeak_secret_key, \
|
|
mock.patch.object(squeak_db, 'set_squeak_decrypted_content', autospec=True) as mock_set_squeak_decrypted_content, \
|
|
mock.patch.object(squeak_core, 'get_decrypted_content', autospec=True) as mock_get_decrypted_content:
|
|
mock_get_squeak.return_value = squeak
|
|
mock_get_squeak_secret_key.return_value = secret_key
|
|
mock_get_decrypted_content.return_value = squeak_content
|
|
squeak_store.unlock_squeak(squeak_hash)
|
|
|
|
mock_set_squeak_decrypted_content.assert_called_once_with(
|
|
squeak_hash, squeak_content)
|
|
|
|
|
|
# @pytest.fixture
|
|
# def unlocked_squeak(squeak_store, saved_squeak, secret_key, squeak_content):
|
|
# saved_squeak_hash = get_hash(saved_squeak)
|
|
# squeak_store.unlock_squeak(saved_squeak_hash, secret_key, squeak_content)
|
|
# yield saved_squeak
|
|
|
|
|
|
# @pytest.fixture
|
|
# def deleted_squeak(squeak_store, unlocked_squeak):
|
|
# unlocked_squeak_hash = get_hash(unlocked_squeak)
|
|
# squeak_store.delete_squeak(unlocked_squeak_hash)
|
|
# yield unlocked_squeak
|
|
|
|
|
|
# def test_get_squeak(squeak_store, saved_squeak):
|
|
# saved_squeak_hash = get_hash(saved_squeak)
|
|
# squeak_entry = squeak_store.get_squeak_entry(saved_squeak_hash)
|
|
|
|
# assert saved_squeak == squeak_store.get_squeak(saved_squeak_hash)
|
|
# assert squeak_entry.content is None
|
|
# assert squeak_store.get_squeak_secret_key(saved_squeak_hash) is None
|
|
|
|
|
|
# def test_unlock_squeak(squeak_store, unlocked_squeak, squeak_content, secret_key):
|
|
# unlocked_squeak_hash = get_hash(unlocked_squeak)
|
|
# squeak_entry = squeak_store.get_squeak_entry(unlocked_squeak_hash)
|
|
|
|
# assert unlocked_squeak == squeak_store.get_squeak(unlocked_squeak_hash)
|
|
# assert squeak_entry.content == squeak_content
|
|
# assert squeak_store.get_squeak_secret_key(
|
|
# unlocked_squeak_hash) == secret_key
|
|
|
|
|
|
# def test_delete_squeak(squeak_store, deleted_squeak):
|
|
# deleted_squeak_hash = get_hash(deleted_squeak)
|
|
# squeak_entry = squeak_store.get_squeak_entry(deleted_squeak_hash)
|
|
|
|
# assert squeak_store.get_squeak(deleted_squeak_hash) is None
|
|
# assert squeak_entry is None
|
|
# assert squeak_store.get_squeak_secret_key(deleted_squeak_hash) is None
|
|
|
|
|
|
def test_get_received_offer(squeak_store, squeak_db, received_offer):
|
|
with mock.patch.object(
|
|
squeak_db,
|
|
'get_received_offer',
|
|
new_callable=mock.PropertyMock,
|
|
) as mock_get_received_offer:
|
|
mock_get_received_offer.return_value = received_offer
|
|
|
|
retrieved_received_offer = squeak_store.get_received_offer(789)
|
|
|
|
assert retrieved_received_offer == received_offer
|
|
mock_get_received_offer.assert_called_once_with(789)
|
|
|
|
|
|
# def test_get_free_secret_key(squeak_store, squeak_core, unlocked_squeak, secret_key, peer_address):
|
|
# unlocked_squeak_hash = get_hash(unlocked_squeak)
|
|
# secret_key_reply = squeak_store.get_secret_key_reply(
|
|
# unlocked_squeak_hash, peer_address, 0, None)
|
|
|
|
# assert secret_key_reply.squeak_hash == unlocked_squeak_hash
|
|
# assert secret_key_reply.secret_key == secret_key
|
|
|
|
|
|
# def test_get_offer_secret_key(squeak_store, squeak_core, unlocked_squeak, secret_key, peer_address, sent_offer, offer):
|
|
# with mock.patch.object(squeak_core, 'create_offer', autospec=True) as mock_create_offer, \
|
|
# mock.patch.object(squeak_core, 'package_offer', autospec=True) as mock_package_offer:
|
|
# mock_create_offer.return_value = sent_offer
|
|
# mock_package_offer.return_value = offer
|
|
# unlocked_squeak_hash = get_hash(unlocked_squeak)
|
|
# secret_key_reply = squeak_store.get_secret_key_reply(
|
|
# unlocked_squeak_hash, peer_address, 1000, None)
|
|
|
|
# assert secret_key_reply.squeak_hash == unlocked_squeak_hash
|
|
# assert secret_key_reply.offer == offer
|
|
|
|
|
|
# def test_pay_offer(
|
|
# squeak_store,
|
|
# squeak_db,
|
|
# squeak_core,
|
|
# unlocked_squeak,
|
|
# block_header,
|
|
# squeak_content,
|
|
# secret_key,
|
|
# peer_address,
|
|
# inserted_received_offer_id,
|
|
# sent_payment,
|
|
# ):
|
|
# with mock.patch.object(squeak_core, 'pay_offer', autospec=True) as mock_pay_offer, \
|
|
# mock.patch.object(squeak_core, 'get_block_header', autospec=True) as mock_get_block_header, \
|
|
# mock.patch.object(squeak_core, 'get_decrypted_content', autospec=True) as mock_get_decrypted_content:
|
|
# mock_pay_offer.return_value = sent_payment
|
|
# mock_get_block_header.return_value = block_header
|
|
# mock_get_decrypted_content.return_value = squeak_content
|
|
# sent_payment_id = squeak_store.pay_offer(inserted_received_offer_id)
|
|
|
|
# retrieved_sent_payment = squeak_db.get_sent_payment(
|
|
# sent_payment_id,
|
|
# )
|
|
|
|
# assert sent_payment_id is not None
|
|
# assert retrieved_sent_payment is not None
|
|
|
|
|
|
# def test_save_received_offer_already_unlocked(
|
|
# squeak_store,
|
|
# unlocked_squeak,
|
|
# offer,
|
|
# peer_address,
|
|
# ):
|
|
# received_offer_id = squeak_store.save_received_offer(
|
|
# offer,
|
|
# peer_address,
|
|
# )
|
|
|
|
# assert received_offer_id is None
|
|
|
|
|
|
# def test_save_received_offer(
|
|
# squeak_store,
|
|
# squeak_db,
|
|
# squeak_core,
|
|
# saved_squeak,
|
|
# offer,
|
|
# received_offer,
|
|
# peer_address,
|
|
# ):
|
|
# with mock.patch.object(squeak_core, 'unpack_offer', autospec=True) as mock_unpack_offer:
|
|
# mock_unpack_offer.return_value = received_offer
|
|
# received_offer_id = squeak_store.save_received_offer(
|
|
# offer,
|
|
# peer_address,
|
|
# )
|
|
|
|
# assert received_offer_id is not None
|
|
# retrieved_received_offer = squeak_db.get_received_offer(received_offer_id)
|
|
|
|
# assert retrieved_received_offer == received_offer._replace(
|
|
# received_offer_id=received_offer_id,
|
|
# )
|