diff --git a/tests/identity.py b/tests/identity.py index 1d3d493..3a9f7d3 100644 --- a/tests/identity.py +++ b/tests/identity.py @@ -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'] diff --git a/tests/link.py b/tests/link.py index 8c0c018..9447068 100644 --- a/tests/link.py +++ b/tests/link.py @@ -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)