@ -20,13 +20,12 @@ from test_framework.util import (
wait_until ,
)
from test_framework . mininode import P2PInterface
import test_framework . messages
from test_framework . messages import (
CAddress ,
msg_addr ,
NODE_NETWORK ,
NODE_WITNESS ,
NODE_GETUTXO , NODE_BLOOM ,
NODE_NETWORK_LIMITED ,
)
def assert_net_servicesnames ( servicesflag , servicenames ) :
@ -36,16 +35,10 @@ def assert_net_servicesnames(servicesflag, servicenames):
: param servicesflag : The services as an integer .
: param servicesnames : The list of decoded services names , as strings .
"""
if servicesflag & NODE_NETWORK :
assert " NETWORK " in servicenames
if servicesflag & NODE_GETUTXO :
assert " GETUTXO " in servicenames
if servicesflag & NODE_BLOOM :
assert " BLOOM " in servicenames
if servicesflag & NODE_WITNESS :
assert " WITNESS " in servicenames
if servicesflag & NODE_NETWORK_LIMITED :
assert " NETWORK_LIMITED " in servicenames
servicesflag_generated = 0
for servicename in servicenames :
servicesflag_generated | = getattr ( test_framework . messages , ' NODE_ ' + servicename )
assert servicesflag_generated == servicesflag
class NetTest ( BitcoinTestFramework ) :
def set_test_params ( self ) :
@ -119,7 +112,7 @@ class NetTest(BitcoinTestFramework):
# check the `servicesnames` field
network_info = [ node . getnetworkinfo ( ) for node in self . nodes ]
for info in network_info :
assert_net_servicesnames ( int ( info [ " localservices " ] ), info [ " localservicesnames " ] )
assert_net_servicesnames ( int ( info [ " localservices " ] , 0x10 ), info [ " localservicesnames " ] )
def _test_getaddednodeinfo ( self ) :
assert_equal ( self . nodes [ 0 ] . getaddednodeinfo ( ) , [ ] )
@ -143,7 +136,7 @@ class NetTest(BitcoinTestFramework):
assert_equal ( peer_info [ 1 ] [ 0 ] [ ' minfeefilter ' ] , Decimal ( " 0.00001000 " ) )
# check the `servicesnames` field
for info in peer_info :
assert_net_servicesnames ( int ( info [ 0 ] [ " services " ] ), info [ 0 ] [ " servicesnames " ] )
assert_net_servicesnames ( int ( info [ 0 ] [ " services " ] , 0x10 ), info [ 0 ] [ " servicesnames " ] )
def _test_getnodeaddresses ( self ) :
self . nodes [ 0 ] . add_p2p_connection ( P2PInterface ( ) )