"""Tests for structures""" import base64 import logging import queue import struct import time import unittest from binascii import unhexlify from minode import message, proofofwork, shared, structure # 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') class TestStructure(unittest.TestCase): """Testing structures serializing and deserializing""" @classmethod def setUpClass(cls): shared.objects = {} 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.data = struct.pack(... # 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())