Add skeleton enforce() method to Enforcer

The `enforce()` method is going to be the main entry point into
oslo.limit for services enforcing usage against unified limits.

Future patches will add methods for collecting information from
keystone, resolving callback functions for project resource usage, and
implement different enforcement models (flat, strict-two-level).

Change-Id: I3cca109213e6d3fad91160ebb632d15499690093
This commit is contained in:
Lance Bragstad 2019-06-17 15:16:12 +00:00
parent 5db5a5930a
commit 996edbae12
2 changed files with 43 additions and 28 deletions

View File

@ -12,28 +12,50 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
class Enforcer(object):
def __init__(self, deltas, callback=None):
def __init__(self, usage_callback):
"""An object for checking usage against resource limits and requests.
:param deltas: An dictionary containing resource names as keys and
requests resource quantities as values.
:type deltas: dictionary
:param callback: A callable function that accepts a project_id string
as a parameter and calculates the current usage of a
resource.
:param usage_callback: A callable function that accepts a project_id
string as a parameter and calculates the current
usage of a resource.
:type callable function:
"""
if not callable(usage_callback):
msg = 'usage_callback must be a callable function.'
raise ValueError(msg)
self.usage_callback = usage_callback
def enforce(self, project_id, deltas, filter_resources=None):
"""Check resource usage against limits and request deltas.
:param project_id: The project to check usage and enforce limits
against.
:type project_id: string
:param deltas: An dictionary containing resource names as keys and
requests resource quantities as values.
:type deltas: dictionary
:param filter_resources: A list of strings containing the resource
names to filter the return values of the
usage_callback. This is a performance
optimization in the event the caller doesn't
want the usage_callback to collect all
resources owned by the service. By default,
no resources will be filtered.
"""
if not isinstance(project_id, six.text_type):
msg = 'project_id must be a string.'
raise ValueError(msg)
if not isinstance(deltas, dict):
msg = 'deltas must be a dictionary.'
raise ValueError(msg)
if callback and not callable(callback):
msg = 'callback must be a callable function.'
if not isinstance(filter_resources, list):
msg = 'filter_resources must be a list.'
raise ValueError(msg)
self.deltas = deltas
self.callback = callback

View File

@ -36,35 +36,28 @@ class TestEnforcer(base.BaseTestCase):
return 8
def test_required_parameters(self):
enforcer = limit.Enforcer(self.deltas)
enforcer = limit.Enforcer(self._get_usage_for_project)
self.assertEqual(self._get_usage_for_project, enforcer.usage_callback)
self.assertEqual(self.deltas, enforcer.deltas)
self.assertIsNone(enforcer.callback)
def test_optional_parameters(self):
callback = self._get_usage_for_project
enforcer = limit.Enforcer(self.deltas, callback=callback)
self.assertEqual(self.deltas, enforcer.deltas)
self.assertEqual(self._get_usage_for_project, enforcer.callback)
def test_callback_must_be_callable(self):
def test_usage_callback_must_be_callable(self):
invalid_callback_types = [uuid.uuid4().hex, 5, 5.1]
for invalid_callback in invalid_callback_types:
self.assertRaises(
ValueError,
limit.Enforcer,
self.deltas,
callback=invalid_callback
invalid_callback
)
def test_deltas_must_be_a_dictionary(self):
project_id = uuid.uuid4().hex
invalid_delta_types = [uuid.uuid4().hex, 5, 5.1, True, False, []]
enforcer = limit.Enforcer(self._get_usage_for_project)
for invalid_delta in invalid_delta_types:
self.assertRaises(
ValueError,
limit.Enforcer,
invalid_delta,
enforcer.enforce,
project_id,
invalid_delta
)