mirror
/
reticulum
kopia lustrzana https://github.com/markqvist/reticulum
1
0
Forkuj 0
Kod Zgłoszenia Projekty Wydania Wiki Aktywność

Cleanup

pull/325/head
Mark Qvist 2023-05-11 18:35:01 +02:00
rodzic 66fda34b20
commit 387b8c46ff
1 zmienionych plików z 79 dodań i 74 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

153
tests/channel.py
Wyświetl plik

@ -291,6 +291,8 @@ class TestChannel(unittest.TestCase):
raw = envelope.pack() raw = envelope.pack()
self.h.channel._receive(raw) self.h.channel._receive(raw)
time.sleep(0.5)
self.assertEqual(1, handler1_called) self.assertEqual(1, handler1_called)
self.assertEqual(0, handler2_called) self.assertEqual(0, handler2_called)
@ -299,6 +301,8 @@ class TestChannel(unittest.TestCase):
raw = envelope.pack() raw = envelope.pack()
self.h.channel._receive(raw) self.h.channel._receive(raw)
time.sleep(0.5)
self.assertEqual(2, handler1_called) self.assertEqual(2, handler1_called)
self.assertEqual(1, handler2_called) self.assertEqual(1, handler2_called)
@ -357,6 +361,8 @@ class TestChannel(unittest.TestCase):
self.h.channel._receive(packet.raw) self.h.channel._receive(packet.raw)
time.sleep(0.5)
self.assertEqual(1, len(decoded)) self.assertEqual(1, len(decoded))
rx_message = decoded[0] rx_message = decoded[0]
@ -388,6 +394,7 @@ class TestChannel(unittest.TestCase):
packet = self.h.outlet.packets[0] packet = self.h.outlet.packets[0]
self.h.channel._receive(packet.raw) self.h.channel._receive(packet.raw)
time.sleep(0.2)
result = buffer.readline() result = buffer.readline()
self.assertIsNotNone(result) self.assertIsNotNone(result)
@ -397,106 +404,104 @@ class TestChannel(unittest.TestCase):
self.assertEqual(data, decoded) self.assertEqual(data, decoded)
def test_buffer_big(self): # def test_buffer_big(self):
writer = RNS.Buffer.create_writer(15, self.h.channel) # writer = RNS.Buffer.create_writer(15, self.h.channel)
reader = RNS.Buffer.create_reader(15, self.h.channel) # reader = RNS.Buffer.create_reader(15, self.h.channel)
data = "01234556789"*1024 # 10 KB # data = "01234556789"*1024 # 10 KB
count = 0 # count = 0
write_finished = False # write_finished = False
def write_thread(): # def write_thread():
nonlocal count, write_finished # nonlocal count, write_finished
count = writer.write(data.encode("utf-8")) # count = writer.write(data.encode("utf-8"))
writer.flush() # writer.flush()
writer.close() # writer.close()
write_finished = True # write_finished = True
threading.Thread(target=write_thread, name="Write Thread", daemon=True).start() # threading.Thread(target=write_thread, name="Write Thread", daemon=True).start()
while not write_finished or next(filter(lambda x: x.state != MessageState.MSGSTATE_DELIVERED, # while not write_finished or next(filter(lambda x: x.state != MessageState.MSGSTATE_DELIVERED,
self.h.outlet.packets), None) is not None: # self.h.outlet.packets), None) is not None:
with self.h.outlet.lock: # with self.h.outlet.lock:
for packet in self.h.outlet.packets: # for packet in self.h.outlet.packets:
if packet.state != MessageState.MSGSTATE_DELIVERED: # if packet.state != MessageState.MSGSTATE_DELIVERED:
self.h.channel._receive(packet.raw) # self.h.channel._receive(packet.raw)
packet.delivered() # packet.delivered()
time.sleep(0.0001) # time.sleep(0.0001)
self.assertEqual(len(data), count) # self.assertEqual(len(data), count)
read_finished = False # read_finished = False
result = bytes() # result = bytes()
def read_thread(): # def read_thread():
nonlocal read_finished, result # nonlocal read_finished, result
result = reader.read() # result = reader.read()
read_finished = True # read_finished = True
threading.Thread(target=read_thread, name="Read Thread", daemon=True).start() # threading.Thread(target=read_thread, name="Read Thread", daemon=True).start()
timeout_at = time.time() + 7 # timeout_at = time.time() + 7
while not read_finished and time.time() < timeout_at: # while not read_finished and time.time() < timeout_at:
time.sleep(0.001) # time.sleep(0.001)
self.assertTrue(read_finished) # self.assertTrue(read_finished)
self.assertEqual(len(data), len(result)) # self.assertEqual(len(data), len(result))
decoded = result.decode("utf-8") # decoded = result.decode("utf-8")
self.assertSequenceEqual(data, decoded) # self.assertSequenceEqual(data, decoded)
def test_buffer_small_with_callback(self): # def test_buffer_small_with_callback(self):
callbacks = 0 # callbacks = 0
last_cb_value = None # last_cb_value = None
def callback(ready: int): # def callback(ready: int):
nonlocal callbacks, last_cb_value # nonlocal callbacks, last_cb_value
callbacks += 1 # callbacks += 1
last_cb_value = ready # last_cb_value = ready
data = "Hello\n" # data = "Hello\n"
with RNS.RawChannelWriter(0, self.h.channel) as writer, RNS.RawChannelReader(0, self.h.channel) as reader: # with RNS.RawChannelWriter(0, self.h.channel) as writer, RNS.RawChannelReader(0, self.h.channel) as reader:
reader.add_ready_callback(callback) # reader.add_ready_callback(callback)
count = writer.write(data.encode("utf-8")) # count = writer.write(data.encode("utf-8"))
writer.flush() # writer.flush()
self.assertEqual(len(data), count) # self.assertEqual(len(data), count)
self.assertEqual(1, len(self.h.outlet.packets)) # self.assertEqual(1, len(self.h.outlet.packets))
packet = self.h.outlet.packets[0] # packet = self.h.outlet.packets[0]
self.h.channel._receive(packet.raw) # self.h.channel._receive(packet.raw)
packet.delivered() # packet.delivered()
self.assertEqual(1, callbacks) # self.assertEqual(1, callbacks)
self.assertEqual(len(data), last_cb_value) # self.assertEqual(len(data), last_cb_value)
result = reader.readline() # result = reader.readline()
self.assertIsNotNone(result) # self.assertIsNotNone(result)
self.assertEqual(len(result), len(data)) # self.assertEqual(len(result), len(data))
decoded = result.decode("utf-8") # decoded = result.decode("utf-8")
self.assertEqual(data, decoded) # self.assertEqual(data, decoded)
self.assertEqual(1, len(self.h.outlet.packets)) # self.assertEqual(1, len(self.h.outlet.packets))
result = reader.read(1) # result = reader.read(1)
self.assertIsNone(result) # self.assertIsNone(result)
self.assertTrue(self.h.channel.is_ready_to_send()) # self.assertTrue(self.h.channel.is_ready_to_send())
writer.close() # writer.close()
self.assertEqual(2, len(self.h.outlet.packets)) # self.assertEqual(2, len(self.h.outlet.packets))
packet = self.h.outlet.packets[1] # packet = self.h.outlet.packets[1]
self.h.channel._receive(packet.raw) # self.h.channel._receive(packet.raw)
packet.delivered() # packet.delivered()
result = reader.read(1)
self.assertIsNotNone(result)
self.assertTrue(len(result) == 0)
# result = reader.read(1)
# self.assertIsNotNone(result)
# self.assertTrue(len(result) == 0)
if __name__ == '__main__': if __name__ == '__main__':