def testProcessDeviceParamsInvalidEncoding(self):
        deviceXML = hostdev._process_device_params(
            libvirtconnection.get().nodeDeviceLookupByName(
                _COMPUTER_DEVICE).XMLDesc()
        )

        self.assertEqual(_COMPUTER_DEVICE_PROCESSED, deviceXML)

    def testProcessNetDeviceParams(self):
        deviceXML = hostdev._process_device_params(
            libvirtconnection.get().nodeDeviceLookupByName(
                _NET_DEVICE).XMLDesc()
        )

        self.assertEqual(_NET_DEVICE_PROCESSED, deviceXML)

def unregister(uuids):
    try:
        uuids = [str(uuid.UUID(s)) for s in uuids]
    except ValueError as e:
        logging.warning("Attempt to unregister invalid uuid %s: %s" %
                        (uuids, e))
        return response.error("secretBadRequestErr")

    con = libvirtconnection.get()
    try:
        for sec_uuid in uuids:
            logging.info("Unregistering secret %r", sec_uuid)
            try:
                virsecret = con.secretLookupByUUIDString(sec_uuid)
            except libvirt.libvirtError as e:
                if e.get_error_code() != libvirt.VIR_ERR_NO_SECRET:
                    raise
                logging.debug("No such secret %r", sec_uuid)
            else:
                virsecret.undefine()
    except libvirt.libvirtError as e:
        logging.error("Could not unregister secrets: %s", e)
        return response.error("secretUnregisterErr")

    return response.success()

def networks():
    """
    Get dict of networks from libvirt

    :returns: dict of networkname={properties}
    :rtype: dict of dict
            { 'ovirtmgmt': { 'bridge': 'ovirtmgmt', 'bridged': True}
              'red': { 'iface': 'red', 'bridged': False}}
    """
    nets = {}
    conn = libvirtconnection.get()
    allNets = ((net, net.name()) for net in conn.listAllNetworks(0))
    for net, netname in allNets:
        if netname.startswith(LIBVIRT_NET_PREFIX):
            netname = netname[len(LIBVIRT_NET_PREFIX):]
            nets[netname] = {}
            xml = etree.fromstring(net.XMLDesc(0))
            interface = xml.find('.//interface')
            if interface is not None:
                nets[netname]['iface'] = interface.get('dev')
                nets[netname]['bridged'] = False
            else:
                nets[netname]['bridge'] = xml.find('.//bridge').get('name')
                nets[netname]['bridged'] = True
    return nets

def attach(queries, reverse=False, callback="attached"):
    def _getDeviceXML(device_xml):
        devXML = xml.etree.ElementTree.fromstring(device_xml)
        caps = devXML.find("capability")
        bus = caps.find("bus").text
        device = caps.find("device").text

        doc = getDOMImplementation().createDocument(None, "hostdev", None)
        hostdev = doc.documentElement
        hostdev.setAttribute("mode", "subsystem")
        hostdev.setAttribute("type", "usb")

        source = doc.createElement("source")
        hostdev.appendChild(source)

        address = doc.createElement("address")
        address.setAttribute("bus", bus)
        address.setAttribute("device", device)
        source.appendChild(address)

        return doc.toxml()

    vm_id = queries["vmId"][0]
    dev_name = queries["devname"][0]

    c = libvirtconnection.get()
    domain = c.lookupByUUIDString(vm_id)
    device_xml = _getDeviceXML(c.nodeDeviceLookupByName(dev_name).XMLDesc())
    if reverse:
        domain.detachDevice(device_xml)
    else:
        domain.attachDevice(device_xml)
    return "%s(\"%s\", \"%s\");" % (callback, vm_id, dev_name)

def main():
    """
    Defines network filters on libvirt
    """
    conn = libvirtconnection.get()
    NoMacSpoofingFilter().defineNwFilter(conn)
    conn.close()

 def testCallSucceeded(self):
     """Positive test - libvirtMock does not raise any errors"""
     with run_libvirt_event_loop():
         LibvirtMock.virConnect.failGetLibVersion = False
         LibvirtMock.virConnect.failNodeDeviceLookupByName = False
         connection = libvirtconnection.get()
         connection.nodeDeviceLookupByName()

    def testListByCaps(self, caps):
        devices = hostdev.list_by_caps(
            libvirtconnection.get().vmContainer, caps)

        for cap in caps:
            self.assertTrue(set(DEVICES_BY_CAPS[cap].keys()).
                            issubset(devices.keys()))

    def testParseSRIOV_VFDeviceParams(self):
        deviceXML = hostdev._parse_device_params(
            libvirtconnection.get().nodeDeviceLookupByName(
                _SRIOV_VF).XMLDesc()
        )

        self.assertEquals(_SRIOV_VF_PARSED, deviceXML)

