def testThreadLeak(self):
     mailer = sm.SPM_MailMonitor(StoragePoolStub(), 100)
     threadCount = len(threading.enumerate())
     mailer.stop()
     mailer.run()
     t = lambda: self.assertEquals(threadCount, len(threading.enumerate()))
     retry(AssertionError, t, timeout=4, sleep=0.1)

 def testZombie(self):
     args = [EXT_SLEEP, "0"]
     sproc = commands.execCmd(args, sync=False)
     sproc.kill()
     try:
         test = lambda: self.assertEquals(utils.getCmdArgs(sproc.pid),
                                          tuple())
         utils.retry(AssertionError, test, tries=10, sleep=0.1)
     finally:
         sproc.wait()

    def testStop(self):
        p = rhandler.PoolHandler()
        procPath = os.path.join("/proc", str(p.process.pid))

        # Make sure handler is running
        self.assertTrue(p.proxy.callCrabRPCFunction(4, "os.path.exists",
                                                    procPath))
        p.stop()
        test = lambda: self.assertFalse(os.path.exists(procPath))

        utils.retry(test, AssertionError, timeout=4, sleep=0.1)

    def _connect(self):
        self._manager = _SuperVdsmManager(address=ADDRESS, authkey='')
        self._manager.register('instance')
        self._manager.register('open')
        self._log.debug("Trying to connect to Super Vdsm")
        try:
            utils.retry(self._manager.connect, Exception, timeout=60, tries=3)
        except Exception as ex:
            msg = "Connect to supervdsm service failed: %s" % ex
            panic(msg)

        self._svdsm = self._manager.instance()

 def retryAssert(self, *args, **kwargs):
     '''Keep retrying an assertion if AssertionError is raised.
        See function utils.retry for the meaning of the arguments.
     '''
     # the utils module only can be imported correctly after
     # hackVdsmModule() is called. Do not import it at the
     # module level.
     from vdsm.utils import retry
     return retry(expectedException=AssertionError, *args, **kwargs)

    def testThreadLeak(self):
        with make_env() as env:
            mailer = sm.SPM_MailMonitor(
                SPUUID, 100,
                inbox=env.inbox,
                outbox=env.outbox,
                monitorInterval=MONITOR_INTERVAL)
            try:
                threadCount = len(threading.enumerate())
                mailer.stop()
                mailer.run()

                t = lambda: self.assertEqual(
                    threadCount, len(threading.enumerate()))
                retry(AssertionError, t, timeout=4, sleep=0.1)
            finally:
                self.assertTrue(
                    mailer.wait(timeout=MAILER_TIMEOUT),
                    msg='mailer.wait: Timeout expired')

 def _recoverThread(self):
     # Trying to run recover process until it works. During that time vdsm
     # stays in recovery mode (_recover=True), means all api requests
     # returns with "vdsm is in initializing process" message.
     utils.retry(self._recoverExistingVms, sleep=5)

 def _connect(self):
     retry(self.start, (socket.error, KeyError), tries=30)

 def __init__(self):
     retry(self.start, (socket.error, KeyError), tries=30)

def get(cif=None):
    """Return current connection to libvirt or open a new one.

    Wrap methods of connection object so that they catch disconnection, and
    take vdsm down.
    """
    def wrapMethod(f):
        def wrapper(*args, **kwargs):
            try:
                ret = f(*args, **kwargs)
                if isinstance(ret, libvirt.virDomain):
                    for name in dir(ret):
                        method = getattr(ret, name)
                        if callable(method) and name[0] != '_':
                            setattr(ret, name, wrapMethod(method))
                return ret
            except libvirt.libvirtError as e:
                edom = e.get_error_domain()
                ecode = e.get_error_code()
                EDOMAINS = (libvirt.VIR_FROM_REMOTE,
                            libvirt.VIR_FROM_RPC)
                ECODES = (libvirt.VIR_ERR_SYSTEM_ERROR,
                          libvirt.VIR_ERR_INTERNAL_ERROR,
                          libvirt.VIR_ERR_NO_CONNECT,
                          libvirt.VIR_ERR_INVALID_CONN)
                if edom in EDOMAINS and ecode in ECODES:
                    cif.log.error('connection to libvirt broken. '
                                  'taking vdsm down. ecode: %d edom: %d',
                                  ecode, edom)
                    cif.prepareForShutdown()
                else:
                    cif.log.debug('Unknown libvirterror: ecode: %d edom: %d '
                                  'level: %d message: %s', ecode, edom,
                                  e.get_error_level(), e.get_error_message())
                raise
        wrapper.__name__ = f.__name__
        wrapper.__doc__ = f.__doc__
        return wrapper

    def req(credentials, user_data):
        passwd = file(constants.P_VDSM_LIBVIRT_PASSWD).readline().rstrip("\n")
        for cred in credentials:
            if cred[0] == libvirt.VIR_CRED_AUTHNAME:
                cred[4] = constants.SASL_USERNAME
            elif cred[0] == libvirt.VIR_CRED_PASSPHRASE:
                cred[4] = passwd
        return 0

    auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE],
            req, None]

    with __connectionLock:
        conn = __connections.get(id(cif))
        if not conn:
            libvirtOpenAuth = functools.partial(libvirt.openAuth,
                                                'qemu:///system', auth, 0)
            logging.debug('trying to connect libvirt')
            conn = utils.retry(libvirtOpenAuth, timeout=10, sleep=0.2)
            __connections[id(cif)] = conn
            if cif is not None:
                for ev in (libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE,
                           libvirt.VIR_DOMAIN_EVENT_ID_REBOOT,
                           libvirt.VIR_DOMAIN_EVENT_ID_RTC_CHANGE,
                           libvirt.VIR_DOMAIN_EVENT_ID_IO_ERROR_REASON,
                           libvirt.VIR_DOMAIN_EVENT_ID_GRAPHICS,
                           libvirt.VIR_DOMAIN_EVENT_ID_BLOCK_JOB,
                           libvirt.VIR_DOMAIN_EVENT_ID_WATCHDOG):
                    conn.domainEventRegisterAny(None, ev,
                                                __eventCallback, (cif, ev))
                for name in dir(libvirt.virConnect):
                    method = getattr(conn, name)
                    if callable(method) and name[0] != '_':
                        setattr(conn, name, wrapMethod(method))
            # In case we're running into troubles with keeping the connections
            # alive we should place here:
            # conn.setKeepAlive(interval=5, count=3)
            # However the values need to be considered wisely to not affect
            # hosts which are hosting a lot of virtual machines

        return conn