work on tests and various fixes

master
Kevin Froman 2019-02-14 17:48:41 -06:00
parent 9ccf870e4d
commit 3357f93fc1
15 changed files with 176 additions and 316 deletions

6
.gitlab-ci.yml Normal file
View File

@ -0,0 +1,6 @@
test:
script:
- apt-get update -qy
- apt-get install -y python3-pip tor
- pip3 install -r requirements.txt
- make test

View File

@ -18,26 +18,20 @@ uninstall:
rm -f $(DESTDIR)$(PREFIX)/bin/onionr rm -f $(DESTDIR)$(PREFIX)/bin/onionr
test: test:
@./onionr.sh stop ./run_tests.sh
@sleep 1
@rm -rf onionr/data-backup
@mv onionr/data onionr/data-backup | true > /dev/null 2>&1
-@cd onionr; ./tests.py;
@rm -rf onionr/data
@mv onionr/data-backup onionr/data | true > /dev/null 2>&1
soft-reset: soft-reset:
@echo "Soft-resetting Onionr..." @echo "Soft-resetting Onionr..."
rm -f onionr/data/blocks/*.dat onionr/data/*.db onionr/data/block-nonces.dat | true > /dev/null 2>&1 rm -f onionr/$(ONIONR_HOME)/blocks/*.dat onionr/data/*.db onionr/$(ONIONR_HOME)/block-nonces.dat | true > /dev/null 2>&1
@./onionr.sh version | grep -v "Failed" --color=always @./onionr.sh version | grep -v "Failed" --color=always
reset: reset:
@echo "Hard-resetting Onionr..." @echo "Hard-resetting Onionr..."
rm -rf onionr/data/ | true > /dev/null 2>&1 rm -rf onionr/$(ONIONR_HOME)/ | true > /dev/null 2>&1
cd onionr/static-data/www/ui/; rm -rf ./dist; python compile.py cd onionr/static-data/www/ui/; rm -rf ./dist; python compile.py
#@./onionr.sh.sh version | grep -v "Failed" --color=always #@./onionr.sh.sh version | grep -v "Failed" --color=always
plugins-reset: plugins-reset:
@echo "Resetting plugins..." @echo "Resetting plugins..."
rm -rf onionr/data/plugins/ | true > /dev/null 2>&1 rm -rf onionr/$(ONIONR_HOME)/plugins/ | true > /dev/null 2>&1
@./onionr.sh version | grep -v "Failed" --color=always @./onionr.sh version | grep -v "Failed" --color=always

View File

@ -69,6 +69,7 @@ class PublicAPI:
self.torAdder = clientAPI._core.hsAddress self.torAdder = clientAPI._core.hsAddress
self.i2pAdder = clientAPI._core.i2pAddress self.i2pAdder = clientAPI._core.i2pAddress
self.bindPort = config.get('client.public.port') self.bindPort = config.get('client.public.port')
self.lastRequest = 0
logger.info('Running public api on %s:%s' % (self.host, self.bindPort)) logger.info('Running public api on %s:%s' % (self.host, self.bindPort))
@app.before_request @app.before_request
@ -98,6 +99,7 @@ class PublicAPI:
resp.headers['X-API'] = onionr.API_VERSION resp.headers['X-API'] = onionr.API_VERSION
# Close connections to limit FD use # Close connections to limit FD use
resp.headers['Connection'] = "close" resp.headers['Connection'] = "close"
self.lastRequest = clientAPI._core._utils.getRoundedEpoch(roundS=5)
return resp return resp
@app.route('/') @app.route('/')
@ -393,6 +395,10 @@ class API:
resp = self.getBlockData(name, decrypt=True, headerOnly=True) resp = self.getBlockData(name, decrypt=True, headerOnly=True)
return Response(resp) return Response(resp)
@app.route('/lastconnect')
def lastConnect():
return Response(str(self.publicAPI.lastRequest))
@app.route('/site/<name>', endpoint='site') @app.route('/site/<name>', endpoint='site')
def site(name): def site(name):
bHash = name bHash = name

View File

@ -21,7 +21,7 @@
''' '''
import sys, os, core, config, json, requests, time, logger, threading, base64, onionr, uuid import sys, os, core, config, json, requests, time, logger, threading, base64, onionr, uuid
import onionrexceptions, onionrpeers, onionrevents as events, onionrplugins as plugins, onionrblockapi as block import onionrexceptions, onionrpeers, onionrevents as events, onionrplugins as plugins, onionrblockapi as block
import onionrdaemontools, onionrsockets, onionr, onionrproofs, proofofmemory import onionrdaemontools, onionrsockets, onionr, onionrproofs
import binascii import binascii
from dependencies import secrets from dependencies import secrets
from defusedxml import minidom from defusedxml import minidom
@ -86,8 +86,6 @@ class OnionrCommunicatorDaemon:
# Loads in and starts the enabled plugins # Loads in and starts the enabled plugins
plugins.reload() plugins.reload()
self.proofofmemory = proofofmemory.ProofOfMemory(self)
# daemon tools are misc daemon functions, e.g. announce to online peers # daemon tools are misc daemon functions, e.g. announce to online peers
# intended only for use by OnionrCommunicatorDaemon # intended only for use by OnionrCommunicatorDaemon
self.daemonTools = onionrdaemontools.DaemonTools(self) self.daemonTools = onionrdaemontools.DaemonTools(self)
@ -630,7 +628,7 @@ class OnionrCommunicatorDaemon:
'''exit if the api server crashes/stops''' '''exit if the api server crashes/stops'''
if self._core._utils.localCommand('ping', silent=False) not in ('pong', 'pong!'): if self._core._utils.localCommand('ping', silent=False) not in ('pong', 'pong!'):
for i in range(8): for i in range(8):
if self._core._utils.localCommand('ping') in ('pong', 'pong!'): if self._core._utils.localCommand('ping') in ('pong', 'pong!') or self.shutdown:
break # break for loop break # break for loop
time.sleep(1) time.sleep(1)
else: else:

View File

@ -85,6 +85,10 @@ class Core:
self.createBlockDB() self.createBlockDB()
if not os.path.exists(self.forwardKeysFile): if not os.path.exists(self.forwardKeysFile):
self.dbCreate.createForwardKeyDB() self.dbCreate.createForwardKeyDB()
if not os.path.exists(self.peerDB):
self.createPeerDB()
if not os.path.exists(self.addressDB):
self.createAddressDB()
if os.path.exists(self.dataDir + '/hs/hostname'): if os.path.exists(self.dataDir + '/hs/hostname'):
with open(self.dataDir + '/hs/hostname', 'r') as hs: with open(self.dataDir + '/hs/hostname', 'r') as hs:
@ -273,15 +277,6 @@ class Core:
Simply return the data associated to a hash Simply return the data associated to a hash
''' '''
'''
try:
# logger.debug('Opening %s' % (str(self.blockDataLocation) + str(hash) + '.dat'))
dataFile = open(self.blockDataLocation + hash + '.dat', 'rb')
data = dataFile.read()
dataFile.close()
except FileNotFoundError:
data = False
'''
data = onionrstorage.getData(self, hash) data = onionrstorage.getData(self, hash)
return data return data

View File

@ -79,8 +79,12 @@ LEVEL_IMPORTANT = 6
_type = OUTPUT_TO_CONSOLE | USE_ANSI # the default settings for logging _type = OUTPUT_TO_CONSOLE | USE_ANSI # the default settings for logging
_level = LEVEL_DEBUG # the lowest level to log _level = LEVEL_DEBUG # the lowest level to log
_outputfile = './output.log' # the file to log to dataFolder = os.getenv('ONIONR_HOME')
if type(dataFolder) is type(None):
dataFolder = 'data/'
if not dataFolder.endswith('/'):
dataFolder += '/'
_outputfile = dataFolder + 'output.log' # the file to log to
def set_settings(type): def set_settings(type):
''' '''
Set the settings for the logger using bitwise operators Set the settings for the logger using bitwise operators

View File

@ -110,12 +110,6 @@ class Onionr:
except: except:
plugins.disable(name, onionr = self, stop_event = False) plugins.disable(name, onionr = self, stop_event = False)
if not os.path.exists(self.onionrCore.peerDB):
self.onionrCore.createPeerDB()
pass
if not os.path.exists(self.onionrCore.addressDB):
self.onionrCore.createAddressDB()
# Get configuration # Get configuration
if type(config.get('client.webpassword')) is type(None): if type(config.get('client.webpassword')) is type(None):
config.set('client.webpassword', base64.b16encode(os.urandom(32)).decode('utf-8'), savefile=True) config.set('client.webpassword', base64.b16encode(os.urandom(32)).decode('utf-8'), savefile=True)
@ -1014,7 +1008,6 @@ class Onionr:
settings = settings | logger.OUTPUT_TO_CONSOLE settings = settings | logger.OUTPUT_TO_CONSOLE
if config.get('log.file.output', True): if config.get('log.file.output', True):
settings = settings | logger.OUTPUT_TO_FILE settings = settings | logger.OUTPUT_TO_FILE
logger.set_file(config.get('log.file.path', '/tmp/onionr.log').replace('data/', dataDir))
logger.set_settings(settings) logger.set_settings(settings)
if not self is None: if not self is None:

View File

@ -18,7 +18,7 @@
along with this program. If not, see <https://www.gnu.org/licenses/>. along with this program. If not, see <https://www.gnu.org/licenses/>.
''' '''
# Misc functions that do not fit in the main api, but are useful # Misc functions that do not fit in the main api, but are useful
import getpass, sys, requests, os, socket, hashlib, logger, sqlite3, config, binascii, time, base64, json, glob, shutil, math, json, re, urllib.parse import getpass, sys, requests, os, socket, hashlib, logger, sqlite3, config, binascii, time, base64, json, glob, shutil, math, json, re, urllib.parse, string
import nacl.signing, nacl.encoding import nacl.signing, nacl.encoding
from onionrblockapi import Block from onionrblockapi import Block
import onionrexceptions import onionrexceptions
@ -309,6 +309,8 @@ class OnionrUtils:
Validate if a string is a valid base32 encoded Ed25519 key Validate if a string is a valid base32 encoded Ed25519 key
''' '''
retVal = False retVal = False
if type(key) is type(None):
return False
try: try:
nacl.signing.SigningKey(seed=key, encoder=nacl.encoding.Base32Encoder) nacl.signing.SigningKey(seed=key, encoder=nacl.encoding.Base32Encoder)
except nacl.exceptions.ValueError: except nacl.exceptions.ValueError:
@ -363,16 +365,9 @@ class OnionrUtils:
retVal = False retVal = False
# Validate address is valid base32 (when capitalized and minus extension); v2/v3 onions and .b32.i2p use base32 # Validate address is valid base32 (when capitalized and minus extension); v2/v3 onions and .b32.i2p use base32
try: for x in idNoDomain.upper():
base64.b32decode(idNoDomain.upper().encode()) if x not in string.ascii_uppercase and x not in '234567':
except binascii.Error: retVal = False
retVal = False
# Validate address is valid base32 (when capitalized and minus extension); v2/v3 onions and .b32.i2p use base32
try:
base64.b32decode(idNoDomain.upper().encode())
except binascii.Error:
retVal = False
return retVal return retVal
except: except:
@ -382,7 +377,7 @@ class OnionrUtils:
'''Check if a string is a valid base10 integer (also returns true if already an int)''' '''Check if a string is a valid base10 integer (also returns true if already an int)'''
try: try:
int(data) int(data)
except ValueError: except (ValueError, TypeError) as e:
return False return False
else: else:
return True return True

View File

@ -1,29 +0,0 @@
'''
Onionr - P2P Anonymous Storage Network
This file handles proof of memory functionality
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
class ProofOfMemory:
def __init__(self, commInst):
self.communicator = commInst
return
def checkRandomPeer(self):
return
def checkPeer(self, peer):
return

View File

@ -0,0 +1 @@
lc4mw2es4se22xsjrccumna3ih53difx64q2t2mgk5ijjk7d6aiacbqd.onion

View File

@ -21,6 +21,7 @@
<br><br><a class='idLink' href='/mail/'>Mail</a> <br><br><a class='idLink' href='/mail/'>Mail</a>
<h2>Stats</h2> <h2>Stats</h2>
<p>Uptime: <span id='uptime'></span></p> <p>Uptime: <span id='uptime'></span></p>
<p>Last Received Connection: <span id='lastIncoming'>Unknown</span></p>
<p>Stored Blocks: <span id='storedBlocks'></span></p> <p>Stored Blocks: <span id='storedBlocks'></span></p>
<p>Blocks in queue: <span id='blockQueue'></span></p> <p>Blocks in queue: <span id='blockQueue'></span></p>
<p>Connected nodes:</p> <p>Connected nodes:</p>

View File

@ -21,6 +21,7 @@ uptimeDisplay = document.getElementById('uptime')
connectedDisplay = document.getElementById('connectedNodes') connectedDisplay = document.getElementById('connectedNodes')
storedBlockDisplay = document.getElementById('storedBlocks') storedBlockDisplay = document.getElementById('storedBlocks')
queuedBlockDisplay = document.getElementById('blockQueue') queuedBlockDisplay = document.getElementById('blockQueue')
lastIncoming = document.getElementById('lastIncoming')
function getStats(){ function getStats(){
stats = JSON.parse(httpGet('getstats', webpass)) stats = JSON.parse(httpGet('getstats', webpass))
@ -28,5 +29,15 @@ function getStats(){
connectedDisplay.innerText = stats['connectedNodes'] connectedDisplay.innerText = stats['connectedNodes']
storedBlockDisplay.innerText = stats['blockCount'] storedBlockDisplay.innerText = stats['blockCount']
queuedBlockDisplay.innerText = stats['blockQueueCount'] queuedBlockDisplay.innerText = stats['blockQueueCount']
var lastConnect = httpGet('/lastconnect')
if (lastConnect > 0){
var humanDate = new Date(0)
humanDate.setUTCSeconds(httpGet('/lastconnect'))
lastConnect = humanDate.toString()
}
else{
lastConnect = 'Unknown'
}
lastIncoming.innerText = lastConnect
} }
getStats() getStats()

View File

@ -1,243 +0,0 @@
#!/usr/bin/env python3
'''
Onionr - P2P Microblogging Platform & Social network
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
import unittest, sys, os, base64, tarfile, shutil, logger
class OnionrTests(unittest.TestCase):
def testPython3(self):
if sys.version_info.major != 3:
logger.debug('Python version: ' + sys.version_info.major)
self.assertTrue(False)
else:
self.assertTrue(True)
def testNone(self):
logger.debug('-'*26 + '\n')
logger.info('Running simple program run test...')
blank = os.system('./onionr.py --version')
if blank != 0:
self.assertTrue(False)
else:
self.assertTrue(True)
def testPeer_a_DBCreation(self):
logger.debug('-'*26 + '\n')
logger.info('Running peer db creation test...')
if os.path.exists('data/peers.db'):
os.remove('data/peers.db')
import core
myCore = core.Core()
myCore.createPeerDB()
if os.path.exists('data/peers.db'):
self.assertTrue(True)
else:
self.assertTrue(False)
def testPeer_b_addPeerToDB(self):
logger.debug('-'*26 + '\n')
logger.info('Running peer db insertion test...')
import core
myCore = core.Core()
if not os.path.exists('data/peers.db'):
myCore.createPeerDB()
if myCore.addPeer('6M5MXL237OK57ITHVYN5WGHANPGOMKS5C3PJLHBBNKFFJQOIDOJA====', '1cSix9Ao/yQSdo0sNif8cm2uTcYnSphb4JdZL/3WkN4=') and not myCore.addPeer('NFXHMYLMNFSAU===', '1cSix9Ao/yQSdo0sNif8cm2uTcYnSphb4JdZL/3WkN4='):
self.assertTrue(True)
else:
self.assertTrue(False)
def testConfig(self):
logger.debug('-'*26 + '\n')
logger.info('Running simple configuration test...')
import config
config.check()
config.reload()
configdata = str(config.get_config())
config.set('testval', 1337)
if not config.get('testval', None) is 1337:
self.assertTrue(False)
config.set('testval')
if not config.get('testval', None) is None:
self.assertTrue(False)
config.save()
config.reload()
if not str(config.get_config()) == configdata:
self.assertTrue(False)
self.assertTrue(True)
'''
def testBlockAPI(self):
logger.debug('-'*26 + '\n')
logger.info('Running BlockAPI test #1...')
content = 'Onionr test block'
from onionrblockapi import Block
hash = Block(type = 'test', content = content).save()
block = Block(hash) # test init
if len(Block.getBlocks(type = 'test')) == 0:
logger.warn('Failed to find test block.')
self.assertTrue(False)
if not block.getContent() == content:
logger.warn('Test block content is invalid! (%s != %s)' % (block.getContent(), content))
self.assertTrue(False)
logger.debug('-'*26 + '\n')
logger.info('Running BlockAPI test #2...')
original_content = 'onionr'
logger.debug('original: %s' % original_content)
blocks = Block.createChain(data = original_content, chunksize = 2, verbose = True)
logger.debug(blocks[1])
child = blocks[0]
merged = Block.mergeChain(child)
logger.debug('merged blocks (child: %s): %s' % (child, merged))
if merged != original_content:
self.assertTrue(False)
self.assertTrue(True)
def testPluginReload(self):
logger.debug('-'*26 + '\n')
logger.info('Running simple plugin reload test...')
import onionrplugins, os
if not onionrplugins.exists('test'):
os.makedirs(onionrplugins.get_plugins_folder('test'))
with open(onionrplugins.get_plugins_folder('test') + '/main.py', 'a') as main:
main.write("print('Running')\n\ndef on_test(pluginapi, data = None):\n print('received test event!')\n return True\n\ndef on_start(pluginapi, data = None):\n print('start event called')\n\ndef on_stop(pluginapi, data = None):\n print('stop event called')\n\ndef on_enable(pluginapi, data = None):\n print('enable event called')\n\ndef on_disable(pluginapi, data = None):\n print('disable event called')\n")
onionrplugins.enable('test')
try:
onionrplugins.reload('test')
self.assertTrue(True)
except:
self.assertTrue(False)
def testPluginStopStart(self):
logger.debug('-'*26 + '\n')
logger.info('Running simple plugin restart test...')
import onionrplugins, os
if not onionrplugins.exists('test'):
os.makedirs(onionrplugins.get_plugins_folder('test'))
with open(onionrplugins.get_plugins_folder('test') + '/main.py', 'a') as main:
main.write("print('Running')\n\ndef on_test(pluginapi, data = None):\n print('received test event!')\n return True\n\ndef on_start(pluginapi, data = None):\n print('start event called')\n\ndef on_stop(pluginapi, data = None):\n print('stop event called')\n\ndef on_enable(pluginapi, data = None):\n print('enable event called')\n\ndef on_disable(pluginapi, data = None):\n print('disable event called')\n")
onionrplugins.enable('test')
try:
onionrplugins.start('test')
onionrplugins.stop('test')
self.assertTrue(True)
except:
self.assertTrue(False)
def testPluginEvent(self):
logger.debug('-'*26 + '\n')
logger.info('Running plugin event test...')
import onionrplugins as plugins, onionrevents as events, os
if not plugins.exists('test'):
os.makedirs(plugins.get_plugins_folder('test'))
with open(plugins.get_plugins_folder('test') + '/main.py', 'a') as main:
main.write("print('Running')\n\ndef on_test(pluginapi, data = None):\n print('received test event!')\n print('thread test started...')\n import time\n time.sleep(1)\n \n return True\n\ndef on_start(pluginapi, data = None):\n print('start event called')\n\ndef on_stop(pluginapi, data = None):\n print('stop event called')\n\ndef on_enable(pluginapi, data = None):\n print('enable event called')\n\ndef on_disable(pluginapi, data = None):\n print('disable event called')\n")
plugins.enable('test')
plugins.start('test')
if not events.call(plugins.get_plugin('test'), 'enable'):
self.assertTrue(False)
logger.debug('preparing to start thread', timestamp = False)
thread = events.event('test', data = {'tests': self})
logger.debug('thread running...', timestamp = False)
thread.join()
logger.debug('thread finished.', timestamp = False)
self.assertTrue(True)
'''
def testQueue(self):
logger.debug('-'*26 + '\n')
logger.info('Running daemon queue test...')
# test if the daemon queue can read/write data
import core
myCore = core.Core()
if not os.path.exists('data/queue.db'):
myCore.daemonQueue()
while True:
command = myCore.daemonQueue()
if command == False:
logger.debug('The queue is empty (false)')
break
else:
logger.debug(command[0])
myCore.daemonQueueAdd('testCommand', 'testData')
command = myCore.daemonQueue()
if command[0] == 'testCommand':
if myCore.daemonQueue() == False:
logger.info('Succesfully added and read command')
def testHashValidation(self):
logger.debug('-'*26 + '\n')
logger.info('Running hash validation test...')
import core
myCore = core.Core()
if not myCore._utils.validateHash("$324dfgfdg") and myCore._utils.validateHash("f2ca1bb6c7e907d06dafe4687e579fce76b37e4e93b7605022da52e6ccc26fd2") and not myCore._utils.validateHash("f2ca1bb6c7e907d06dafe4687e579fce76b37e4e93b7605022da52e6ccc26fd$"):
self.assertTrue(True)
else:
self.assertTrue(False)
def testAddAdder(self):
logger.debug('-'*26 + '\n')
logger.info('Running address add+remove test')
import core
myCore = core.Core()
if not os.path.exists('data/address.db'):
myCore.createAddressDB()
if myCore.addAddress('facebookcorewwwi.onion') and not myCore.removeAddress('invalid'):
if myCore.removeAddress('facebookcorewwwi.onion'):
self.assertTrue(True)
else:
self.assertTrue(False)
else:
self.assertTrue(False) # <- annoying :(
def testCrypto(self):
logger.info('running cryptotests')
if os.system('python3 cryptotests.py') == 0:
self.assertTrue(True)
else:
self.assertTrue(False)
unittest.main()

View File

@ -0,0 +1,70 @@
#!/usr/bin/env python3
import sys, os
sys.path.append(".")
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 urllib.request import pathname2url
import core, onionr
core.Core()
class OnionrTests(unittest.TestCase):
def test_peer_db_creation(self):
try:
dburi = 'file:{}?mode=rw'.format(pathname2url(TEST_DIR + 'peers.db'))
conn = sqlite3.connect(dburi, uri=True, timeout=30)
cursor = conn.cursor()
conn.close()
except sqlite3.OperationalError:
self.assertTrue(False)
else:
self.assertTrue(True)
def test_block_db_creation(self):
try:
dburi = 'file:{}?mode=rw'.format(pathname2url(TEST_DIR + 'blocks.db'))
conn = sqlite3.connect(dburi, uri=True, timeout=30)
cursor = conn.cursor()
conn.close()
except sqlite3.OperationalError:
self.assertTrue(False)
else:
self.assertTrue(True)
def test_forward_keys_db_creation(self):
try:
dburi = 'file:{}?mode=rw'.format(pathname2url(TEST_DIR + 'forward-keys.db'))
conn = sqlite3.connect(dburi, uri=True, timeout=30)
cursor = conn.cursor()
conn.close()
except sqlite3.OperationalError:
self.assertTrue(False)
else:
self.assertTrue(True)
def test_address_db_creation(self):
try:
dburi = 'file:{}?mode=rw'.format(pathname2url(TEST_DIR + 'address.db'))
conn = sqlite3.connect(dburi, uri=True, timeout=30)
cursor = conn.cursor()
conn.close()
except sqlite3.OperationalError:
self.assertTrue(False)
else:
self.assertTrue(True)
def blacklist_db_creation(self):
try:
dburi = 'file:{}?mode=rw'.format(pathname2url(TEST_DIR + 'blacklist.db'))
conn = sqlite3.connect(dburi, uri=True, timeout=30)
cursor = conn.cursor()
conn.close()
except sqlite3.OperationalError:
self.assertTrue(False)
else:
self.assertTrue(True)
unittest.main()

View File

@ -0,0 +1,58 @@
#!/usr/bin/env python3
import sys, os
sys.path.append(".")
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 urllib.request import pathname2url
import core, onionr
core.Core()
class OnionrValidations(unittest.TestCase):
def test_peer_validator(self):
# Test hidden service domain validities
c = core.Core()
valid = ['facebookcorewwwi.onion', 'vww6ybal4bd7szmgncyruucpgfkqahzddi37ktceo3ah7ngmcopnpyyd.onion',
'5bvb5ncnfr4dlsfriwczpzcvo65kn7fnnlnt2ln7qvhzna2xaldq.b32.i2p']
invalid = [None, 'dsfewjirji0ejipdfs', '', ' ', '\n', '\r\n', 'f$ce%^okc+rewwwi.onion']
for x in valid:
print('testing', x)
self.assertTrue(c._utils.validateID(x))
for x in invalid:
print('testing', x)
self.assertFalse(c._utils.validateID(x))
def test_pubkey_validator(self):
# Test ed25519 public key validity
valid = 'JZ5VE72GUS3C7BOHDRIYZX4B5U5EJMCMLKHLYCVBQQF3UKHYIRRQ===='
invalid = [None, '', ' ', 'dfsg', '\n', 'JZ5VE72GUS3C7BOHDRIYZX4B5U5EJMCMLKHLYCVBQQF3UKHYIR$Q====']
c = core.Core()
print('testing', valid)
self.assertTrue(c._utils.validatePubKey(valid))
for x in invalid:
print('testing', x)
self.assertFalse(c._utils.validatePubKey(x))
def test_integer_string(self):
valid = ["1", "100", 100, "-5", -5]
invalid = ['test', "1d3434", "1e100", None]
c = core.Core()
for x in valid:
print('testing', x)
self.assertTrue(c._utils.isIntegerString(x))
for x in invalid:
print('testing', x)
self.assertFalse(c._utils.isIntegerString(x))
unittest.main()