Bitmessage/PyBitmessage-2024-12-03
Archived
2
0
Fork 0
Code Issues 168 Pull Requests 53 Actions Packages Projects Releases 5 Wiki Activity

Test for class object Processor

Browse Source
This commit is contained in:
anand k 2024-05-16 08:12:08 +05:30
parent e571ba8a51
commit 3073ef1e3b
No known key found for this signature in database
GPG Key ID: 515AC24FA525DDE0
3 changed files with 237 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/class_objectProcessor.py
View File

@ -315,7 +315,7 @@ class objectProcessor(threading.Thread):
dataToStore = data[20:readPosition]
sha = hashlib.new('sha512')
sha.update(
'\x04' + publicSigningKey + '\x04' + publicEncryptionKey)
b'\x04' + publicSigningKey + b'\x04' + publicEncryptionKey)
ripe = RIPEMD160Hash(sha.digest()).digest()
if logger.isEnabledFor(logging.DEBUG):
@ -354,9 +354,9 @@ class objectProcessor(threading.Thread):
' Sanity check failed.')
return
readPosition += 4
publicSigningKey = '\x04' + data[readPosition:readPosition + 64]
publicSigningKey = b'\x04' + data[readPosition:readPosition + 64]
readPosition += 64
publicEncryptionKey = '\x04' + data[readPosition:readPosition + 64]
publicEncryptionKey = b'\x04' + data[readPosition:readPosition + 64]
readPosition += 64
specifiedNonceTrialsPerByteLength = decodeVarint(
data[readPosition:readPosition + 10])[1]
@ -513,9 +513,9 @@ class objectProcessor(threading.Thread):
return
readPosition += sendersStreamNumberLength
readPosition += 4
pubSigningKey = '\x04' + decryptedData[readPosition:readPosition + 64]
pubSigningKey = b'\x04' + decryptedData[readPosition:readPosition + 64]
readPosition += 64
pubEncryptionKey = '\x04' + decryptedData[readPosition:readPosition + 64]
pubEncryptionKey = b'\x04' + decryptedData[readPosition:readPosition + 64]
readPosition += 64
if sendersAddressVersionNumber >= 3:
requiredAverageProofOfWorkNonceTrialsPerByte, varintLength = \
@ -862,10 +862,10 @@ class objectProcessor(threading.Thread):
)
readPosition += sendersStreamLength
readPosition += 4
sendersPubSigningKey = '\x04' + \
sendersPubSigningKey = b'\x04' + \
decryptedData[readPosition:readPosition + 64]
readPosition += 64
sendersPubEncryptionKey = '\x04' + \
sendersPubEncryptionKey = b'\x04' + \
decryptedData[readPosition:readPosition + 64]
readPosition += 64
if sendersAddressVersion >= 3:

21
src/mockbm/class_objectProcessor.py Normal file
View File

@ -0,0 +1,21 @@
"""Mock function/class used in objectProcessor"""
def mockFunctionToPass():
pass
def sqlQueryMock(*args):
return [(None, "mockData"),]
def sqlExecuteMock(*args):
return
class SqlReadyMock(object):
"""Mock helper_sql.sql_ready event with dummy class"""
@staticmethod
def wait(time):
"""Don't wait, return immediately"""
return

209
src/tests/test_objectProcessor.py Normal file
View File

