from collections import defaultdict
import socket
import os
import re
from cfme.fixtures.pytest_store import store
from cfme.utils.log import logger
_ports = defaultdict(dict)
_dns_cache = {}
ip_address = re.compile(
r"^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}"
r"(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$")
[docs]def random_port(tcp=True):
"""Get a random port number for making a socket
Args:
tcp: Return a TCP port number if True, UDP if False
This may not be reliable at all due to an inherent race condition. This works
by creating a socket on an ephemeral port, inspecting it to see what port was used,
closing it, and returning that port number. In the time between closing the socket
and opening a new one, it's possible for the OS to reopen that port for another purpose.
In practical testing, this race condition did not result in a failure to (re)open the
returned port number, making this solution squarely "good enough for now".
"""
# Port 0 will allocate an ephemeral port
socktype = socket.SOCK_STREAM if tcp else socket.SOCK_DGRAM
s = socket.socket(socket.AF_INET, socktype)
s.bind(('', 0))
addr, port = s.getsockname()
s.close()
return port
[docs]def my_ip_address(http=False):
"""Get the ip address of the host running tests using the service listed in cfme_data['ip_echo']
The ip echo endpoint is expected to write the ip address to the socket and close the
connection. See a working example of this in :py:func:`ip_echo_socket`.
"""
# the pytest store does this work, it's included here for convenience
return store.my_ip_address
[docs]def ip_echo_socket(port=32123):
"""A simple socket server, for use with :py:func:`my_ip_address`"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('', port))
s.listen(0)
while True:
conn, addr = s.accept()
conn.sendall(addr[0])
conn.close()
[docs]def net_check(port, addr=None, force=False):
"""Checks the availablility of a port"""
port = int(port)
if not addr:
addr = store.current_appliance.hostname
if port not in _ports[addr] or force:
# First try DNS resolution
try:
addr = socket.gethostbyname(addr)
# Then try to connect to the port
try:
socket.create_connection((addr, port), timeout=10)
_ports[addr][port] = True
except socket.error:
_ports[addr][port] = False
except:
_ports[addr][port] = False
return _ports[addr][port]
[docs]def net_check_remote(port, addr=None, machine_addr=None, ssh_creds=None, force=False):
"""Checks the availability of a port from outside using another machine (over SSH)"""
from cfme.utils.ssh import SSHClient
port = int(port)
if not addr:
addr = my_ip_address()
if port not in _ports[addr] or force:
if not machine_addr:
machine_addr = store.current_appliance.hostname
if not ssh_creds:
ssh_client = store.current_appliance.ssh_client
else:
ssh_client = SSHClient(
hostname=machine_addr,
username=ssh_creds['username'],
password=ssh_creds['password']
)
with ssh_client:
# on exception => fails with return code 1
cmd = '''python -c "
import sys, socket
addr = socket.gethostbyname('%s')
socket.create_connection((addr, %d), timeout=10)
sys.exit(0)
"''' % (addr, port)
result = ssh_client.run_command(cmd)
_ports[addr][port] = result.success
return _ports[addr][port]
[docs]def resolve_hostname(hostname, force=False):
"""Cached DNS resolver. If the hostname does not resolve to an IP, returns None."""
if hostname not in _dns_cache or force:
try:
_dns_cache[hostname] = socket.gethostbyname(hostname)
except socket.gaierror:
_dns_cache[hostname] = None
return _dns_cache[hostname]
[docs]def resolve_ips(host_iterable, force_dns=False):
"""Takes list of hostnames, ips and another things. If the item is not an IP, it will be tried
to be converted to an IP. If that succeeds, it is appended to the set together with original
hostname. If it can't be resolved, just the original hostname is appended.
"""
result = set([])
for host in map(str, host_iterable):
result.add(host) # It is already an IP address
if ip_address.match(host) is None:
ip = resolve_hostname(host, force=force_dns)
if ip is not None:
result.add(ip)
return result
[docs]def is_pingable(ip_addr):
"""verifies the specified ip_address is reachable or not.
Args:
ip_addr: ip_address to verify the PING.
returns: return True is ip_address is pinging else returns False.
"""
try:
status = os.system("ping -c1 -w2 {}".format(ip_addr))
if status == 0:
logger.info('IP: %s is UP !', ip_addr)
return True
logger.info('IP: %s is DOWN !', ip_addr)
return False
except Exception as e:
logger.exception(e)
return False