2022-06-10 03:15:49 +02:00
|
|
|
"""TestAPIThread class definition"""
|
|
|
|
|
2021-12-11 17:24:54 +01:00
|
|
|
import time
|
|
|
|
|
2021-12-11 19:01:10 +01:00
|
|
|
from six.moves import queue, xmlrpc_client
|
2021-12-11 17:24:54 +01:00
|
|
|
|
2022-06-10 03:15:49 +02:00
|
|
|
from .partial import TestPartialRun
|
2021-12-11 19:01:10 +01:00
|
|
|
from .samples import sample_statusbar_msg # any
|
|
|
|
|
2021-12-11 17:24:54 +01:00
|
|
|
|
2022-06-10 03:15:49 +02:00
|
|
|
class TestAPIThread(TestPartialRun):
|
2021-12-11 17:24:54 +01:00
|
|
|
"""Test case running the API thread"""
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
def setUpClass(cls):
|
2022-06-10 03:15:49 +02:00
|
|
|
super(TestAPIThread, cls).setUpClass()
|
2021-12-11 17:24:54 +01:00
|
|
|
|
|
|
|
import helper_sql
|
2021-12-11 19:01:10 +01:00
|
|
|
import queues
|
2021-12-11 17:24:54 +01:00
|
|
|
|
2021-12-11 19:01:10 +01:00
|
|
|
# pylint: disable=too-few-public-methods
|
2021-12-11 17:24:54 +01:00
|
|
|
class SqlReadyMock(object):
|
2021-12-11 19:01:10 +01:00
|
|
|
"""Mock helper_sql.sql_ready event with dummy class"""
|
2021-12-11 17:24:54 +01:00
|
|
|
@staticmethod
|
|
|
|
def wait():
|
2021-12-11 19:01:10 +01:00
|
|
|
"""Don't wait, return immediately"""
|
2021-12-11 17:24:54 +01:00
|
|
|
return
|
|
|
|
|
|
|
|
helper_sql.sql_ready = SqlReadyMock
|
2021-12-11 19:01:10 +01:00
|
|
|
cls.queues = queues
|
|
|
|
|
2022-06-10 03:15:49 +02:00
|
|
|
cls.config.set('bitmessagesettings', 'apiusername', 'username')
|
|
|
|
cls.config.set('bitmessagesettings', 'apipassword', 'password')
|
|
|
|
cls.config.set('inventory', 'storage', 'filesystem')
|
2021-12-11 17:24:54 +01:00
|
|
|
|
|
|
|
import api
|
|
|
|
cls.thread = api.singleAPI()
|
|
|
|
cls.thread.daemon = True
|
|
|
|
cls.thread.start()
|
|
|
|
time.sleep(3)
|
|
|
|
cls.api = xmlrpc_client.ServerProxy(
|
|
|
|
"http://username:password@127.0.0.1:8442/")
|
|
|
|
|
|
|
|
def test_connection(self):
|
|
|
|
"""API command 'helloWorld'"""
|
|
|
|
self.assertEqual(
|
|
|
|
self.api.helloWorld('hello', 'world'), 'hello-world')
|
|
|
|
|
2021-12-11 19:01:10 +01:00
|
|
|
def test_statusbar(self):
|
|
|
|
"""Check UISignalQueue after issuing the 'statusBar' command"""
|
|
|
|
self.queues.UISignalQueue.queue.clear()
|
|
|
|
self.assertEqual(
|
|
|
|
self.api.statusBar(sample_statusbar_msg), 'success')
|
|
|
|
try:
|
|
|
|
cmd, data = self.queues.UISignalQueue.get(block=False)
|
|
|
|
except queue.Empty:
|
|
|
|
self.fail('UISignalQueue is empty!')
|
|
|
|
|
|
|
|
self.assertEqual(cmd, 'updateStatusBar')
|
|
|
|
self.assertEqual(data, sample_statusbar_msg)
|