pharos-validator/src/validation_tool/src/util.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Todd Gaunt and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import ipaddress
  11 import logging
  12 import os
  13
  14 class cd:
  15     """Context manager for changing the current working directory"""
  16     def __init__(self, new_path):
  17         self.new_path = os.path.expanduser(new_path)
  18
  19     def __enter__(self):
  20         self.saved_path = os.getcwd()
  21         os.chdir(self.new_path)
  22
  23     def __exit__(self, etype, value, traceback):
  24         os.chdir(self.saved_path)
  25
  26 def approxsize(x, y, deviation):
  27     """Approximately compares 'x' to 'y' with in % of 'deviation'"""
  28     logger = logging.getLogger(__name__)
  29
  30     dev = (y * .01 * deviation)
  31
  32     if x >= round(y - dev, 0) and x <= round(y + dev, 0):
  33         logger.debug("{} is approximately {}".format(x, y))
  34         return True
  35     else:
  36         logger.debug("{} is not approximately {}".format(x, y))
  37         return False
  38
  39 def read_line(sock):
  40     """Reads from a socket until a \n character or 512 bytes have been read,
  41     whichever comes first"""
  42     c = ""
  43     recvline = ""
  44     reads = 0
  45     while (c != "\n" and reads < 512):
  46         # Decode bytes to str, sockets output bytes which aren't pretty
  47         c = sock.recv(1).decode("utf-8")
  48         #print("char: '" + c + "'") # Debugging code
  49         recvline += c
  50         reads += 1
  51     return recvline
  52
  53 def read_msg(sock):
  54     """Reads a message prefixed with a number and a newline char, eg. "20\n"
  55     then reads x lines, where x is equal to the number in the first line."""
  56     # Read the socket once initially for the line count
  57     buf = read_line(sock)
  58     buf = buf[:-1] # Cut off the '\n' character
  59     length = int(buf)
  60
  61     lines = []
  62     for i in range(length):
  63         lines.append(read_line(sock))
  64     return "".join(lines)
  65
  66 def send_msg(sock, msg):
  67     """Sends a message to a socket"""
  68     # Encode python str to bytes beforehand, sockets only deal in bytes
  69     msg = bytes(msg, "utf-8")
  70     totalsent = 0
  71     while totalsent < len(msg):
  72         sent = sock.send(msg[totalsent:])
  73         if sent == 0:
  74             return -1
  75         totalsent = totalsent + sent
  76     return totalsent
  77
  78 def get_addr(interface):
  79     """Get the address of the machine that this program is running on"""
  80     return netifaces.ifaddresses(interface)[netifaces.AF_INET][0]["addr"]
  81
  82 def gen_ip_range(cidr, excluded, minimum, maximum ):
  83     """Takes a network cidr number, and then a min max value, and creates a list
  84     of ip addresses avalable on [a,b]. Also removes "excluded" addresses
  85     from the range"""
  86     logger = logging.getLogger(__name__)
  87     # Generate a list of available ip addresses for the dhcp server
  88     ip_range = list(map(lambda x: x.exploded, ipaddress.ip_network(cidr).hosts()))
  89
  90     for addr in excluded:
  91         # Remove the value from the list, if it isn't in the list then whatever
  92         try:
  93             ip_range.remove(addr)
  94         except ValueError:
  95             logger.debug("{} not in ip_range, cannot remove".format(addr))
  96
  97     # Remove all values before the minimum usable value
  98     for i in range(len(ip_range)):
  99         if ip_range[i] == minimum:
 100             ip_range = ip_range[i::]
 101             break
 102     # Remove all values after the maximum usable value
 103     for i in range(len(ip_range)):
 104         if ip_range[i] == maximum:
 105             ip_range = ip_range[0:i+1]
 106             break
 107     return ip_range