[libvirt-test-API][PATCH] Add network_dhcp_leases test case

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

 



The network_dhcp_leases.py uses DHCPLeases() to validate new API
virNetworkGetDHCPLeases of libvirt.
---
 repos/network/network_dhcp_leases.py | 277 +++++++++++++++++++++++++++++++++++
 1 file changed, 277 insertions(+)
 create mode 100644 repos/network/network_dhcp_leases.py

diff --git a/repos/network/network_dhcp_leases.py b/repos/network/network_dhcp_leases.py
new file mode 100644
index 0000000..29ee529
--- /dev/null
+++ b/repos/network/network_dhcp_leases.py
@@ -0,0 +1,277 @@
+#!/usr/bin/env python
+#test DHCPLeases() API for libvirt
+
+import os
+import time
+import libvirt
+from libvirt import libvirtError
+from utils import utils
+from src import sharedmod
+
+required_params = ('networkname',)
+optional_params = {'macaddr': ''}
+
+LEASE_FILE = "/var/lib/libvirt/dnsmasq/"
+
+def check_ip(ipaddr, logger):
+    """
+       return a string according to ip address type, return 'ipv4' for ipv4,
+       return 'ipv6' for ipv6, return False for others
+    """
+    addr4 = ipaddr.strip().split('.')
+    addr6 = ipaddr.strip().split(':')
+    if len(addr4) == 4:
+        iptype = "ipv4"
+    elif len(addr6) == 6:
+        iptype = "ipv6"
+    else:
+        return False
+    return iptype
+
+def get_network_type(ipaddr,logger):
+    """
+       return 0 or 1 for ipv4/ipv6, this function will be used in
+       check_ipv4_values()/check_ipv6_values()
+    """
+    if check_ip(ipaddr, logger) == "ipv4":
+        return 0
+    elif check_ip(ipaddr, logger) == "ipv6":
+        return 1
+
+def get_bridge_name(network,logger):
+    """
+       get bridge name under specified network from specified network conf
+    """
+    CONF_NETWORK = LEASE_FILE + network + ".conf"
+    GREP_BRIDGE = "grep \"^interface=\" %s | awk -F\"=\" '{print $2}'"
+    status, output = utils.exec_cmd(GREP_BRIDGE % CONF_NETWORK, shell=True)
+    if not status:
+        pass
+    else:
+        logger.error("\"" + GREP_BRIDGE + "\"" + "error")
+        logger.error(output)
+        return False
+    return output[0]
+
+def get_ip_prefix(network, iptype, logger):
+    """
+       get ip prefix according to IP type
+    """
+    br = get_bridge_name(network, logger)
+    PREFIX = "ip -4 -o ad show %s | awk '{print $4}'|awk -F\"/\" '{print $2}'"
+    PREFIX_6 = "ip -6 -o ad show %s|awk '{print $4}'|awk -F\"/\" '{print $2}'"
+    if iptype == "ipv4":
+        status, output = utils.exec_cmd(PREFIX % br, shell=True)
+    elif iptype == "ipv6":
+        status, output = utils.exec_cmd(PREFIX_6 % br, shell=True)
+    if not status:
+        pass
+    else:
+        logger.error("\"" + GREP_BRIDGE + "\"" + "error")
+        logger.error(output)
+        return False
+    return output[0]
+
+def get_info_from_dnsmasq(network,macaddr,logger):
+    """
+       generate dict for lease info from virtual network's lease file
+    """
+    title = ['expirytime','mac','ipaddr','hostname','clientid']
+    output_list = []
+    lease_dnsmasq = []
+    temp = []
+    remove_list = []
+    GREP_MAC = "grep -w %s" + " " + LEASE_FILE_DNSMASQ
+    CAT_FILE = "cat" + " " + LEASE_FILE_DNSMASQ
+
+    status, output = utils.exec_cmd(CAT_FILE, shell=True)
+    if not status:
+        for i in range(0, len(output)):
+            output_list = []
+            output_str = output[i]
+            for item in output_str.split(" "):
+                output_list.append(item)
+            lease_dnsmasq.append(dict(zip(title,output_list)))
+
+        #due to no mac field in IPv6 line, so do nothing here temporarily.
+        if macaddr != None:
+             pass
+
+        #remove bridge duid line
+        for i in range(0, len(lease_dnsmasq)):
+            if lease_dnsmasq[i]['expirytime'] == 'duid':
+                remove_list.append(lease_dnsmasq[i])
+
+        for i in range(0, len(remove_list)):
+                lease_dnsmasq.remove(remove_list[i])
+
+        #remove expiry leases
+        for i in range(0, len(lease_dnsmasq)):
+            temp = int(lease_dnsmasq[i]['expirytime'])
+            lease_dnsmasq[i]['expirytime'] = temp
+
+        remove_list = []
+        for i in range(0, len(lease_dnsmasq)):
+            if time.time() >= int(lease_dnsmasq[i]['expirytime']):
+                remove_list.append(lease_dnsmasq[i])
+
+        for i in range(0, len(remove_list)):
+                lease_dnsmasq.remove(remove_list[i])
+
+        #replace * to None
+        for i in range(0, len(lease_dnsmasq)):
+            if lease_dnsmasq[i]['hostname'] == "*":
+                lease_dnsmasq[i]['hostname'] = None
+            if lease_dnsmasq[i]['clientid'] == "*":
+                lease_dnsmasq[i]['clientid'] = None
+
+        return lease_dnsmasq
+    else:
+        logger.error("\"" + CAT_FILE + "\"" + "error")
+        logger.error(output)
+        return False
+
+def compare_values(op1, op2, network, iptype, logger):
+    """
+       check all printed values from API
+    """
+    dnsmasq = op1
+    api = op2
+    temp = int(api['expirytime'])
+    api['expirytime'] = temp
+
+    for j in range(0,len(dnsmasq)):
+        if dnsmasq[j]['hostname'] == api['hostname'] and \
+           dnsmasq[j]['expirytime'] == api['expirytime']:
+            if dnsmasq[j]['ipaddr'] == api['ipaddr'] and \
+               dnsmasq[j]['clientid'] == api['clientid']:
+
+                if iptype == "ipv4":
+                    logger.debug("PASS: hostname: %s expirytime: %s ipaddr: %s" \
+                           % (api['hostname'],api['expirytime'],api['ipaddr']))
+                    logger.debug("Unsupported: clientid: %s in IPv4" \
+                            % (api['clientid']))
+                elif iptype == "ipv6":
+                    logger.debug("PASS: hostname: %s expirytime: %s ipaddr: %s \
+clientid: %s" % (api['hostname'],api['expirytime'],api['ipaddr'],\
+api['clientid']))
+
+                if iptype == "ipv4" and api['mac'] == dnsmasq[j]['mac']:
+                    logger.debug("PASS: mac: %s" % api['mac'])
+                elif iptype == "ipv6" and api['iaid'] == dnsmasq[j]['mac']:
+                    logger.debug("PASS: iaid: %s" % api['iaid'])
+                else:
+                    logger.error("Fail: mac/iaid: %s/%s" % (api['mac'], \
+                                 api['iaid']))
+                    return False
+
+                break
+            else:
+                if j == len(dnsmasq) - 1:
+                    logger.debug("Last loop %d, FAIL: %s" % (j,api))
+                    logger.debug("failed on ipaddr or clientid")
+                    return False
+                else:
+                    logger.debug("Skipped loop %d,Warning: ipaddr: %s \
+clientid: %s" % (j,api['ipaddr'],api['clientid']))
+                    continue
+        else:
+            if j == len(dnsmasq) - 1:
+                logger.error("Fail: hostname: %s expirytime: %s ipaddr: %s \
+clientid: %s"  % (api['hostname'],api['expirytime'],api['ipaddr'], \
+api['clientid']))
+                logger.error("Last loop %d, FAIL: %s" % (j,api))
+                return False
+            else:
+                logger.debug("Skipped loop %d,Warning: hostname: \
+%s expirytime: %s" % (j,api['hostname'],api['expirytime']))
+                continue
+    if not api['iface'] == get_bridge_name(network,logger):
+        logger.error("FAIL: iface: %s" % api['iface'])
+        return False
+    else:
+        logger.debug("PASS: iface: %s" % api['iface'])
+    if not api['type'] == get_network_type(api['ipaddr'],logger):
+        logger.error("FAIL: type: %s" % api['type'])
+        return False
+    else:
+        logger.debug("PASS: type: %s" % api['type'])
+
+    if not api['prefix'] == int(get_ip_prefix(network, iptype ,logger)):
+        logger.error("FAIL: prefix: %s" % api['prefix'])
+        logger.error("FAIL: %s" % api)
+        return False
+    else:
+        logger.debug("PASS: prefix: %s" % api['prefix'])
+    if iptype == "ipv4":
+        if not api['iaid'] == None:
+            logger.error("FAIL: iaid: %s" % api['iaid'])
+            return False
+        else:
+            logger.debug("Unsupported: iaid: %s in IPv4" % api['iaid'])
+            logger.debug("PASS: %s" % api)
+    elif iptype == "ipv6":
+         logger.debug("Ignoring mac checking on IPv6 line %s" % api['mac'])
+         logger.debug("PASS: %s" % api)
+
+    return True
+
+def check_values(op1, op2, network, logger):
+     """
+        check each line accorting to ip type, if ipv4 go to check_ipv4_values
+        if ipv6, go to check_ipv6_values.
+     """
+     networkname = network
+     dnsmasq = op1
+     api = op2
+
+     for i in range(0, len(api)):
+         if check_ip(api[i]['ipaddr'],logger) == "ipv4":
+             if not compare_values(dnsmasq,api[i],networkname,"ipv4",logger):
+                 return False
+         elif check_ip(api[i]['ipaddr'],logger) == "ipv6":
+             if not compare_values(dnsmasq,api[i],networkname,"ipv6",logger):
+                 return False
+         else:
+             logger.error("invalid list element for ipv4 and ipv6")
+             return False
+     return True
+
+def network_dhcp_leases(params):
+    """
+       test API for DHCPLeases in class virNetwork
+    """
+    global LEASE_FILE_DNSMASQ
+    logger = params['logger']
+    networkname = params['networkname']
+    LEASE_FILE_DNSMASQ = "/var/lib/libvirt/dnsmasq/" + networkname + ".leases"
+    mac_value = params.get('macaddr', None)
+    conn = sharedmod.libvirtobj['conn']
+    logger.info("The given mac is %s" % (mac_value))
+
+    if not os.path.exists(LEASE_FILE_DNSMASQ):
+        logger.error("leases file for %s is not exist" % networkname)
+        logger.error("%s" % LEASE_FILE_DNSMASQ)
+        return 1
+    dhcp_lease_dns = get_info_from_dnsmasq(networkname, mac_value, logger)
+    logger.info("From dnsmasq: %s" % (dhcp_lease_dns))
+    if not dhcp_lease_dns:
+        return 1
+
+    netobj = conn.networkLookupByName(networkname)
+
+    try:
+        dhcp_lease_api = netobj.DHCPLeases(mac_value,0)
+        if not dhcp_lease_api and dhcp_lease_dns:
+            logger.info("From API: %s" % (dhcp_lease_api))
+            return 1 
+        logger.info("From API: %s" % (dhcp_lease_api))
+        if not check_values(dhcp_lease_dns,dhcp_lease_api,networkname,logger):
+           return 1
+
+    except libvirtError, e:
+        logger.error("API error message: %s, error code is %s" \
+                     % (e.message, e.get_error_code()))
+        return 1
+
+    return 0
-- 
1.8.3.1

--
libvir-list mailing list
libvir-list@xxxxxxxxxx
https://www.redhat.com/mailman/listinfo/libvir-list




[Index of Archives]     [Virt Tools]     [Libvirt Users]     [Lib OS Info]     [Fedora Users]     [Fedora Desktop]     [Fedora SELinux]     [Big List of Linux Books]     [Yosemite News]     [KDE Users]     [Fedora Tools]