Synnefo » snf-ganeti

  """Test class that supports adding/waiting on threads"""

  def setUp(self):

    unittest.TestCase.setUp(self)

    self.threads = []

  def _addThread(self, *args, **kwargs):

    _ThreadedTestCase.setUp(self)

    self.lock = threading.Lock()

    self.cond = locking._PipeCondition(self.lock)

  def testAcquireRelease(self):

    self.assert_(not self.cond._is_owned())

  def setUp(self):

    _ThreadedTestCase.setUp(self)

    self.sl = locking.SharedLock()

  def testSequenceAndOwnership(self):

    self.assert_(not self.sl._is_owned())

  def setUp(self):

    _ThreadedTestCase.setUp(self)

  @locking.ssynchronized(_decoratorlock)

  def _doItExclusive(self):

  def setUp(self):

    _ThreadedTestCase.setUp(self)

    self._setUpLS()

  def _setUpLS(self):

    """Helper to (re)initialize the lock set"""

    self.instances=['i1', 'i2', 'i3']

    self.GL = locking.GanetiLockManager(nodes=self.nodes,

                                        instances=self.instances)

  def tearDown(self):

    # Don't try this at home...