@ -0,0 +1,209 @@
"""Test cases for ObjectProcessor"""
from binascii import hexlify
import threading
import state
import pybitmessage.class_objectProcessor as op
try:
from unittest.mock import patch, MagicMock
except ImportError:
from mock import patch, MagicMock, Mock
from .partial import TestPartialRun
from mockbm.class_objectProcessor import (
mockFunctionToPass,
sqlExecuteMock,
sqlQueryMock,
SqlReadyMock,
)
from .samples import (
sample_pubsigningkey,
sample_pubencryptionkey,
sample_privencryptionkey,
sample_deterministic_addr4,
sample_deterministic_addr3,
sample_privsigningkey
)
class TestObjectProcessor(TestPartialRun):
"""Test class for ObjectProcessor"""
@classmethod
def setUpClass(cls):
super(TestObjectProcessor, cls).setUpClass()
op.sqlQuery = sqlQueryMock
op.sqlExecute = sqlExecuteMock
op.sql_ready = SqlReadyMock
op.shared.reloadBroadcastSendersForWhichImWatching = mockFunctionToPass
op.shared.reloadMyAddressHashes = mockFunctionToPass
cls.queues = op.queues
cls.processor = op.objectProcessor()
cls.processor.start()
@classmethod
def tearDownClass(cls):
super(TestObjectProcessor, cls).tearDownClass()
for thread in threading.enumerate():
if thread.name == "objectProcessor":
op.queues.objectProcessorQueue.put(('checkShutdownVariable', 'no data'))
@patch("pybitmessage.class_objectProcessor.logger.debug")
@patch("pybitmessage.class_objectProcessor.queues.UISignalQueue.put")
def test_checkackdata(self, mock_UISignalQueue_put, mock_logger):
"""test checkdata"""
data = b"\x00" * 8 # nonce
data += b"\x00" * 8 # expiresTime
data += b"\x00" * 4 # getpubkey object type
data += b"\x04" # addressVersion
data += b"\x01" # streamNumber
data += b"\00" * 4 # behaviour bitfield
data += sample_pubsigningkey
data += sample_pubencryptionkey
# ackdata not in state.ackdataForWhichImWatching
state.ackdataForWhichImWatching = {}
self.processor.checkackdata(data)
mock_logger.assert_called_with('This object is not an acknowledgement bound for me.')
# ackdata is in state.ackdataForWhichImWatching
state.ackdataForWhichImWatching = {data[16:]: data}
self.processor.checkackdata(data)
self.assertEqual(len(state.ackdataForWhichImWatching), 0)
mock_UISignalQueue_put.assert_called_once()
@patch("pybitmessage.class_objectProcessor.protocol.checkIPAddress")
@patch("pybitmessage.class_objectProcessor.decodeVarint")
@patch("pybitmessage.class_objectProcessor.knownnodes.addKnownNode")
def test_processonion(self, mock_addknownNode, mock_decodeVarint, mock_checkIPAddress):
"""Test processonion"""
data = b"\x00" * 100
# decode varient called 3 times returns -> addressversion & lenth
# stream & length, port & length
mock_decodeVarint.side_effect = [(3, 1), (2, 1), (8080, 4), ]
mock_checkIPAddress.return_value = "127.0.0.1"
self.processor.processonion(data)
mock_addknownNode.assert_called_once()
@patch("pybitmessage.class_objectProcessor.queues.workerQueue.put")
def test_processgetpubkey_V3(
self, mock_workerqueue_put
):
"""Test processgetpukey with address version 3"""
HASH = b"\x01" * 20
gpk = b"\x00" * 8 # nonce
gpk += b"\x00" * 8 # expiresTime
gpk += b"\x00" * 4 # getpubkey object type
gpk += b"\x03" # addressVersion 3
gpk += b"\x01" # streamNumber
gpk += HASH # hash
# set address corresponding to hash
op.shared.myAddressesByHash[HASH] = sample_deterministic_addr3
self.processor.processgetpubkey(gpk)
mock_workerqueue_put.assert_called_once()
@patch("pybitmessage.class_objectProcessor.queues.workerQueue.put")
def test_processgetpubkey_V4(
self, mock_workerqueue_put
):
"""Test processgetpukey with address version 4"""
TAG = b"\x01" * 32
gpk = b"\x00" * 8 # nonce
gpk += b"\x00" * 8 # expiresTime
gpk += b"\x00" * 4 # getpubkey object type
gpk += b"\x04" # addressVersion 4
gpk += b"\x01" # streamNumber
gpk += TAG # tag
# set address corresponding to tag
op.shared.myAddressesByTag[TAG] = sample_deterministic_addr4
self.processor.processgetpubkey(gpk)
mock_workerqueue_put.assert_called_once()
def test_processpubkey_version2(
self,
):
"""Test processpubkey with version 2"""
ppk = b"\x00" * 8 # nonce
ppk += b"\x00" * 8 # expiresTime
ppk += b"\x00\x00\x00\x01" # getpubkey object type
ppk += b"\x02" # addressVersion
ppk += b"\x01" # streamNumber
ppk += b"\00" * 4 # behaviour bitfield
ppk += sample_pubsigningkey
ppk += sample_pubencryptionkey
self.processor.processpubkey(ppk)
def test_processpubkey_version3(
self,
):
"""Test processpubkey with version 3"""
ppk = b"\x00" * 8 # nonce
ppk += b"\x00" * 8 # expiresTime
ppk += b"\x00\x00\x00\x01" # getpubkey object type
ppk += b"\x03" # addressVersion
ppk += b"\x01" # streamNumber
ppk += b"\00" * 4 # behaviour bitfield
ppk += sample_pubsigningkey
ppk += sample_pubencryptionkey
ppk += b"\x01" # nonce_trials_per_byte
ppk += b"\x01" # extra_bytes
signature = op.highlevelcrypto.sign(ppk[8:-2], hexPrivkey=hexlify(sample_privsigningkey)) # signature
ppk += hex(len(signature)).encode() # sig_length
ppk += signature
self.processor.processpubkey(ppk)
def test_processpubkey_version4(
self
):
"""Test processpubkey with version 4"""
Tag = b"\00" * 32
pk = b"\x00" * 8 # nonce
pk += b"\x00" * 8 # expiresTime
pk += b"\x00\x00\x00\x01" # getpubkey object type
pk += b"\x04" # addressVersion
pk += b"\x01" # streamNumber
pk += Tag # tag
data_to_encrypt = b"\00" * 4 # behaviour bitfield
data_to_encrypt += sample_pubsigningkey
data_to_encrypt += sample_pubencryptionkey
data_to_encrypt += b"\x01" # nonce_trials_per_byte
data_to_encrypt += b"\x01" # extra_bytes
signature = op.highlevelcrypto.sign(pk[8:], hexPrivkey=hexlify(sample_privsigningkey)) # signature
data_to_encrypt += hex(len(signature)).encode() # sig_length
data_to_encrypt += signature
cryptor = op.highlevelcrypto.makeCryptor(sample_privencryptionkey)
encrypted_data = cryptor.encrypt(data_to_encrypt, hexlify(sample_pubencryptionkey))
pk += encrypted_data
state.neededPubkeys[Tag] = (sample_deterministic_addr4, cryptor)
self.processor.processpubkey(pk)
@patch("pybitmessage.class_objectProcessor.objectProcessor.sendMessages")
def test_possibleNewPubkey_with_addressV3(self, mock_sendMessages):
"""Test possibleNewPubkey with version 3 address"""
state.neededPubkeys = {sample_deterministic_addr3: "pubkey2"}
self.processor.possibleNewPubkey(sample_deterministic_addr3)
mock_sendMessages.assert_called_with(sample_deterministic_addr3)
@patch("pybitmessage.class_objectProcessor.highlevelcrypto.double_sha512")
@patch("pybitmessage.class_objectProcessor.objectProcessor.sendMessages")
def test_possibleNewPubkey_with_addressV4(self, mock_sendMessages, mock_double_sha512):
"""Test possibleNewPubkey with version 4 address"""
fake_encrypted_data = "\x01" * 64
state.neededPubkeys = {fake_encrypted_data[32:]: "pubkey"}
mock_double_sha512.return_value = fake_encrypted_data
self.processor.possibleNewPubkey(sample_deterministic_addr4)
mock_sendMessages.assert_called_with(sample_deterministic_addr4)