Add API endpoint to return inbound and outbound connections

This commit is contained in:
813492291816 2022-11-03 17:10:42 -04:00
parent eebbcf79d1
commit e6ecaa5e7d
No known key found for this signature in database
GPG Key ID: B14DF20410E5A5BC
2 changed files with 34 additions and 0 deletions

View File

@ -91,6 +91,7 @@ from bmconfigparser import config
from debug import logger from debug import logger
from helper_sql import SqlBulkExecute, sqlExecute, sqlQuery, sqlStoredProcedure, sql_ready from helper_sql import SqlBulkExecute, sqlExecute, sqlQuery, sqlStoredProcedure, sql_ready
from inventory import Inventory from inventory import Inventory
from network import BMConnectionPool
from network.threads import StoppableThread from network.threads import StoppableThread
from six.moves import queue from six.moves import queue
from version import softwareVersion from version import softwareVersion
@ -1441,6 +1442,33 @@ class BMRPCDispatcher(object):
'softwareVersion': softwareVersion 'softwareVersion': softwareVersion
} }
@command('listConnections')
def HandleListConnections(self):
"""
Returns bitmessage connection information as dict with keys *inbound,
*outbound.
"""
inboundConnections = []
outboundConnections = []
for i in BMConnectionPool().inboundConnections.values():
inboundConnections.append({
'host': i.destination.host,
'port': i.destination.port,
'fullyEstablished': i.fullyEstablished,
'userAgent': str(i.userAgent)
})
for i in BMConnectionPool().outboundConnections.values():
outboundConnections.append({
'host': i.destination.host,
'port': i.destination.port,
'fullyEstablished': i.fullyEstablished,
'userAgent': str(i.userAgent)
})
return {
'inbound': inboundConnections,
'outbound': outboundConnections
}
@command('helloWorld') @command('helloWorld')
def HandleHelloWorld(self, a, b): def HandleHelloWorld(self, a, b):
"""Test two string params""" """Test two string params"""

View File

@ -157,6 +157,12 @@ class TestAPI(TestAPIProto):
else: else:
self.assertGreater(status["networkConnections"], 0) 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): def test_list_addresses(self):
"""Checking the return of API command 'listAddresses'""" """Checking the return of API command 'listAddresses'"""
self.assertEqual( self.assertEqual(