Skip to content

Commit

Permalink
test: Add rpc_bind test to default-run tests
Browse files Browse the repository at this point in the history 
Skip the parts that cannot be run on the host due to lack
of IPv6 support or a second interface to bind on, and warn
appropriately.

Without no strong requirements (besides being Linux only, in which case
the test is skipped) left, just add this test to the default in
test_runner.

Includes suggested changes by John Newbery.
  • Loading branch information
@laanwj @furszy
laanwj authored and furszy committed Aug 10, 2021
1 parent 61a08af commit 4a034d8
Show file tree
Hide file tree
Showing 2 changed files with 64 additions and 44 deletions.
104 changes: 61 additions & 43 deletions test/functional/rpc_bind.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,9 @@
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test running pivxd with the -rpcbind and -rpcallowip options."""

import socket
import sys

from test_framework.netutil import addr_to_hex, all_interfaces, get_bind_addrs
from test_framework.netutil import addr_to_hex, all_interfaces, get_bind_addrs, test_ipv6_local
from test_framework.test_framework import PivxTestFramework, SkipTest
from test_framework.util import (
assert_equal,
Expand All @@ -28,6 +27,11 @@ def set_test_params(self):
def setup_network(self):
self.add_nodes(self.num_nodes, None)

def add_options(self, parser):
parser.add_option("--ipv4", action='store_true', dest="run_ipv4", help="Run ipv4 tests only", default=False)
parser.add_option("--ipv6", action='store_true', dest="run_ipv6", help="Run ipv6 tests only", default=False)
parser.add_option("--nonloopback", action='store_true', dest="run_nonloopback", help="Run non-loopback tests only", default=False)

def run_bind_test(self, allow_ips, connect_to, addresses, expected):
'''
Start a node with requested rpcallowip and rpcbind parameters,
Expand Down Expand Up @@ -62,55 +66,69 @@ def run_allowip_test(self, allow_ips, rpchost, rpcport):

def run_test(self):
# due to OS-specific network stats queries, this test works only on Linux
if sum([self.options.run_ipv4, self.options.run_ipv6, self.options.run_nonloopback]) > 1:
raise AssertionError("Only one of --ipv4, --ipv6 and --nonloopback can be set")

self.log.info("Check for linux")
if not sys.platform.startswith('linux'):
raise SkipTest("This test can only be run on linux.")
# find the first non-loopback interface for testing
non_loopback_ip = None

self.log.info("Check for ipv6")
have_ipv6 = test_ipv6_local()
if not have_ipv6 and not self.options.run_ipv4:
raise SkipTest("This test requires ipv6 support.")

self.log.info("Check for non-loopback interface")
self.non_loopback_ip = None
for name,ip in all_interfaces():
if ip != '127.0.0.1':
non_loopback_ip = ip
self.non_loopback_ip = ip
break
if non_loopback_ip is None:
raise SkipTest("This test requires at least one non-loopback IPv4 interface.")
try:
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s.connect(("::1",1))
s.close
except OSError:
raise SkipTest("This test requires IPv6 support.")

self.log.info("Using interface %s for testing" % non_loopback_ip)

defaultport = rpc_port(0)

# check default without rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(None, '127.0.0.1', [],
[('127.0.0.1', defaultport), ('::1', defaultport)])
# check default with rpcallowip (IPv6 any)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
[('::0', defaultport)])
# check only IPv4 localhost (explicit)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
[('127.0.0.1', defaultport)])
# check only IPv4 localhost (explicit) with alternative port
self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
[('127.0.0.1', 32171)])
# check only IPv4 localhost (explicit) with multiple alternative ports on same host
self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
[('127.0.0.1', 32171), ('127.0.0.1', 32172)])
# check only IPv6 localhost (explicit)
self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
[('::1', defaultport)])
# check both IPv4 and IPv6 localhost (explicit)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
[('127.0.0.1', defaultport), ('::1', defaultport)])
if self.non_loopback_ip is None and self.options.run_nonloopback:
raise SkipTest("This test requires a non-loopback ip address.")

self.defaultport = rpc_port(0)

if not self.options.run_nonloopback:
self._run_loopback_tests()
if not self.options.run_ipv4 and not self.options.run_ipv6:
self._run_nonloopback_tests()

def _run_loopback_tests(self):
if self.options.run_ipv4:
# check only IPv4 localhost (explicit)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
[('127.0.0.1', self.defaultport)])
# check only IPv4 localhost (explicit) with alternative port
self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
[('127.0.0.1', 32171)])
# check only IPv4 localhost (explicit) with multiple alternative ports on same host
self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
[('127.0.0.1', 32171), ('127.0.0.1', 32172)])
else:
# check default without rpcallowip (IPv4 and IPv6 localhost)
self.run_bind_test(None, '127.0.0.1', [],
[('127.0.0.1', self.defaultport), ('::1', self.defaultport)])
# check default with rpcallowip (IPv6 any)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
[('::0', self.defaultport)])
# check only IPv6 localhost (explicit)
self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
[('::1', self.defaultport)])
# check both IPv4 and IPv6 localhost (explicit)
self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
[('127.0.0.1', self.defaultport), ('::1', self.defaultport)])

def _run_nonloopback_tests(self):
self.log.info("Using interface %s for testing" % self.non_loopback_ip)

# check only non-loopback interface
self.run_bind_test([non_loopback_ip], non_loopback_ip, [non_loopback_ip],
[(non_loopback_ip, defaultport)])
self.run_bind_test([self.non_loopback_ip], self.non_loopback_ip, [self.non_loopback_ip],
[(self.non_loopback_ip, self.defaultport)])

# Check that with invalid rpcallowip, we are denied
self.run_allowip_test([non_loopback_ip], non_loopback_ip, defaultport)
assert_raises_rpc_error(-342, "non-JSON HTTP response with '403 Forbidden' from server", self.run_allowip_test, ['1.1.1.1'], non_loopback_ip, defaultport)
self.run_allowip_test([self.non_loopback_ip], self.non_loopback_ip, self.defaultport)
assert_raises_rpc_error(-342, "non-JSON HTTP response with '403 Forbidden' from server", self.run_allowip_test, ['1.1.1.1'], self.non_loopback_ip, self.defaultport)

if __name__ == '__main__':
RPCBindTest().main()
4 changes: 3 additions & 1 deletion test/functional/test_runner.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,9 @@
'rpc_fundrawtransaction.py', # ~ 227 sec
'mining_pos_coldStaking.py', # ~ 220 sec
'wallet_import_rescan.py', # ~ 204 sec
'rpc_bind.py --ipv4',
'rpc_bind.py --ipv6',
'rpc_bind.py --nonloopback',
'p2p_invalid_block.py', # ~ 213 sec
'p2p_addr_relay.py',
'p2p_addrv2_relay.py',
Expand Down Expand Up @@ -187,7 +190,6 @@
# vv Tests less than 60s vv
#'p2p_feefilter.py',
'feature_abortnode.py',
'rpc_bind.py',
# vv Tests less than 30s vv
#'example_test.py',
'feature_notifications.py',
Expand Down

0 comments on commit 4a034d8

Please sign in to comment.