mirror of
https://github.com/yzernik/squeaknode.git
synced 2026-08-13 12:33:25 +02:00
2cabb8197a
* Use only named volumes in docker-compose * Try fixing itest with new docker compose * Update development doc
927 lines
31 KiB
Python
927 lines
31 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.
|
|
from __future__ import print_function
|
|
|
|
import datetime
|
|
import time
|
|
|
|
import pytest
|
|
from squeak.core import CheckSqueak
|
|
from squeak.core import CSqueak
|
|
|
|
from proto import lnd_pb2 as ln
|
|
from proto import squeak_admin_pb2
|
|
from tests.util import channel
|
|
from tests.util import clear_sell_price
|
|
from tests.util import create_contact_profile
|
|
from tests.util import create_saved_peer
|
|
from tests.util import create_signing_profile
|
|
from tests.util import delete_profile
|
|
from tests.util import delete_squeak
|
|
from tests.util import download_squeak
|
|
from tests.util import download_squeak_secret_key
|
|
from tests.util import free_price
|
|
from tests.util import get_balance
|
|
from tests.util import get_external_address
|
|
from tests.util import get_hash
|
|
from tests.util import get_network
|
|
from tests.util import get_peer_by_address
|
|
from tests.util import get_profile_by_pubkey
|
|
from tests.util import get_pubkey_squeak_displays
|
|
from tests.util import get_search_squeaks
|
|
from tests.util import get_sell_price
|
|
from tests.util import get_squeak_display
|
|
from tests.util import get_squeak_profile
|
|
from tests.util import import_signing_profile
|
|
from tests.util import make_squeak
|
|
from tests.util import peer_connection
|
|
from tests.util import send_coins
|
|
from tests.util import set_sell_price
|
|
from tests.util import subscribe_squeak_ancestor_entries
|
|
from tests.util import subscribe_squeak_entry
|
|
|
|
|
|
def test_get_network(admin_stub):
|
|
# Get the network
|
|
network = get_network(admin_stub)
|
|
|
|
assert network == "simnet"
|
|
|
|
|
|
def test_get_sell_price(admin_stub):
|
|
# Get the sell price
|
|
price = get_sell_price(admin_stub)
|
|
|
|
assert price.price_msat == 1000000
|
|
|
|
set_sell_price(admin_stub, 98765)
|
|
price = get_sell_price(admin_stub)
|
|
|
|
assert price.price_msat == 98765
|
|
|
|
clear_sell_price(admin_stub)
|
|
price = get_sell_price(admin_stub)
|
|
|
|
assert price.price_msat == 1000000
|
|
|
|
|
|
def test_get_external_address(admin_stub):
|
|
# Get the external address
|
|
external_address = get_external_address(admin_stub)
|
|
|
|
print(external_address)
|
|
# assert external_address.host is not None and len(external_address.host) > 0
|
|
assert external_address.host == 'myexternaladdress.com'
|
|
# assert external_address.port > 0
|
|
assert external_address.port == 8765
|
|
|
|
|
|
def test_reprocess_received_payments(admin_stub):
|
|
# Reprocess received payments
|
|
reprocess_received_payments_response = admin_stub.ReprocessReceivedPayments(
|
|
squeak_admin_pb2.ReprocessReceivedPaymentsRequest()
|
|
)
|
|
|
|
assert reprocess_received_payments_response is not None
|
|
|
|
|
|
def test_get_profile(admin_stub, signing_profile_id):
|
|
# Get the squeak profile
|
|
squeak_profile = get_squeak_profile(admin_stub, signing_profile_id)
|
|
# address = squeak_profile.address
|
|
pubkey = squeak_profile.pubkey
|
|
name = squeak_profile.profile_name
|
|
|
|
# Get the same squeak profile by address
|
|
# get_squeak_profile_by_pubkey_response = admin_stub.GetSqueakProfileByAddress(
|
|
# squeak_admin_pb2.GetSqueakProfileByAddressRequest(
|
|
# pubkey=pubkey,
|
|
# )
|
|
# )
|
|
profile = get_profile_by_pubkey(admin_stub, pubkey)
|
|
# assert address == get_squeak_profile_by_address_response.squeak_profile.address
|
|
assert pubkey == profile.pubkey
|
|
assert name == profile.profile_name
|
|
|
|
# Get the same squeak profile by name
|
|
get_squeak_profile_by_name_response = admin_stub.GetSqueakProfileByName(
|
|
squeak_admin_pb2.GetSqueakProfileByNameRequest(
|
|
name=name,
|
|
)
|
|
)
|
|
assert pubkey == get_squeak_profile_by_name_response.squeak_profile.pubkey
|
|
assert name == get_squeak_profile_by_name_response.squeak_profile.profile_name
|
|
|
|
|
|
def test_make_squeak(admin_stub, signing_profile_id):
|
|
# Create a new squeak using the new profile
|
|
make_squeak_content = "Hello from the profile on the server!"
|
|
make_squeak_hash = make_squeak(
|
|
admin_stub, signing_profile_id, make_squeak_content)
|
|
assert len(make_squeak_hash) == 32 * 2
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
admin_stub, make_squeak_hash)
|
|
assert get_squeak_display_entry.squeak_hash == make_squeak_hash
|
|
assert (
|
|
get_squeak_display_entry.content_str == "Hello from the profile on the server!"
|
|
)
|
|
# assert get_squeak_display_response.squeak_display_entry.author_address == signing_profile_address
|
|
assert get_squeak_display_entry.is_author_known
|
|
assert get_squeak_display_entry.HasField("author")
|
|
assert len(
|
|
get_squeak_display_entry.author.profile_image) > 0
|
|
assert not get_squeak_display_entry.is_reply
|
|
assert not bool(get_squeak_display_entry.reply_to)
|
|
assert len(get_squeak_display_entry.secret_key_hex) == 32 * 2
|
|
|
|
# Block time should be within the past hour
|
|
block_time = datetime.datetime.fromtimestamp(
|
|
get_squeak_display_entry.block_time,
|
|
)
|
|
one_hour = datetime.timedelta(hours=1)
|
|
assert block_time > datetime.datetime.now() - one_hour
|
|
assert block_time < datetime.datetime.now() + one_hour
|
|
|
|
# Get the squeak profile
|
|
squeak_profile = get_squeak_profile(admin_stub, signing_profile_id)
|
|
squeak_profile_pubkey = squeak_profile.pubkey
|
|
squeak_profile_name = squeak_profile.profile_name
|
|
|
|
# Get all squeak displays for the known address
|
|
# get_address_squeak_display_response = admin_stub.GetAddressSqueakDisplays(
|
|
# squeak_admin_pb2.GetAddressSqueakDisplaysRequest(
|
|
# pubkey=squeak_profile_pubkey,
|
|
# limit=100,
|
|
# ),
|
|
# )
|
|
address_squeak_displays = get_pubkey_squeak_displays(
|
|
admin_stub, squeak_profile_pubkey, 100)
|
|
|
|
assert len(address_squeak_displays) == 1
|
|
for (
|
|
squeak_display_entry
|
|
) in address_squeak_displays:
|
|
assert squeak_display_entry.author.profile_name == squeak_profile_name
|
|
assert squeak_display_entry.author.pubkey == squeak_profile_pubkey
|
|
|
|
# check serialized squeak hex string
|
|
serialized_squeak_hex = get_squeak_display_entry.serialized_squeak_hex
|
|
# print("serialized_squeak_hex: {}".format(serialized_squeak_hex))
|
|
assert len(serialized_squeak_hex) > 200
|
|
serialized_squeak = bytes.fromhex(serialized_squeak_hex)
|
|
deserialized_squeak = CSqueak.deserialize(serialized_squeak)
|
|
assert get_hash(deserialized_squeak) == make_squeak_hash
|
|
CheckSqueak(deserialized_squeak)
|
|
|
|
|
|
def test_make_reply_squeak(
|
|
admin_stub, saved_squeak_hash, signing_profile_id
|
|
):
|
|
# Make another squeak as a reply
|
|
reply_1_squeak_hash = make_squeak(
|
|
admin_stub,
|
|
signing_profile_id,
|
|
"Reply #1",
|
|
saved_squeak_hash,
|
|
)
|
|
|
|
# Make a second squeak as a reply
|
|
reply_2_squeak_hash = make_squeak(
|
|
admin_stub,
|
|
signing_profile_id,
|
|
"Reply #2",
|
|
reply_1_squeak_hash,
|
|
)
|
|
|
|
# Get the squeak and check that the reply field is correct
|
|
get_reply_squeak_display_entry = get_squeak_display(
|
|
admin_stub, reply_2_squeak_hash)
|
|
assert (
|
|
get_reply_squeak_display_entry.squeak_hash == reply_2_squeak_hash
|
|
)
|
|
assert (
|
|
get_reply_squeak_display_entry.reply_to == reply_1_squeak_hash
|
|
)
|
|
|
|
# Get the ancestors of the latest reply squeak
|
|
get_ancestors_response = admin_stub.GetAncestorSqueakDisplays(
|
|
squeak_admin_pb2.GetAncestorSqueakDisplaysRequest(
|
|
squeak_hash=reply_2_squeak_hash,
|
|
)
|
|
)
|
|
assert len(get_ancestors_response.squeak_display_entries) == 3
|
|
|
|
# Get the replies of the original squeak
|
|
get_replies_response = admin_stub.GetReplySqueakDisplays(
|
|
squeak_admin_pb2.GetReplySqueakDisplaysRequest(
|
|
squeak_hash=saved_squeak_hash,
|
|
limit=100,
|
|
)
|
|
)
|
|
assert len(get_replies_response.squeak_display_entries) == 1
|
|
assert reply_1_squeak_hash in [
|
|
entry.squeak_hash
|
|
for entry in get_replies_response.squeak_display_entries
|
|
]
|
|
|
|
|
|
def test_make_private_squeak(admin_stub, signing_profile_id, recipient_contact_profile_id):
|
|
make_squeak_content = "This is a private squeak!"
|
|
make_squeak_hash = make_squeak(
|
|
admin_stub,
|
|
signing_profile_id,
|
|
make_squeak_content,
|
|
recipient_profile_id=recipient_contact_profile_id,
|
|
)
|
|
assert len(make_squeak_hash) == 32 * 2
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
admin_stub, make_squeak_hash)
|
|
assert get_squeak_display_entry.squeak_hash == make_squeak_hash
|
|
assert (
|
|
get_squeak_display_entry.content_str == "This is a private squeak!"
|
|
)
|
|
assert get_squeak_display_entry.is_author_known
|
|
assert get_squeak_display_entry.HasField("author")
|
|
assert len(
|
|
get_squeak_display_entry.author.profile_image) > 0
|
|
assert not get_squeak_display_entry.is_reply
|
|
assert not bool(get_squeak_display_entry.reply_to)
|
|
assert len(get_squeak_display_entry.secret_key_hex) == 32 * 2
|
|
|
|
|
|
def test_make_signing_profile(admin_stub):
|
|
# Create a new signing profile
|
|
profile_name = "test_signing_profile_name"
|
|
profile_id = create_signing_profile(admin_stub, profile_name)
|
|
|
|
# Get the new squeak profile
|
|
squeak_profile = get_squeak_profile(admin_stub, profile_id)
|
|
assert squeak_profile.profile_name == profile_name
|
|
squeak_profile_pubkey = squeak_profile.pubkey
|
|
|
|
# Get all signing profiles
|
|
get_signing_profiles_response = admin_stub.GetSigningProfiles(
|
|
squeak_admin_pb2.GetSigningProfilesRequest()
|
|
)
|
|
signing_profile_names = [
|
|
profile.profile_name
|
|
for profile in get_signing_profiles_response.squeak_profiles
|
|
]
|
|
assert profile_name in signing_profile_names
|
|
|
|
# Get squeak profile by address
|
|
# get_profile_by_address_response = admin_stub.GetSqueakProfileByAddress(
|
|
# squeak_admin_pb2.GetSqueakProfileByAddressRequest(
|
|
# pubkey=squeak_profile_pubkey
|
|
# )
|
|
# )
|
|
profile = get_profile_by_pubkey(admin_stub, squeak_profile_pubkey)
|
|
assert profile.profile_name == profile_name
|
|
|
|
# Export the private key, delete the profile, and re-import it.
|
|
get_private_key_response = admin_stub.GetSqueakProfilePrivateKey(
|
|
squeak_admin_pb2.GetSqueakProfilePrivateKeyRequest(
|
|
profile_id=profile_id,
|
|
)
|
|
)
|
|
private_key = get_private_key_response.private_key
|
|
|
|
delete_profile(admin_stub, profile_id)
|
|
new_profile_id = import_signing_profile(
|
|
admin_stub,
|
|
"imported_profile_name",
|
|
private_key,
|
|
)
|
|
|
|
# Get the new imported profile
|
|
squeak_profile = get_squeak_profile(admin_stub, new_profile_id)
|
|
assert squeak_profile.profile_name == "imported_profile_name"
|
|
assert squeak_profile.pubkey == squeak_profile_pubkey
|
|
|
|
|
|
def test_make_contact_profile(admin_stub, public_key):
|
|
# Create a new contact profile
|
|
contact_name = "test_contact_profile_name"
|
|
contact_profile_id = create_contact_profile(
|
|
admin_stub, contact_name, public_key)
|
|
|
|
# Get all contact profiles
|
|
get_contact_profiles_response = admin_stub.GetContactProfiles(
|
|
squeak_admin_pb2.GetContactProfilesRequest()
|
|
)
|
|
contact_profile_names = [
|
|
profile.profile_name
|
|
for profile in get_contact_profiles_response.squeak_profiles
|
|
]
|
|
contact_profile_ids = [
|
|
profile.profile_id
|
|
for profile in get_contact_profiles_response.squeak_profiles
|
|
]
|
|
assert contact_name in contact_profile_names
|
|
assert contact_profile_id in contact_profile_ids
|
|
|
|
|
|
def test_make_signing_profile_empty_name(admin_stub):
|
|
# Try to create a new signing profile with an empty name
|
|
with pytest.raises(Exception) as excinfo:
|
|
create_signing_profile(admin_stub, "")
|
|
assert "Profile name cannot be empty." in str(excinfo.value)
|
|
|
|
|
|
def test_make_contact_profile_empty_name(admin_stub, public_key):
|
|
# Try to create a new contact profile with an empty name
|
|
with pytest.raises(Exception) as excinfo:
|
|
create_contact_profile(admin_stub, "", public_key)
|
|
assert "Profile name cannot be empty." in str(excinfo.value)
|
|
|
|
|
|
def test_set_profile_following(admin_stub, contact_profile_id):
|
|
# Set the profile to be following
|
|
admin_stub.SetSqueakProfileFollowing(
|
|
squeak_admin_pb2.SetSqueakProfileFollowingRequest(
|
|
profile_id=contact_profile_id,
|
|
following=True,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile again
|
|
squeak_profile = get_squeak_profile(admin_stub, contact_profile_id)
|
|
assert squeak_profile.following
|
|
|
|
# Set the profile to be not following
|
|
admin_stub.SetSqueakProfileFollowing(
|
|
squeak_admin_pb2.SetSqueakProfileFollowingRequest(
|
|
profile_id=contact_profile_id,
|
|
following=False,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile again
|
|
squeak_profile = get_squeak_profile(admin_stub, contact_profile_id)
|
|
assert not squeak_profile.following
|
|
|
|
|
|
def test_rename_profile(admin_stub, contact_profile_id, random_name):
|
|
# Rename the profile to something new
|
|
admin_stub.RenameSqueakProfile(
|
|
squeak_admin_pb2.RenameSqueakProfileRequest(
|
|
profile_id=contact_profile_id,
|
|
profile_name=random_name,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile
|
|
squeak_profile = get_squeak_profile(admin_stub, contact_profile_id)
|
|
assert squeak_profile.profile_name == random_name
|
|
|
|
|
|
def test_set_profile_image(admin_stub, contact_profile_id, random_image, random_image_base64_string):
|
|
# print("random_image: {}".format(random_image))
|
|
# print("random_image_base64_string: {}".format(random_image_base64_string))
|
|
# Set the profile image to something new
|
|
admin_stub.SetSqueakProfileImage(
|
|
squeak_admin_pb2.SetSqueakProfileImageRequest(
|
|
profile_id=contact_profile_id,
|
|
profile_image=random_image_base64_string,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile
|
|
squeak_profile = get_squeak_profile(admin_stub, contact_profile_id)
|
|
assert squeak_profile.profile_image == random_image_base64_string
|
|
assert squeak_profile.has_custom_profile_image
|
|
|
|
# Clear the profile image
|
|
admin_stub.ClearSqueakProfileImage(
|
|
squeak_admin_pb2.ClearSqueakProfileImageRequest(
|
|
profile_id=contact_profile_id,
|
|
)
|
|
)
|
|
|
|
# Get the squeak profile
|
|
squeak_profile = get_squeak_profile(admin_stub, contact_profile_id)
|
|
assert squeak_profile.profile_image != random_image_base64_string
|
|
assert not squeak_profile.has_custom_profile_image
|
|
|
|
|
|
def test_delete_profile(admin_stub, random_name, public_key, contact_profile_id):
|
|
# Delete the profile
|
|
delete_profile(admin_stub, contact_profile_id)
|
|
|
|
# Try to get the profile and fail
|
|
squeak_profile = get_squeak_profile(admin_stub, contact_profile_id)
|
|
assert squeak_profile is None
|
|
|
|
get_squeak_profile_by_name_response = admin_stub.GetSqueakProfileByName(
|
|
squeak_admin_pb2.GetSqueakProfileByNameRequest(
|
|
name=random_name,
|
|
)
|
|
)
|
|
assert not get_squeak_profile_by_name_response.HasField("squeak_profile")
|
|
|
|
# get_squeak_profile_by_pubkey_response = admin_stub.GetSqueakProfileByAddress(
|
|
# squeak_admin_pb2.GetSqueakProfileByAddressRequest(
|
|
# pubkey=public_key.to_bytes().hex(),
|
|
# )
|
|
# )
|
|
profile = get_profile_by_pubkey(admin_stub, public_key.to_bytes().hex())
|
|
# assert not get_squeak_profile_by_pubkey_response.HasField(
|
|
# "squeak_profile")
|
|
assert profile is None
|
|
|
|
|
|
def test_get_profile_private_key(admin_stub, signing_profile_id):
|
|
# Get the private key
|
|
private_key_response = admin_stub.GetSqueakProfilePrivateKey(
|
|
squeak_admin_pb2.GetSqueakProfilePrivateKeyRequest(
|
|
profile_id=signing_profile_id,
|
|
)
|
|
)
|
|
|
|
# print(private_key_response.private_key)
|
|
assert len(private_key_response.private_key) > 0
|
|
|
|
|
|
def test_get_following_squeaks(
|
|
admin_stub, saved_squeak_hash, signing_profile_id
|
|
):
|
|
# Set the profile to be following
|
|
admin_stub.SetSqueakProfileFollowing(
|
|
squeak_admin_pb2.SetSqueakProfileFollowingRequest(
|
|
profile_id=signing_profile_id,
|
|
following=True,
|
|
)
|
|
)
|
|
|
|
# Get all squeak displays for the known address
|
|
get_timeline_squeak_display_response = admin_stub.GetTimelineSqueakDisplays(
|
|
squeak_admin_pb2.GetTimelineSqueakDisplaysRequest(
|
|
limit=100,
|
|
)
|
|
)
|
|
assert len(get_timeline_squeak_display_response.squeak_display_entries) >= 1
|
|
for (
|
|
squeak_display_entry
|
|
) in get_timeline_squeak_display_response.squeak_display_entries:
|
|
# TODO: check the profile id of the squeak display entry
|
|
# assert squeak_display_entry.profile_id == signing_profile_id
|
|
pass
|
|
|
|
|
|
def test_delete_squeak(admin_stub, saved_squeak_hash):
|
|
# Delete the squeak
|
|
delete_squeak(admin_stub, saved_squeak_hash)
|
|
# Try to get the squeak display item
|
|
squeak_display_entry = get_squeak_display(
|
|
admin_stub, saved_squeak_hash)
|
|
|
|
assert squeak_display_entry is None
|
|
|
|
|
|
def test_create_peer(admin_stub):
|
|
# Add a new peer
|
|
peer_id = create_saved_peer(
|
|
admin_stub,
|
|
"fake_peer_name",
|
|
"fake_host",
|
|
1234,
|
|
)
|
|
|
|
# Get the new peer
|
|
get_peer_response = admin_stub.GetPeer(
|
|
squeak_admin_pb2.GetPeerRequest(
|
|
peer_id=peer_id,
|
|
)
|
|
)
|
|
assert get_peer_response.squeak_peer.peer_address.host == "fake_host"
|
|
assert get_peer_response.squeak_peer.peer_address.port == 1234
|
|
|
|
# Get all peers
|
|
get_peers_response = admin_stub.GetPeers(
|
|
squeak_admin_pb2.GetPeersRequest())
|
|
peer_hosts = [
|
|
squeak_peer.peer_address.host
|
|
for squeak_peer in get_peers_response.squeak_peers
|
|
]
|
|
assert "fake_host" in peer_hosts
|
|
|
|
# Get the new peer by address
|
|
get_peer_by_address_response = get_peer_by_address(
|
|
admin_stub,
|
|
host="fake_host",
|
|
port=1234,
|
|
)
|
|
assert get_peer_by_address_response.peer_name == "fake_peer_name"
|
|
assert get_peer_response.squeak_peer.autoconnect
|
|
|
|
|
|
def test_set_peer_autoconnect(admin_stub, peer_id):
|
|
# Set the peer to be autoconnect
|
|
admin_stub.SetPeerAutoconnect(
|
|
squeak_admin_pb2.SetPeerAutoconnectRequest(
|
|
peer_id=peer_id,
|
|
autoconnect=True,
|
|
)
|
|
)
|
|
|
|
# Get the peer again
|
|
get_peer_response = admin_stub.GetPeer(
|
|
squeak_admin_pb2.GetPeerRequest(
|
|
peer_id=peer_id,
|
|
)
|
|
)
|
|
assert get_peer_response.squeak_peer.autoconnect
|
|
|
|
# Set the peer to be not autoconnect
|
|
admin_stub.SetPeerAutoconnect(
|
|
squeak_admin_pb2.SetPeerAutoconnectRequest(
|
|
peer_id=peer_id,
|
|
autoconnect=False,
|
|
)
|
|
)
|
|
|
|
# Get the peer again
|
|
get_peer_response = admin_stub.GetPeer(
|
|
squeak_admin_pb2.GetPeerRequest(
|
|
peer_id=peer_id,
|
|
)
|
|
)
|
|
assert not get_peer_response.squeak_peer.autoconnect
|
|
|
|
|
|
def test_rename_peer(admin_stub, peer_id, random_name):
|
|
# Rename the peer
|
|
admin_stub.RenamePeer(
|
|
squeak_admin_pb2.RenamePeerRequest(
|
|
peer_id=peer_id,
|
|
peer_name=random_name,
|
|
)
|
|
)
|
|
|
|
# Get the peer again
|
|
get_peer_response = admin_stub.GetPeer(
|
|
squeak_admin_pb2.GetPeerRequest(
|
|
peer_id=peer_id,
|
|
)
|
|
)
|
|
assert get_peer_response.squeak_peer.peer_name == random_name
|
|
|
|
|
|
def test_delete_peer(admin_stub, peer_id):
|
|
# Delete the peer
|
|
admin_stub.DeletePeer(
|
|
squeak_admin_pb2.DeletePeerRequest(
|
|
peer_id=peer_id,
|
|
)
|
|
)
|
|
|
|
get_peer_response = admin_stub.GetPeer(
|
|
squeak_admin_pb2.GetPeerRequest(
|
|
peer_id=peer_id,
|
|
)
|
|
)
|
|
assert not get_peer_response.HasField("squeak_peer")
|
|
|
|
|
|
def test_send_coins(admin_stub, other_admin_stub):
|
|
new_address_response = admin_stub.LndNewAddress(ln.NewAddressRequest())
|
|
get_balance_response = get_balance(
|
|
other_admin_stub,
|
|
)
|
|
print("Get balance: {}".format(get_balance_response))
|
|
print("--------------------")
|
|
print("--------------------")
|
|
print("--------------------")
|
|
print("--------------------")
|
|
print("--------------------")
|
|
print("--------------------")
|
|
print("--------------------")
|
|
print("--------------------", flush=True)
|
|
send_coins_response = send_coins(
|
|
other_admin_stub, new_address_response.address, 55555555
|
|
)
|
|
time.sleep(10)
|
|
get_transactions_response = admin_stub.LndGetTransactions(
|
|
ln.GetTransactionsRequest()
|
|
)
|
|
|
|
assert any(
|
|
[
|
|
transaction.tx_hash == send_coins_response.txid
|
|
for transaction in get_transactions_response.transactions
|
|
]
|
|
)
|
|
|
|
|
|
def test_buy_squeak(
|
|
admin_stub,
|
|
other_admin_stub,
|
|
signing_profile_id,
|
|
saved_squeak_hash,
|
|
admin_peer,
|
|
):
|
|
# Download squeak
|
|
download_squeak(other_admin_stub, saved_squeak_hash)
|
|
# Download secret key
|
|
download_squeak_secret_key(other_admin_stub, saved_squeak_hash)
|
|
|
|
# Get the sent offers from the seller node
|
|
get_sent_offers_response = admin_stub.GetSentOffers(
|
|
squeak_admin_pb2.GetSentOffersRequest(),
|
|
)
|
|
squeak_hashes = [
|
|
sent_offer.squeak_hash
|
|
for sent_offer in get_sent_offers_response.sent_offers
|
|
]
|
|
assert saved_squeak_hash in squeak_hashes
|
|
|
|
# Get the buy offer
|
|
get_buy_offers_response = other_admin_stub.GetBuyOffers(
|
|
squeak_admin_pb2.GetBuyOffersRequest(
|
|
squeak_hash=saved_squeak_hash,
|
|
)
|
|
)
|
|
# print(get_buy_offers_response)
|
|
assert len(get_buy_offers_response.offers) > 0
|
|
|
|
offer = get_buy_offers_response.offers[0]
|
|
|
|
print("Tring to connect LND peer with offer: {}".format(offer))
|
|
with peer_connection(
|
|
other_admin_stub, offer.node_host, offer.node_pubkey
|
|
), channel(other_admin_stub, offer.node_pubkey, 1000000):
|
|
|
|
print("Channel context manager opened.")
|
|
|
|
# Pay the offer
|
|
list_channels_response = other_admin_stub.LndListChannels(
|
|
ln.ListChannelsRequest(
|
|
active_only=True,
|
|
)
|
|
)
|
|
print("list_channels_response: {}".format(list_channels_response))
|
|
|
|
# Pay the offer
|
|
pay_offer_response = other_admin_stub.PayOffer(
|
|
squeak_admin_pb2.PayOfferRequest(
|
|
offer_id=offer.offer_id,
|
|
)
|
|
)
|
|
# print(pay_offer_response)
|
|
assert pay_offer_response.sent_payment_id > 0
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
other_admin_stub, saved_squeak_hash)
|
|
assert (
|
|
get_squeak_display_entry.content_str == "Hello from the profile on the server!"
|
|
)
|
|
|
|
# Get all sent payments
|
|
get_sent_payments_response = other_admin_stub.GetSentPayments(
|
|
squeak_admin_pb2.GetSentPaymentsRequest(
|
|
limit=10,
|
|
),
|
|
)
|
|
squeak_hashes = [
|
|
sent_payment.squeak_hash
|
|
for sent_payment in get_sent_payments_response.sent_payments
|
|
]
|
|
assert saved_squeak_hash in squeak_hashes
|
|
|
|
# Get the single sent payment
|
|
for sent_payment in get_sent_payments_response.sent_payments:
|
|
if sent_payment.squeak_hash == saved_squeak_hash:
|
|
sent_payment_id = sent_payment.sent_payment_id
|
|
get_sent_payment_response = other_admin_stub.GetSentPayment(
|
|
squeak_admin_pb2.GetSentPaymentRequest(
|
|
sent_payment_id=sent_payment_id,
|
|
),
|
|
)
|
|
assert saved_squeak_hash == get_sent_payment_response.sent_payment.squeak_hash
|
|
assert get_sent_payment_response.sent_payment.price_msat == 1000000
|
|
assert get_sent_payment_response.sent_payment.valid
|
|
|
|
# Get the received payment from the seller node
|
|
get_received_payments_response = admin_stub.GetReceivedPayments(
|
|
squeak_admin_pb2.GetReceivedPaymentsRequest(
|
|
limit=100,
|
|
),
|
|
)
|
|
# print(
|
|
# "get_received_payments_response: {}".format(
|
|
# get_received_payments_response)
|
|
# )
|
|
payment_hashes = [
|
|
received_payment.payment_hash
|
|
for received_payment in get_received_payments_response.received_payments
|
|
]
|
|
assert sent_payment.payment_hash in payment_hashes
|
|
for received_payment in get_received_payments_response.received_payments:
|
|
received_payment_time_ms = received_payment.time_ms
|
|
# print("received_payment_time_s: {}".format(
|
|
# received_payment_time_s))
|
|
received_payment_time = datetime.datetime.fromtimestamp(
|
|
received_payment_time_ms / 1000,
|
|
)
|
|
five_minutes = datetime.timedelta(minutes=5)
|
|
assert received_payment_time > datetime.datetime.now() - five_minutes
|
|
assert received_payment_time < datetime.datetime.now()
|
|
assert len(received_payment.peer_address.host) > 4
|
|
|
|
# Subscribe to received payments starting from index zero
|
|
subscribe_received_payments_response = admin_stub.SubscribeReceivedPayments(
|
|
squeak_admin_pb2.SubscribeReceivedPaymentsRequest(
|
|
payment_index=0,
|
|
),
|
|
)
|
|
for payment in subscribe_received_payments_response:
|
|
# print("Got payment from subscription: {}".format(payment))
|
|
assert payment.received_payment_id == 1
|
|
break
|
|
|
|
# Get the payment summary from the seller node
|
|
get_payment_summary_response = admin_stub.GetPaymentSummary(
|
|
squeak_admin_pb2.GetPaymentSummaryRequest(),
|
|
)
|
|
# print(
|
|
# "get_payment_summary_response from seller: {}".format(
|
|
# get_payment_summary_response)
|
|
# )
|
|
assert get_payment_summary_response.payment_summary.num_received_payments > 0
|
|
assert get_payment_summary_response.payment_summary.amount_earned_msat > 0
|
|
|
|
# Get the payment summary from the buyer node
|
|
get_payment_summary_response = other_admin_stub.GetPaymentSummary(
|
|
squeak_admin_pb2.GetPaymentSummaryRequest(),
|
|
)
|
|
# print(
|
|
# "get_payment_summary_response from buyer: {}".format(
|
|
# get_payment_summary_response)
|
|
# )
|
|
assert get_payment_summary_response.payment_summary.num_sent_payments > 0
|
|
assert get_payment_summary_response.payment_summary.amount_spent_msat > 0
|
|
|
|
print("Channel context manager closed.")
|
|
|
|
|
|
def test_download_free_squeak(
|
|
admin_stub,
|
|
other_admin_stub,
|
|
saved_squeak_hash,
|
|
admin_peer,
|
|
):
|
|
with free_price(admin_stub):
|
|
# Download squeak
|
|
download_result = download_squeak(other_admin_stub, saved_squeak_hash)
|
|
# Download secret key
|
|
download_squeak_secret_key(other_admin_stub, saved_squeak_hash)
|
|
print('download_result:')
|
|
print(download_result)
|
|
assert download_result.number_downloaded == 1
|
|
assert download_result.number_requested == 1
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
other_admin_stub, saved_squeak_hash)
|
|
assert (
|
|
get_squeak_display_entry.content_str == "Hello from the profile on the server!"
|
|
)
|
|
|
|
|
|
def test_download_single_squeak(
|
|
admin_stub,
|
|
other_admin_stub,
|
|
signing_profile_id,
|
|
saved_squeak_hash,
|
|
admin_peer,
|
|
):
|
|
|
|
with subscribe_squeak_entry(other_admin_stub, saved_squeak_hash) as subscription_queue, \
|
|
subscribe_squeak_ancestor_entries(other_admin_stub, saved_squeak_hash) as ancestor_subscription_queue:
|
|
|
|
# Get the squeak display item (should be empty)
|
|
squeak_display_entry = get_squeak_display(
|
|
other_admin_stub, saved_squeak_hash)
|
|
assert squeak_display_entry is None
|
|
|
|
# Get buy offers for the squeak hash (should be empty)
|
|
get_buy_offers_response = other_admin_stub.GetBuyOffers(
|
|
squeak_admin_pb2.GetBuyOffersRequest(
|
|
squeak_hash=saved_squeak_hash,
|
|
)
|
|
)
|
|
# print(get_buy_offers_response)
|
|
assert len(get_buy_offers_response.offers) == 0
|
|
|
|
# Download squeak
|
|
download_result = download_squeak(other_admin_stub, saved_squeak_hash)
|
|
# Download secret key
|
|
download_squeak_secret_key(other_admin_stub, saved_squeak_hash)
|
|
assert download_result.number_downloaded == 1
|
|
assert download_result.number_requested == 1
|
|
|
|
# Get the squeak display item
|
|
squeak_display_entry = get_squeak_display(
|
|
other_admin_stub, saved_squeak_hash)
|
|
assert squeak_display_entry is not None
|
|
|
|
# Get the buy offer
|
|
get_buy_offers_response = other_admin_stub.GetBuyOffers(
|
|
squeak_admin_pb2.GetBuyOffersRequest(
|
|
squeak_hash=saved_squeak_hash,
|
|
)
|
|
)
|
|
# print(get_buy_offers_response)
|
|
assert len(get_buy_offers_response.offers) > 0
|
|
|
|
item = subscription_queue.get()
|
|
print("subscription_queue item:")
|
|
print(item)
|
|
assert item.squeak_hash == saved_squeak_hash
|
|
|
|
item = ancestor_subscription_queue.get()
|
|
print("ancestor_subscription_queue item:")
|
|
print(item)
|
|
assert item[0].squeak_hash == saved_squeak_hash
|
|
|
|
|
|
def test_like_squeak(admin_stub, saved_squeak_hash):
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
admin_stub, saved_squeak_hash)
|
|
assert (
|
|
get_squeak_display_entry.liked_time_ms == 0
|
|
)
|
|
|
|
# Like the squeak
|
|
admin_stub.LikeSqueak(
|
|
squeak_admin_pb2.LikeSqueakRequest(
|
|
squeak_hash=saved_squeak_hash,
|
|
)
|
|
)
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
admin_stub, saved_squeak_hash)
|
|
assert (
|
|
get_squeak_display_entry.liked_time_ms > 0
|
|
)
|
|
|
|
# Unlike the squeak
|
|
admin_stub.UnlikeSqueak(
|
|
squeak_admin_pb2.UnlikeSqueakRequest(
|
|
squeak_hash=saved_squeak_hash,
|
|
)
|
|
)
|
|
|
|
# Get the squeak display item
|
|
get_squeak_display_entry = get_squeak_display(
|
|
admin_stub, saved_squeak_hash)
|
|
assert (
|
|
get_squeak_display_entry.liked_time_ms == 0
|
|
)
|
|
|
|
|
|
def test_search(admin_stub, signing_profile_id):
|
|
make_squeak_content = "Just some random weird text."
|
|
make_squeak_hash = make_squeak(
|
|
admin_stub, signing_profile_id, make_squeak_content)
|
|
assert len(make_squeak_hash) == 32 * 2
|
|
|
|
# Get all squeak displays for the given search text.
|
|
search_results = get_search_squeaks(
|
|
admin_stub,
|
|
"Weird",
|
|
)
|
|
assert len(search_results) == 1
|
|
|
|
# Get all squeak displays for other search text that shouldn't be there.
|
|
missing_search_results = get_search_squeaks(
|
|
admin_stub,
|
|
"strange",
|
|
)
|
|
assert len(missing_search_results) == 0
|