MiNode/minode/tests/test_structure.py

195 lines
7.7 KiB
Python
Raw Normal View History

2023-08-01 23:15:00 +02:00
"""Tests for structures"""
import base64
import logging
import queue
2021-08-01 18:57:30 +02:00
import struct
import time
import unittest
2021-08-01 18:57:30 +02:00
from binascii import unhexlify
from minode import message, proofofwork, shared, structure
2021-08-01 18:57:30 +02:00
# host pregenerated by pybitmessage.protocol.encodeHost()
# for one of bootstrap servers, port 8080,
# everything else is like in test_message: 1626611891, 1, 1
sample_addr_data = unhexlify(
'0000000060f420b3000000010000000000000001'
'260753000201300000000000000057ae1f90')
# data for an object with expires_time 1697063939
# structure.Object(
# b'\x00' * 8, expires_time, 42, 1, 2, b'HELLO').to_bytes()
sample_object_data = unhexlify(
'000000000000000000000000652724030000002a010248454c4c4f')
logging.basicConfig(
level=shared.log_level,
format='[%(asctime)s] [%(levelname)s] %(message)s')
2021-08-01 18:57:30 +02:00
class TestStructure(unittest.TestCase):
"""Testing structures serializing and deserializing"""
@classmethod
def setUpClass(cls):
shared.objects = {}
2021-08-01 18:57:30 +02:00
def test_varint(self):
"""Test varint serializing and deserializing"""
s = structure.VarInt(0)
self.assertEqual(s.to_bytes(), b'\x00')
s = structure.VarInt.from_bytes(b'\x00')
self.assertEqual(s.n, 0)
s = structure.VarInt(42)
self.assertEqual(s.to_bytes(), b'*')
s = structure.VarInt.from_bytes(b'*')
self.assertEqual(s.n, 42)
s = structure.VarInt(252)
self.assertEqual(s.to_bytes(), unhexlify('fc'))
s = structure.VarInt.from_bytes(unhexlify('fc'))
self.assertEqual(s.n, 252)
s = structure.VarInt(253)
self.assertEqual(s.to_bytes(), unhexlify('fd00fd'))
s = structure.VarInt.from_bytes(unhexlify('fd00fd'))
self.assertEqual(s.n, 253)
s = structure.VarInt(100500)
self.assertEqual(s.to_bytes(), unhexlify('fe00018894'))
s = structure.VarInt.from_bytes(unhexlify('fe00018894'))
self.assertEqual(s.n, 100500)
s = structure.VarInt(65535)
self.assertEqual(s.to_bytes(), unhexlify('fdffff'))
s = structure.VarInt.from_bytes(unhexlify('fdffff'))
self.assertEqual(s.n, 65535)
s = structure.VarInt(4294967295)
self.assertEqual(s.to_bytes(), unhexlify('feffffffff'))
s = structure.VarInt.from_bytes(unhexlify('feffffffff'))
self.assertEqual(s.n, 4294967295)
s = structure.VarInt(4294967296)
self.assertEqual(s.to_bytes(), unhexlify('ff0000000100000000'))
s = structure.VarInt.from_bytes(unhexlify('ff0000000100000000'))
self.assertEqual(s.n, 4294967296)
s = structure.VarInt(18446744073709551615)
self.assertEqual(s.to_bytes(), unhexlify('ffffffffffffffffff'))
s = structure.VarInt.from_bytes(unhexlify('ffffffffffffffffff'))
self.assertEqual(s.n, 18446744073709551615)
def test_address(self):
"""Check address encoding in structure.NetAddrNoPrefix()"""
addr = structure.NetAddrNoPrefix(1, '127.0.0.1', 8444)
self.assertEqual(
addr.to_bytes()[8:24],
b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF'
+ struct.pack('>L', 2130706433))
addr = structure.NetAddrNoPrefix(1, '191.168.1.1', 8444)
self.assertEqual(
addr.to_bytes()[8:24],
unhexlify('00000000000000000000ffffbfa80101'))
addr = structure.NetAddrNoPrefix(1, '1.1.1.1', 8444)
self.assertEqual(
addr.to_bytes()[8:24],
unhexlify('00000000000000000000ffff01010101'))
addr = structure.NetAddrNoPrefix(
1, '0102:0304:0506:0708:090A:0B0C:0D0E:0F10', 8444)
self.assertEqual(
addr.to_bytes()[8:24],
unhexlify('0102030405060708090a0b0c0d0e0f10'))
addr = structure.NetAddr.from_bytes(sample_addr_data)
self.assertEqual(addr.host, '2607:5300:201:3000::57ae')
self.assertEqual(addr.port, 8080)
self.assertEqual(addr.stream, 1)
self.assertEqual(addr.services, 1)
addr = structure.NetAddr(1, '2607:5300:201:3000::57ae', 8080, 1)
self.assertEqual(addr.to_bytes()[8:], sample_addr_data[8:])
def test_network_group(self):
"""Test various types of network groups"""
test_ip = '1.2.3.4'
self.assertEqual(
b'\x01\x02', structure.NetAddrNoPrefix.network_group(test_ip))
self.assertEqual(
structure.NetAddrNoPrefix.network_group('8.8.8.8'),
structure.NetAddrNoPrefix.network_group('8.8.4.4'))
self.assertNotEqual(
structure.NetAddrNoPrefix.network_group('1.1.1.1'),
structure.NetAddrNoPrefix.network_group('8.8.8.8'))
test_ip = '0102:0304:0506:0708:090A:0B0C:0D0E:0F10'
self.assertEqual(
b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C',
structure.NetAddrNoPrefix.network_group(test_ip))
for test_ip in (
'bootstrap8444.bitmessage.org', 'quzwelsuziwqgpt2.onion', None
):
self.assertEqual(
test_ip, structure.NetAddrNoPrefix.network_group(test_ip))
def test_object(self):
"""Create and check objects"""
obj = structure.Object.from_message(
message.Message(b'object', sample_object_data))
self.assertEqual(obj.object_type, 42)
self.assertEqual(obj.stream_number, 2)
self.assertEqual(obj.expires_time, 1697063939)
self.assertEqual(obj.object_payload, b'HELLO')
obj = structure.Object(
int(time.time() + 3000000), 42, 1, 1, object_payload=b'HELLO')
self.assertFalse(obj.is_valid())
obj.expires_time = int(time.time() - 11000)
self.assertFalse(obj.is_valid())
obj = structure.Object(
int(time.time() + 300), 42, 1, 2, object_payload=b'HELLO')
vector = obj.vector
proofofwork._worker(obj) # pylint: disable=protected-access
obj = shared.objects.popitem()[1]
self.assertNotEqual(obj.vector, vector)
self.assertFalse(obj.is_expired())
self.assertFalse(obj.is_valid())
shared.stream = 2
self.assertTrue(obj.is_valid())
obj.object_payload = \
b'TIGER, tiger, burning bright. In the forests of the night'
self.assertFalse(obj.is_valid())
def test_proofofwork(self):
"""Check the main proofofwork call and worker"""
shared.vector_advertise_queue = queue.Queue()
obj = structure.Object(
int(time.time() + 300), 42, 1, shared.stream,
object_payload=b'HELLO')
start_time = time.time()
proofofwork.do_pow_and_publish(obj)
try:
vector = shared.vector_advertise_queue.get(timeout=300)
except queue.Empty:
self.fail("Couldn't make work in 300 sec")
else:
time.sleep(1)
try:
result = shared.objects[vector]
except KeyError:
self.fail(
"Couldn't found object with vector %s"
" %s sec after pow start" % (
base64.b16encode(vector), time.time() - start_time))
self.assertTrue(result.is_valid())
self.assertEqual(result.object_type, 42)
self.assertEqual(result.object_payload, b'HELLO')
q = queue.Queue()
# pylint: disable=protected-access
proofofwork._pow_worker(obj.pow_target(), obj.pow_initial_hash(), q)
try:
nonce = q.get(timeout=5)
except queue.Empty:
self.fail("No nonce found in the queue")
obj = structure.Object(
obj.expires_time, obj.object_type, obj.version, obj.stream_number,
object_payload=obj.object_payload, nonce=nonce)
self.assertTrue(obj.is_valid())