def start(cif):
    global _operations

    _scheduler.start()
    _executor.start()

    def per_vm_operation(func, period):
        disp = VmDispatcher(cif.getVMs, _executor, func, _timeout_from(period))
        return Operation(disp, period)

    _operations = [
        # needs dispatching becuse updating the volume stats needs the
        # access the storage, thus can block.
        per_vm_operation(UpdateVolumes, config.getint("irs", "vol_size_sample_interval")),
        # needs dispatching becuse access FS and libvirt data
        per_vm_operation(NumaInfoMonitor, config.getint("vars", "vm_sample_numa_interval")),
        # Job monitoring need QEMU monitor access.
        per_vm_operation(BlockjobMonitor, config.getint("vars", "vm_sample_jobs_interval")),
        # libvirt sampling using bulk stats can block, but unresponsive
        # domains are handled inside VMBulkSampler for performance reasons;
        # thus, does not need dispatching.
        Operation(
            sampling.VMBulkSampler(libvirtconnection.get(cif), cif.getVMs, sampling.stats_cache),
            config.getint("vars", "vm_sample_interval"),
        ),
        # we do this only until we get high water mark notifications
        # from qemu. Access storage and/or qemu monitor, so can block,
        # thus we need dispatching.
        per_vm_operation(DriveWatermarkMonitor, config.getint("vars", "vm_watermark_interval")),
    ]

    for op in _operations:
        op.start()

    def testProcessSRIOV_VFDeviceParams(self):
        deviceXML = hostdev._process_device_params(
            libvirtconnection.get().nodeDeviceLookupByName(
                _SRIOV_VF).XMLDesc()
        )

        self.assertEqual(_SRIOV_VF_PROCESSED, deviceXML)

    def testParseNetDeviceParams(self):
        deviceXML = hostdev._parse_device_params(
            libvirtconnection.get().nodeDeviceLookupByName(
                _NET_DEVICE).XMLDesc()
        )

        self.assertEquals(_NET_DEVICE_PARSED, deviceXML)

def main():
    portProfile = os.environ.get('vmfex')
    if portProfile is not None:
        handleDirectPool(libvirtconnection.get())
        doc = hooking.read_domxml()
        interface, = doc.getElementsByTagName('interface')
        attachProfileToInterfaceXml(interface, portProfile)
        hooking.write_domxml(doc)

def removeNetwork(network):
    netName = netinfo.LIBVIRT_NET_PREFIX + network
    conn = libvirtconnection.get()

    net = conn.networkLookupByName(netName)
    if net.isActive():
        net.destroy()
    if net.isPersistent():
        net.undefine()

def flush():
    conn = libvirtconnection.get()
    allNets = ((net, net.name()) for net in conn.listAllNetworks(0))
    for net, netname in allNets:
        if netname.startswith(netinfo.LIBVIRT_NET_PREFIX):
            if net.isActive():
                net.destroy()
            if net.isPersistent():
                net.undefine()

def removeNetwork(network):
    netName = LIBVIRT_NET_PREFIX + network
    conn = libvirtconnection.get()

    net = _netlookup_by_name(conn, netName)
    if net:
        if net.isActive():
            net.destroy()
        if net.isPersistent():
            net.undefine()

def getNetworkDef(network):
    netName = netinfo.LIBVIRT_NET_PREFIX + network
    conn = libvirtconnection.get()
    try:
        net = conn.networkLookupByName(netName)
        return net.XMLDesc(0)
    except libvirt.libvirtError as e:
        if e.get_error_code() == libvirt.VIR_ERR_NO_NETWORK:
            return

        raise

def getMemoryStatsByNumaCell(cell):
    """
    Get the memory stats of a specified numa node, the unit is MiB.

    :param cell: the index of numa node
    :type cell: int
    :return: dict like {'total': '49141', 'free': '46783'}
    """
    cellMemInfo = libvirtconnection.get().getMemoryStats(cell, 0)
    cellMemInfo['total'] = str(cellMemInfo['total'] / 1024)
    cellMemInfo['free'] = str(cellMemInfo['free'] / 1024)
    return cellMemInfo

def _list_domains():
    conn = libvirtconnection.get()
    for dom_uuid in conn.listDomainsID():
        try:
            dom_obj = conn.lookupByID(dom_uuid)
            dom_xml = dom_obj.XMLDesc(0)
        except libvirt.libvirtError as e:
            if e.get_error_code() == libvirt.VIR_ERR_NO_DOMAIN:
                logging.exception("domain %s is dead", dom_uuid)
            else:
                raise
        else:
            yield dom_obj, dom_xml

 def testCallFailedConnectionUp(self):
     """
     libvirtMock will raise an error when nodeDeviceLookupByName is called.
     When getLibVersion is called
     (used by libvirtconnection to recognize disconnections)
     it will not raise an error -> in that case an error should be raised
     ('Unknown libvirterror').
     """
     connection = libvirtconnection.get(killOnFailure=True)
     LibvirtMock.virConnect.failNodeDeviceLookupByName = True
     LibvirtMock.virConnect.failGetLibVersion = False
     self.assertRaises(LibvirtMock.libvirtError,
                       connection.nodeDeviceLookupByName)