Onionr/tests/gossip-unittests/test_peer_exchange.py

79 lines
1.8 KiB
Python
Raw Permalink Normal View History

2022-04-23 05:41:57 +00:00
import os, uuid
from unittest.mock import Mock
from time import sleep
import secrets
TEST_DIR = 'testdata/%s-%s' % (secrets.token_hex(2), os.path.basename(__file__)) + '/'
2022-04-23 05:41:57 +00:00
print("Test directory:", TEST_DIR)
os.environ["ONIONR_HOME"] = TEST_DIR
from threading import Thread
import asyncio
import unittest
import sys
sys.path.append(".")
sys.path.append("src/")
from unittest.mock import patch
from ordered_set import OrderedSet
from gossip import peerset
import onionrblocks
import blockdb
from gossip.server import gossip_server
from gossip.peer import Peer
from filepaths import gossip_server_socket_file
BLOCK_MAX_SIZE = 1024 * 2000
2022-06-26 05:34:49 +00:00
BLOCK_SIZE_LEN = len(str(BLOCK_MAX_SIZE))
2022-04-23 05:41:57 +00:00
BLOCK_ID_SIZE = 128
BLOCK_STREAM_OFFSET_DIGITS = 8
MAX_PEERS = 10
TRANSPORT_SIZE_BYTES = 64
class MockPeer(Peer):
def __init__(self):
self.transport_address = secrets.token_hex(16)
def __hash__(self):
return hash(self.transport_address)
class OnionrServerPeerExchange(unittest.TestCase):
def test_peer_exchange(self):
Thread(target=gossip_server, daemon=True).start()
peers = []
for _ in range(10):
p = MockPeer()
peers.append(p)
peerset.gossip_peer_set.add(p)
sleep(0.03)
async def exchange_client():
reader, writer = await asyncio.open_unix_connection(
gossip_server_socket_file)
writer.write(int(3).to_bytes(1, 'big'))
await writer.drain()
for i in range(MAX_PEERS):
peer = await reader.readline()
for p in peers:
if p.transport_address == peer.decode('utf-8').strip():
break
else:
raise AssertionError("Peer received not in list " + peer.decode('utf-8'))
asyncio.run(exchange_client())
unittest.main()