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

Test more deterministic commands, add timeouts in get from queue

Browse Source
This commit is contained in:
Lee Miller 2023-12-27 04:09:21 +02:00
parent d8e0c52ecd
commit c45f1eae14
Signed by untrusted user: lee.miller
GPG Key ID: 4F97A5EA88F4AB63
1 changed files with 35 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
src/tests/test_addressgenerator.py
View File

@ -2,11 +2,15 @@
from binascii import unhexlify
from six.moves import queue
from .partial import TestPartialRun
from .samples import (
sample_seed, sample_deterministic_addr3, sample_deterministic_addr4,
sample_deterministic_ripe)
TEST_LABEL = 'test'
class TestAddressGenerator(TestPartialRun):
"""Test case for AddressGenerator thread"""
@ -39,26 +43,44 @@ class TestAddressGenerator(TestPartialRun):
def _execute(self, command, *args):
self.command_queue.put((command,) + args)
try:
return self.return_queue.get()[0]
except IndexError:
return self.return_queue.get(timeout=30)[0]
except (IndexError, queue.Empty):
self.fail('Failed to execute command %s' % command)
def test_createChan(self):
"""Test createChan command"""
def test_deterministic(self):
"""Test deterministic commands"""
self.command_queue.put((
'getDeterministicAddress', 3, 1,
TEST_LABEL, 1, sample_seed, False))
self.assertEqual(sample_deterministic_addr3, self.return_queue.get())
self.assertEqual(
sample_deterministic_addr3,
self._execute('createChan', 3, 1, 'test', sample_seed, True))
self._execute(
'createDeterministicAddresses', 3, 1, TEST_LABEL, 2,
sample_seed, False, 0, 0))
try:
self.assertEqual(
self.worker_queue.get(),
('sendOutOrStoreMyV3Pubkey', unhexlify(sample_deterministic_ripe)))
self.worker_queue.get(timeout=30),
('sendOutOrStoreMyV3Pubkey',
unhexlify(sample_deterministic_ripe)))
self.worker_queue.get(timeout=30) # get the next addr's task
except queue.Empty:
self.fail('No commands in the worker queue')
self.assertEqual(
sample_deterministic_addr4,
self._execute('createChan', 4, 1, 'test', sample_seed, True))
self._execute('createChan', 4, 1, TEST_LABEL, sample_seed, True))
try:
self.assertEqual(
self.worker_queue.get(),
('sendOutOrStoreMyV4Pubkey', sample_deterministic_addr4))
except queue.Empty:
self.fail('No commands in the worker queue')
self.assertEqual(
self.config.get(sample_deterministic_addr4, 'label'), 'test')
self.config.get(sample_deterministic_addr4, 'label'), TEST_LABEL)
self.assertTrue(
self.config.getboolean(sample_deterministic_addr4, 'chan'))
self.assertTrue(