Onionr/tests/gossip-unittests/test_client_announce.py

122 lines
2.9 KiB
Python
Raw Permalink Normal View History

2022-05-19 04:39:14 +00:00
import os, uuid
from unittest.mock import Mock
from time import sleep
import secrets
import socket
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
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
from gossip.peerset import gossip_peer_set
from gossip.client.announce import do_announce
BLOCK_MAX_SIZE = 1024 * 2000
2022-06-26 05:34:49 +00:00
BLOCK_SIZE_LEN = len(str(BLOCK_MAX_SIZE))
2022-05-19 04:39:14 +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)
def get_socket(self, timeout):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(30)
s.connect(server_file)
return s
server_file = TEST_DIR + 'test_serv.sock'
reced_address = ['']
def _announce_server():
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
s.bind(server_file)
while True:
s.listen(1)
conn, _ = s.accept()
with conn:
conn.recv(1)
for _ in range(TRANSPORT_SIZE_BYTES):
dat = conn.recv(1).decode('utf-8')
if dat == '\n':
conn.sendall(int(1).to_bytes(1, 'big'))
break
reced_address[0] += dat
else:
conn.sendall(int(0).to_bytes(1, 'big'))
Thread(target=_announce_server, daemon=True).start()
class OnionrClientAnnounce(unittest.TestCase):
def test_client_announce_too_long(self):
import onionrplugins
our_address = "testtransport" * 100
def event_func(event, *args, data={}, **kwargs):
data['callback'](data['peer'], "testtransport" * 100)
onionrplugins.events.event = event_func
reced_address[0] = ''
gossip_peer_set.clear()
p = MockPeer()
gossip_peer_set.append(p)
do_announce()
self.assertNotEqual(our_address, reced_address[0])
def test_client_announce(self):
import onionrplugins
our_address = "testtransport"
def event_func(event, *args, data={}, **kwargs):
data['callback'](data['peer'], "testtransport")
onionrplugins.events.event = event_func
gossip_peer_set.clear()
reced_address[0] = ''
p = MockPeer()
gossip_peer_set.append(p)
do_announce()
self.assertEqual(our_address, reced_address[0])
unittest.main()