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

V0.6 test api #1781

Closed
kdcis wants to merge 8 commits from v0.6-test-api into v0.6
pull from: v0.6-test-api
merge into: Bitmessage:v0.6
Conversation 0 Commits 8 Files Changed 6 +281 -130
8 changed files with 281 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 71b5c87f75 - Show all commits

2
setup.py
View File

@ -168,7 +168,7 @@ if __name__ == "__main__":
'pybitmessage = pybitmessage.bitmessagemain:main' 'pybitmessage = pybitmessage.bitmessagemain:main'
] if sys.platform[:3] == 'win' else [] ] if sys.platform[:3] == 'win' else []
}, },
scripts=['src/pybitmessage'], scripts=['src/pybitmessage3'],
cmdclass={'install': InstallCmd}, cmdclass={'install': InstallCmd},
command_options={ command_options={
'build_sphinx': { 'build_sphinx': {

1
src/helper_random.py
View File

@ -2,7 +2,6 @@
import os import os
import random import random
from pyelliptic.openssl import OpenSSL from pyelliptic.openssl import OpenSSL
NoneType = type(None) NoneType = type(None)

6
src/multiqueue.py
View File

@ -3,6 +3,7 @@ A queue with multiple internal subqueues.
Elements are added into a random subqueue, and retrieval rotates Elements are added into a random subqueue, and retrieval rotates
""" """
import helper_random
import sys import sys
if sys.version_info[0] == 3: if sys.version_info[0] == 3:
import queue as Queue import queue as Queue
@ -11,11 +12,6 @@ else:
from collections import deque from collections import deque
if sys.version_info[0] == 3:
from . import helper_random
else:
import helper_random
class MultiQueue(Queue.Queue): class MultiQueue(Queue.Queue):
"""A base queue class""" """A base queue class"""

6
src/pybitmessage
View File

@ -3,9 +3,13 @@
import os import os
import pkg_resources import pkg_resources
dist = pkg_resources.get_distribution('pybitmessage') dist = pkg_resources.get_distribution('pybitmessage')
script_file = os.path.join(dist.location, dist.key, 'bitmessagemain.py') script_file = os.path.join(dist.location, dist.key, 'bitmessagemain.py')
new_globals = globals() new_globals = globals()
new_globals.update(__file__=script_file) new_globals.update(__file__=script_file)
execfile(script_file, new_globals) try:
execfile(script_file, new_globals)
except NameError:
exec(compile(open(script_file, "rb").read(), script_file, 'exec'), new_globals)

15
src/pybitmessage3 Normal file
View File

@ -0,0 +1,15 @@
#!/usr/bin/python2.7
import os
import pkg_resources
dist = pkg_resources.get_distribution('pybitmessage')
script_file = os.path.join(dist.location, dist.key, 'bitmessagemain.py')
new_globals = globals()
new_globals.update(__file__=script_file)
try:
execfile(script_file, new_globals)
except NameError:
exec(compile(open(script_file, "rb").read(), script_file, 'exec'), new_globals)

360
src/tests/test_api.py
View File

@ -1,3 +1,4 @@
# pylint: disable=E1101
""" """
Tests using API. Tests using API.
""" """
@ -5,9 +6,7 @@ Tests using API.
import base64 import base64
import json import json
import time import time
from .common import skip_python3 import sys
skip_python3()
try: # nosec try: # nosec
from xmlrpclib import ServerProxy, ProtocolError from xmlrpclib import ServerProxy, ProtocolError
@ -16,6 +15,8 @@ except ImportError:
from .test_process import TestProcessProto, TestProcessShutdown from .test_process import TestProcessProto, TestProcessShutdown
PY3 = sys.version_info[0] >= 3
class TestAPIProto(TestProcessProto): class TestAPIProto(TestProcessProto):
"""Test case logic for testing API""" """Test case logic for testing API"""
@ -63,12 +64,18 @@ class TestAPIShutdown(TestAPIProto, TestProcessShutdown):
class TestAPI(TestAPIProto): class TestAPI(TestAPIProto):
"""Main API test case""" """Main API test case"""
_seed = base64.encodestring( if PY3:
'TIGER, tiger, burning bright. In the forests of the night' _seed = base64.encodebytes(
) b'TIGER, tiger, burning bright. In the forests of the night')
else:
_seed = base64.encodestring(
'TIGER, tiger, burning bright. In the forests of the night')
def _add_random_address(self, label): def _add_random_address(self, label):
return self.api.createRandomAddress(base64.encodestring(label)) if PY3:
return self.api.createRandomAddress(base64.encodebytes(bytes(label, 'UTF-8')).decode('utf-8'))
else:
return self.api.createRandomAddress(base64.encodestring(label))
def test_user_password(self): def test_user_password(self):
"""Trying to connect with wrong username/password""" """Trying to connect with wrong username/password"""
@ -119,108 +126,202 @@ class TestAPI(TestAPIProto):
def test_create_deterministic_addresses(self): def test_create_deterministic_addresses(self):
"""Test creation of deterministic addresses""" """Test creation of deterministic addresses"""
self.assertEqual( if PY3:
self.api.getDeterministicAddress(self._seed, 4, 1), self.assertEqual(
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK') self.api.getDeterministicAddress(self._seed.decode('utf-8'), 4, 1),
self.assertEqual( 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
self.api.getDeterministicAddress(self._seed, 3, 1), self.assertEqual(
'BM-2DBPTgeSawWYZceFD69AbDT5q4iUWtj1ZN') self.api.getDeterministicAddress(self._seed.decode('utf-8'), 3, 1),
self.assertRegexpMatches( 'BM-2DBPTgeSawWYZceFD69AbDT5q4iUWtj1ZN')
self.api.getDeterministicAddress(self._seed, 2, 1), self.assertRegex(
r'^API Error 0002:') self.api.getDeterministicAddress(self._seed.decode('utf-8'), 2, 1),
r'^API Error 0002:')
# This is here until the streams will be implemented # This is here until the streams will be implemented
self.assertRegexpMatches( self.assertRegex(
self.api.getDeterministicAddress(self._seed, 3, 2), self.api.getDeterministicAddress(self._seed.decode('utf-8'), 3, 2),
r'API Error 0003:') r'API Error 0003:')
self.assertRegexpMatches( self.assertRegex(
self.api.createDeterministicAddresses(self._seed, 1, 4, 2), self.api.createDeterministicAddresses(self._seed.decode('utf-8'), 1, 4, 2),
r'API Error 0003:') r'API Error 0003:')
self.assertRegexpMatches( self.assertRegex(
self.api.createDeterministicAddresses('', 1), self.api.createDeterministicAddresses('', 1),
r'API Error 0001:') r'API Error 0001:')
self.assertRegexpMatches( self.assertRegex(
self.api.createDeterministicAddresses(self._seed, 1, 2), self.api.createDeterministicAddresses(self._seed.decode('utf-8'), 1, 2),
r'API Error 0002:') r'API Error 0002:')
self.assertRegexpMatches( self.assertRegex(
self.api.createDeterministicAddresses(self._seed, 0), self.api.createDeterministicAddresses(self._seed.decode('utf-8'), 0),
r'API Error 0004:') r'API Error 0004:')
self.assertRegexpMatches( self.assertRegex(
self.api.createDeterministicAddresses(self._seed, 1000), self.api.createDeterministicAddresses(self._seed.decode('utf-8'), 1000),
r'API Error 0005:') r'API Error 0005:')
addresses = json.loads( addresses = json.loads(
self.api.createDeterministicAddresses(self._seed, 2, 4) self.api.createDeterministicAddresses(self._seed.decode('utf-8'), 2, 4)
)['addresses'] )['addresses']
self.assertEqual(len(addresses), 2) self.assertEqual(len(addresses), 2)
self.assertEqual(addresses[0], 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK') self.assertEqual(addresses[0], 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
for addr in addresses: for addr in addresses:
self.assertEqual(self.api.deleteAddress(addr), 'success') self.assertEqual(self.api.deleteAddress(addr), 'success')
else:
self.assertEqual(
self.api.getDeterministicAddress(self._seed, 4, 1),
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
self.assertEqual(
self.api.getDeterministicAddress(self._seed, 3, 1),
'BM-2DBPTgeSawWYZceFD69AbDT5q4iUWtj1ZN')
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], 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
for addr in addresses:
self.assertEqual(self.api.deleteAddress(addr), 'success')
def test_create_random_address(self): def test_create_random_address(self):
"""API command 'createRandomAddress': basic BM-address validation""" """API command 'createRandomAddress': basic BM-address validation"""
addr = self._add_random_address('random_1') addr = self._add_random_address('random_1')
self.assertRegexpMatches(addr, r'^BM-') if PY3:
self.assertRegexpMatches(addr[3:], r'[a-zA-Z1-9]+$') self.assertRegex(addr, r'^BM-')
self.assertRegex(addr[3:], r'[a-zA-Z1-9]+$')
else:
self.assertRegexpMatches(addr, r'^BM-')
self.assertRegexpMatches(addr[3:], r'[a-zA-Z1-9]+$')
# Whitepaper says "around 36 character" # Whitepaper says "around 36 character"
self.assertLessEqual(len(addr[3:]), 40) self.assertLessEqual(len(addr[3:]), 40)
self.assertEqual(self.api.deleteAddress(addr), 'success') self.assertEqual(self.api.deleteAddress(addr), 'success')
def test_addressbook(self): def test_addressbook(self):
"""Testing API commands for addressbook manipulations""" """Testing API commands for addressbook manipulations"""
# Initially it's empty if PY3:
self.assertEqual( # Initially it's empty
json.loads(self.api.listAddressBookEntries()).get('addresses'), self.assertEqual(
[] json.loads(self.api.listAddressBookEntries()).get('addresses'),
) []
# Add known address )
self.api.addAddressBookEntry( # Add known address
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK', self.api.addAddressBookEntry(
base64.encodestring('tiger_4') 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK',
) base64.encodebytes('tiger_4'.encode('UTF-8')).decode('utf-8')
# Check addressbook entry )
entries = json.loads( # Check addressbook entry
self.api.listAddressBookEntries()).get('addresses')[0] entries = json.loads(
self.assertEqual( self.api.listAddressBookEntries()).get('addresses')[0]
entries['address'], 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK') self.assertEqual(
self.assertEqual( entries['address'], 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
base64.decodestring(entries['label']), 'tiger_4') self.assertEqual(
# Remove known address base64.decodebytes(bytes(entries['label'], 'utf-8')).decode('utf-8'), 'tiger_4')
self.api.deleteAddressBookEntry( # Remove known address
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK') self.api.deleteAddressBookEntry(
# Addressbook should be empty again 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
self.assertEqual( # Addressbook should be empty again
json.loads(self.api.listAddressBookEntries()).get('addresses'), self.assertEqual(
[] json.loads(self.api.listAddressBookEntries()).get('addresses'),
) []
)
else:
# Initially it's empty
self.assertEqual(
json.loads(self.api.listAddressBookEntries()).get('addresses'),
[]
)
# Add known address
self.api.addAddressBookEntry(
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK',
base64.encodestring('tiger_4')
)
# Check addressbook entry
entries = json.loads(
self.api.listAddressBookEntries()).get('addresses')[0]
self.assertEqual(
entries['address'], 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
self.assertEqual(
base64.decodestring(entries['label']), 'tiger_4')
# Remove known address
self.api.deleteAddressBookEntry(
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK')
# Addressbook should be empty again
self.assertEqual(
json.loads(self.api.listAddressBookEntries()).get('addresses'),
[]
)
def test_subscriptions(self): def test_subscriptions(self):
"""Testing the API commands related to subscriptions""" """Testing the API commands related to subscriptions"""
for s in json.loads(self.api.listSubscriptions())['subscriptions']: if PY3:
# special address, added when sqlThread starts for s in json.loads(self.api.listSubscriptions())['subscriptions']:
if s['address'] == 'BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw': # special address, added when sqlThread starts
self.assertEqual( if s['address'] == 'BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw':
base64.decodestring(s['label']), self.assertEqual(
'Bitmessage new releases/announcements') base64.decodebytes(bytes(s['label'], 'utf-8')).decode('utf-8'),
self.assertTrue(s['enabled']) 'Bitmessage new releases/announcements')
break self.assertTrue(s['enabled'])
break
else:
self.fail(
'Could not find Bitmessage new releases/announcements'
' in subscriptions')
self.assertEqual(
self.api.deleteSubscription('BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw'),
'Deleted subscription if it existed.')
self.assertEqual(
json.loads(self.api.listSubscriptions())['subscriptions'], [])
else: else:
self.fail( for s in json.loads(self.api.listSubscriptions())['subscriptions']:
'Could not find Bitmessage new releases/announcements' # special address, added when sqlThread starts
' in subscriptions') if s['address'] == 'BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw':
self.assertEqual( self.assertEqual(
self.api.deleteSubscription('BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw'), base64.decodestring(s['label']),
'Deleted subscription if it existed.') 'Bitmessage new releases/announcements')
self.assertEqual( self.assertTrue(s['enabled'])
json.loads(self.api.listSubscriptions())['subscriptions'], []) break
else:
self.fail(
'Could not find Bitmessage new releases/announcements'
' in subscriptions')
self.assertEqual(
self.api.deleteSubscription('BM-GtovgYdgs7qXPkoYaRgrLFuFKz1SFpsw'),
'Deleted subscription if it existed.')
self.assertEqual(
json.loads(self.api.listSubscriptions())['subscriptions'], [])
def test_send(self): def test_send(self):
"""Test message sending""" """Test message sending"""
# self.api.createDeterministicAddresses(self._seed, 1, 4) if PY3:
addr = self._add_random_address('random_2') addr = str(self._add_random_address('random_2'))
msg = base64.encodestring('test message') msg = str(base64.encodebytes('test message'.encode('UTF-8')).decode('utf-8'))
msg_subject = base64.encodestring('test_subject') msg_subject = str(base64.encodebytes('test_subject'.encode('UTF-8')).decode('utf-8'))
else:
addr = self._add_random_address('random_2')
msg = base64.encodestring('test message')
msg_subject = base64.encodestring('test_subject')
ackdata = self.api.sendMessage( ackdata = self.api.sendMessage(
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK', addr, msg_subject, msg) 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK', addr, msg_subject, msg)
try: try:
@ -277,10 +378,16 @@ class TestAPI(TestAPIProto):
def test_send_broadcast(self): def test_send_broadcast(self):
"""Test broadcast sending""" """Test broadcast sending"""
addr = self._add_random_address('random_2') if PY3:
msg = base64.encodestring('test broadcast') addr = self._add_random_address('random_2')
ackdata = self.api.sendBroadcast( msg = base64.encodebytes('test broadcast'.encode('UTF-8')).decode('utf-8')
addr, base64.encodestring('test_subject'), msg) ackdata = self.api.sendBroadcast(
addr, base64.encodebytes('test_subject'.encode('UTF-8')).decode('utf-8'), msg)
else:
addr = self._add_random_address('random_2')
msg = base64.encodestring('test broadcast')
ackdata = self.api.sendBroadcast(
addr, base64.encodestring('test_subject'), msg)
try: try:
int(ackdata, 16) int(ackdata, 16)
status = self.api.getStatus(ackdata) status = self.api.getStatus(ackdata)
@ -324,25 +431,48 @@ class TestAPI(TestAPIProto):
def test_chan(self): def test_chan(self):
"""Testing chan creation/joining""" """Testing chan creation/joining"""
# Create chan with known address if PY3:
self.assertEqual( # Create chan with known address
self.api.createChan(self._seed), self.assertEqual(
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK' self.api.createChan(self._seed.decode("utf-8")),
) 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK'
# cleanup )
self.assertEqual( # cleanup
self.api.leaveChan('BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK'), self.assertEqual(
'success' self.api.leaveChan('BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK'),
) 'success'
# Join chan with addresses of version 3 or 4 )
for addr in ( # Join chan with addresses of version 3 or 4
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK', for addr in (
'BM-2DBPTgeSawWYZceFD69AbDT5q4iUWtj1ZN' 'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK',
): 'BM-2DBPTgeSawWYZceFD69AbDT5q4iUWtj1ZN'
self.assertEqual(self.api.joinChan(self._seed, addr), 'success') ):
self.assertEqual(self.api.leaveChan(addr), 'success') self.assertEqual(self.api.joinChan(self._seed.decode("utf-8"), addr), 'success')
# Joining with wrong address should fail self.assertEqual(self.api.leaveChan(addr), 'success')
self.assertRegexpMatches( # Joining with wrong address should fail
self.api.joinChan(self._seed, 'BM-2cWzSnwjJ7yRP3nLEW'), self.assertRegex(
r'^API Error 0008:' self.api.joinChan(self._seed.decode("utf-8"), 'BM-2cWzSnwjJ7yRP3nLEW'),
) r'^API Error 0008:'
)
else:
self.assertEqual(
self.api.createChan(self._seed),
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK'
)
# cleanup
self.assertEqual(
self.api.leaveChan('BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK'),
'success'
)
# Join chan with addresses of version 3 or 4
for addr in (
'BM-2cWzSnwjJ7yRP3nLEWUV5LisTZyREWSzUK',
'BM-2DBPTgeSawWYZceFD69AbDT5q4iUWtj1ZN'
):
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:'
)

16
src/tests/test_logger.py
View File

@ -1,11 +1,14 @@
# pylint: disable=E1101
""" """
Testing the logger configuration Testing the logger configuration
""" """
import os import os
import tempfile import tempfile
import sys
from .test_process import TestProcessProto from .test_process import TestProcessProto
PY3 = sys.version_info[0] >= 3
class TestLogger(TestProcessProto): class TestLogger(TestProcessProto):
@ -43,7 +46,10 @@ handlers=default
cls.log_file = os.path.join(cls.home, 'debug.log') cls.log_file = os.path.join(cls.home, 'debug.log')
with open(os.path.join(cls.home, 'logging.dat'), 'wb') as dst: with open(os.path.join(cls.home, 'logging.dat'), 'wb') as dst:
dst.write(cls.conf_template.format(cls.log_file, cls.pattern)) if PY3:
dst.write(bytes(cls.conf_template.format(cls.log_file, cls.pattern), 'utf-8'))
else:
dst.write(cls.conf_template.format(cls.log_file, cls.pattern))
super(TestLogger, cls).setUpClass() super(TestLogger, cls).setUpClass()
@ -52,5 +58,9 @@ handlers=default
self._stop_process() self._stop_process()
data = open(self.log_file).read() data = open(self.log_file).read()
self.assertRegexpMatches(data, self.pattern) if PY3:
self.assertRegexpMatches(data, 'Loaded logger configuration') self.assertRegex(data, self.pattern)
self.assertRegex(data, 'Loaded logger configuration')
else:
self.assertRegexpMatches(data, self.pattern)
self.assertRegexpMatches(data, 'Loaded logger configuration')

5
src/tests/test_process.py
View File

@ -12,10 +12,7 @@ import unittest
import psutil import psutil
from .common import cleanup, put_signal_file, skip_python3 from .common import cleanup, put_signal_file
skip_python3()
class TestProcessProto(unittest.TestCase): class TestProcessProto(unittest.TestCase):