1  """Tests for AddressGenerator (with thread or not)"""
  2  
  3  from binascii import unhexlify
  4  
  5  import six
  6  from six.moves import queue
  7  
  8  from .partial import TestPartialRun
  9  from .samples import (
 10      sample_seed, sample_deterministic_addr3, sample_deterministic_addr4,
 11      sample_deterministic_ripe)
 12  
 13  TEST_LABEL = 'test'
 14  
 15  
 16  class TestAddressGenerator(TestPartialRun):
 17      """Test case for AddressGenerator thread"""
 18  
 19      @classmethod
 20      def setUpClass(cls):
 21          super(TestAddressGenerator, cls).setUpClass()
 22  
 23          import defaults
 24          import queues
 25          from class_addressGenerator import addressGenerator
 26  
 27          cls.state.enableGUI = False
 28  
 29          cls.command_queue = queues.addressGeneratorQueue
 30          cls.return_queue = queues.apiAddressGeneratorReturnQueue
 31          cls.worker_queue = queues.workerQueue
 32  
 33          cls.config.set(
 34              'bitmessagesettings', 'defaultnoncetrialsperbyte',
 35              str(defaults.networkDefaultProofOfWorkNonceTrialsPerByte))
 36          cls.config.set(
 37              'bitmessagesettings', 'defaultpayloadlengthextrabytes',
 38              str(defaults.networkDefaultPayloadLengthExtraBytes))
 39  
 40          thread = addressGenerator()
 41          thread.daemon = True
 42          thread.start()
 43  
 44      def _execute(self, command, *args):
 45          self.command_queue.put((command,) + args)
 46          try:
 47              return self.return_queue.get(timeout=30)[0]
 48          except (IndexError, queue.Empty):
 49              self.fail('Failed to execute command %s' % command)
 50  
 51      def test_deterministic(self):
 52          """Test deterministic commands"""
 53          self.command_queue.put((
 54              'getDeterministicAddress', 3, 1,
 55              TEST_LABEL, 1, sample_seed, False))
 56          self.assertEqual(sample_deterministic_addr3, self.return_queue.get())
 57  
 58          self.assertEqual(
 59              sample_deterministic_addr3,
 60              self._execute(
 61                  'createDeterministicAddresses', 3, 1, TEST_LABEL, 2,
 62                  sample_seed, False, 0, 0))
 63  
 64          try:
 65              self.assertEqual(
 66                  self.worker_queue.get(timeout=30),
 67                  ('sendOutOrStoreMyV3Pubkey',
 68                   unhexlify(sample_deterministic_ripe)))
 69  
 70              self.worker_queue.get(timeout=30)  # get the next addr's task
 71          except queue.Empty:
 72              self.fail('No commands in the worker queue')
 73  
 74          self.assertEqual(
 75              sample_deterministic_addr4,
 76              self._execute('createChan', 4, 1, TEST_LABEL, sample_seed, True))
 77          try:
 78              self.assertEqual(
 79                  self.worker_queue.get(),
 80                  ('sendOutOrStoreMyV4Pubkey', sample_deterministic_addr4))
 81          except queue.Empty:
 82              self.fail('No commands in the worker queue')
 83          self.assertEqual(
 84              self.config.get(sample_deterministic_addr4, 'label'), TEST_LABEL)
 85          self.assertTrue(
 86              self.config.getboolean(sample_deterministic_addr4, 'chan'))
 87          self.assertTrue(
 88              self.config.getboolean(sample_deterministic_addr4, 'enabled'))
 89  
 90      def test_random(self):
 91          """Test random address"""
 92          self.command_queue.put((
 93              'createRandomAddress', 4, 1, 'test_random', 1, '', False, 0, 0))
 94          addr = self.return_queue.get()
 95          six.assertRegex(self, addr, r'^BM-')
 96          six.assertRegex(self, addr[3:], r'[a-zA-Z1-9]+$')
 97          self.assertLessEqual(len(addr[3:]), 40)
 98  
 99          self.assertEqual(
100              self.worker_queue.get(), ('sendOutOrStoreMyV4Pubkey', addr))
101          self.assertEqual(self.config.get(addr, 'label'), 'test_random')
102          self.assertTrue(self.config.getboolean(addr, 'enabled'))