Remove tests associated with removed code
This commit is contained in:
Kevin F
parent
15931ccc32
commit
44d5eeab2a
@ -1,8 +1,8 @@
|
||||
from . import getopenport
|
||||
from .getopenport import get_open_port
|
||||
import os
|
||||
from shutil import which
|
||||
|
||||
|
||||
def tor_binary():
|
||||
"""Return tor binary path or none if not exists"""
|
||||
tor_path = './tor'
|
||||
|
||||
@ -79,7 +79,7 @@ class OnionrUser:
|
||||
|
||||
if saveUser and not publicKey == getourkeypair.get_keypair():
|
||||
try:
|
||||
keydb.addkeys.add_peer(publicKey)
|
||||
keydb.addkeys.add_pub_key(publicKey)
|
||||
except (AssertionError, ValueError) as _:
|
||||
pass
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
from onionrutils import localcommand
|
||||
def check_ui(test_manager):
|
||||
endpoints = ['/', '/mail/', '/friends/', '/board/']
|
||||
endpoints = ['/']
|
||||
for point in endpoints:
|
||||
result = localcommand.local_command(point)
|
||||
if not result: raise ValueError
|
||||
|
||||
@ -1,49 +0,0 @@
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
TEST_DIR = 'testdata/-%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from unittest.mock import patch
|
||||
|
||||
from utils import createdirs
|
||||
|
||||
from subprocess import Popen
|
||||
import subprocess
|
||||
from time import sleep
|
||||
|
||||
from helium import start_firefox, click, Text, Config
|
||||
|
||||
|
||||
from onionrcommands.openwebinterface import get_url
|
||||
from onionrutils import escapeansi
|
||||
BROWSER_HEADLESS = os.getenv('ONIONR_TEST_HEADLESS')
|
||||
|
||||
Config.implicit_wait_secs = 30
|
||||
|
||||
def start_onionr():
|
||||
testargs = ["onionr.py", "start"]
|
||||
with patch.object(sys, 'argv', testargs):
|
||||
parser.register()
|
||||
|
||||
|
||||
class OnionrTests(unittest.TestCase):
|
||||
def test_circles_home_load(self):
|
||||
Popen(['./onionr.sh', 'start'])
|
||||
while b'http' not in Popen(['./onionr.sh', 'url'], stdout=subprocess.PIPE).communicate()[0]:
|
||||
sleep(1)
|
||||
url = 'http' + escapeansi.escape_ANSI(Popen(['./onionr.sh', 'url'], stdout=subprocess.PIPE).communicate()[0].decode().split('http')[1])
|
||||
web_driver = start_firefox(url=url, headless=BROWSER_HEADLESS)
|
||||
if not Text('Circles').exists():
|
||||
click('Get Started')
|
||||
sleep(2)
|
||||
click('Circles')
|
||||
sleep(5)
|
||||
if not Text('Circle Name').exists():
|
||||
Popen(['./onionr.sh', 'stop']).wait()
|
||||
web_driver.quit()
|
||||
raise ValueError
|
||||
Popen(['./onionr.sh', 'stop']).wait()
|
||||
web_driver.quit()
|
||||
|
||||
unittest.main()
|
||||
@ -1,69 +0,0 @@
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
TEST_DIR = 'testdata/-%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from unittest.mock import patch
|
||||
|
||||
from utils import createdirs
|
||||
|
||||
from subprocess import Popen
|
||||
import subprocess
|
||||
from time import sleep
|
||||
|
||||
from helium import start_firefox, click, Text, Config
|
||||
|
||||
|
||||
from onionrcommands.openwebinterface import get_url
|
||||
from onionrutils import escapeansi
|
||||
BROWSER_HEADLESS = os.getenv('ONIONR_TEST_HEADLESS')
|
||||
|
||||
Config.implicit_wait_secs = 30
|
||||
|
||||
def start_onionr():
|
||||
testargs = ["onionr.py", "start"]
|
||||
with patch.object(sys, 'argv', testargs):
|
||||
parser.register()
|
||||
|
||||
|
||||
class OnionrTests(unittest.TestCase):
|
||||
def test_mail_home_load(self):
|
||||
Popen(['./onionr.sh', 'start'])
|
||||
while b'http' not in Popen(['./onionr.sh', 'url'], stdout=subprocess.PIPE).communicate()[0]:
|
||||
sleep(1)
|
||||
|
||||
url = 'http' + escapeansi.escape_ANSI(Popen(['./onionr.sh', 'url'], stdout=subprocess.PIPE).communicate()[0].decode().split('http')[1])
|
||||
web_driver = start_firefox(url=url, headless=BROWSER_HEADLESS)
|
||||
if not Text('Mail').exists():
|
||||
click('Get Started')
|
||||
sleep(2)
|
||||
click('Mail')
|
||||
sleep(5)
|
||||
if not Text('Mail').exists():
|
||||
Popen(['./onionr.sh', 'stop']).wait()
|
||||
web_driver.quit()
|
||||
raise ValueError
|
||||
Popen(['./onionr.sh', 'stop']).wait()
|
||||
web_driver.quit()
|
||||
|
||||
def test_mail(self):
|
||||
Popen(['./onionr.sh', 'start'])
|
||||
while b'http' not in Popen(['./onionr.sh', 'url'], stdout=subprocess.PIPE).communicate()[0]:
|
||||
sleep(1)
|
||||
|
||||
url = 'http' + escapeansi.escape_ANSI(Popen(['./onionr.sh', 'url'], stdout=subprocess.PIPE).communicate()[0].decode().split('http')[1])
|
||||
web_driver = start_firefox(url=url, headless=BROWSER_HEADLESS)
|
||||
if not Text('Mail').exists():
|
||||
click('Get Started')
|
||||
sleep(2)
|
||||
click('Mail')
|
||||
sleep(5)
|
||||
if not Text('Mail').exists():
|
||||
Popen(['./onionr.sh', 'stop']).wait()
|
||||
web_driver.quit()
|
||||
raise ValueError
|
||||
Popen(['./onionr.sh', 'stop']).wait()
|
||||
web_driver.quit()
|
||||
|
||||
unittest.main()
|
||||
@ -1,38 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import uuid
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import unittest
|
||||
|
||||
from utils import identifyhome, createdirs, bettersleep
|
||||
from onionrsetup import setup_config, setup_default_plugins
|
||||
|
||||
createdirs.create_dirs()
|
||||
setup_config()
|
||||
setup_config()
|
||||
setup_default_plugins()
|
||||
|
||||
import config
|
||||
config.set("general.minimum_block_pow", 2)
|
||||
config.set('general.minimum_send_pow', 2)
|
||||
config.save()
|
||||
from onionrblocks import BlockList, insert
|
||||
|
||||
|
||||
class TestBlockList(unittest.TestCase):
|
||||
def test_block_list(self):
|
||||
block_list = BlockList()
|
||||
self.assertEqual(len(block_list.get()), 0)
|
||||
bl = insert('test')
|
||||
bettersleep.better_sleep(0.8)
|
||||
self.assertIn(bl, block_list.get())
|
||||
bl2 = insert('test2')
|
||||
bettersleep.better_sleep(0.8)
|
||||
self.assertIn(bl2, block_list.get())
|
||||
self.assertIn(bl, block_list.get())
|
||||
|
||||
unittest.main()
|
||||
@ -1,44 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid, hashlib
|
||||
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
|
||||
|
||||
import onionrstorage
|
||||
from onionrutils import bytesconverter
|
||||
import onionrcrypto
|
||||
from onionrblocks import onionrblockapi
|
||||
|
||||
from onionrsetup import setup_config, setup_default_plugins
|
||||
|
||||
setup_config()
|
||||
setup_default_plugins()
|
||||
|
||||
import config
|
||||
config.set("general.minimum_block_pow", 2)
|
||||
config.set('general.minimum_send_pow', 2)
|
||||
config.save()
|
||||
import onionrblocks
|
||||
|
||||
class OnionrBlockTests(unittest.TestCase):
|
||||
def test_plaintext_insert(self):
|
||||
|
||||
message = 'hello world'
|
||||
bl = onionrblocks.insert(message)
|
||||
self.assertTrue(bl.startswith('0'))
|
||||
self.assertIn(bytesconverter.str_to_bytes(message), onionrstorage.getData(bl))
|
||||
|
||||
def test_encrypted_insert(self):
|
||||
|
||||
message = 'hello world2'
|
||||
bl = onionrblocks.insert(message, asymPeer=onionrcrypto.pub_key)
|
||||
self.assertIn(bytesconverter.str_to_bytes(message), onionrblockapi.Block(bl, decrypt=True).bcontent)
|
||||
|
||||
unittest.main()
|
||||
@ -9,17 +9,10 @@ os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from utils import createdirs
|
||||
from onionrcommands import parser
|
||||
import onionrsetup as setup
|
||||
from netcontroller.torcontrol import customtorrc
|
||||
class OnionrTests(unittest.TestCase):
|
||||
def test_version_command(self):
|
||||
testargs = ["onionr.py", "version"]
|
||||
with patch.object(sys, 'argv', testargs):
|
||||
parser.register()
|
||||
def test_site_list(self):
|
||||
createdirs.create_dirs()
|
||||
setup.setup_config()
|
||||
testargs = ["onionr.py", "list-sites"]
|
||||
with patch.object(sys, 'argv', testargs):
|
||||
parser.register()
|
||||
|
||||
unittest.main()
|
||||
|
||||
@ -1,31 +0,0 @@
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from utils import createdirs, identifyhome
|
||||
import onionrsetup as setup
|
||||
from netcontroller.torcontrol import customtorrc
|
||||
createdirs.create_dirs()
|
||||
setup.setup_config()
|
||||
|
||||
class TestCustomTorrc(unittest.TestCase):
|
||||
def test_torrc_get(self):
|
||||
torrc = identifyhome.identify_home() + '/torrc-custom'
|
||||
self.assertEqual(customtorrc.get_custom_torrc(), '\n')
|
||||
with open(torrc, 'w') as torrc_file:
|
||||
torrc_file.write('test')
|
||||
self.assertEqual(customtorrc.get_custom_torrc(), '\ntest')
|
||||
os.remove(torrc)
|
||||
|
||||
def test_torrc_set(self):
|
||||
torrc = identifyhome.identify_home() + '/torrc-custom'
|
||||
customtorrc.set_custom_torrc('test2')
|
||||
with open(torrc, 'r') as torrc_file:
|
||||
self.assertEqual(torrc_file.read().splitlines()[2], 'test2')
|
||||
os.remove(torrc)
|
||||
|
||||
unittest.main()
|
||||
@ -1,44 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid, sqlite3
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
from urllib.request import pathname2url
|
||||
from coredb import keydb
|
||||
|
||||
|
||||
class OnionrTests(unittest.TestCase):
|
||||
|
||||
def test_address_add(self):
|
||||
testAddresses = ['facebookcorewwwi.onion', '56kmnycrvepfarolhnx6t2dvmldfeyg7jdymwgjb7jjzg47u2lqw2sad.onion', '5bvb5ncnfr4dlsfriwczpzcvo65kn7fnnlnt2ln7qvhzna2xaldq.b32.i2p']
|
||||
for address in testAddresses:
|
||||
keydb.addkeys.add_address(address)
|
||||
dbAddresses = keydb.listkeys.list_adders()
|
||||
for address in testAddresses:
|
||||
self.assertIn(address, dbAddresses)
|
||||
|
||||
invalidAddresses = [None, '', ' ', '\t', '\n', ' test ', 24, 'fake.onion', 'fake.b32.i2p']
|
||||
for address in invalidAddresses:
|
||||
try:
|
||||
keydb.addkeys.add_address(address)
|
||||
except TypeError:
|
||||
pass
|
||||
dbAddresses = keydb.listkeys.list_adders()
|
||||
for address in invalidAddresses:
|
||||
self.assertNotIn(address, dbAddresses)
|
||||
|
||||
def test_address_info(self):
|
||||
adder = 'nytimes3xbfgragh.onion'
|
||||
keydb.addkeys.add_address(adder)
|
||||
self.assertNotEqual(keydb.transportinfo.get_address_info(adder, 'success'), 1000)
|
||||
keydb.transportinfo.set_address_info(adder, 'success', 1000)
|
||||
self.assertEqual(keydb.transportinfo.get_address_info(adder, 'success'), 1000)
|
||||
|
||||
unittest.main()
|
||||
@ -1,25 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
import json
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
|
||||
from utils import identifyhome, createdirs
|
||||
from onionrsetup import setup_config
|
||||
from coredb import blockmetadb
|
||||
from onionrexceptions import BlockMetaEntryExists
|
||||
createdirs.create_dirs()
|
||||
setup_config()
|
||||
|
||||
class TestDuplicateMetaEntry(unittest.TestCase):
|
||||
def test_no_duplicate(self):
|
||||
bl_hash = '0c88c7d4515363310f0a2522706c49f3f21def5f6fd69af1f91a1849239e7ea6'
|
||||
blockmetadb.add_to_block_DB(bl_hash)
|
||||
self.assertRaises(
|
||||
BlockMetaEntryExists, blockmetadb.add_to_block_DB, bl_hash)
|
||||
|
||||
unittest.main()
|
||||
@ -1,35 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import uuid
|
||||
import ipaddress
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import unittest
|
||||
from utils import identifyhome, createdirs
|
||||
from onionrsetup import setup_config
|
||||
|
||||
createdirs.create_dirs()
|
||||
setup_config()
|
||||
|
||||
|
||||
from lan.getip import lan_ips, best_ip
|
||||
|
||||
class TestGetLanIps(unittest.TestCase):
|
||||
def test_get_lan_ips(self):
|
||||
self.assertGreater(len(lan_ips), 0)
|
||||
for ip in lan_ips:
|
||||
ip = ipaddress.IPv4Address(ip)
|
||||
if not ip.is_private or ip.is_multicast or ip.is_reserved:
|
||||
raise ValueError
|
||||
|
||||
def test_best_lan_ip(self):
|
||||
ip = best_ip
|
||||
ip = ipaddress.IPv4Address(ip)
|
||||
if not ip.is_private or ip.is_multicast or ip.is_reserved:
|
||||
raise ValueError
|
||||
|
||||
|
||||
unittest.main()
|
||||
@ -1,53 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import uuid
|
||||
from threading import Thread
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import unittest, json
|
||||
|
||||
from utils import identifyhome, createdirs
|
||||
from onionrsetup import setup_config
|
||||
createdirs.create_dirs()
|
||||
setup_config()
|
||||
from utils import bettersleep
|
||||
from lan.discover import lan_ips, MCAST_GRP, MCAST_PORT
|
||||
from lan.discover import learn_services, advertise_service
|
||||
import socket
|
||||
import logger
|
||||
from socket import SHUT_RDWR
|
||||
from etc import onionrvalues
|
||||
lan_ips = ['']
|
||||
|
||||
|
||||
class TestLanLearn(unittest.TestCase):
|
||||
def test_lan_learn(self):
|
||||
if onionrvalues.IS_QUBES:
|
||||
logger.info('Cannot run LAN tests on Qubes')
|
||||
return
|
||||
test_ip = '192.168.1.30'
|
||||
def multicast():
|
||||
port = 1349
|
||||
MULTICAST_TTL = 3
|
||||
ips = '-'.join([test_ip]) + f'-{port}'
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||||
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, MULTICAST_TTL)
|
||||
sock.sendto(f"onionr-{ips}".encode('utf-8'), (MCAST_GRP, MCAST_PORT))
|
||||
bettersleep.better_sleep(1)
|
||||
try:
|
||||
sock.shutdown(SHUT_RDWR)
|
||||
except OSError:
|
||||
pass
|
||||
sock.close()
|
||||
test_list = [test_ip]
|
||||
|
||||
Thread(target=learn_services, daemon=True).start()
|
||||
bettersleep.better_sleep(3)
|
||||
multicast()
|
||||
self.assertIn(test_ip, test_list)
|
||||
|
||||
unittest.main()
|
||||
@ -1,64 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import uuid
|
||||
from threading import Thread
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import unittest, json
|
||||
|
||||
from utils import identifyhome, createdirs
|
||||
from onionrsetup import setup_config
|
||||
createdirs.create_dirs()
|
||||
setup_config()
|
||||
import struct
|
||||
from utils import bettersleep
|
||||
from lan.discover import lan_ips, MCAST_GRP, MCAST_PORT, IS_ALL_GROUPS
|
||||
from lan.discover import advertise_service
|
||||
import socket
|
||||
from socket import SHUT_RDWR
|
||||
import logger
|
||||
from etc import onionrvalues
|
||||
|
||||
|
||||
class TestLanLearn(unittest.TestCase):
|
||||
def test_lan_broadcast(self):
|
||||
test_ip = '192.168.1.30'
|
||||
if onionrvalues.IS_QUBES:
|
||||
logger.info('Cannot run LAN tests on Qubes')
|
||||
return
|
||||
def multicast():
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
if IS_ALL_GROUPS:
|
||||
# on this port, receives ALL multicast groups
|
||||
sock.bind(('', MCAST_PORT))
|
||||
else:
|
||||
# on this port, listen ONLY to MCAST_GRP
|
||||
sock.bind((MCAST_GRP, MCAST_PORT))
|
||||
mreq = struct.pack("4sl", socket.inet_aton(MCAST_GRP), socket.INADDR_ANY)
|
||||
|
||||
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
|
||||
|
||||
service_ips = sock.recv(200).decode('utf-8')
|
||||
if not service_ips: raise ValueError
|
||||
|
||||
self.assertIn('onionr', service_ips)
|
||||
self.assertEqual(len(set(set(service_ips.replace('onionr', '').split('-')) ^ set(lan_ips))), 1)
|
||||
for x in set(set(service_ips.replace('onionr', '').split('-')) ^ set(lan_ips)):
|
||||
self.assertEqual('', x)
|
||||
|
||||
try:
|
||||
sock.shutdown(SHUT_RDWR)
|
||||
except OSError:
|
||||
pass
|
||||
sock.close()
|
||||
test_list = [test_ip]
|
||||
|
||||
Thread(target=advertise_service, daemon=True).start()
|
||||
bettersleep.better_sleep(1)
|
||||
multicast()
|
||||
|
||||
unittest.main()
|
||||
@ -1,35 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
|
||||
from utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
from utils import networkmerger
|
||||
from coredb import keydb
|
||||
import onionrsetup as setup
|
||||
setup.setup_config()
|
||||
class NetworkMergerTest(unittest.TestCase):
|
||||
def test_valid_merge(self):
|
||||
adders = 'facebookcorewwwi.onion,mporbyyjhmz2c62shctbi3ngrslne5lpcyav6uzhxok45iblodhgjoad.onion'
|
||||
networkmerger.mergeAdders(adders)
|
||||
added = keydb.listkeys.list_adders()
|
||||
self.assertIn('mporbyyjhmz2c62shctbi3ngrslne5lpcyav6uzhxok45iblodhgjoad.onion', added)
|
||||
self.assertNotIn('inwalidkcorewwi.onion', added)
|
||||
self.assertIn('facebookcorewwwi.onion', added)
|
||||
|
||||
def test_invalid_mergeself(self):
|
||||
adders = 'facebookc0rewwi.onion,sdfsdfsdf.onion, ssdf324, null, \n'
|
||||
networkmerger.mergeAdders(adders)
|
||||
added = keydb.listkeys.list_adders()
|
||||
for adder in adders:
|
||||
self.assertNotIn(adder, added)
|
||||
|
||||
unittest.main()
|
||||
@ -13,7 +13,6 @@ createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
|
||||
from utils import networkmerger
|
||||
from coredb import keydb
|
||||
import onionrsetup as setup
|
||||
from onionrthreads import add_onionr_thread
|
||||
|
||||
@ -1,68 +0,0 @@
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
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 utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
from onionrpeers import peerprofiles
|
||||
import onionrexceptions
|
||||
from coredb import keydb
|
||||
from onionrutils import stringvalidators, epoch
|
||||
TEST_PEER = '3n5wclq4w4pfkcfmjcpqrjluctpm2tzt7etfblavf42cntv6hrerkzyb.onion'
|
||||
|
||||
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()
|
||||
@ -1,64 +0,0 @@
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
from time import sleep
|
||||
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
import config
|
||||
import logger
|
||||
import onionrsetup as setup
|
||||
import filepaths
|
||||
import onionrexceptions
|
||||
|
||||
from onionrsetup import setup_config, setup_default_plugins
|
||||
|
||||
setup_config()
|
||||
setup_default_plugins()
|
||||
|
||||
import config
|
||||
config.set("general.minimum_block_pow", 2)
|
||||
config.set('general.minimum_send_pow', 2)
|
||||
config.save()
|
||||
import onionrblocks
|
||||
|
||||
from onionrblocks import storagecounter
|
||||
import onionrstorage
|
||||
|
||||
def _test_setup():
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
createdirs.create_dirs()
|
||||
setup.setup_config()
|
||||
|
||||
class TestStorageCounter(unittest.TestCase):
|
||||
def test_basic_amount(self):
|
||||
_test_setup()
|
||||
|
||||
self.assertIsNotNone(config.get('allocations.disk'))
|
||||
self.assertGreaterEqual(config.get('allocations.disk'), 1000000)
|
||||
|
||||
def test_insert_too_much(self):
|
||||
_test_setup()
|
||||
config.set('allocations.disk', 1000)
|
||||
self.assertRaises(onionrexceptions.DiskAllocationReached, onionrblocks.insert, "test")
|
||||
|
||||
def test_count(self):
|
||||
_test_setup()
|
||||
counter = storagecounter.StorageCounter()
|
||||
start_value = counter.amount
|
||||
b_hash = onionrblocks.insert("test")
|
||||
sleep(0.1)
|
||||
self.assertGreater(counter.amount, start_value)
|
||||
onionrstorage.removeblock.remove_block(b_hash)
|
||||
sleep(0.1)
|
||||
self.assertEqual(counter.amount, start_value)
|
||||
|
||||
|
||||
unittest.main()
|
||||
@ -1,53 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import uuid
|
||||
import binascii
|
||||
from base64 import b32decode
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import unittest
|
||||
|
||||
from streamfill import extract_ed25519_from_onion_address
|
||||
|
||||
class TestStreamfillExtractEd25519(unittest.TestCase):
|
||||
def test_extract_normal_onion(self):
|
||||
hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
|
||||
correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
|
||||
self.assertEqual(correct, hardcodedCorrect)
|
||||
self.assertEqual(correct, extract_ed25519_from_onion_address('pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion'))
|
||||
def test_extract_normal_onion_bytes(self):
|
||||
hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
|
||||
correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
|
||||
self.assertEqual(correct, hardcodedCorrect)
|
||||
self.assertEqual(correct, extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion'))
|
||||
def test_extract_no_ext(self):
|
||||
hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
|
||||
correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
|
||||
self.assertEqual(correct, hardcodedCorrect)
|
||||
self.assertEqual(correct, extract_ed25519_from_onion_address('pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'))
|
||||
def test_extract_no_ext_bytes(self):
|
||||
hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
|
||||
correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
|
||||
self.assertEqual(correct, hardcodedCorrect)
|
||||
self.assertEqual(correct, extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'))
|
||||
def test_fail_length(self):
|
||||
hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
|
||||
correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
|
||||
self.assertEqual(correct, hardcodedCorrect)
|
||||
try:
|
||||
extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscry')
|
||||
except binascii.Error:
|
||||
pass
|
||||
def test_fail_length_onion(self):
|
||||
hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
|
||||
correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
|
||||
self.assertEqual(correct, hardcodedCorrect)
|
||||
try:
|
||||
extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscry.onion')
|
||||
except binascii.Error:
|
||||
pass
|
||||
|
||||
unittest.main()
|
||||
@ -1,60 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
import subprocess
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import uuid
|
||||
import binascii
|
||||
from base64 import b32decode
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import unittest
|
||||
|
||||
from streamfill import identify_neighbors, extract_ed25519_from_onion_address
|
||||
|
||||
class TestStreamfillNeighbors(unittest.TestCase):
|
||||
def test_neighbor_closeness_consistent(self):
|
||||
main = '7uifxsgidchopmdwmtip6x4ydra6cpf2ov4ghj2lzx5uydyssduh5qid.onion'
|
||||
others = ['bxxajpimlonmbxb5jzjre3go3dvfobqyayqwpksd6zpjz4s4mknstwyd.onion', '2zofaifd6s3flwbv5wl4vtgnesbprc4f2ptljl4a47dfkvrbmw3e5iqd.onion', '6umslj7jtzu27n4jgf3byn55ztz5mkoqocx32zwjya6rbnxqjpyysyyd.onion']
|
||||
main_num = int.from_bytes(extract_ed25519_from_onion_address(main), 'big')
|
||||
|
||||
test_data = identify_neighbors(main, others, 3)
|
||||
|
||||
my_result = []
|
||||
for i in others:
|
||||
my_result.append((i, abs(main_num - int.from_bytes(extract_ed25519_from_onion_address(i), 'big'))))
|
||||
my_result.sort(key=lambda p: p[1])
|
||||
|
||||
final = []
|
||||
for i in my_result:
|
||||
final.append(i[0])
|
||||
self.assertListEqual(final, test_data)
|
||||
|
||||
|
||||
def test_neighbor_closeness_random(self):
|
||||
onions = []
|
||||
p = subprocess.Popen(["scripts/generate-onions.py", '100'],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE)
|
||||
for line in iter(p.stdout.readline, b''):
|
||||
line = line.decode()
|
||||
onions.append(line.strip())
|
||||
p.terminate()
|
||||
main = '7uifxsgidchopmdwmtip6x4ydra6cpf2ov4ghj2lzx5uydyssduh5qid.onion'
|
||||
main_num = int.from_bytes(extract_ed25519_from_onion_address(main), 'big')
|
||||
|
||||
test_data = identify_neighbors(main, onions, 100)
|
||||
|
||||
my_result = []
|
||||
for i in onions:
|
||||
my_result.append((i, abs(main_num - int.from_bytes(extract_ed25519_from_onion_address(i), 'big'))))
|
||||
my_result.sort(key=lambda p: p[1])
|
||||
|
||||
final = []
|
||||
for i in my_result:
|
||||
final.append(i[0])
|
||||
self.assertListEqual(final, test_data)
|
||||
|
||||
|
||||
unittest.main()
|
||||
@ -1,44 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
from onionrblocks import time_insert
|
||||
from onionrblocks import onionrblockapi
|
||||
from onionrsetup import setup_config, setup_default_plugins
|
||||
|
||||
setup_config()
|
||||
setup_default_plugins()
|
||||
|
||||
import config
|
||||
config.set("general.minimum_block_pow", 2)
|
||||
config.set('general.minimum_send_pow', 2)
|
||||
config.save()
|
||||
|
||||
class TestTimeInsert(unittest.TestCase):
|
||||
def test_time_insert_none(self):
|
||||
bl = time_insert('test')
|
||||
self.assertTrue(bl)
|
||||
bl = onionrblockapi.Block(bl)
|
||||
self.assertIs(bl.bmetadata['dly'], 0)
|
||||
|
||||
def test_time_insert_10(self):
|
||||
bl = time_insert('test', delay=10)
|
||||
self.assertTrue(bl)
|
||||
bl = onionrblockapi.Block(bl)
|
||||
self.assertIs(bl.bmetadata['dly'], 10)
|
||||
|
||||
def test_negative(self):
|
||||
self.assertRaises(ValueError, time_insert, 'test', delay=-1)
|
||||
self.assertRaises(ValueError, time_insert, 'test', delay=-10)
|
||||
|
||||
|
||||
|
||||
unittest.main()
|
||||
@ -1,22 +0,0 @@
|
||||
from unittest.mock import patch
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
from onionrcommands import parser
|
||||
from onionrsetup import setup_config
|
||||
setup_config()
|
||||
import config
|
||||
class TestToggleBootstrap(unittest.TestCase):
|
||||
def test_toggle_bootstrap(self):
|
||||
testargs = ["onionr.py", "togglebootstrap"]
|
||||
self.assertTrue(config.get('general.use_bootstrap_list'))
|
||||
with patch.object(sys, 'argv', testargs):
|
||||
parser.register()
|
||||
config.reload()
|
||||
self.assertFalse(config.get('general.use_bootstrap_list'))
|
||||
|
||||
unittest.main()
|
||||
@ -1,40 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys, os
|
||||
sys.path.append(".")
|
||||
sys.path.append("src/")
|
||||
import unittest, uuid
|
||||
TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
|
||||
print("Test directory:", TEST_DIR)
|
||||
os.environ["ONIONR_HOME"] = TEST_DIR
|
||||
import hashlib
|
||||
from utils import createdirs
|
||||
createdirs.create_dirs()
|
||||
from onionrcrypto import getourkeypair
|
||||
getourkeypair.get_keypair()
|
||||
from communicatorutils import uploadblocks
|
||||
|
||||
def hash_generator():
|
||||
hasher = hashlib.sha3_256()
|
||||
hasher.update(os.urandom(15))
|
||||
return hasher.hexdigest()
|
||||
|
||||
test_hashes = []
|
||||
for x in range(100): test_hashes.append(hash_generator())
|
||||
|
||||
class UploadSessionTest(unittest.TestCase):
|
||||
def test_init_fail(self):
|
||||
s = test_hashes.pop()
|
||||
s = uploadblocks.session.UploadSession(s)
|
||||
self.assertEqual(s.total_fail_count, 0)
|
||||
|
||||
def test_init_success(self):
|
||||
s = test_hashes.pop()
|
||||
s = uploadblocks.session.UploadSession(s)
|
||||
self.assertEqual(s.total_success_count, 0)
|
||||
|
||||
def test_invalid(self):
|
||||
invalid = [None, 1, -1, 0, 'ab43c5b8c7b9b037d4f02fa6bc77dbb522bfcbcd7e8ea2953bf2252c6e9232a8b', lambda: None, True, False]
|
||||
for x in invalid:
|
||||
self.assertRaises((ValueError, AttributeError), uploadblocks.session.UploadSession, x)
|
||||
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user