added peerprofiles and filepaths tests

master
Kevin Froman 2019-09-14 01:29:31 -05:00
parent 4f6f83f383
commit 376b2cc2d6
9 changed files with 103 additions and 10 deletions

View File

@ -1,4 +1,4 @@
FROM ubuntu:bionic
FROM ubuntu:disco
#Base settings
ENV HOME /root

View File

@ -24,7 +24,7 @@ from onionrutils import localcommand, epoch
from .. import dbfiles
import dbcreator
def daemon_queue():
def daemon_queue()->str:
'''
Gives commands to the communication proccess/daemon by reading an sqlite3 database
@ -51,7 +51,7 @@ def daemon_queue():
return retData
def daemon_queue_add(command, data='', responseID=''):
def daemon_queue_add(command: str, data='', responseID: str =''):
'''
Add a command to the daemon queue, used by the communication daemon (communicator.py)
'''

View File

@ -94,7 +94,7 @@ class PrivateEndpoints:
# returns node stats
while True:
try:
return Response(client_api._too_many.get(SerializedData).getStats())
return Response(client_api._too_many.get(SerializedData).get_stats())
except AttributeError as e:
pass

View File

@ -17,6 +17,7 @@
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
from typing import Callable
from .. import onionrstatistics, version, daemonlaunch, keyadders, openwebinterface
from .. import banblocks # Command to blacklist a block by its hash
from .. import filecommands # commands to share files with onionr
@ -28,7 +29,7 @@ from .. import softreset # command to delete onionr blocks
import onionrexceptions
from onionrutils import importnewblocks # func to import new blocks
import onionrevents as events
def get_arguments():
def get_arguments()->dict:
"""This is a function because we need to be able to dynamically modify them with plugins"""
args = {
('blacklist', 'blacklist-block', 'remove-block', 'removeblock', 'banblock', 'ban-block'): banblocks.ban_block,
@ -64,7 +65,7 @@ def get_help(arg: str) -> str:
if arg in argument: return arguments[argument].onionr_help
raise KeyError
def get_func(argument):
def get_func(argument: str) -> Callable:
"""Returns the function for a given command argument"""
argument = argument.lower()
args = get_arguments()

View File

@ -19,6 +19,9 @@
'''
from coredb import keydb
from onionrutils import epoch
from onionrutils import stringvalidators
import onionrblacklist
import onionrexceptions
UPDATE_DELAY = 300
@ -27,6 +30,7 @@ class PeerProfiles:
PeerProfiles
'''
def __init__(self, address):
if not stringvalidators.validate_transport(address): raise onionrexceptions.InvalidAddress
self.address = address # node address
self.score = None
self.friendSigCount = 0
@ -38,7 +42,9 @@ class PeerProfiles:
self.getConnectTime()
self.last_updated = {'connect_time': UPDATE_DELAY, 'score': UPDATE_DELAY} # Last time a given value was updated
return
if not address in keydb.listkeys.list_adders() and not onionrblacklist.OnionrBlackList().inBlacklist(address):
keydb.addkeys.add_address(address)
def loadScore(self):
'''Load the node's score from the database'''
@ -49,10 +55,13 @@ class PeerProfiles:
self.score = self.success
def getConnectTime(self):
"""set the connectTime variable for when we last connected to them, using the db value"""
try:
self.connectTime = int(keydb.transportinfo.get_address_info(self.address, 'lastConnect'))
except (KeyError, ValueError, TypeError) as e:
pass
else:
return self.connectTime
def update_connect_time(self):
if epoch.get_epoch() - self.last_updated['connect_time'] >= UPDATE_DELAY:

View File

@ -32,7 +32,7 @@ class SerializedData:
}
'''
def getStats(self):
def get_stats(self):
'''Return statistics about our node'''
stats = {}
try:

View File

@ -23,7 +23,8 @@ import dbcreator, filepaths
home = identifyhome.identify_home()
def create_dirs():
"""Creates onionr data-related directories in order of the hardcoded list below"""
"""Creates onionr data-related directories in order of the hardcoded list below,
then trigger creation of DBs"""
gen_dirs = [home, filepaths.block_data_location, filepaths.contacts_location, filepaths.export_location]
for path in gen_dirs:
if not os.path.exists(path):

16
tests/test_filepaths.py Normal file
View File

@ -0,0 +1,16 @@
import sys, os
sys.path.append(".")
sys.path.append("onionr/")
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 filepaths
from utils import identifyhome
class TestFilePaths(unittest.TestCase):
def test_filepaths_main(self):
home = identifyhome.identify_home()
self.assertTrue(filepaths.home.startswith(home))
unittest.main()

View File

@ -0,0 +1,66 @@
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()