""" Tests using API. """ import base64 import json import time from binascii import hexlify from six.moves import xmlrpc_client # nosec import psutil from .samples import ( sample_deterministic_addr3, sample_deterministic_addr4, sample_seed, sample_inbox_msg_ids, sample_subscription_addresses, sample_subscription_name ) from .test_process import TestProcessProto class TestAPIProto(TestProcessProto): """Test case logic for testing API""" _process_cmd = ['pybitmessage', '-t'] @classmethod def setUpClass(cls): """Setup XMLRPC proxy for pybitmessage API""" super(TestAPIProto, cls).setUpClass() cls.addresses = [] cls.api = xmlrpc_client.ServerProxy( "http://username:password@127.0.0.1:8442/") for _ in range(5): if cls._get_readline('.api_started'): return time.sleep(1) class TestAPIShutdown(TestAPIProto): """Separate test case for API command 'shutdown'""" def test_shutdown(self): """Shutdown the pybitmessage""" self.assertEqual(self.api.shutdown(), 'done') try: self.process.wait(20) except psutil.TimeoutExpired: self.fail( '%s has not stopped in 20 sec' % ' '.join(self._process_cmd)) # TODO: uncovered API commands # disseminatePreEncryptedMsg # disseminatePubkey # getMessageDataByDestinationHash class TestAPI(TestAPIProto): """Main API test case""" _seed = base64.encodestring(sample_seed) def _add_random_address(self, label): addr = self.api.createRandomAddress(base64.encodestring(label)) return addr def test_user_password(self): """Trying to connect with wrong username/password""" api_wrong = xmlrpc_client.ServerProxy( "http://test:wrong@127.0.0.1:8442/") with self.assertRaises(xmlrpc_client.ProtocolError): api_wrong.clientStatus() def test_connection(self): """API command 'helloWorld'""" self.assertEqual( self.api.helloWorld('hello', 'world'), 'hello-world' ) def test_arithmetic(self): """API command 'add'""" self.assertEqual(self.api.add(69, 42), 111) def test_invalid_method(self): """Issuing nonexistent command 'test'""" self.assertEqual( self.api.test(), 'API Error 0020: Invalid method: test' ) def test_message_inbox(self): """Test message inbox methods""" self.assertEqual( len(json.loads( self.api.getAllInboxMessages())["inboxMessages"]), 4, # Custom AssertError message for details json.loads(self.api.getAllInboxMessages())["inboxMessages"] ) self.assertEqual( len(json.loads( self.api.getAllInboxMessageIds())["inboxMessageIds"]), 4 ) self.assertEqual( len(json.loads( self.api.getInboxMessageById( hexlify(sample_inbox_msg_ids[2])))["inboxMessage"]), 1 ) self.assertEqual( len(json.loads( self.api.getInboxMessagesByReceiver( sample_deterministic_addr4))["inboxMessages"]), 4 ) def test_message_trash(self): """Test message inbox methods""" messages_before_delete = len( json.loads(self.api.getAllInboxMessageIds())["inboxMessageIds"]) for msgid in sample_inbox_msg_ids[:2]: self.assertEqual( self.api.trashMessage(hexlify(msgid)), 'Trashed message (assuming message existed).' ) self.assertEqual(len( json.loads(self.api.getAllInboxMessageIds())["inboxMessageIds"] ), messages_before_delete - 2) for msgid in sample_inbox_msg_ids[:2]: self.assertEqual( self.api.undeleteMessage(hexlify(msgid)), 'Undeleted message' ) self.assertEqual(len( json.loads(self.api.getAllInboxMessageIds())["inboxMessageIds"] ), messages_before_delete) def test_clientstatus_consistency(self): """If networkStatus is notConnected networkConnections should be 0""" status = json.loads(self.api.clientStatus()) if status["networkStatus"] == "notConnected": self.assertEqual(status["networkConnections"], 0) else: self.assertGreater(status["networkConnections"], 0) def test_listconnections_consistency(self): """Checking the return of API command 'listConnections'""" result = json.loads(self.api.listConnections()) self.assertGreaterEqual(len(result["inbound"]), 0) self.assertGreaterEqual(len(result["outbound"]), 0) def test_list_addresses(self): """Checking the return of API command 'listAddresses'""" self.assertEqual( json.loads(self.api.listAddresses()).get('addresses'), self.addresses ) def test_decode_address(self): """Checking the return of API command 'decodeAddress'""" result = json.loads( self.api.decodeAddress(sample_deterministic_addr4)) self.assertEqual(result.get('status'), 'success') self.assertEqual(result['addressVersion'], 4) self.assertEqual(result['streamNumber'], 1) def test_create_deterministic_addresses(self): """Test creation of deterministic addresses""" self.assertEqual( self.api.getDeterministicAddress(self._seed, 4, 1), sample_deterministic_addr4) self.assertEqual( self.api.getDeterministicAddress(self._seed, 3, 1), sample_deterministic_addr3) self.assertRegexpMatches( self.api.getDeterministicAddress(self._seed, 2, 1), r'^API Error 0002:') # This is here until the streams will be implemented self.assertRegexpMatches( self.api.getDeterministicAddress(self._seed, 3, 2), r'API Error 0003:') self.assertRegexpMatches( self.api.createDeterministicAddresses(self._seed, 1, 4, 2), r'API Error 0003:') self.assertRegexpMatches( self.api.createDeterministicAddresses('', 1), r'API Error 0001:') self.assertRegexpMatches( self.api.createDeterministicAddresses(self._seed, 1, 2), r'API Error 0002:') self.assertRegexpMatches( self.api.createDeterministicAddresses(self._seed, 0), r'API Error 0004:') self.assertRegexpMatches( self.api.createDeterministicAddresses(self._seed, 1000), r'API Error 0005:') addresses = json.loads( self.api.createDeterministicAddresses(self._seed, 2, 4) )['addresses'] self.assertEqual(len(addresses), 2) self.assertEqual(addresses[0], sample_deterministic_addr4) for addr in addresses: self.assertEqual(self.api.deleteAddress(addr), 'success') def test_create_random_address(self): """API command 'createRandomAddress': basic BM-address validation""" addr = self._add_random_address('random_1') self.assertRegexpMatches(addr, r'^BM-') self.assertRegexpMatches(addr[3:], r'[a-zA-Z1-9]+$') # Whitepaper says "around 36 character" self.assertLessEqual(len(addr[3:]), 40) self.assertEqual(self.api.deleteAddress(addr), 'success') def test_addressbook(self): """Testing API commands for addressbook manipulations""" # Initially it's empty self.assertEqual( json.loads(self.api.listAddressBookEntries()).get('addresses'), [] ) # Add known address self.api.addAddressBookEntry( sample_deterministic_addr4, base64.encodestring('tiger_4') ) # Check addressbook entry entries = json.loads( self.api.listAddressBookEntries()).get('addresses')[0] self.assertEqual( entries['address'], sample_deterministic_addr4) self.assertEqual( base64.decodestring(entries['label']), 'tiger_4') # Try sending to this address (#1898) addr = self._add_random_address('random_2') # TODO: it was never deleted msg = base64.encodestring('test message') msg_subject = base64.encodestring('test_subject') result = self.api.sendMessage( sample_deterministic_addr4, addr, msg_subject, msg) self.assertNotRegexpMatches(result, r'^API Error') self.api.deleteAddress(addr) # Remove known address self.api.deleteAddressBookEntry(sample_deterministic_addr4) # Addressbook should be empty again self.assertEqual( json.loads(self.api.listAddressBookEntries()).get('addresses'), [] ) def test_subscriptions(self): """Testing the API commands related to subscriptions""" self.assertEqual( self.api.addSubscription( sample_subscription_addresses[0], sample_subscription_name.encode('base64')), 'Added subscription.' ) added_subscription = {'label': None, 'enabled': False} # check_address for sub in json.loads(self.api.listSubscriptions())['subscriptions']: # special address, added when sqlThread starts if sub['address'] == sample_subscription_addresses[0]: added_subscription = sub self.assertEqual( base64.decodestring(sub['label']), sample_subscription_name ) self.assertTrue(sub['enabled']) break self.assertEqual( base64.decodestring(added_subscription['label']) if added_subscription['label'] else None, sample_subscription_name) self.assertTrue(added_subscription['enabled']) for s in json.loads(self.api.listSubscriptions())['subscriptions']: # special address, added when sqlThread starts if s['address'] == sample_subscription_addresses[1]: self.assertEqual( base64.decodestring(s['label']), 'Bitmessage new releases/announcements') self.assertTrue(s['enabled']) break else: self.fail( 'Could not find Bitmessage new releases/announcements' ' in subscriptions') self.assertEqual( self.api.deleteSubscription(sample_subscription_addresses[0]), 'Deleted subscription if it existed.') self.assertEqual( self.api.deleteSubscription(sample_subscription_addresses[1]), 'Deleted subscription if it existed.') self.assertEqual( json.loads(self.api.listSubscriptions())['subscriptions'], []) def test_send(self): """Test message sending""" addr = self._add_random_address('random_2') msg = base64.encodestring('test message') msg_subject = base64.encodestring('test_subject') ackdata = self.api.sendMessage( sample_deterministic_addr4, addr, msg_subject, msg) try: # Check ackdata and message status int(ackdata, 16) status = self.api.getStatus(ackdata) if status == 'notfound': raise KeyError self.assertIn( status, ( 'msgqueued', 'awaitingpubkey', 'msgsent', 'ackreceived', 'doingpubkeypow', 'doingmsgpow', 'msgsentnoackexpected' )) # Find the message in sent for m in json.loads( self.api.getSentMessagesByAddress(addr))['sentMessages']: if m['ackData'] == ackdata: sent_msg = m['message'] break else: raise KeyError except ValueError: self.fail('sendMessage returned error or ackData is not hex') except KeyError: self.fail('Could not find sent message in sent messages') else: # Check found message try: self.assertEqual(sent_msg, msg.strip()) except UnboundLocalError: self.fail('Could not find sent message in sent messages') # self.assertEqual(inbox_msg, msg.strip()) self.assertEqual(json.loads( self.api.getSentMessageByAckData(ackdata) )['sentMessage'][0]['message'], sent_msg) # Trash the message self.assertEqual( self.api.trashSentMessageByAckData(ackdata), 'Trashed sent message (assuming message existed).') # Empty trash self.assertEqual(self.api.deleteAndVacuum(), 'done') # The message should disappear self.assertIsNone(json.loads( self.api.getSentMessageByAckData(ackdata))) finally: self.assertEqual(self.api.deleteAddress(addr), 'success') def test_send_broadcast(self): """Test broadcast sending""" addr = self._add_random_address('random_2') msg = base64.encodestring('test broadcast') ackdata = self.api.sendBroadcast( addr, base64.encodestring('test_subject'), msg) try: int(ackdata, 16) status = self.api.getStatus(ackdata) if status == 'notfound': raise KeyError self.assertIn(status, ( 'doingbroadcastpow', 'broadcastqueued', 'broadcastsent')) start = time.time() while status != 'broadcastsent': spent = int(time.time() - start) if spent > 30: self.fail('PoW is taking too much time: %ss' % spent) time.sleep(1) # wait for PoW to get final msgid on next step status = self.api.getStatus(ackdata) # Find the message and its ID in sent for m in json.loads(self.api.getAllSentMessages())['sentMessages']: if m['ackData'] == ackdata: sent_msg = m['message'] sent_msgid = m['msgid'] break else: raise KeyError except ValueError: self.fail('sendBroadcast returned error or ackData is not hex') except KeyError: self.fail('Could not find sent broadcast in sent messages') else: # Check found message and its ID try: self.assertEqual(sent_msg, msg.strip()) except UnboundLocalError: self.fail('Could not find sent message in sent messages') self.assertEqual(json.loads( self.api.getSentMessageById(sent_msgid) )['sentMessage'][0]['message'], sent_msg) self.assertIn( {'msgid': sent_msgid}, json.loads( self.api.getAllSentMessageIds())['sentMessageIds']) # Trash the message by ID self.assertEqual( self.api.trashSentMessage(sent_msgid), 'Trashed sent message (assuming message existed).') self.assertEqual(self.api.deleteAndVacuum(), 'done') self.assertIsNone(json.loads( self.api.getSentMessageById(sent_msgid))) # Try sending from disabled address self.assertEqual(self.api.enableAddress(addr, False), 'success') result = self.api.sendBroadcast( addr, base64.encodestring('test_subject'), msg) self.assertRegexpMatches(result, r'^API Error 0014:') finally: self.assertEqual(self.api.deleteAddress(addr), 'success') # sending from an address without private key # (Bitmessage new releases/announcements) result = self.api.sendBroadcast( 'BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw', base64.encodestring('test_subject'), msg) self.assertRegexpMatches(result, r'^API Error 0013:') def test_chan(self): """Testing chan creation/joining""" # Create chan with known address self.assertEqual( self.api.createChan(self._seed), sample_deterministic_addr4) # cleanup self.assertEqual( self.api.leaveChan(sample_deterministic_addr4), 'success') # Join chan with addresses of version 3 or 4 for addr in (sample_deterministic_addr4, sample_deterministic_addr3): self.assertEqual(self.api.joinChan(self._seed, addr), 'success') self.assertEqual(self.api.leaveChan(addr), 'success') # Joining with wrong address should fail self.assertRegexpMatches( self.api.joinChan(self._seed, 'BM-2cWzSnwjJ7yRP3nLEW'), r'^API Error 0008:' )