Changeset View
Changeset View
Standalone View
Standalone View
test/functional/test_framework/netutil.py
| Show All 26 Lines | |||||
| STATE_LISTEN = '0A' | STATE_LISTEN = '0A' | ||||
| # STATE_CLOSING = '0B' | # STATE_CLOSING = '0B' | ||||
| def get_socket_inodes(pid): | def get_socket_inodes(pid): | ||||
| ''' | ''' | ||||
| Get list of socket inodes for process pid. | Get list of socket inodes for process pid. | ||||
| ''' | ''' | ||||
| base = '/proc/%i/fd' % pid | base = '/proc/{}/fd'.format(pid) | ||||
| inodes = [] | inodes = [] | ||||
| for item in os.listdir(base): | for item in os.listdir(base): | ||||
| target = os.readlink(os.path.join(base, item)) | target = os.readlink(os.path.join(base, item)) | ||||
| if target.startswith('socket:'): | if target.startswith('socket:'): | ||||
| inodes.append(int(target[8:-1])) | inodes.append(int(target[8:-1])) | ||||
| return inodes | return inodes | ||||
| def _remove_empty(array): | def _remove_empty(array): | ||||
| return [x for x in array if x != ''] | return [x for x in array if x != ''] | ||||
| def _convert_ip_port(array): | def _convert_ip_port(array): | ||||
| host, port = array.split(':') | host, port = array.split(':') | ||||
| # convert host from mangled-per-four-bytes form as used by kernel | # convert host from mangled-per-four-bytes form as used by kernel | ||||
| host = unhexlify(host) | host = unhexlify(host) | ||||
| host_out = '' | host_out = '' | ||||
| for x in range(0, len(host) // 4): | for x in range(0, len(host) // 4): | ||||
| (val,) = struct.unpack('=I', host[x * 4:(x + 1) * 4]) | (val,) = struct.unpack('=I', host[x * 4:(x + 1) * 4]) | ||||
| host_out += '%08x' % val | host_out += '{:08x}'.format(val) | ||||
| return host_out, int(port, 16) | return host_out, int(port, 16) | ||||
| def netstat(typ='tcp'): | def netstat(typ='tcp'): | ||||
| ''' | ''' | ||||
| Function to return a list with status of tcp connections at linux systems | Function to return a list with status of tcp connections at linux systems | ||||
| To get pid of all network process running on system, you must run this script | To get pid of all network process running on system, you must run this script | ||||
| ▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Lines | elif ':' in addr: # IPv6 | ||||
| else: # two bytes per component | else: # two bytes per component | ||||
| val = int(comp, 16) | val = int(comp, 16) | ||||
| sub[x].append(val >> 8) | sub[x].append(val >> 8) | ||||
| sub[x].append(val & 0xff) | sub[x].append(val & 0xff) | ||||
| nullbytes = 16 - len(sub[0]) - len(sub[1]) | nullbytes = 16 - len(sub[0]) - len(sub[1]) | ||||
| assert((x == 0 and nullbytes == 0) or (x == 1 and nullbytes > 0)) | assert((x == 0 and nullbytes == 0) or (x == 1 and nullbytes > 0)) | ||||
| addr = sub[0] + ([0] * nullbytes) + sub[1] | addr = sub[0] + ([0] * nullbytes) + sub[1] | ||||
| else: | else: | ||||
| raise ValueError('Could not parse address %s' % addr) | raise ValueError('Could not parse address {}'.format(addr)) | ||||
| return hexlify(bytearray(addr)).decode('ascii') | return hexlify(bytearray(addr)).decode('ascii') | ||||
| def test_ipv6_local(): | def test_ipv6_local(): | ||||
| ''' | ''' | ||||
| Check for (local) IPv6 support. | Check for (local) IPv6 support. | ||||
| ''' | ''' | ||||
| import socket | import socket | ||||
| Show All 9 Lines | |||||