test_api.py

"""
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:'
        )