test_api.py
1 """ 2 Tests using API. 3 """ 4 5 import base64 6 import json 7 import time 8 from binascii import hexlify 9 10 import psutil 11 import six 12 from six.moves import xmlrpc_client # nosec 13 14 from .samples import ( 15 sample_deterministic_addr3, sample_deterministic_addr4, sample_seed, 16 sample_inbox_msg_ids, 17 sample_subscription_addresses, sample_subscription_name 18 ) 19 20 from .test_process import TestProcessProto 21 22 23 class TestAPIProto(TestProcessProto): 24 """Test case logic for testing API""" 25 _process_cmd = ['pybitmessage', '-t'] 26 27 @classmethod 28 def setUpClass(cls): 29 """Setup XMLRPC proxy for pybitmessage API""" 30 super(TestAPIProto, cls).setUpClass() 31 cls.addresses = [] 32 cls.api = xmlrpc_client.ServerProxy( 33 "http://username:password@127.0.0.1:8442/") 34 for _ in range(5): 35 if cls._get_readline('.api_started'): 36 return 37 time.sleep(1) 38 39 40 class TestAPIShutdown(TestAPIProto): 41 """Separate test case for API command 'shutdown'""" 42 def test_shutdown(self): 43 """Shutdown the pybitmessage""" 44 self.assertEqual(self.api.shutdown(), 'done') 45 try: 46 self.process.wait(20) 47 except psutil.TimeoutExpired: 48 self.fail( 49 '%s has not stopped in 20 sec' % ' '.join(self._process_cmd)) 50 51 52 # TODO: uncovered API commands 53 # disseminatePreEncryptedMsg 54 # disseminatePubkey 55 # getMessageDataByDestinationHash 56 57 58 class TestAPI(TestAPIProto): 59 """Main API test case""" 60 _seed = base64.encodestring(sample_seed) 61 62 def _add_random_address(self, label): 63 addr = self.api.createRandomAddress(base64.encodestring(label)) 64 return addr 65 66 def test_user_password(self): 67 """Trying to connect with wrong username/password""" 68 api_wrong = xmlrpc_client.ServerProxy( 69 "http://test:wrong@127.0.0.1:8442/") 70 with self.assertRaises(xmlrpc_client.ProtocolError): 71 api_wrong.clientStatus() 72 73 def test_connection(self): 74 """API command 'helloWorld'""" 75 self.assertEqual( 76 self.api.helloWorld('hello', 'world'), 77 'hello-world' 78 ) 79 80 def test_arithmetic(self): 81 """API command 'add'""" 82 self.assertEqual(self.api.add(69, 42), 111) 83 84 def test_invalid_method(self): 85 """Issuing nonexistent command 'test'""" 86 self.assertEqual( 87 self.api.test(), 88 'API Error 0020: Invalid method: test' 89 ) 90 91 def test_message_inbox(self): 92 """Test message inbox methods""" 93 self.assertEqual( 94 len(json.loads( 95 self.api.getAllInboxMessages())["inboxMessages"]), 96 4, 97 # Custom AssertError message for details 98 json.loads(self.api.getAllInboxMessages())["inboxMessages"] 99 ) 100 self.assertEqual( 101 len(json.loads( 102 self.api.getAllInboxMessageIds())["inboxMessageIds"]), 103 4 104 ) 105 self.assertEqual( 106 len(json.loads( 107 self.api.getInboxMessageById( 108 hexlify(sample_inbox_msg_ids[2])))["inboxMessage"]), 109 1 110 ) 111 self.assertEqual( 112 len(json.loads( 113 self.api.getInboxMessagesByReceiver( 114 sample_deterministic_addr4))["inboxMessages"]), 115 4 116 ) 117 118 def test_message_trash(self): 119 """Test message inbox methods""" 120 121 messages_before_delete = len( 122 json.loads(self.api.getAllInboxMessageIds())["inboxMessageIds"]) 123 for msgid in sample_inbox_msg_ids[:2]: 124 self.assertEqual( 125 self.api.trashMessage(hexlify(msgid)), 126 'Trashed message (assuming message existed).' 127 ) 128 self.assertEqual(len( 129 json.loads(self.api.getAllInboxMessageIds())["inboxMessageIds"] 130 ), messages_before_delete - 2) 131 for msgid in sample_inbox_msg_ids[:2]: 132 self.assertEqual( 133 self.api.undeleteMessage(hexlify(msgid)), 134 'Undeleted message' 135 ) 136 self.assertEqual(len( 137 json.loads(self.api.getAllInboxMessageIds())["inboxMessageIds"] 138 ), messages_before_delete) 139 140 def test_clientstatus_consistency(self): 141 """If networkStatus is notConnected networkConnections should be 0""" 142 status = json.loads(self.api.clientStatus()) 143 if status["networkStatus"] == "notConnected": 144 self.assertEqual(status["networkConnections"], 0) 145 else: 146 self.assertGreater(status["networkConnections"], 0) 147 148 def test_listconnections_consistency(self): 149 """Checking the return of API command 'listConnections'""" 150 result = json.loads(self.api.listConnections()) 151 self.assertGreaterEqual(len(result["inbound"]), 0) 152 self.assertGreaterEqual(len(result["outbound"]), 0) 153 154 def test_list_addresses(self): 155 """Checking the return of API command 'listAddresses'""" 156 self.assertEqual( 157 json.loads(self.api.listAddresses()).get('addresses'), 158 self.addresses 159 ) 160 161 def test_decode_address(self): 162 """Checking the return of API command 'decodeAddress'""" 163 result = json.loads( 164 self.api.decodeAddress(sample_deterministic_addr4)) 165 self.assertEqual(result.get('status'), 'success') 166 self.assertEqual(result['addressVersion'], 4) 167 self.assertEqual(result['streamNumber'], 1) 168 169 def test_create_deterministic_addresses(self): 170 """Test creation of deterministic addresses""" 171 self.assertEqual( 172 self.api.getDeterministicAddress(self._seed, 4, 1), 173 sample_deterministic_addr4) 174 self.assertEqual( 175 self.api.getDeterministicAddress(self._seed, 3, 1), 176 sample_deterministic_addr3) 177 six.assertRegex( 178 self, self.api.getDeterministicAddress(self._seed, 2, 1), 179 r'^API Error 0002:') 180 181 # This is here until the streams will be implemented 182 six.assertRegex( 183 self, self.api.getDeterministicAddress(self._seed, 3, 2), 184 r'API Error 0003:') 185 six.assertRegex( 186 self, self.api.createDeterministicAddresses(self._seed, 1, 4, 2), 187 r'API Error 0003:') 188 189 six.assertRegex( 190 self, self.api.createDeterministicAddresses('', 1), 191 r'API Error 0001:') 192 six.assertRegex( 193 self, self.api.createDeterministicAddresses(self._seed, 1, 2), 194 r'API Error 0002:') 195 six.assertRegex( 196 self, self.api.createDeterministicAddresses(self._seed, 0), 197 r'API Error 0004:') 198 six.assertRegex( 199 self, self.api.createDeterministicAddresses(self._seed, 1000), 200 r'API Error 0005:') 201 202 addresses = json.loads( 203 self.api.createDeterministicAddresses(self._seed, 2, 4) 204 )['addresses'] 205 self.assertEqual(len(addresses), 2) 206 self.assertEqual(addresses[0], sample_deterministic_addr4) 207 for addr in addresses: 208 self.assertEqual(self.api.deleteAddress(addr), 'success') 209 210 def test_create_random_address(self): 211 """API command 'createRandomAddress': basic BM-address validation""" 212 addr = self._add_random_address('random_1') 213 six.assertRegex(self, addr, r'^BM-') 214 six.assertRegex(self, addr[3:], r'[a-zA-Z1-9]+$') 215 # Whitepaper says "around 36 character" 216 self.assertLessEqual(len(addr[3:]), 40) 217 self.assertEqual(self.api.deleteAddress(addr), 'success') 218 219 def test_addressbook(self): 220 """Testing API commands for addressbook manipulations""" 221 # Initially it's empty 222 self.assertEqual( 223 json.loads(self.api.listAddressBookEntries()).get('addresses'), 224 [] 225 ) 226 # Add known address 227 self.api.addAddressBookEntry( 228 sample_deterministic_addr4, 229 base64.encodestring('tiger_4') 230 ) 231 # Check addressbook entry 232 entries = json.loads( 233 self.api.listAddressBookEntries()).get('addresses')[0] 234 self.assertEqual( 235 entries['address'], sample_deterministic_addr4) 236 self.assertEqual( 237 base64.decodestring(entries['label']), 'tiger_4') 238 # Try sending to this address (#1898) 239 addr = self._add_random_address('random_2') 240 # TODO: it was never deleted 241 msg = base64.encodestring('test message') 242 msg_subject = base64.encodestring('test_subject') 243 result = self.api.sendMessage( 244 sample_deterministic_addr4, addr, msg_subject, msg) 245 self.assertNotRegexpMatches(result, r'^API Error') 246 self.api.deleteAddress(addr) 247 # Remove known address 248 self.api.deleteAddressBookEntry(sample_deterministic_addr4) 249 # Addressbook should be empty again 250 self.assertEqual( 251 json.loads(self.api.listAddressBookEntries()).get('addresses'), 252 [] 253 ) 254 255 def test_subscriptions(self): 256 """Testing the API commands related to subscriptions""" 257 258 self.assertEqual( 259 self.api.addSubscription( 260 sample_subscription_addresses[0], 261 sample_subscription_name.encode('base64')), 262 'Added subscription.' 263 ) 264 265 added_subscription = {'label': None, 'enabled': False} 266 # check_address 267 for sub in json.loads(self.api.listSubscriptions())['subscriptions']: 268 # special address, added when sqlThread starts 269 if sub['address'] == sample_subscription_addresses[0]: 270 added_subscription = sub 271 self.assertEqual( 272 base64.decodestring(sub['label']), sample_subscription_name 273 ) 274 self.assertTrue(sub['enabled']) 275 break 276 277 self.assertEqual( 278 base64.decodestring(added_subscription['label']) 279 if added_subscription['label'] else None, 280 sample_subscription_name) 281 self.assertTrue(added_subscription['enabled']) 282 283 for s in json.loads(self.api.listSubscriptions())['subscriptions']: 284 # special address, added when sqlThread starts 285 if s['address'] == sample_subscription_addresses[1]: 286 self.assertEqual( 287 base64.decodestring(s['label']), 288 'Bitmessage new releases/announcements') 289 self.assertTrue(s['enabled']) 290 break 291 else: 292 self.fail( 293 'Could not find Bitmessage new releases/announcements' 294 ' in subscriptions') 295 self.assertEqual( 296 self.api.deleteSubscription(sample_subscription_addresses[0]), 297 'Deleted subscription if it existed.') 298 self.assertEqual( 299 self.api.deleteSubscription(sample_subscription_addresses[1]), 300 'Deleted subscription if it existed.') 301 self.assertEqual( 302 json.loads(self.api.listSubscriptions())['subscriptions'], []) 303 304 def test_blackwhitelist(self): 305 """Test API commands for managing the black/white list""" 306 # Initially it's black 307 self.assertEqual(self.api.getBlackWhitelistKind(), 'black') 308 # Initially they are empty 309 self.assertEqual( 310 json.loads(self.api.listBlacklistEntries()).get('addresses'), []) 311 self.assertEqual( 312 json.loads(self.api.listWhitelistEntries()).get('addresses'), []) 313 314 # For the Blacklist: 315 # Add known address 316 self.api.addBlacklistEntry( 317 sample_deterministic_addr4, 318 base64.encodestring('tiger_4') 319 ) 320 # Check list entry 321 entry = json.loads(self.api.listBlacklistEntries()).get('addresses')[0] 322 self.assertEqual(entry['address'], sample_deterministic_addr4) 323 self.assertEqual(base64.decodestring(entry['label']), 'tiger_4') 324 # Remove known address 325 self.api.deleteBlacklistEntry(sample_deterministic_addr4) 326 self.assertEqual( 327 json.loads(self.api.listBlacklistEntries()).get('addresses'), []) 328 329 # Only two kinds - black and white 330 six.assertRegex( 331 self, self.api.setBlackWhitelistKind('yellow'), 332 r'^API Error 0028:') 333 # Change kind 334 self.api.setBlackWhitelistKind('white') 335 self.assertEqual(self.api.getBlackWhitelistKind(), 'white') 336 337 # For the Whitelist: 338 # Add known address 339 self.api.addWhitelistEntry( 340 sample_deterministic_addr4, 341 base64.encodestring('tiger_4') 342 ) 343 # Check list entry 344 entry = json.loads(self.api.listWhitelistEntries()).get('addresses')[0] 345 self.assertEqual(entry['address'], sample_deterministic_addr4) 346 self.assertEqual(base64.decodestring(entry['label']), 'tiger_4') 347 # Remove known address 348 self.api.deleteWhitelistEntry(sample_deterministic_addr4) 349 self.assertEqual( 350 json.loads(self.api.listWhitelistEntries()).get('addresses'), []) 351 352 def test_send(self): 353 """Test message sending""" 354 addr = self._add_random_address('random_2') 355 msg = base64.encodestring('test message') 356 msg_subject = base64.encodestring('test_subject') 357 ackdata = self.api.sendMessage( 358 sample_deterministic_addr4, addr, msg_subject, msg) 359 try: 360 # Check ackdata and message status 361 int(ackdata, 16) 362 status = self.api.getStatus(ackdata) 363 if status == 'notfound': 364 raise KeyError 365 self.assertIn( 366 status, ( 367 'msgqueued', 'awaitingpubkey', 'msgsent', 'ackreceived', 368 'doingpubkeypow', 'doingmsgpow', 'msgsentnoackexpected' 369 )) 370 # Find the message in sent 371 for m in json.loads( 372 self.api.getSentMessagesByAddress(addr))['sentMessages']: 373 if m['ackData'] == ackdata: 374 sent_msg = m['message'] 375 break 376 else: 377 raise KeyError 378 except ValueError: 379 self.fail('sendMessage returned error or ackData is not hex') 380 except KeyError: 381 self.fail('Could not find sent message in sent messages') 382 else: 383 # Check found message 384 try: 385 self.assertEqual(sent_msg, msg.strip()) 386 except UnboundLocalError: 387 self.fail('Could not find sent message in sent messages') 388 # self.assertEqual(inbox_msg, msg.strip()) 389 self.assertEqual(json.loads( 390 self.api.getSentMessageByAckData(ackdata) 391 )['sentMessage'][0]['message'], sent_msg) 392 # Trash the message 393 self.assertEqual( 394 self.api.trashSentMessageByAckData(ackdata), 395 'Trashed sent message (assuming message existed).') 396 # Empty trash 397 self.assertEqual(self.api.deleteAndVacuum(), 'done') 398 # The message should disappear 399 self.assertIsNone(json.loads( 400 self.api.getSentMessageByAckData(ackdata))) 401 finally: 402 self.assertEqual(self.api.deleteAddress(addr), 'success') 403 404 def test_send_broadcast(self): 405 """Test broadcast sending""" 406 addr = self._add_random_address('random_2') 407 msg = base64.encodestring('test broadcast') 408 ackdata = self.api.sendBroadcast( 409 addr, base64.encodestring('test_subject'), msg) 410 411 try: 412 int(ackdata, 16) 413 status = self.api.getStatus(ackdata) 414 if status == 'notfound': 415 raise KeyError 416 self.assertIn(status, ( 417 'doingbroadcastpow', 'broadcastqueued', 'broadcastsent')) 418 419 start = time.time() 420 while status != 'broadcastsent': 421 spent = int(time.time() - start) 422 if spent > 30: 423 self.fail('PoW is taking too much time: %ss' % spent) 424 time.sleep(1) # wait for PoW to get final msgid on next step 425 status = self.api.getStatus(ackdata) 426 427 # Find the message and its ID in sent 428 for m in json.loads(self.api.getAllSentMessages())['sentMessages']: 429 if m['ackData'] == ackdata: 430 sent_msg = m['message'] 431 sent_msgid = m['msgid'] 432 break 433 else: 434 raise KeyError 435 except ValueError: 436 self.fail('sendBroadcast returned error or ackData is not hex') 437 except KeyError: 438 self.fail('Could not find sent broadcast in sent messages') 439 else: 440 # Check found message and its ID 441 try: 442 self.assertEqual(sent_msg, msg.strip()) 443 except UnboundLocalError: 444 self.fail('Could not find sent message in sent messages') 445 self.assertEqual(json.loads( 446 self.api.getSentMessageById(sent_msgid) 447 )['sentMessage'][0]['message'], sent_msg) 448 self.assertIn( 449 {'msgid': sent_msgid}, json.loads( 450 self.api.getAllSentMessageIds())['sentMessageIds']) 451 # Trash the message by ID 452 self.assertEqual( 453 self.api.trashSentMessage(sent_msgid), 454 'Trashed sent message (assuming message existed).') 455 self.assertEqual(self.api.deleteAndVacuum(), 'done') 456 self.assertIsNone(json.loads( 457 self.api.getSentMessageById(sent_msgid))) 458 # Try sending from disabled address 459 self.assertEqual(self.api.enableAddress(addr, False), 'success') 460 result = self.api.sendBroadcast( 461 addr, base64.encodestring('test_subject'), msg) 462 six.assertRegex(self, result, r'^API Error 0014:') 463 finally: 464 self.assertEqual(self.api.deleteAddress(addr), 'success') 465 466 # sending from an address without private key 467 # (Bitmessage new releases/announcements) 468 result = self.api.sendBroadcast( 469 'BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw', 470 base64.encodestring('test_subject'), msg) 471 six.assertRegex(self, result, r'^API Error 0013:') 472 473 def test_chan(self): 474 """Testing chan creation/joining""" 475 # Create chan with known address 476 self.assertEqual( 477 self.api.createChan(self._seed), sample_deterministic_addr4) 478 # cleanup 479 self.assertEqual( 480 self.api.leaveChan(sample_deterministic_addr4), 'success') 481 # Join chan with addresses of version 3 or 4 482 for addr in (sample_deterministic_addr4, sample_deterministic_addr3): 483 self.assertEqual(self.api.joinChan(self._seed, addr), 'success') 484 self.assertEqual(self.api.leaveChan(addr), 'success') 485 # Joining with wrong address should fail 486 six.assertRegex( 487 self, self.api.joinChan(self._seed, 'BM-2cWzSnwjJ7yRP3nLEW'), 488 r'^API Error 0008:' 489 )