Skip to content

Commit

Permalink
Test and refactor.
Browse files Browse the repository at this point in the history
* Import modules, not names.
* Stricten port tests.
* Rename check_NAT_status to get_NAT_status.
* Test is_ipv6_address.
* Fix docstrings and test get_peer_url.
  • Loading branch information
Renelvon committed Oct 28, 2014
1 parent ecd53dc commit ace5cb9
Show file tree
Hide file tree
Showing 3 changed files with 46 additions and 16 deletions.
28 changes: 16 additions & 12 deletions node/network_util.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from IPy import IP, IPint
import IPy
import requests
from requests.exceptions import RequestException
import stun


Expand Down Expand Up @@ -32,7 +31,7 @@ def init_additional_STUN_servers(servers=_ADDITIONAL_STUN_SERVERS):
stun.stun_servers_list = tuple(server_set)


def check_NAT_status():
def get_NAT_status():
nat_type, external_ip, external_port = stun.get_ip_info()
return {'nat_type': nat_type,
'external_ip': external_ip,
Expand All @@ -52,31 +51,36 @@ def is_valid_protocol(protocol):


def is_private_ip_address(addr):
return is_loopback_addr(addr) or IP(addr).iptype() != 'PUBLIC'
return is_loopback_addr(addr) or IPy.IP(addr).iptype() != 'PUBLIC'


def get_my_ip():
try:
r = requests.get('https://icanhazip.com')
return r.text.strip()
except (AttributeError, RequestException) as e:
except (AttributeError, requests.RequestException) as e:
print '[Requests] error: %s' % e
return None


def is_ipv6_address(ip):
address = IPint(ip)
return address.version == 6
return IPy.IP(ip).version() == 6


def get_peer_url(address, port, protocol='tcp'):
"""
Returns a url for a peer that can be used by ZMQ
Return a URL which can be used by ZMQ.
@param address: A string which can be an IPv4 address, an IPv6 address
or a DNS name
@param address: An IPv4 address, an IPv6 address or a DNS name.
@type address: str
@param port: the port that will be used to connect to the peer
@param port: The port that will be used to connect to the peer
@type port: int
@param protocol: The connection protocol
@type protocol: str
@rtype: str
"""
try:
# is_ipv6_address will throw an exception for a DNS name
Expand All @@ -85,7 +89,7 @@ def get_peer_url(address, port, protocol='tcp'):
is_ipv6 = False

if is_ipv6:
# an IPv6 address must be enclosed in brackets
# An IPv6 address must be enclosed in brackets.
return '%s://[%s]:%s' % (protocol, address, port)
else:
return '%s://%s:%s' % (protocol, address, port)
2 changes: 1 addition & 1 deletion node/openbazaar.py
Original file line number Diff line number Diff line change
Expand Up @@ -341,7 +341,7 @@ def start(arguments):
nat_status = None
if not arguments.disable_stun_check:
print "Checking NAT Status..."
nat_status = network_util.check_NAT_status()
nat_status = network_util.get_NAT_status()
elif not arguments.dev_mode and network_util.is_private_ip_address(arguments.server_ip):
print "openbazaar: Could not start. The given/default server IP address",
print arguments.server_ip, "is not a public ip address."
Expand Down
32 changes: 29 additions & 3 deletions test/test_network_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,10 @@ def test_is_loopback_addr(self):

def test_is_valid_port(self):
self.assertTrue(network_util.is_valid_port(1))
self.assertTrue(network_util.is_valid_port(65335))
self.assertTrue(network_util.is_valid_port(2**16 - 1))

self.assertFalse(network_util.is_valid_port(-1))
self.assertFalse(network_util.is_valid_port(70000))
self.assertFalse(network_util.is_valid_port(0))
self.assertFalse(network_util.is_valid_port(2**16))

def test_is_valid_protocol(self):
self.assertTrue(network_util.is_valid_protocol('tcp'))
Expand All @@ -57,6 +57,32 @@ def test_is_private_ip_address(self):

self.assertFalse(network_util.is_private_ip_address('8.8.8.8'))

def test_is_ipv6_address(self):
self.assertTrue(network_util.is_ipv6_address('2a00::'))
self.assertFalse(network_util.is_ipv6_address('8.8.8.8'))

def test_get_peer_url(self):
self.assertEqual(
network_util.get_peer_url('8.8.8.8', 1234),
'tcp://8.8.8.8:1234'
)
self.assertEqual(
network_util.get_peer_url('8.8.8.8', 1234, protocol='udp'),
'udp://8.8.8.8:1234'
)
self.assertEqual(
network_util.get_peer_url('2a00::', 1234),
'tcp://[2a00::]:1234'
)
self.assertEqual(
network_util.get_peer_url('2a00::', 1234, protocol='udp'),
'udp://[2a00::]:1234'
)
self.assertEqual(
network_util.get_peer_url('www.openbazaar.com', 1234),
'tcp://www.openbazaar.com:1234'
)


if __name__ == '__main__':
unittest.main()

0 comments on commit ace5cb9

Please sign in to comment.