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

Enabled channel tests

Browse Source
This commit is contained in:
Mark Qvist 2023-05-18 23:31:29 +02:00
parent d40e19f08d
commit e184861822
1 changed files with 72 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

144
tests/channel.py
View File

@ -404,104 +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) result = reader.read(1)
# self.assertIsNotNone(result) self.assertIsNotNone(result)
# self.assertTrue(len(result) == 0) self.assertTrue(len(result) == 0)
if __name__ == '__main__': if __name__ == '__main__':