test_resource_manager.py 文件源码

def test_lock_unavailable_resource_timeout(self):
        """Lock an already locked resource & validate failure after timeout.

        * Validates the DB initial state.
        * Locks an already locked resource, using resource client.
        * Validates a ResourceUnavailableError is raised.
        * Validates 'lock_resources' duration is greater then the timeout.
        """
        resources_num = DemoResourceData.objects.filter(~Q(owner=""),
                                  name=self.LOCKED1_NAME).count()

        self.assertEquals(resources_num, 1, "Expected 1 locked "
                          "resource with name %r in DB found %d"
                          % (self.LOCKED1_NAME, resources_num))

        descriptor = Descriptor(DemoResource, name=self.LOCKED1_NAME)

        start_time = time.time()
        self.assertRaises(ResourceUnavailableError,
                          self.client._lock_resources,
                          descriptors=[descriptor],
                          timeout=self.LOCK_TIMEOUT)

        duration = time.time() - start_time
        self.assertGreaterEqual(duration, self.LOCK_TIMEOUT, "Waiting for "
                                "resources took %.2f seconds, but should take "
                                "at least %d" % (duration, self.LOCK_TIMEOUT))