67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
import sys, os
|
|
sys.path.append(".")
|
|
sys.path.append("onionr/")
|
|
import unittest, uuid
|
|
import base64
|
|
|
|
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
|
print("Test directory:", TEST_DIR)
|
|
os.environ["ONIONR_HOME"] = TEST_DIR
|
|
from onionrpeers import peerprofiles
|
|
import onionrexceptions
|
|
from coredb import keydb
|
|
from utils import createdirs
|
|
from onionrutils import stringvalidators, epoch
|
|
TEST_PEER = '3n5wclq4w4pfkcfmjcpqrjluctpm2tzt7etfblavf42cntv6hrerkzyb.onion'
|
|
createdirs.create_dirs()
|
|
|
|
def rand_fake_adder_generator():
|
|
rand_bytes = os.urandom(35)
|
|
return base64.b32encode(rand_bytes).decode().lower() + '.onion'
|
|
|
|
test_peers = []
|
|
for x in range(100):
|
|
p = rand_fake_adder_generator()
|
|
assert stringvalidators.validate_transport(p)
|
|
test_peers.append(p)
|
|
|
|
class TestPeerProfiles(unittest.TestCase):
|
|
def test_invalid_init(self):
|
|
self.assertRaises(onionrexceptions.InvalidAddress, peerprofiles.PeerProfiles, "invalid")
|
|
def test_valid_init(self):
|
|
peerprofiles.PeerProfiles(test_peers.pop())
|
|
|
|
def test_load_score(self):
|
|
p = peerprofiles.PeerProfiles(test_peers.pop())
|
|
self.assertEqual(p.score, 0)
|
|
|
|
def test_inc_score(self):
|
|
p = peerprofiles.PeerProfiles(test_peers.pop())
|
|
s = 0
|
|
for x in range(2):
|
|
s += 1
|
|
p.addScore(1)
|
|
self.assertEqual(p.score, s)
|
|
|
|
def test_inc_score_with_db(self):
|
|
p = peerprofiles.PeerProfiles(test_peers.pop())
|
|
s = 0
|
|
for x in range(2):
|
|
p.last_updated['score'] = epoch.get_epoch() - peerprofiles.UPDATE_DELAY
|
|
s += 1
|
|
p.addScore(1)
|
|
self.assertEqual(p.score, keydb.transportinfo.get_address_info(p.address, 'success'))
|
|
|
|
def test_inc_score_with_sync_delay(self):
|
|
p = peerprofiles.PeerProfiles(test_peers.pop())
|
|
s = 0
|
|
for x in range(2):
|
|
s += 1
|
|
p.addScore(1)
|
|
if x == 0:
|
|
self.assertEqual(p.score, keydb.transportinfo.get_address_info(p.address, 'success'))
|
|
else:
|
|
self.assertNotEqual(p.score, keydb.transportinfo.get_address_info(p.address, 'success'))
|
|
|
|
unittest.main()
|