test_resource_manager.py 文件源码

def test_lock_already_locked_resource(self):
        """Lock an already locked resource & validate failure.

        * Validates the DB initial state.
        * Locks an already locked resource, using resource client.
        * Validates a ResourceUnavailableError is raised.
        """
        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)
        self.assertRaises(ResourceUnavailableError,
                          self.client._lock_resources,
                          descriptors=[descriptor],
                          timeout=self.LOCK_TIMEOUT)