William/Reticulum
1
0
Fork 0
mirror of https://github.com/markqvist/Reticulum.git synced 2024-11-05 05:40:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Updated tests

Browse Source
This commit is contained in:
Mark Qvist 2022-06-09 19:27:11 +02:00
parent 513a82e363
commit 06fefebc08
2 changed files with 111 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tests/identity.py
View File

@ -51,8 +51,7 @@ class TestIdentity(unittest.TestCase):
self.assertEqual(True, id2.validate(signature, msg))
t += time.time() - start
# print("Sign/validate: "+self.size_str(b)+", "+self.size_str(b, "b")+"/s")
print("Sign/validate chunks < MTU: "+self.size_str(b/t, "b")+"/s")
print("Sign/validate chunks < MTU: "+self.size_str(b/t, "b")+"ps")
for i in range(1, 500):
mlen = 16*1024
@ -67,8 +66,7 @@ class TestIdentity(unittest.TestCase):
self.assertEqual(True, id2.validate(signature, msg))
t += time.time() - start
# print("Sign/validate: "+self.size_str(b)+", "+self.size_str(b, "b")+"/s")
print("Sign/validate 16KB chunks: "+self.size_str(b/t, "b")+"/s")
print("Sign/validate 16KB chunks: "+self.size_str(b/t, "b")+"ps")
def test_encrypt(self):
print("")
@ -99,8 +97,8 @@ class TestIdentity(unittest.TestCase):
self.assertEqual(msg, id1.decrypt(token))
d_t += time.time() - d_start
print("Encrypt chunks < MTU: "+self.size_str(b/e_t, "b")+"/s")
print("Decrypt chunks < MTU: "+self.size_str(b/d_t, "b")+"/s")
print("Encrypt chunks < MTU: "+self.size_str(b/e_t, "b")+"ps")
print("Decrypt chunks < MTU: "+self.size_str(b/d_t, "b")+"ps")
print("")
# Test encrypt and decrypt of large chunks
@ -120,8 +118,8 @@ class TestIdentity(unittest.TestCase):
self.assertEqual(msg, id1.decrypt(token))
d_t += time.time() - d_start
print("Encrypt "+self.size_str(mlen)+" chunks: "+self.size_str(b/e_t, "b")+"/s")
print("Decrypt "+self.size_str(mlen)+" chunks: "+self.size_str(b/d_t, "b")+"/s")
print("Encrypt "+self.size_str(mlen)+" chunks: "+self.size_str(b/e_t, "b")+"ps")
print("Decrypt "+self.size_str(mlen)+" chunks: "+self.size_str(b/d_t, "b")+"ps")
def size_str(self, num, suffix='B'):
units = ['','K','M','G','T','P','E','Z']

108
tests/link.py
View File

@ -97,10 +97,79 @@ class TestLink(unittest.TestCase):
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.CLOSED)
def test_resource(self):
def test_micro_resource(self):
init_rns()
print("")
print("Micro resource test")
# TODO: Load this from public bytes only
id1 = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
self.assertEqual(id1.hash, bytes.fromhex(fixed_keys[0][1]))
dest = RNS.Destination(id1, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "link", "establish")
self.assertEqual(dest.hash, bytes.fromhex("be0c90339fce3db5b4e5"))
l1 = RNS.Link(dest)
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.ACTIVE)
resource_timeout = 120
resource_size = 128
data = os.urandom(resource_size)
print("Sending "+self.size_str(resource_size)+" resource...")
resource = RNS.Resource(data, l1, timeout=resource_timeout)
start = time.time()
while resource.status < RNS.Resource.COMPLETE:
time.sleep(0.01)
t = time.time() - start
self.assertEqual(resource.status, RNS.Resource.COMPLETE)
print("Resource completed at "+self.size_str(resource_size/t, "b")+"ps")
l1.teardown()
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.CLOSED)
def test_small_resource(self):
init_rns()
print("")
print("Small resource test")
# TODO: Load this from public bytes only
id1 = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
self.assertEqual(id1.hash, bytes.fromhex(fixed_keys[0][1]))
dest = RNS.Destination(id1, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "link", "establish")
self.assertEqual(dest.hash, bytes.fromhex("be0c90339fce3db5b4e5"))
l1 = RNS.Link(dest)
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.ACTIVE)
resource_timeout = 120
resource_size = 1000*1000
data = os.urandom(resource_size)
print("Sending "+self.size_str(resource_size)+" resource...")
resource = RNS.Resource(data, l1, timeout=resource_timeout)
start = time.time()
while resource.status < RNS.Resource.COMPLETE:
time.sleep(0.01)
t = time.time() - start
self.assertEqual(resource.status, RNS.Resource.COMPLETE)
print("Resource completed at "+self.size_str(resource_size/t, "b")+"ps")
l1.teardown()
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.CLOSED)
def test_medium_resource(self):
init_rns()
print("")
print("Medium resource test")
# TODO: Load this from public bytes only
id1 = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
@ -131,6 +200,39 @@ class TestLink(unittest.TestCase):
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.CLOSED)
def test_large_resource(self):
init_rns()
print("")
print("Large resource test")
# TODO: Load this from public bytes only
id1 = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
self.assertEqual(id1.hash, bytes.fromhex(fixed_keys[0][1]))
dest = RNS.Destination(id1, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "link", "establish")
self.assertEqual(dest.hash, bytes.fromhex("be0c90339fce3db5b4e5"))
l1 = RNS.Link(dest)
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.ACTIVE)
resource_timeout = 120
resource_size = 35*1000*1000
data = os.urandom(resource_size)
print("Sending "+self.size_str(resource_size)+" resource...")
resource = RNS.Resource(data, l1, timeout=resource_timeout)
start = time.time()
while resource.status < RNS.Resource.COMPLETE:
time.sleep(0.01)
t = time.time() - start
self.assertEqual(resource.status, RNS.Resource.COMPLETE)
print("Resource completed at "+self.size_str(resource_size/t, "b")+"ps")
l1.teardown()
time.sleep(0.5)
self.assertEqual(l1.status, RNS.Link.CLOSED)
def size_str(self, num, suffix='B'):
@ -216,7 +318,7 @@ def resource_profiling():
time.sleep(0.5)
resource_timeout = 120
resource_size = 4*1000*1000
resource_size = 5*1000*1000
data = os.urandom(resource_size)
print("Sending "+size_str(resource_size)+" resource...")
resource = RNS.Resource(data, l1, timeout=resource_timeout)