Rearchitect quota checking to partially fix bug 938317.

This is a rearchitecting/rewriting of quota handling to correct the
quota atomicity issues highlighted by bug 938317.  Partially implements
blueprint quota-refactor as well.

This change is fairly substantial.  To make it easier to review, it has been
broken up into 3 parts.  This is the first part.

Change-Id: I805f5750c08de17487e59fe33fad0bed203188a6

8 files changed, 3026 insertions, 40 deletions

diff --git a/nova/db/api.py b/nova/db/api.py
index fed92072d..f2f74cc55 100644
--- a/nova/db/api.py
+++ b/nova/db/api.py
@@ -251,9 +251,10 @@ def floating_ip_create(context, values):
     return IMPL.floating_ip_create(context, values)
 
 
-def floating_ip_count_by_project(context, project_id):
+def floating_ip_count_by_project(context, project_id, session=None):
     """Count floating ips used by project."""
-    return IMPL.floating_ip_count_by_project(context, project_id)
+    return IMPL.floating_ip_count_by_project(context, project_id,
+                                             session=session)
 
 
 def floating_ip_deallocate(context, address):
@@ -520,9 +521,10 @@ def instance_create(context, values):
     return IMPL.instance_create(context, values)
 
 
-def instance_data_get_for_project(context, project_id):
+def instance_data_get_for_project(context, project_id, session=None):
     """Get (instance_count, total_cores, total_ram) for project."""
-    return IMPL.instance_data_get_for_project(context, project_id)
+    return IMPL.instance_data_get_for_project(context, project_id,
+                                              session=session)
 
 
 def instance_destroy(context, instance_id):
@@ -900,11 +902,6 @@ def quota_destroy(context, project_id, resource):
     return IMPL.quota_destroy(context, project_id, resource)
 
 
-def quota_destroy_all_by_project(context, project_id):
-    """Destroy all quotas associated with a given project."""
-    return IMPL.quota_get_all_by_project(context, project_id)
-
-
 ###################
 
 
@@ -941,6 +938,93 @@ def quota_class_destroy_all_by_name(context, class_name):
 ###################
 
 
+def quota_usage_create(context, project_id, resource, in_use, reserved,
+                       until_refresh):
+    """Create a quota usage for the given project and resource."""
+    return IMPL.quota_usage_create(context, project_id, resource,
+                                   in_use, reserved, until_refresh)
+
+
+def quota_usage_get(context, project_id, resource):
+    """Retrieve a quota usage or raise if it does not exist."""
+    return IMPL.quota_usage_get(context, project_id, resource)
+
+
+def quota_usage_get_all_by_project(context, project_id):
+    """Retrieve all usage associated with a given resource."""
+    return IMPL.quota_usage_get_all_by_project(context, project_id)
+
+
+def quota_usage_update(context, class_name, resource, in_use, reserved,
+                       until_refresh):
+    """Update a quota usage or raise if it does not exist."""
+    return IMPL.quota_usage_update(context, project_id, resource,
+                                   in_use, reserved, until_refresh)
+
+
+def quota_usage_destroy(context, project_id, resource):
+    """Destroy the quota usage or raise if it does not exist."""
+    return IMPL.quota_usage_destroy(context, project_id, resource)
+
+
+###################
+
+
+def reservation_create(context, uuid, usage, project_id, resource, delta,
+                       expire):
+    """Create a reservation for the given project and resource."""
+    return IMPL.reservation_create(context, uuid, usage, project_id,
+                                   resource, delta, expire)
+
+
+def reservation_get(context, uuid):
+    """Retrieve a reservation or raise if it does not exist."""
+    return IMPL.reservation_get(context, uuid)
+
+
+def reservation_get_all_by_project(context, project_id):
+    """Retrieve all reservations associated with a given project."""
+    return IMPL.reservation_get_all_by_project(context, project_id)
+
+
+def reservation_destroy(context, uuid):
+    """Destroy the reservation or raise if it does not exist."""
+    return IMPL.reservation_destroy(context, uuid)
+
+
+###################
+
+
+def quota_reserve(context, resources, quotas, deltas, expire,
+                  until_refresh, max_age):
+    """Check quotas and create appropriate reservations."""
+    return IMPL.quota_reserve(context, resources, quotas, deltas, expire,
+                              until_refresh, max_age)
+
+
+def reservation_commit(context, reservations):
+    """Commit quota reservations."""
+    return IMPL.reservation_commit(context, reservations)
+
+
+def reservation_rollback(context, reservations):
+    """Roll back quota reservations."""
+    return IMPL.reservation_rollback(context, reservations)
+
+
+def quota_destroy_all_by_project(context, project_id):
+    """Destroy all quotas associated with a given project."""
+    return IMPL.quota_get_all_by_project(context, project_id)
+
+
+def reservation_expire(context):
+    """Roll back any expired reservations."""
+    return IMPL.reservation_expire(context)
+
+
+###################
+
+
 def volume_allocate_iscsi_target(context, volume_id, host):
     """Atomically allocate a free iscsi_target from the pool."""
     return IMPL.volume_allocate_iscsi_target(context, volume_id, host)
@@ -956,9 +1040,10 @@ def volume_create(context, values):
     return IMPL.volume_create(context, values)
 
 
-def volume_data_get_for_project(context, project_id):
+def volume_data_get_for_project(context, project_id, session=None):
     """Get (volume_count, gigabytes) for project."""
-    return IMPL.volume_data_get_for_project(context, project_id)
+    return IMPL.volume_data_get_for_project(context, project_id,
+                                            session=session)
 
 
 def volume_destroy(context, volume_id):
@@ -1161,9 +1246,10 @@ def security_group_destroy(context, security_group_id):
     return IMPL.security_group_destroy(context, security_group_id)
 
 
-def security_group_count_by_project(context, project_id):
+def security_group_count_by_project(context, project_id, session=None):
     """Count number of security groups in a project."""
-    return IMPL.security_group_count_by_project(context, project_id)
+    return IMPL.security_group_count_by_project(context, project_id,
+                                                session=session)
 
 
 ####################
diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py
index 4574e8033..3a4e8ea1b 100644
--- a/nova/db/sqlalchemy/api.py
+++ b/nova/db/sqlalchemy/api.py
@@ -677,10 +677,11 @@ def floating_ip_create(context, values):
 
 
 @require_context
-def floating_ip_count_by_project(context, project_id):
+def floating_ip_count_by_project(context, project_id, session=None):
     authorize_project_context(context, project_id)
     # TODO(tr3buchet): why leave auto_assigned floating IPs out?
-    return model_query(context, models.FloatingIp, read_deleted="no").\
+    return model_query(context, models.FloatingIp, read_deleted="no",
+                       session=session).\
                    filter_by(project_id=project_id).\
                    filter_by(auto_assigned=False).\
                    count()
@@ -1295,12 +1296,13 @@ def instance_create(context, values):
 
 
 @require_admin_context
-def instance_data_get_for_project(context, project_id):
+def instance_data_get_for_project(context, project_id, session=None):
     result = model_query(context,
                          func.count(models.Instance.id),
                          func.sum(models.Instance.vcpus),
                          func.sum(models.Instance.memory_mb),
-                         read_deleted="no").\
+                         read_deleted="no",
+                         session=session).\
                      filter_by(project_id=project_id).\
                      first()
     # NOTE(vish): convert None to 0
@@ -2293,19 +2295,6 @@ def quota_destroy(context, project_id, resource):
         quota_ref.delete(session=session)
 
 
-@require_admin_context
-def quota_destroy_all_by_project(context, project_id):
-    session = get_session()
-    with session.begin():
-        quotas = model_query(context, models.Quota, session=session,
-                             read_deleted="no").\
-                         filter_by(project_id=project_id).\
-                         all()
-
-        for quota_ref in quotas:
-            quota_ref.delete(session=session)
-
-
 ###################
 
 
@@ -2383,6 +2372,370 @@ def quota_class_destroy_all_by_name(context, class_name):
 ###################
 
 
+@require_context
+def quota_usage_get(context, project_id, resource, session=None):
+    result = model_query(context, models.QuotaUsage, session=session,
+                         read_deleted="no").\
+                     filter_by(project_id=project_id).\
+                     filter_by(resource=resource).\
+                     first()
+
+    if not result:
+        raise exception.QuotaUsageNotFound(project_id=project_id)
+
+    return result
+
+
+@require_context
+def quota_usage_get_all_by_project(context, project_id):
+    authorize_project_context(context, project_id)
+
+    rows = model_query(context, models.QuotaUsage, read_deleted="no").\
+                   filter_by(project_id=project_id).\
+                   all()
+
+    result = {'project_id': project_id}
+    for row in rows:
+        result[row.resource] = dict(in_use=row.in_use, reserved=row.reserved)
+
+    return result
+
+
+@require_admin_context
+def quota_usage_create(context, project_id, resource, in_use, reserved,
+                       until_refresh, session=None, save=True):
+    quota_usage_ref = models.QuotaUsage()
+    quota_usage_ref.project_id = project_id
+    quota_usage_ref.resource = resource
+    quota_usage_ref.in_use = in_use
+    quota_usage_ref.reserved = reserved
+    quota_usage_ref.until_refresh = until_refresh
+
+    # Allow us to hold the save operation until later; keeps the
+    # transaction in quota_reserve() from breaking too early
+    if save:
+        quota_usage_ref.save(session=session)
+
+    return quota_usage_ref
+
+
+@require_admin_context
+def quota_usage_update(context, project_id, resource, in_use, reserved,
+                       until_refresh, session=None):
+    def do_update(session):
+        quota_usage_ref = quota_usage_get(context, project_id, resource,
+                                          session=session)
+        quota_usage_ref.in_use = in_use
+        quota_usage_ref.reserved = reserved
+        quota_usage_ref.until_refresh = until_refresh
+        quota_usage_ref.save(session=session)
+
+    if session:
+        # Assume caller started a transaction
+        do_update(session)
+    else:
+        session = get_session()
+        with session.begin():
+            do_update(session)
+
+
+@require_admin_context
+def quota_usage_destroy(context, project_id, resource):
+    session = get_session()
+    with session.begin():
+        quota_usage_ref = quota_usage_get(context, project_id, resource,
+                                          session=session)
+        quota_usage_ref.delete(session=session)
+
+
+###################
+
+
+@require_context
+def reservation_get(context, uuid, session=None):
+    result = model_query(context, models.Reservation, session=session,
+                         read_deleted="no").\
+                     filter_by(uuid=uuid).\
+                     first()
+
+    if not result:
+        raise exception.ReservationNotFound(uuid=uuid)
+
+    return result
+
+
+@require_context
+def reservation_get_all_by_project(context, project_id):
+    authorize_project_context(context, project_id)
+
+    rows = model_query(context, models.QuotaUsage, read_deleted="no").\
+                   filter_by(project_id=project_id).\
+                   all()
+
+    result = {'project_id': project_id}
+    for row in rows:
+        result.setdefault(row.resource, {})
+        result[row.resource][row.uuid] = row.delta
+
+    return result
+
+
+@require_admin_context
+def reservation_create(context, uuid, usage, project_id, resource, delta,
+                       expire, session=None):
+    reservation_ref = models.Reservation()
+    reservation_ref.uuid = uuid
+    reservation_ref.usage = usage
+    reservation_ref.project_id = project_id
+    reservation_ref.resource = resource
+    reservation_ref.delta = delta
+    reservation_ref.expire = expire
+    reservation_ref.save(session=session)
+    return reservation_ref
+
+
+@require_admin_context
+def reservation_destroy(context, uuid):
+    session = get_session()
+    with session.begin():
+        reservation_ref = reservation_get(context, uuid, session=session)
+        reservation_ref.delete(session=session)
+
+
+###################
+
+
+def _get_quota_usages(context, session, keys):
+    # Broken out for testability
+    rows = model_query(context, models.QuotaUsage,
+                       read_deleted="no",
+                       session=session).\
+                   filter_by(project_id=context.project_id).\
+                   filter(models.QuotaUsage.resource.in_(keys)).\
+                   with_lockmode('update').\
+                   all()
+    return dict((row.resource, row) for row in rows)
+
+
+@require_context
+def quota_reserve(context, resources, quotas, deltas, expire,
+                  until_refresh, max_age):
+    elevated = context.elevated()
+    session = get_session()
+    with session.begin():
+        # Get the current usages
+        usages = _get_quota_usages(context, session, deltas.keys())
+
+        # Handle usage refresh
+        work = set(deltas.keys())
+        while work:
+            resource = work.pop()
+
+            # Do we need to refresh the usage?
+            refresh = False
+            if resource not in usages:
+                # Note we're inhibiting save...
+                usages[resource] = quota_usage_create(elevated,
+                                                      context.project_id,
+                                                      resource,
+                                                      0, 0,
+                                                      until_refresh or None,
+                                                      session=session,
+                                                      save=False)
+                refresh = True
+            elif usages[resource].until_refresh is not None:
+                usages[resource].until_refresh -= 1
+                if usages[resource].until_refresh <= 0:
+                    refresh = True
+            elif max_age and (usages[resource].updated_at -
+                              utils.utcnow()).seconds >= max_age:
+                refresh = True
+
+            # OK, refresh the usage
+            if refresh:
+                # Grab the sync routine
+                sync = resources[resource].sync
+
+                updates = sync(elevated, context.project_id, session)
+                for res, in_use in updates.items():
+                    # Make sure we have a destination for the usage!
+                    if res not in usages:
+                        # Note we're inhibiting save...
+                        usages[res] = quota_usage_create(elevated,
+                                                         context.project_id,
+                                                         res,
+                                                         0, 0,
+                                                         until_refresh or None,
+                                                         session=session,
+                                                         save=False)
+
+                    # Update the usage
+                    usages[res].in_use = in_use
+                    usages[res].until_refresh = until_refresh or None
+
+                    # Because more than one resource may be refreshed
+                    # by the call to the sync routine, and we don't
+                    # want to double-sync, we make sure all refreshed
+                    # resources are dropped from the work set.
+                    work.discard(res)
+
+                    # NOTE(Vek): We make the assumption that the sync
+                    #            routine actually refreshes the
+                    #            resources that it is the sync routine
+                    #            for.  We don't check, because this is
+                    #            a best-effort mechanism.
+
+        # Check for deltas that would go negative
+        unders = [resource for resource, delta in deltas.items()
+                  if delta < 0 and
+                  delta + usages[resource].in_use < 0]
+
+        # Now, let's check the quotas
+        # NOTE(Vek): We're only concerned about positive increments.
+        #            If a project has gone over quota, we want them to
+        #            be able to reduce their usage without any
+        #            problems.
+        overs = [resource for resource, delta in deltas.items()
+                 if quotas[resource] >= 0 and delta >= 0 and
+                 quotas[resource] < delta + usages[resource].total]
+
+        # NOTE(Vek): The quota check needs to be in the transaction,
+        #            but the transaction doesn't fail just because
+        #            we're over quota, so the OverQuota raise is
+        #            outside the transaction.  If we did the raise
+        #            here, our usage updates would be discarded, but
+        #            they're not invalidated by being over-quota.
+
+        # Create the reservations
+        if not unders and not overs:
+            reservations = []
+            for resource, delta in deltas.items():
+                reservation = reservation_create(elevated,
+                                                 str(utils.gen_uuid()),
+                                                 usages[resource],
+                                                 context.project_id,
+                                                 resource, delta, expire,
+                                                 session=session)
+                reservations.append(reservation.uuid)
+
+                # Also update the reserved quantity
+                # NOTE(Vek): Again, we are only concerned here about
+                #            positive increments.  Here, though, we're
+                #            worried about the following scenario:
+                #
+                #            1) User initiates resize down.
+                #            2) User allocates a new instance.
+                #            3) Resize down fails or is reverted.
+                #            4) User is now over quota.
+                #
+                #            To prevent this, we only update the
+                #            reserved value if the delta is positive.
+                if delta > 0:
+                    usages[resource].reserved += delta
+
+        # Apply updates to the usages table
+        for usage_ref in usages.values():
+            usage_ref.save(session=session)
+
+    if unders:
+        raise exception.InvalidQuotaValue(unders=sorted(unders))
+    if overs:
+        usages = dict((k, dict(in_use=v['in_use'], reserved=v['reserved']))
+                      for k, v in usages.items())
+        raise exception.OverQuota(overs=sorted(overs), quotas=quotas,
+                                  usages=usages)
+
+    return reservations
+
+
+def _quota_reservations(session, context, reservations):
+    """Return the relevant reservations."""
+
+    # Get the listed reservations
+    return model_query(context, models.Reservation,
+                       read_deleted="no",
+                       session=session).\
+                   options(joinedload('usage')).\
+                   filter(models.Reservation.uuid.in_(reservations)).\
+                   with_lockmode('update').\
+                   all()
+
+
+@require_context
+def reservation_commit(context, reservations):
+    session = get_session()
+    with session.begin():
+        for reservation in _quota_reservations(session, context, reservations):
+            if reservation.delta >= 0:
+                reservation.usage.reserved -= reservation.delta
+            reservation.usage.in_use += reservation.delta
+
+            reservation.usage.save(session=session)
+            reservation.delete(session=session)
+
+
+@require_context
+def reservation_rollback(context, reservations):
+    session = get_session()
+    with session.begin():
+        for reservation in _quota_reservations(session, context, reservations):
+            if reservation.delta >= 0:
+                reservation.usage.reserved -= reservation.delta
+                reservation.usage.save(session=session)
+
+            reservation.delete(session=session)
+
+
+@require_admin_context
+def quota_destroy_all_by_project(context, project_id):
+    session = get_session()
+    with session.begin():
+        quotas = model_query(context, models.Quota, session=session,
+                             read_deleted="no").\
+                         filter_by(project_id=project_id).\
+                         all()
+
+        for quota_ref in quotas:
+            quota_ref.delete(session=session)
+
+        quota_usages = model_query(context, models.QuotaUsage,
+                                   session=session, read_deleted="no").\
+                               filter_by(project_id=project_id).\
+                               all()
+
+        for quota_usage_ref in quota_usages:
+            quota_usage_ref.delete(session=session)
+
+        reservations = model_query(context, models.Reservation,
+                                   session=session, read_deleted="no").\
+                               filter_by(project_id=project_id).\
+                               all()
+
+        for reservation_ref in reservations:
+            reservation_ref.delete(session=session)
+
+
+@require_admin_context
+def reservation_expire(context):
+    session = get_session()
+    with session.begin():
+        results = model_query(context, models.Reservation, session=session,
+                              read_deleted="no").\
+                          filter(models.Reservation.expire < utils.utcnow()).\
+                          all()
+
+        if results:
+            for reservation in results:
+                if reservation.delta >= 0:
+                    reservation.usage.reserved -= reservation.delta
+                    reservation.usage.save(session=session)
+
+                reservation.delete(session=session)
+
+
+###################
+
+
 @require_admin_context
 def volume_allocate_iscsi_target(context, volume_id, host):
     session = get_session()
@@ -2438,11 +2791,12 @@ def volume_create(context, values):
 
 
 @require_admin_context
-def volume_data_get_for_project(context, project_id):
+def volume_data_get_for_project(context, project_id, session=None):
     result = model_query(context,
                          func.count(models.Volume.id),
                          func.sum(models.Volume.size),
-                         read_deleted="no").\
+                         read_deleted="no",
+                         session=session).\
                      filter_by(project_id=project_id).\
                      first()
 
@@ -3010,9 +3364,10 @@ def security_group_destroy(context, security_group_id):
 
 
 @require_context
-def security_group_count_by_project(context, project_id):
+def security_group_count_by_project(context, project_id, session=None):
     authorize_project_context(context, project_id)
-    return model_query(context, models.SecurityGroup, read_deleted="no").\
+    return model_query(context, models.SecurityGroup, read_deleted="no",
+                       session=session).\
                    filter_by(project_id=project_id).\
                    count()
 
diff --git a/nova/db/sqlalchemy/migrate_repo/versions/097_quota_usages_reservations.py b/nova/db/sqlalchemy/migrate_repo/versions/097_quota_usages_reservations.py
new file mode 100644
index 000000000..f56cc71b9
--- /dev/null
+++ b/nova/db/sqlalchemy/migrate_repo/versions/097_quota_usages_reservations.py
@@ -0,0 +1,106 @@
+# Copyright 2012 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from sqlalchemy import Boolean, Column, DateTime
+from sqlalchemy import MetaData, Integer, String, Table, ForeignKey
+
+from nova import log as logging
+
+LOG = logging.getLogger(__name__)
+
+
+def upgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+
+    # New tables
+    quota_usages = Table('quota_usages', meta,
+            Column('created_at', DateTime(timezone=False)),
+            Column('updated_at', DateTime(timezone=False)),
+            Column('deleted_at', DateTime(timezone=False)),
+            Column('deleted', Boolean(create_constraint=True, name=None)),
+            Column('id', Integer(), primary_key=True),
+            Column('project_id',
+                   String(length=255, convert_unicode=True,
+                          assert_unicode=None, unicode_error=None,
+                          _warn_on_bytestring=False),
+                   index=True),
+            Column('resource',
+                   String(length=255, convert_unicode=True,
+                          assert_unicode=None, unicode_error=None,
+                          _warn_on_bytestring=False)),
+            Column('in_use', Integer(), nullable=False),
+            Column('reserved', Integer(), nullable=False),
+            Column('until_refresh', Integer(), nullable=True),
+            mysql_engine='InnoDB',
+            mysql_charset='utf8',
+            )
+
+    try:
+        quota_usages.create()
+    except Exception:
+        LOG.error(_("Table |%s| not created!"), repr(quota_usages))
+        raise
+
+    reservations = Table('reservations', meta,
+            Column('created_at', DateTime(timezone=False)),
+            Column('updated_at', DateTime(timezone=False)),
+            Column('deleted_at', DateTime(timezone=False)),
+            Column('deleted', Boolean(create_constraint=True, name=None)),
+            Column('id', Integer(), primary_key=True),
+            Column('uuid',
+                   String(length=36, convert_unicode=True,
+                          assert_unicode=None, unicode_error=None,
+                          _warn_on_bytestring=False), nullable=False),
+            Column('usage_id', Integer(), ForeignKey('quota_usages.id'),
+                   nullable=False),
+            Column('project_id',
+                   String(length=255, convert_unicode=True,
+                          assert_unicode=None, unicode_error=None,
+                          _warn_on_bytestring=False),
+                   index=True),
+            Column('resource',
+                   String(length=255, convert_unicode=True,
+                          assert_unicode=None, unicode_error=None,
+                          _warn_on_bytestring=False)),
+            Column('delta', Integer(), nullable=False),
+            Column('expire', DateTime(timezone=False)),
+            mysql_engine='InnoDB',
+            mysql_charset='utf8',
+            )
+
+    try:
+        reservations.create()
+    except Exception:
+        LOG.error(_("Table |%s| not created!"), repr(reservations))
+        raise
+
+
+def downgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+
+    quota_usages = Table('quota_usages', meta, autoload=True)
+    try:
+        quota_usages.drop()
+    except Exception:
+        LOG.error(_("quota_usages table not dropped"))
+        raise
+
+    reservations = Table('reservations', meta, autoload=True)
+    try:
+        reservations.drop()
+    except Exception:
+        LOG.error(_("reservations table not dropped"))
+        raise
diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py
index 056ecd411..8a446091b 100644
--- a/nova/db/sqlalchemy/models.py
+++ b/nova/db/sqlalchemy/models.py
@@ -439,6 +439,47 @@ class QuotaClass(BASE, NovaBase):
     hard_limit = Column(Integer, nullable=True)
 
 
+class QuotaUsage(BASE, NovaBase):
+    """Represents the current usage for a given resource."""
+
+    __tablename__ = 'quota_usages'
+    id = Column(Integer, primary_key=True)
+
+    project_id = Column(String(255), index=True)
+    resource = Column(String(255))
+
+    in_use = Column(Integer)
+    reserved = Column(Integer)
+
+    @property
+    def total(self):
+        return self.in_use + self.reserved
+
+    until_refresh = Column(Integer, nullable=True)
+
+
+class Reservation(BASE, NovaBase):
+    """Represents a resource reservation for quotas."""
+
+    __tablename__ = 'reservations'
+    id = Column(Integer, primary_key=True)
+    uuid = Column(String(36), nullable=False)
+
+    usage_id = Column(Integer, ForeignKey('quota_usages.id'), nullable=False)
+    usage = relationship(QuotaUsage,
+                         backref=backref('reservations'),
+                         foreign_keys=usage_id,
+                         primaryjoin='and_('
+                              'Reservation.usage_id == QuotaUsage.id,'
+                              'Reservation.deleted == False)')
+
+    project_id = Column(String(255), index=True)
+    resource = Column(String(255))
+
+    delta = Column(Integer)
+    expire = Column(DateTime, nullable=False)
+
+
 class Snapshot(BASE, NovaBase):
     """Represents a block storage device that can be attached to a vm."""
     __tablename__ = 'snapshots'
diff --git a/nova/exception.py b/nova/exception.py
index af4850367..bcca2a6dc 100644
--- a/nova/exception.py
+++ b/nova/exception.py
@@ -689,10 +689,23 @@ class AccessKeyNotFound(NotFound):
     message = _("Access Key %(access_key)s could not be found.")
 
 
+class InvalidReservationExpiration(Invalid):
+    message = _("Invalid reservation expiration %(expire)s.")
+
+
+class InvalidQuotaValue(Invalid):
+    message = _("Change would make usage less than 0 for the following "
+                "resources: %(unders)s")
+
+
 class QuotaNotFound(NotFound):
     message = _("Quota could not be found")
 
 
+class QuotaResourceUnknown(QuotaNotFound):
+    message = _("Unknown quota resources %(unknown)s.")
+
+
 class ProjectQuotaNotFound(QuotaNotFound):
     message = _("Quota for project %(project_id)s could not be found.")
 
@@ -701,6 +714,18 @@ class QuotaClassNotFound(QuotaNotFound):
     message = _("Quota class %(class_name)s could not be found.")
 
 
+class QuotaUsageNotFound(QuotaNotFound):
+    message = _("Quota usage for project %(project_id)s could not be found.")
+
+
+class ReservationNotFound(QuotaNotFound):
+    message = _("Quota reservation %(uuid)s could not be found.")
+
+
+class OverQuota(NovaException):
+    message = _("Quota exceeded for resources: %(overs)s")
+
+
 class SecurityGroupNotFound(NotFound):
     message = _("Security group %(security_group_id)s not found.")
 
diff --git a/nova/quota.py b/nova/quota.py
index 486fe3271..a5c751bec 100644
--- a/nova/quota.py
+++ b/nova/quota.py
@@ -18,10 +18,18 @@
 
 """Quotas for instances, volumes, and floating ips."""
 
+import datetime
+
 from nova import db
+from nova import exception
 from nova import flags
+from nova import log as logging
 from nova.openstack.common import cfg
+from nova.openstack.common import importutils
+from nova import utils
+
 
+LOG = logging.getLogger(__name__)
 
 quota_opts = [
     cfg.IntOpt('quota_instances',
@@ -63,6 +71,18 @@ quota_opts = [
     cfg.IntOpt('quota_key_pairs',
                default=100,
                help='number of key pairs per user'),
+    cfg.IntOpt('reservation_expire',
+               default=86400,
+               help='number of seconds until a reservation expires'),
+    cfg.IntOpt('until_refresh',
+               default=0,
+               help='count of reservations until usage is refreshed'),
+    cfg.IntOpt('max_age',
+               default=0,
+               help='number of seconds between subsequent usage refreshes'),
+    cfg.StrOpt('quota_driver',
+               default='nova.quota.DbQuotaDriver',
+               help='default driver to use for quota checks'),
     ]
 
 FLAGS = flags.FLAGS
@@ -250,3 +270,727 @@ def allowed_injected_file_content_bytes(context, requested_bytes):
 def allowed_injected_file_path_bytes(context):
     """Return the number of bytes allowed in an injected file path."""
     return FLAGS.quota_injected_file_path_bytes
+
+
+class DbQuotaDriver(object):
+    """
+    Driver to perform necessary checks to enforce quotas and obtain
+    quota information.  The default driver utilizes the local
+    database.
+    """
+
+    def get_by_project(self, context, project_id, resource):
+        """Get a specific quota by project."""
+
+        return db.quota_get(context, project_id, resource)
+
+    def get_by_class(self, context, quota_class, resource):
+        """Get a specific quota by quota class."""
+
+        return db.quota_class_get(context, quota_class, resource)
+
+    def get_defaults(self, context, resources):
+        """Given a list of resources, retrieve the default quotas.
+
+        :param context: The request context, for access checks.
+        :param resources: A dictionary of the registered resources.
+        """
+
+        quotas = {}
+        for resource in resources.values():
+            quotas[resource.name] = resource.default
+
+        return quotas
+
+    def get_class_quotas(self, context, resources, quota_class,
+                         defaults=True):
+        """
+        Given a list of resources, retrieve the quotas for the given
+        quota class.
+
+        :param context: The request context, for access checks.
+        :param resources: A dictionary of the registered resources.
+        :param quota_class: The name of the quota class to return
+                            quotas for.
+        :param defaults: If True, the default value will be reported
+                         if there is no specific value for the
+                         resource.
+        """
+
+        quotas = {}
+        class_quotas = db.quota_class_get_all_by_name(context, quota_class)
+        for resource in resources.values():
+            if defaults or resource.name in class_quotas:
+                quotas[resource.name] = class_quotas.get(resource.name,
+                                                         resource.default)
+
+        return quotas
+
+    def get_project_quotas(self, context, resources, project_id,
+                           quota_class=None, defaults=True,
+                           usages=True):
+        """
+        Given a list of resources, retrieve the quotas for the given
+        project.
+
+        :param context: The request context, for access checks.
+        :param resources: A dictionary of the registered resources.
+        :param project_id: The ID of the project to return quotas for.
+        :param quota_class: If project_id != context.project_id, the
+                            quota class cannot be determined.  This
+                            parameter allows it to be specified.  It
+                            will be ignored if project_id ==
+                            context.project_id.
+        :param defaults: If True, the quota class value (or the
+                         default value, if there is no value from the
+                         quota class) will be reported if there is no
+                         specific value for the resource.
+        :param usages: If True, the current in_use and reserved counts
+                       will also be returned.
+        """
+
+        quotas = {}
+        project_quotas = db.quota_get_all_by_project(context, project_id)
+        if usages:
+            project_usages = db.quota_usage_get_all_by_project(context,
+                                                               project_id)
+
+        # Get the quotas for the appropriate class.  If the project ID
+        # matches the one in the context, we use the quota_class from
+        # the context, otherwise, we use the provided quota_class (if
+        # any)
+        if project_id == context.project_id:
+            quota_class = context.quota_class
+        if quota_class:
+            class_quotas = db.quota_class_get_all_by_name(context, quota_class)
+        else:
+            class_quotas = {}
+
+        for resource in resources.values():
+            # Omit default/quota class values
+            if not defaults and resource.name not in project_quotas:
+                continue
+
+            quotas[resource.name] = dict(
+                limit=project_quotas.get(resource.name, class_quotas.get(
+                        resource.name, resource.default)),
+                )
+
+            # Include usages if desired.  This is optional because one
+            # internal consumer of this interface wants to access the
+            # usages directly from inside a transaction.
+            if usages:
+                usage = project_usages.get(resource.name, {})
+                quotas[resource.name].update(
+                    in_use=usage.get('in_use', 0),
+                    reserved=usage.get('reserved', 0),
+                    )
+
+        return quotas
+
+    def _get_quotas(self, context, resources, keys, has_sync):
+        """
+        A helper method which retrieves the quotas for the specific
+        resources identified by keys, and which apply to the current
+        context.
+
+        :param context: The request context, for access checks.
+        :param resources: A dictionary of the registered resources.
+        :param keys: A list of the desired quotas to retrieve.
+        :param has_sync: If True, indicates that the resource must
+                         have a sync attribute; if False, indicates
+                         that the resource must NOT have a sync
+                         attribute.
+        """
+
+        # Filter resources
+        if has_sync:
+            sync_filt = lambda x: hasattr(x, 'sync')
+        else:
+            sync_filt = lambda x: not hasattr(x, 'sync')
+        desired = set(keys)
+        sub_resources = dict((k, v) for k, v in resources.items()
+                             if k in desired and sync_filt(v))
+
+        # Make sure we accounted for all of them...
+        if len(keys) != len(sub_resources):
+            unknown = desired - set(sub_resources.keys())
+            raise exception.QuotaResourceUnknown(unknown=sorted(unknown))
+
+        # Grab and return the quotas (without usages)
+        quotas = self.get_project_quotas(context, sub_resources,
+                                         context.project_id,
+                                         context.quota_class, usages=False)
+
+        return dict((k, v['limit']) for k, v in quotas.items())
+
+    def limit_check(self, context, resources, values):
+        """Check simple quota limits.
+
+        For limits--those quotas for which there is no usage
+        synchronization function--this method checks that a set of
+        proposed values are permitted by the limit restriction.
+
+        This method will raise a QuotaResourceUnknown exception if a
+        given resource is unknown or if it is not a simple limit
+        resource.
+
+        If any of the proposed values is over the defined quota, an
+        OverQuota exception will be raised with the sorted list of the
+        resources which are too high.  Otherwise, the method returns
+        nothing.
+
+        :param context: The request context, for access checks.
+        :param resources: A dictionary of the registered resources.
+        :param values: A dictionary of the values to check against the
+                       quota.
+        """
+
+        # Ensure no value is less than zero
+        unders = [key for key, val in values.items() if val < 0]
+        if unders:
+            raise exception.InvalidQuotaValue(unders=sorted(unders))
+
+        # Get the applicable quotas
+        quotas = self._get_quotas(context, resources, values.keys(),
+                                  has_sync=False)
+
+        # Check the quotas and construct a list of the resources that
+        # would be put over limit by the desired values
+        overs = [key for key, val in values.items()
+                 if quotas[key] >= 0 and quotas[key] < val]
+        if overs:
+            raise exception.OverQuota(overs=sorted(overs), quotas=quotas,
+                                      usages={})
+
+    def reserve(self, context, resources, deltas, expire=None):
+        """Check quotas and reserve resources.
+
+        For counting quotas--those quotas for which there is a usage
+        synchronization function--this method checks quotas against
+        current usage and the desired deltas.
+
+        This method will raise a QuotaResourceUnknown exception if a
+        given resource is unknown or if it does not have a usage
+        synchronization function.
+
+        If any of the proposed values is over the defined quota, an
+        OverQuota exception will be raised with the sorted list of the
+        resources which are too high.  Otherwise, the method returns a
+        list of reservation UUIDs which were created.
+
+        :param context: The request context, for access checks.
+        :param resources: A dictionary of the registered resources.
+        :param deltas: A dictionary of the proposed delta changes.
+        :param expire: An optional parameter specifying an expiration
+                       time for the reservations.  If it is a simple
+                       number, it is interpreted as a number of
+                       seconds and added to the current time; if it is
+                       a datetime.timedelta object, it will also be
+                       added to the current time.  A datetime.datetime
+                       object will be interpreted as the absolute
+                       expiration time.  If None is specified, the
+                       default expiration time set by
+                       --default-reservation-expire will be used (this
+                       value will be treated as a number of seconds).
+        """
+
+        # Set up the reservation expiration
+        if expire is None:
+            expire = FLAGS.reservation_expire
+        if isinstance(expire, (int, long)):
+            expire = datetime.timedelta(seconds=expire)
+        if isinstance(expire, datetime.timedelta):
+            expire = utils.utcnow() + expire
+        if not isinstance(expire, datetime.datetime):
+            raise exception.InvalidReservationExpiration(expire=expire)
+
+        # Get the applicable quotas.
+        # NOTE(Vek): We're not worried about races at this point.
+        #            Yes, the admin may be in the process of reducing
+        #            quotas, but that's a pretty rare thing.
+        quotas = self._get_quotas(context, resources, deltas.keys(),
+                                  has_sync=True)
+
+        # NOTE(Vek): Most of the work here has to be done in the DB
+        #            API, because we have to do it in a transaction,
+        #            which means access to the session.  Since the
+        #            session isn't available outside the DBAPI, we
+        #            have to do the work there.
+        return db.quota_reserve(context, resources, quotas, deltas, expire,
+                                FLAGS.until_refresh, FLAGS.max_age)
+
+    def commit(self, context, reservations):
+        """Commit reservations.
+
+        :param context: The request context, for access checks.
+        :param reservations: A list of the reservation UUIDs, as
+                             returned by the reserve() method.
+        """
+
+        db.reservation_commit(context, reservations)
+
+    def rollback(self, context, reservations):
+        """Roll back reservations.
+
+        :param context: The request context, for access checks.
+        :param reservations: A list of the reservation UUIDs, as
+                             returned by the reserve() method.
+        """
+
+        db.reservation_rollback(context, reservations)
+
+    def destroy_all_by_project(self, context, project_id):
+        """
+        Destroy all quotas, usages, and reservations associated with a
+        project.
+
+        :param context: The request context, for access checks.
+        :param project_id: The ID of the project being deleted.
+        """
+
+        db.quota_destroy_all_by_project(context, project_id)
+
+    def expire(self, context):
+        """Expire reservations.
+
+        Explores all currently existing reservations and rolls back
+        any that have expired.
+
+        :param context: The request context, for access checks.
+        """
+
+        db.reservation_expire(context)
+
+
+class BaseResource(object):
+    """Describe a single resource for quota checking."""
+
+    def __init__(self, name, flag=None):
+        """
+        Initializes a Resource.
+
+        :param name: The name of the resource, i.e., "instances".
+        :param flag: The name of the flag or configuration option
+                     which specifies the default value of the quota
+                     for this resource.
+        """
+
+        self.name = name
+        self.flag = flag
+
+    def quota(self, driver, context, **kwargs):
+        """
+        Given a driver and context, obtain the quota for this
+        resource.
+
+        :param driver: A quota driver.
+        :param context: The request context.
+        :param project_id: The project to obtain the quota value for.
+                           If not provided, it is taken from the
+                           context.  If it is given as None, no
+                           project-specific quota will be searched
+                           for.
+        :param quota_class: The quota class corresponding to the
+                            project, or for which the quota is to be
+                            looked up.  If not provided, it is taken
+                            from the context.  If it is given as None,
+                            no quota class-specific quota will be
+                            searched for.  Note that the quota class
+                            defaults to the value in the context,
+                            which may not correspond to the project if
+                            project_id is not the same as the one in
+                            the context.
+        """
+
+        # Get the project ID
+        project_id = kwargs.get('project_id', context.project_id)
+
+        # Ditto for the quota class
+        quota_class = kwargs.get('quota_class', context.quota_class)
+
+        # Look up the quota for the project
+        if project_id:
+            try:
+                return driver.get_by_project(context, project_id, self.name)
+            except exception.ProjectQuotaNotFound:
+                pass
+
+        # Try for the quota class
+        if quota_class:
+            try:
+                return driver.get_by_class(context, quota_class, self.name)
+            except exception.QuotaClassNotFound:
+                pass
+
+        # OK, return the default
+        return self.default
+
+    @property
+    def default(self):
+        """Return the default value of the quota."""
+
+        return FLAGS[self.flag] if self.flag else -1
+
+
+class ReservableResource(BaseResource):
+    """Describe a reservable resource."""
+
+    def __init__(self, name, sync, flag=None):
+        """
+        Initializes a ReservableResource.
+
+        Reservable resources are those resources which directly
+        correspond to objects in the database, i.e., instances, cores,
+        etc.  A ReservableResource must be constructed with a usage
+        synchronization function, which will be called to determine the
+        current counts of one or more resources.
+
+        The usage synchronization function will be passed three
+        arguments: an admin context, the project ID, and an opaque
+        session object, which should in turn be passed to the
+        underlying database function.  Synchronization functions
+        should return a dictionary mapping resource names to the
+        current in_use count for those resources; more than one
+        resource and resource count may be returned.  Note that
+        synchronization functions may be associated with more than one
+        ReservableResource.
+
+        :param name: The name of the resource, i.e., "instances".
+        :param sync: A callable which returns a dictionary to
+                     resynchronize the in_use count for one or more
+                     resources, as described above.
+        :param flag: The name of the flag or configuration option
+                     which specifies the default value of the quota
+                     for this resource.
+        """
+
+        super(ReservableResource, self).__init__(name, flag=flag)
+        self.sync = sync
+
+
+class AbsoluteResource(BaseResource):
+    """Describe a non-reservable resource."""
+
+    pass
+
+
+class CountableResource(AbsoluteResource):
+    """
+    Describe a resource where the counts aren't based solely on the
+    project ID.
+    """
+
+    def __init__(self, name, count, flag=None):
+        """
+        Initializes a CountableResource.
+
+        Countable resources are those resources which directly
+        correspond to objects in the database, i.e., instances, cores,
+        etc., but for which a count by project ID is inappropriate.  A
+        CountableResource must be constructed with a counting
+        function, which will be called to determine the current counts
+        of the resource.
+
+        The counting function will be passed the context, along with
+        the extra positional and keyword arguments that are passed to
+        Quota.count().  It should return an integer specifying the
+        count.
+
+        Note that this counting is not performed in a transaction-safe
+        manner.  This resource class is a temporary measure to provide
+        required functionality, until a better approach to solving
+        this problem can be evolved.
+
+        :param name: The name of the resource, i.e., "instances".
+        :param count: A callable which returns the count of the
+                      resource.  The arguments passed are as described
+                      above.
+        :param flag: The name of the flag or configuration option
+                     which specifies the default value of the quota
+                     for this resource.
+        """
+
+        super(CountableResource, self).__init__(name, flag=flag)
+        self.count = count
+
+
+class QuotaEngine(object):
+    """Represent the set of recognized quotas."""
+
+    def __init__(self, quota_driver_class=None):
+        """Initialize a Quota object."""
+
+        if not quota_driver_class:
+            quota_driver_class = FLAGS.quota_driver
+
+        if isinstance(quota_driver_class, basestring):
+            quota_driver_class = importutils.import_object(quota_driver_class)
+
+        self._resources = {}
+        self._driver = quota_driver_class
+
+    def __contains__(self, resource):
+        return resource in self._resources
+
+    def register_resource(self, resource):
+        """Register a resource."""
+
+        self._resources[resource.name] = resource
+
+    def register_resources(self, resources):
+        """Register a list of resources."""
+
+        for resource in resources:
+            self.register_resource(resource)
+
+    def get_by_project(self, context, project_id, resource):
+        """Get a specific quota by project."""
+
+        return self._driver.get_by_project(context, project_id, resource)
+
+    def get_by_class(self, context, quota_class, resource):
+        """Get a specific quota by quota class."""
+
+        return self._driver.get_by_class(context, quota_class, resource)
+
+    def get_defaults(self, context):
+        """Retrieve the default quotas.
+
+        :param context: The request context, for access checks.
+        """
+
+        return self._driver.get_defaults(context, self._resources)
+
+    def get_class_quotas(self, context, quota_class, defaults=True):
+        """Retrieve the quotas for the given quota class.
+
+        :param context: The request context, for access checks.
+        :param quota_class: The name of the quota class to return
+                            quotas for.
+        :param defaults: If True, the default value will be reported
+                         if there is no specific value for the
+                         resource.
+        """
+
+        return self._driver.get_class_quotas(context, self._resources,
+                                             quota_class, defaults=defaults)
+
+    def get_project_quotas(self, context, project_id, quota_class=None,
+                           defaults=True, usages=True):
+        """Retrieve the quotas for the given project.
+
+        :param context: The request context, for access checks.
+        :param project_id: The ID of the project to return quotas for.
+        :param quota_class: If project_id != context.project_id, the
+                            quota class cannot be determined.  This
+                            parameter allows it to be specified.
+        :param defaults: If True, the quota class value (or the
+                         default value, if there is no value from the
+                         quota class) will be reported if there is no
+                         specific value for the resource.
+        :param usages: If True, the current in_use and reserved counts
+                       will also be returned.
+        """
+
+        return self._driver.get_project_quotas(context, self._resources,
+                                              project_id,
+                                              quota_class=quota_class,
+                                              defaults=defaults,
+                                              usages=usages)
+
+    def count(self, context, resource, *args, **kwargs):
+        """Count a resource.
+
+        For countable resources, invokes the count() function and
+        returns its result.  Arguments following the context and
+        resource are passed directly to the count function declared by
+        the resource.
+
+        :param context: The request context, for access checks.
+        :param resource: The name of the resource, as a string.
+        """
+
+        # Get the resource
+        res = self._resources.get(resource)
+        if not res or not hasattr(res, 'count'):
+            raise exception.QuotaResourceUnknown(unknown=[resource])
+
+        return res.count(context, *args, **kwargs)
+
+    def limit_check(self, context, **values):
+        """Check simple quota limits.
+
+        For limits--those quotas for which there is no usage
+        synchronization function--this method checks that a set of
+        proposed values are permitted by the limit restriction.  The
+        values to check are given as keyword arguments, where the key
+        identifies the specific quota limit to check, and the value is
+        the proposed value.
+
+        This method will raise a QuotaResourceUnknown exception if a
+        given resource is unknown or if it is not a simple limit
+        resource.
+
+        If any of the proposed values is over the defined quota, an
+        OverQuota exception will be raised with the sorted list of the
+        resources which are too high.  Otherwise, the method returns
+        nothing.
+
+        :param context: The request context, for access checks.
+        """
+
+        return self._driver.limit_check(context, self._resources, values)
+
+    def reserve(self, context, expire=None, **deltas):
+        """Check quotas and reserve resources.
+
+        For counting quotas--those quotas for which there is a usage
+        synchronization function--this method checks quotas against
+        current usage and the desired deltas.  The deltas are given as
+        keyword arguments, and current usage and other reservations
+        are factored into the quota check.
+
+        This method will raise a QuotaResourceUnknown exception if a
+        given resource is unknown or if it does not have a usage
+        synchronization function.
+
+        If any of the proposed values is over the defined quota, an
+        OverQuota exception will be raised with the sorted list of the
+        resources which are too high.  Otherwise, the method returns a
+        list of reservation UUIDs which were created.
+
+        :param context: The request context, for access checks.
+        :param expire: An optional parameter specifying an expiration
+                       time for the reservations.  If it is a simple
+                       number, it is interpreted as a number of
+                       seconds and added to the current time; if it is
+                       a datetime.timedelta object, it will also be
+                       added to the current time.  A datetime.datetime
+                       object will be interpreted as the absolute
+                       expiration time.  If None is specified, the
+                       default expiration time set by
+                       --default-reservation-expire will be used (this
+                       value will be treated as a number of seconds).
+        """
+
+        reservations = self._driver.reserve(context, self._resources, deltas,
+                                            expire=expire)
+
+        LOG.debug(_("Created reservations %(reservations)s") % locals())
+
+        return reservations
+
+    def commit(self, context, reservations):
+        """Commit reservations.
+
+        :param context: The request context, for access checks.
+        :param reservations: A list of the reservation UUIDs, as
+                             returned by the reserve() method.
+        """
+
+        try:
+            self._driver.commit(context, reservations)
+        except Exception:
+            # NOTE(Vek): Ignoring exceptions here is safe, because the
+            # usage resynchronization and the reservation expiration
+            # mechanisms will resolve the issue.  The exception is
+            # logged, however, because this is less than optimal.
+            LOG.exception(_("Failed to commit reservations "
+                            "%(reservations)s") % locals())
+
+    def rollback(self, context, reservations):
+        """Roll back reservations.
+
+        :param context: The request context, for access checks.
+        :param reservations: A list of the reservation UUIDs, as
+                             returned by the reserve() method.
+        """
+
+        try:
+            self._driver.rollback(context, reservations)
+        except Exception:
+            # NOTE(Vek): Ignoring exceptions here is safe, because the
+            # usage resynchronization and the reservation expiration
+            # mechanisms will resolve the issue.  The exception is
+            # logged, however, because this is less than optimal.
+            LOG.exception(_("Failed to roll back reservations "
+                            "%(reservations)s") % locals())
+
+    def destroy_all_by_project(self, context, project_id):
+        """
+        Destroy all quotas, usages, and reservations associated with a
+        project.
+
+        :param context: The request context, for access checks.
+        :param project_id: The ID of the project being deleted.
+        """
+
+        self._driver.destroy_all_by_project(context, project_id)
+
+    def expire(self, context):
+        """Expire reservations.
+
+        Explores all currently existing reservations and rolls back
+        any that have expired.
+
+        :param context: The request context, for access checks.
+        """
+
+        self._driver.expire(context)
+
+    @property
+    def resources(self):
+        return sorted(self._resources.keys())
+
+
+def _sync_instances(context, project_id, session):
+    return dict(zip(('instances', 'cores', 'ram'),
+                    db.instance_data_get_for_project(
+                context, project_id, session=session)))
+
+
+def _sync_volumes(context, project_id, session):
+    return dict(zip(('volumes', 'gigabytes'),
+                    db.volume_data_get_for_project(
+                context, project_id, session=session)))
+
+
+def _sync_floating_ips(context, project_id, session):
+    return dict(floating_ips=db.floating_ip_count_by_project(
+            context, project_id, session=session))
+
+
+def _sync_security_groups(context, project_id, session):
+    return dict(security_groups=db.security_group_count_by_project(
+            context, project_id, session=session))
+
+
+QUOTAS = QuotaEngine()
+
+
+resources = [
+    ReservableResource('instances', _sync_instances, 'quota_instances'),
+    ReservableResource('cores', _sync_instances, 'quota_cores'),
+    ReservableResource('ram', _sync_instances, 'quota_ram'),
+    ReservableResource('volumes', _sync_volumes, 'quota_volumes'),
+    ReservableResource('gigabytes', _sync_volumes, 'quota_gigabytes'),
+    ReservableResource('floating_ips', _sync_floating_ips,
+                       'quota_floating_ips'),
+    AbsoluteResource('metadata_items', 'quota_metadata_items'),
+    AbsoluteResource('injected_files', 'quota_injected_files'),
+    AbsoluteResource('injected_file_content_bytes',
+                     'quota_injected_file_content_bytes'),
+    AbsoluteResource('injected_file_path_bytes',
+                     'quota_injected_file_path_bytes'),
+    ReservableResource('security_groups', _sync_security_groups,
+                       'quota_security_groups'),
+    CountableResource('security_group_rules',
+                      db.security_group_rule_count_by_group,
+                      'quota_security_group_rules'),
+    CountableResource('key_pairs', db.key_pair_count_by_user,
+                      'quota_key_pairs'),
+    ]
+
+
+QUOTAS.register_resources(resources)
diff --git a/nova/scheduler/manager.py b/nova/scheduler/manager.py
index 1dd6a438e..dfb7da704 100644
--- a/nova/scheduler/manager.py
+++ b/nova/scheduler/manager.py
@@ -33,6 +33,7 @@ from nova.notifier import api as notifier
 from nova.openstack.common import cfg
 from nova.openstack.common import excutils
 from nova.openstack.common import importutils
+from nova import quota
 
 
 LOG = logging.getLogger(__name__)
@@ -44,6 +45,8 @@ scheduler_driver_opt = cfg.StrOpt('scheduler_driver',
 FLAGS = flags.FLAGS
 FLAGS.register_opt(scheduler_driver_opt)
 
+QUOTAS = quota.QUOTAS
+
 
 class SchedulerManager(manager.Manager):
     """Chooses a host to run instances on."""
@@ -228,3 +231,7 @@ class SchedulerManager(manager.Manager):
                                  'ephemeral_gb': sum(ephemeral)}
 
         return {'resource': resource, 'usage': usage}
+
+    @manager.periodic_task
+    def _expire_reservations(self, context):
+        QUOTAS.expire(context)
diff --git a/nova/tests/test_quota.py b/nova/tests/test_quota.py
index 65ee7471a..88b3f7535 100644
--- a/nova/tests/test_quota.py
+++ b/nova/tests/test_quota.py
@@ -16,17 +16,22 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import datetime
+
 from nova import compute
+from nova.compute import instance_types
 from nova import context
 from nova import db
+from nova.db.sqlalchemy import api as sqa_api
+from nova.db.sqlalchemy import models as sqa_models
+from nova import exception
 from nova import flags
 from nova import quota
-from nova import exception
 from nova import rpc
+from nova.scheduler import driver as scheduler_driver
 from nova import test
+from nova import utils
 from nova import volume
-from nova.compute import instance_types
-from nova.scheduler import driver as scheduler_driver
 
 
 FLAGS = flags.FLAGS
@@ -258,7 +263,7 @@ class GetQuotaTestCase(test.TestCase):
                 ))
 
 
-class QuotaTestCase(test.TestCase):
+class OldQuotaTestCase(test.TestCase):
 
     class StubImageService(object):
 
@@ -266,7 +271,7 @@ class QuotaTestCase(test.TestCase):
             return {"properties": {}}
 
     def setUp(self):
-        super(QuotaTestCase, self).setUp()
+        super(OldQuotaTestCase, self).setUp()
         self.flags(connection_type='fake',
                    quota_instances=2,
                    quota_cores=4,
@@ -275,7 +280,9 @@ class QuotaTestCase(test.TestCase):
                    quota_floating_ips=1,
                    network_manager='nova.network.manager.FlatDHCPManager')
 
-        self.network = self.network = self.start_service('network')
+        # Apparently needed by the RPC tests...
+        self.network = self.start_service('network')
+
         self.user_id = 'admin'
         self.project_id = 'admin'
         self.context = context.RequestContext(self.user_id,
@@ -677,3 +684,1618 @@ class QuotaTestCase(test.TestCase):
         db.quota_class_create(self.context, 'foo', 'floating_ips', -1)
         items = quota.allowed_floating_ips(self.context, 100)
         self.assertEqual(items, 100)
+
+
+class FakeContext(object):
+    def __init__(self, project_id, quota_class):
+        self.is_admin = False
+        self.user_id = 'fake_user'
+        self.project_id = project_id
+        self.quota_class = quota_class
+
+    def elevated(self):
+        elevated = self.__class__(self.project_id, self.quota_class)
+        elevated.is_admin = True
+        return elevated
+
+
+class FakeDriver(object):
+    def __init__(self, by_project=None, by_class=None, reservations=None):
+        self.called = []
+        self.by_project = by_project or {}
+        self.by_class = by_class or {}
+        self.reservations = reservations or []
+
+    def get_by_project(self, context, project_id, resource):
+        self.called.append(('get_by_project', context, project_id, resource))
+        try:
+            return self.by_project[project_id][resource]
+        except KeyError:
+            raise exception.ProjectQuotaNotFound(project_id=project_id)
+
+    def get_by_class(self, context, quota_class, resource):
+        self.called.append(('get_by_class', context, quota_class, resource))
+        try:
+            return self.by_class[quota_class][resource]
+        except KeyError:
+            raise exception.QuotaClassNotFound(class_name=quota_class)
+
+    def get_defaults(self, context, resources):
+        self.called.append(('get_defaults', context, resources))
+        return resources
+
+    def get_class_quotas(self, context, resources, quota_class,
+                         defaults=True):
+        self.called.append(('get_class_quotas', context, resources,
+                            quota_class, defaults))
+        return resources
+
+    def get_project_quotas(self, context, resources, project_id,
+                           quota_class=None, defaults=True, usages=True):
+        self.called.append(('get_project_quotas', context, resources,
+                            project_id, quota_class, defaults, usages))
+        return resources
+
+    def limit_check(self, context, resources, values):
+        self.called.append(('limit_check', context, resources, values))
+
+    def reserve(self, context, resources, deltas, expire=None):
+        self.called.append(('reserve', context, resources, deltas, expire))
+        return self.reservations
+
+    def commit(self, context, reservations):
+        self.called.append(('commit', context, reservations))
+
+    def rollback(self, context, reservations):
+        self.called.append(('rollback', context, reservations))
+
+    def destroy_all_by_project(self, context, project_id):
+        self.called.append(('destroy_all_by_project', context, project_id))
+
+    def expire(self, context):
+        self.called.append(('expire', context))
+
+
+class BaseResourceTestCase(test.TestCase):
+    def test_no_flag(self):
+        resource = quota.BaseResource('test_resource')
+
+        self.assertEqual(resource.name, 'test_resource')
+        self.assertEqual(resource.flag, None)
+        self.assertEqual(resource.default, -1)
+
+    def test_with_flag(self):
+        # We know this flag exists, so use it...
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+
+        self.assertEqual(resource.name, 'test_resource')
+        self.assertEqual(resource.flag, 'quota_instances')
+        self.assertEqual(resource.default, 10)
+
+    def test_with_flag_no_quota(self):
+        self.flags(quota_instances=-1)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+
+        self.assertEqual(resource.name, 'test_resource')
+        self.assertEqual(resource.flag, 'quota_instances')
+        self.assertEqual(resource.default, -1)
+
+    def test_quota_no_project_no_class(self):
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+        driver = FakeDriver()
+        context = FakeContext(None, None)
+        quota_value = resource.quota(driver, context)
+
+        self.assertEqual(quota_value, 10)
+
+    def test_quota_with_project_no_class(self):
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+        driver = FakeDriver(by_project=dict(
+                test_project=dict(test_resource=15),
+                ))
+        context = FakeContext('test_project', None)
+        quota_value = resource.quota(driver, context)
+
+        self.assertEqual(quota_value, 15)
+
+    def test_quota_no_project_with_class(self):
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+        driver = FakeDriver(by_class=dict(
+                test_class=dict(test_resource=20),
+                ))
+        context = FakeContext(None, 'test_class')
+        quota_value = resource.quota(driver, context)
+
+        self.assertEqual(quota_value, 20)
+
+    def test_quota_with_project_with_class(self):
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+        driver = FakeDriver(by_project=dict(
+                test_project=dict(test_resource=15),
+                ),
+                            by_class=dict(
+                test_class=dict(test_resource=20),
+                ))
+        context = FakeContext('test_project', 'test_class')
+        quota_value = resource.quota(driver, context)
+
+        self.assertEqual(quota_value, 15)
+
+    def test_quota_override_project_with_class(self):
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+        driver = FakeDriver(by_project=dict(
+                test_project=dict(test_resource=15),
+                override_project=dict(test_resource=20),
+                ))
+        context = FakeContext('test_project', 'test_class')
+        quota_value = resource.quota(driver, context,
+                                     project_id='override_project')
+
+        self.assertEqual(quota_value, 20)
+
+    def test_quota_with_project_override_class(self):
+        self.flags(quota_instances=10)
+        resource = quota.BaseResource('test_resource', 'quota_instances')
+        driver = FakeDriver(by_class=dict(
+                test_class=dict(test_resource=15),
+                override_class=dict(test_resource=20),
+                ))
+        context = FakeContext('test_project', 'test_class')
+        quota_value = resource.quota(driver, context,
+                                     quota_class='override_class')
+
+        self.assertEqual(quota_value, 20)
+
+
+class QuotaEngineTestCase(test.TestCase):
+    def test_init(self):
+        quota_obj = quota.QuotaEngine()
+
+        self.assertEqual(quota_obj._resources, {})
+        self.assertTrue(isinstance(quota_obj._driver, quota.DbQuotaDriver))
+
+    def test_init_override_string(self):
+        quota_obj = quota.QuotaEngine(
+            quota_driver_class='nova.tests.test_quota.FakeDriver')
+
+        self.assertEqual(quota_obj._resources, {})
+        self.assertTrue(isinstance(quota_obj._driver, FakeDriver))
+
+    def test_init_override_obj(self):
+        quota_obj = quota.QuotaEngine(quota_driver_class=FakeDriver)
+
+        self.assertEqual(quota_obj._resources, {})
+        self.assertEqual(quota_obj._driver, FakeDriver)
+
+    def test_register_resource(self):
+        quota_obj = quota.QuotaEngine()
+        resource = quota.AbsoluteResource('test_resource')
+        quota_obj.register_resource(resource)
+
+        self.assertEqual(quota_obj._resources, dict(test_resource=resource))
+
+    def test_register_resources(self):
+        quota_obj = quota.QuotaEngine()
+        resources = [
+            quota.AbsoluteResource('test_resource1'),
+            quota.AbsoluteResource('test_resource2'),
+            quota.AbsoluteResource('test_resource3'),
+            ]
+        quota_obj.register_resources(resources)
+
+        self.assertEqual(quota_obj._resources, dict(
+                test_resource1=resources[0],
+                test_resource2=resources[1],
+                test_resource3=resources[2],
+                ))
+
+    def test_sync_predeclared(self):
+        quota_obj = quota.QuotaEngine()
+
+        def spam(*args, **kwargs):
+            pass
+
+        resource = quota.ReservableResource('test_resource', spam)
+        quota_obj.register_resource(resource)
+
+        self.assertEqual(resource.sync, spam)
+
+    def test_sync_multi(self):
+        quota_obj = quota.QuotaEngine()
+
+        def spam(*args, **kwargs):
+            pass
+
+        resources = [
+            quota.ReservableResource('test_resource1', spam),
+            quota.ReservableResource('test_resource2', spam),
+            quota.ReservableResource('test_resource3', spam),
+            quota.ReservableResource('test_resource4', spam),
+            ]
+        quota_obj.register_resources(resources[:2])
+
+        self.assertEqual(resources[0].sync, spam)
+        self.assertEqual(resources[1].sync, spam)
+        self.assertEqual(resources[2].sync, spam)
+        self.assertEqual(resources[3].sync, spam)
+
+    def test_get_by_project(self):
+        context = FakeContext('test_project', 'test_class')
+        driver = FakeDriver(by_project=dict(
+                test_project=dict(test_resource=42)))
+        quota_obj = quota.QuotaEngine(quota_driver_class=driver)
+        result = quota_obj.get_by_project(context, 'test_project',
+                                          'test_resource')
+
+        self.assertEqual(driver.called, [
+                ('get_by_project', context, 'test_project', 'test_resource'),
+                ])
+        self.assertEqual(result, 42)
+
+    def test_get_by_class(self):
+        context = FakeContext('test_project', 'test_class')
+        driver = FakeDriver(by_class=dict(
+                test_class=dict(test_resource=42)))
+        quota_obj = quota.QuotaEngine(quota_driver_class=driver)
+        result = quota_obj.get_by_class(context, 'test_class', 'test_resource')
+
+        self.assertEqual(driver.called, [
+                ('get_by_class', context, 'test_class', 'test_resource'),
+                ])
+        self.assertEqual(result, 42)
+
+    def _make_quota_obj(self, driver):
+        quota_obj = quota.QuotaEngine(quota_driver_class=driver)
+        resources = [
+            quota.AbsoluteResource('test_resource4'),
+            quota.AbsoluteResource('test_resource3'),
+            quota.AbsoluteResource('test_resource2'),
+            quota.AbsoluteResource('test_resource1'),
+            ]
+        quota_obj.register_resources(resources)
+
+        return quota_obj
+
+    def test_get_defaults(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        result = quota_obj.get_defaults(context)
+
+        self.assertEqual(driver.called, [
+                ('get_defaults', context, quota_obj._resources),
+                ])
+        self.assertEqual(result, quota_obj._resources)
+
+    def test_get_class_quotas(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        result1 = quota_obj.get_class_quotas(context, 'test_class')
+        result2 = quota_obj.get_class_quotas(context, 'test_class', False)
+
+        self.assertEqual(driver.called, [
+                ('get_class_quotas', context, quota_obj._resources,
+                 'test_class', True),
+                ('get_class_quotas', context, quota_obj._resources,
+                 'test_class', False),
+                ])
+        self.assertEqual(result1, quota_obj._resources)
+        self.assertEqual(result2, quota_obj._resources)
+
+    def test_get_project_quotas(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        result1 = quota_obj.get_project_quotas(context, 'test_project')
+        result2 = quota_obj.get_project_quotas(context, 'test_project',
+                                               quota_class='test_class',
+                                               defaults=False,
+                                               usages=False)
+
+        self.assertEqual(driver.called, [
+                ('get_project_quotas', context, quota_obj._resources,
+                 'test_project', None, True, True),
+                ('get_project_quotas', context, quota_obj._resources,
+                 'test_project', 'test_class', False, False),
+                ])
+        self.assertEqual(result1, quota_obj._resources)
+        self.assertEqual(result2, quota_obj._resources)
+
+    def test_count_no_resource(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        self.assertRaises(exception.QuotaResourceUnknown,
+                          quota_obj.count, context, 'test_resource5',
+                          True, foo='bar')
+
+    def test_count_wrong_resource(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        self.assertRaises(exception.QuotaResourceUnknown,
+                          quota_obj.count, context, 'test_resource1',
+                          True, foo='bar')
+
+    def test_count(self):
+        def fake_count(context, *args, **kwargs):
+            self.assertEqual(args, (True,))
+            self.assertEqual(kwargs, dict(foo='bar'))
+            return 5
+
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        quota_obj.register_resource(quota.CountableResource('test_resource5',
+                                                            fake_count))
+        result = quota_obj.count(context, 'test_resource5', True, foo='bar')
+
+        self.assertEqual(result, 5)
+
+    def test_limit_check(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        quota_obj.limit_check(context, test_resource1=4, test_resource2=3,
+                              test_resource3=2, test_resource4=1)
+
+        self.assertEqual(driver.called, [
+                ('limit_check', context, quota_obj._resources, dict(
+                        test_resource1=4,
+                        test_resource2=3,
+                        test_resource3=2,
+                        test_resource4=1,
+                        )),
+                ])
+
+    def test_reserve(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver(reservations=[
+                'resv-01', 'resv-02', 'resv-03', 'resv-04',
+                ])
+        quota_obj = self._make_quota_obj(driver)
+        result1 = quota_obj.reserve(context, test_resource1=4,
+                                    test_resource2=3, test_resource3=2,
+                                    test_resource4=1)
+        result2 = quota_obj.reserve(context, expire=3600,
+                                    test_resource1=1, test_resource2=2,
+                                    test_resource3=3, test_resource4=4)
+
+        self.assertEqual(driver.called, [
+                ('reserve', context, quota_obj._resources, dict(
+                        test_resource1=4,
+                        test_resource2=3,
+                        test_resource3=2,
+                        test_resource4=1,
+                        ), None),
+                ('reserve', context, quota_obj._resources, dict(
+                        test_resource1=1,
+                        test_resource2=2,
+                        test_resource3=3,
+                        test_resource4=4,
+                        ), 3600),
+                ])
+        self.assertEqual(result1, [
+                'resv-01', 'resv-02', 'resv-03', 'resv-04',
+                ])
+        self.assertEqual(result2, [
+                'resv-01', 'resv-02', 'resv-03', 'resv-04',
+                ])
+
+    def test_commit(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        quota_obj.commit(context, ['resv-01', 'resv-02', 'resv-03'])
+
+        self.assertEqual(driver.called, [
+                ('commit', context, ['resv-01', 'resv-02', 'resv-03']),
+                ])
+
+    def test_rollback(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        quota_obj.rollback(context, ['resv-01', 'resv-02', 'resv-03'])
+
+        self.assertEqual(driver.called, [
+                ('rollback', context, ['resv-01', 'resv-02', 'resv-03']),
+                ])
+
+    def test_destroy_all_by_project(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        quota_obj.destroy_all_by_project(context, 'test_project')
+
+        self.assertEqual(driver.called, [
+                ('destroy_all_by_project', context, 'test_project'),
+                ])
+
+    def test_expire(self):
+        context = FakeContext(None, None)
+        driver = FakeDriver()
+        quota_obj = self._make_quota_obj(driver)
+        quota_obj.expire(context)
+
+        self.assertEqual(driver.called, [
+                ('expire', context),
+                ])
+
+    def test_resources(self):
+        quota_obj = self._make_quota_obj(None)
+
+        self.assertEqual(quota_obj.resources,
+                         ['test_resource1', 'test_resource2',
+                          'test_resource3', 'test_resource4'])
+
+
+class DbQuotaDriverTestCase(test.TestCase):
+    def setUp(self):
+        super(DbQuotaDriverTestCase, self).setUp()
+
+        self.flags(quota_instances=10,
+                   quota_cores=20,
+                   quota_ram=50 * 1024,
+                   quota_volumes=10,
+                   quota_gigabytes=1000,
+                   quota_floating_ips=10,
+                   quota_metadata_items=128,
+                   quota_injected_files=5,
+                   quota_injected_file_content_bytes=10 * 1024,
+                   quota_injected_file_path_bytes=255,
+                   quota_security_groups=10,
+                   quota_security_group_rules=20,
+                   reservation_expire=86400,
+                   until_refresh=0,
+                   max_age=0,
+                   )
+
+        self.driver = quota.DbQuotaDriver()
+
+        self.calls = []
+
+        utils.set_time_override()
+
+    def tearDown(self):
+        utils.clear_time_override()
+        super(DbQuotaDriverTestCase, self).tearDown()
+
+    def test_get_defaults(self):
+        # Use our pre-defined resources
+        result = self.driver.get_defaults(None, quota.QUOTAS._resources)
+
+        self.assertEqual(result, dict(
+                instances=10,
+                cores=20,
+                ram=50 * 1024,
+                volumes=10,
+                gigabytes=1000,
+                floating_ips=10,
+                metadata_items=128,
+                injected_files=5,
+                injected_file_content_bytes=10 * 1024,
+                injected_file_path_bytes=255,
+                security_groups=10,
+                security_group_rules=20,
+                key_pairs=100,
+                ))
+
+    def _stub_quota_class_get_all_by_name(self):
+        # Stub out quota_class_get_all_by_name
+        def fake_qcgabn(context, quota_class):
+            self.calls.append('quota_class_get_all_by_name')
+            self.assertEqual(quota_class, 'test_class')
+            return dict(
+                instances=5,
+                ram=25 * 1024,
+                gigabytes=500,
+                metadata_items=64,
+                injected_file_content_bytes=5 * 1024,
+                )
+        self.stubs.Set(db, 'quota_class_get_all_by_name', fake_qcgabn)
+
+    def test_get_class_quotas(self):
+        self._stub_quota_class_get_all_by_name()
+        result = self.driver.get_class_quotas(None, quota.QUOTAS._resources,
+                                              'test_class')
+
+        self.assertEqual(self.calls, ['quota_class_get_all_by_name'])
+        self.assertEqual(result, dict(
+                instances=5,
+                cores=20,
+                ram=25 * 1024,
+                volumes=10,
+                gigabytes=500,
+                floating_ips=10,
+                metadata_items=64,
+                injected_files=5,
+                injected_file_content_bytes=5 * 1024,
+                injected_file_path_bytes=255,
+                security_groups=10,
+                security_group_rules=20,
+                key_pairs=100,
+                ))
+
+    def test_get_class_quotas_no_defaults(self):
+        self._stub_quota_class_get_all_by_name()
+        result = self.driver.get_class_quotas(None, quota.QUOTAS._resources,
+                                              'test_class', False)
+
+        self.assertEqual(self.calls, ['quota_class_get_all_by_name'])
+        self.assertEqual(result, dict(
+                instances=5,
+                ram=25 * 1024,
+                gigabytes=500,
+                metadata_items=64,
+                injected_file_content_bytes=5 * 1024,
+                ))
+
+    def _stub_get_by_project(self):
+        def fake_qgabp(context, project_id):
+            self.calls.append('quota_get_all_by_project')
+            self.assertEqual(project_id, 'test_project')
+            return dict(
+                cores=10,
+                gigabytes=50,
+                injected_files=2,
+                injected_file_path_bytes=127,
+                )
+
+        def fake_qugabp(context, project_id):
+            self.calls.append('quota_usage_get_all_by_project')
+            self.assertEqual(project_id, 'test_project')
+            return dict(
+                instances=dict(in_use=2, reserved=2),
+                cores=dict(in_use=4, reserved=4),
+                ram=dict(in_use=10 * 1024, reserved=0),
+                volumes=dict(in_use=2, reserved=0),
+                gigabytes=dict(in_use=10, reserved=0),
+                floating_ips=dict(in_use=2, reserved=0),
+                metadata_items=dict(in_use=0, reserved=0),
+                injected_files=dict(in_use=0, reserved=0),
+                injected_file_content_bytes=dict(in_use=0, reserved=0),
+                injected_file_path_bytes=dict(in_use=0, reserved=0),
+                )
+
+        self.stubs.Set(db, 'quota_get_all_by_project', fake_qgabp)
+        self.stubs.Set(db, 'quota_usage_get_all_by_project', fake_qugabp)
+
+        self._stub_quota_class_get_all_by_name()
+
+    def test_get_project_quotas(self):
+        self._stub_get_by_project()
+        result = self.driver.get_project_quotas(
+            FakeContext('test_project', 'test_class'),
+            quota.QUOTAS._resources, 'test_project')
+
+        self.assertEqual(self.calls, [
+                'quota_get_all_by_project',
+                'quota_usage_get_all_by_project',
+                'quota_class_get_all_by_name',
+                ])
+        self.assertEqual(result, dict(
+                instances=dict(
+                    limit=5,
+                    in_use=2,
+                    reserved=2,
+                    ),
+                cores=dict(
+                    limit=10,
+                    in_use=4,
+                    reserved=4,
+                    ),
+                ram=dict(
+                    limit=25 * 1024,
+                    in_use=10 * 1024,
+                    reserved=0,
+                    ),
+                volumes=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=0,
+                    ),
+                gigabytes=dict(
+                    limit=50,
+                    in_use=10,
+                    reserved=0,
+                    ),
+                floating_ips=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=0,
+                    ),
+                metadata_items=dict(
+                    limit=64,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_files=dict(
+                    limit=2,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_content_bytes=dict(
+                    limit=5 * 1024,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_path_bytes=dict(
+                    limit=127,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                security_groups=dict(
+                    limit=10,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                security_group_rules=dict(
+                    limit=20,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                key_pairs=dict(
+                    limit=100,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                ))
+
+    def test_get_project_quotas_alt_context_no_class(self):
+        self._stub_get_by_project()
+        result = self.driver.get_project_quotas(
+            FakeContext('other_project', 'other_class'),
+            quota.QUOTAS._resources, 'test_project')
+
+        self.assertEqual(self.calls, [
+                'quota_get_all_by_project',
+                'quota_usage_get_all_by_project',
+                ])
+        self.assertEqual(result, dict(
+                instances=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=2,
+                    ),
+                cores=dict(
+                    limit=10,
+                    in_use=4,
+                    reserved=4,
+                    ),
+                ram=dict(
+                    limit=50 * 1024,
+                    in_use=10 * 1024,
+                    reserved=0,
+                    ),
+                volumes=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=0,
+                    ),
+                gigabytes=dict(
+                    limit=50,
+                    in_use=10,
+                    reserved=0,
+                    ),
+                floating_ips=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=0,
+                    ),
+                metadata_items=dict(
+                    limit=128,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_files=dict(
+                    limit=2,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_content_bytes=dict(
+                    limit=10 * 1024,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_path_bytes=dict(
+                    limit=127,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                security_groups=dict(
+                    limit=10,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                security_group_rules=dict(
+                    limit=20,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                key_pairs=dict(
+                    limit=100,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                ))
+
+    def test_get_project_quotas_alt_context_with_class(self):
+        self._stub_get_by_project()
+        result = self.driver.get_project_quotas(
+            FakeContext('other_project', 'other_class'),
+            quota.QUOTAS._resources, 'test_project', quota_class='test_class')
+
+        self.assertEqual(self.calls, [
+                'quota_get_all_by_project',
+                'quota_usage_get_all_by_project',
+                'quota_class_get_all_by_name',
+                ])
+        self.assertEqual(result, dict(
+                instances=dict(
+                    limit=5,
+                    in_use=2,
+                    reserved=2,
+                    ),
+                cores=dict(
+                    limit=10,
+                    in_use=4,
+                    reserved=4,
+                    ),
+                ram=dict(
+                    limit=25 * 1024,
+                    in_use=10 * 1024,
+                    reserved=0,
+                    ),
+                volumes=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=0,
+                    ),
+                gigabytes=dict(
+                    limit=50,
+                    in_use=10,
+                    reserved=0,
+                    ),
+                floating_ips=dict(
+                    limit=10,
+                    in_use=2,
+                    reserved=0,
+                    ),
+                metadata_items=dict(
+                    limit=64,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_files=dict(
+                    limit=2,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_content_bytes=dict(
+                    limit=5 * 1024,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_path_bytes=dict(
+                    limit=127,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                security_groups=dict(
+                    limit=10,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                security_group_rules=dict(
+                    limit=20,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                key_pairs=dict(
+                    limit=100,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                ))
+
+    def test_get_project_quotas_no_defaults(self):
+        self._stub_get_by_project()
+        result = self.driver.get_project_quotas(
+            FakeContext('test_project', 'test_class'),
+            quota.QUOTAS._resources, 'test_project', defaults=False)
+
+        self.assertEqual(self.calls, [
+                'quota_get_all_by_project',
+                'quota_usage_get_all_by_project',
+                'quota_class_get_all_by_name',
+                ])
+        self.assertEqual(result, dict(
+                cores=dict(
+                    limit=10,
+                    in_use=4,
+                    reserved=4,
+                    ),
+                gigabytes=dict(
+                    limit=50,
+                    in_use=10,
+                    reserved=0,
+                    ),
+                injected_files=dict(
+                    limit=2,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                injected_file_path_bytes=dict(
+                    limit=127,
+                    in_use=0,
+                    reserved=0,
+                    ),
+                ))
+
+    def test_get_project_quotas_no_usages(self):
+        self._stub_get_by_project()
+        result = self.driver.get_project_quotas(
+            FakeContext('test_project', 'test_class'),
+            quota.QUOTAS._resources, 'test_project', usages=False)
+
+        self.assertEqual(self.calls, [
+                'quota_get_all_by_project',
+                'quota_class_get_all_by_name',
+                ])
+        self.assertEqual(result, dict(
+                instances=dict(
+                    limit=5,
+                    ),
+                cores=dict(
+                    limit=10,
+                    ),
+                ram=dict(
+                    limit=25 * 1024,
+                    ),
+                volumes=dict(
+                    limit=10,
+                    ),
+                gigabytes=dict(
+                    limit=50,
+                    ),
+                floating_ips=dict(
+                    limit=10,
+                    ),
+                metadata_items=dict(
+                    limit=64,
+                    ),
+                injected_files=dict(
+                    limit=2,
+                    ),
+                injected_file_content_bytes=dict(
+                    limit=5 * 1024,
+                    ),
+                injected_file_path_bytes=dict(
+                    limit=127,
+                    ),
+                security_groups=dict(
+                    limit=10,
+                    ),
+                security_group_rules=dict(
+                    limit=20,
+                    ),
+                key_pairs=dict(
+                    limit=100,
+                    ),
+                ))
+
+    def _stub_get_project_quotas(self):
+        def fake_get_project_quotas(context, resources, project_id,
+                                    quota_class=None, defaults=True,
+                                    usages=True):
+            self.calls.append('get_project_quotas')
+            return dict((k, dict(limit=v.default))
+                        for k, v in resources.items())
+
+        self.stubs.Set(self.driver, 'get_project_quotas',
+                       fake_get_project_quotas)
+
+    def test_get_quotas_has_sync_unknown(self):
+        self._stub_get_project_quotas()
+        self.assertRaises(exception.QuotaResourceUnknown,
+                          self.driver._get_quotas,
+                          None, quota.QUOTAS._resources,
+                          ['unknown'], True)
+        self.assertEqual(self.calls, [])
+
+    def test_get_quotas_no_sync_unknown(self):
+        self._stub_get_project_quotas()
+        self.assertRaises(exception.QuotaResourceUnknown,
+                          self.driver._get_quotas,
+                          None, quota.QUOTAS._resources,
+                          ['unknown'], False)
+        self.assertEqual(self.calls, [])
+
+    def test_get_quotas_has_sync_no_sync_resource(self):
+        self._stub_get_project_quotas()
+        self.assertRaises(exception.QuotaResourceUnknown,
+                          self.driver._get_quotas,
+                          None, quota.QUOTAS._resources,
+                          ['metadata_items'], True)
+        self.assertEqual(self.calls, [])
+
+    def test_get_quotas_no_sync_has_sync_resource(self):
+        self._stub_get_project_quotas()
+        self.assertRaises(exception.QuotaResourceUnknown,
+                          self.driver._get_quotas,
+                          None, quota.QUOTAS._resources,
+                          ['instances'], False)
+        self.assertEqual(self.calls, [])
+
+    def test_get_quotas_has_sync(self):
+        self._stub_get_project_quotas()
+        result = self.driver._get_quotas(FakeContext('test_project',
+                                                     'test_class'),
+                                         quota.QUOTAS._resources,
+                                         ['instances', 'cores', 'ram',
+                                          'volumes', 'gigabytes',
+                                          'floating_ips', 'security_groups'],
+                                         True)
+
+        self.assertEqual(self.calls, ['get_project_quotas'])
+        self.assertEqual(result, dict(
+                instances=10,
+                cores=20,
+                ram=50 * 1024,
+                volumes=10,
+                gigabytes=1000,
+                floating_ips=10,
+                security_groups=10,
+                ))
+
+    def test_get_quotas_no_sync(self):
+        self._stub_get_project_quotas()
+        result = self.driver._get_quotas(FakeContext('test_project',
+                                                     'test_class'),
+                                         quota.QUOTAS._resources,
+                                         ['metadata_items', 'injected_files',
+                                          'injected_file_content_bytes',
+                                          'injected_file_path_bytes',
+                                          'security_group_rules'], False)
+
+        self.assertEqual(self.calls, ['get_project_quotas'])
+        self.assertEqual(result, dict(
+                metadata_items=128,
+                injected_files=5,
+                injected_file_content_bytes=10 * 1024,
+                injected_file_path_bytes=255,
+                security_group_rules=20,
+                ))
+
+    def test_limit_check_under(self):
+        self._stub_get_project_quotas()
+        self.assertRaises(exception.InvalidQuotaValue,
+                          self.driver.limit_check,
+                          FakeContext('test_project', 'test_class'),
+                          quota.QUOTAS._resources,
+                          dict(metadata_items=-1))
+
+    def test_limit_check_over(self):
+        self._stub_get_project_quotas()
+        self.assertRaises(exception.OverQuota,
+                          self.driver.limit_check,
+                          FakeContext('test_project', 'test_class'),
+                          quota.QUOTAS._resources,
+                          dict(metadata_items=129))
+
+    def test_limit_check_unlimited(self):
+        self.flags(quota_metadata_items=-1)
+        self._stub_get_project_quotas()
+        self.driver.limit_check(FakeContext('test_project', 'test_class'),
+                                quota.QUOTAS._resources,
+                                dict(metadata_items=32767))
+
+    def test_limit_check(self):
+        self._stub_get_project_quotas()
+        self.driver.limit_check(FakeContext('test_project', 'test_class'),
+                                quota.QUOTAS._resources,
+                                dict(metadata_items=128))
+
+    def _stub_quota_reserve(self):
+        def fake_quota_reserve(context, resources, quotas, deltas, expire,
+                               until_refresh, max_age):
+            self.calls.append(('quota_reserve', expire, until_refresh,
+                               max_age))
+            return ['resv-1', 'resv-2', 'resv-3']
+        self.stubs.Set(db, 'quota_reserve', fake_quota_reserve)
+
+    def test_reserve_bad_expire(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        self.assertRaises(exception.InvalidReservationExpiration,
+                          self.driver.reserve,
+                          FakeContext('test_project', 'test_class'),
+                          quota.QUOTAS._resources,
+                          dict(instances=2), expire='invalid')
+        self.assertEqual(self.calls, [])
+
+    def test_reserve_default_expire(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        result = self.driver.reserve(FakeContext('test_project', 'test_class'),
+                                     quota.QUOTAS._resources,
+                                     dict(instances=2))
+
+        expire = utils.utcnow() + datetime.timedelta(seconds=86400)
+        self.assertEqual(self.calls, [
+                'get_project_quotas',
+                ('quota_reserve', expire, 0, 0),
+                ])
+        self.assertEqual(result, ['resv-1', 'resv-2', 'resv-3'])
+
+    def test_reserve_int_expire(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        result = self.driver.reserve(FakeContext('test_project', 'test_class'),
+                                     quota.QUOTAS._resources,
+                                     dict(instances=2), expire=3600)
+
+        expire = utils.utcnow() + datetime.timedelta(seconds=3600)
+        self.assertEqual(self.calls, [
+                'get_project_quotas',
+                ('quota_reserve', expire, 0, 0),
+                ])
+        self.assertEqual(result, ['resv-1', 'resv-2', 'resv-3'])
+
+    def test_reserve_timedelta_expire(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        expire_delta = datetime.timedelta(seconds=60)
+        result = self.driver.reserve(FakeContext('test_project', 'test_class'),
+                                     quota.QUOTAS._resources,
+                                     dict(instances=2), expire=expire_delta)
+
+        expire = utils.utcnow() + expire_delta
+        self.assertEqual(self.calls, [
+                'get_project_quotas',
+                ('quota_reserve', expire, 0, 0),
+                ])
+        self.assertEqual(result, ['resv-1', 'resv-2', 'resv-3'])
+
+    def test_reserve_datetime_expire(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        expire = utils.utcnow() + datetime.timedelta(seconds=120)
+        result = self.driver.reserve(FakeContext('test_project', 'test_class'),
+                                     quota.QUOTAS._resources,
+                                     dict(instances=2), expire=expire)
+
+        self.assertEqual(self.calls, [
+                'get_project_quotas',
+                ('quota_reserve', expire, 0, 0),
+                ])
+        self.assertEqual(result, ['resv-1', 'resv-2', 'resv-3'])
+
+    def test_reserve_until_refresh(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        self.flags(until_refresh=500)
+        expire = utils.utcnow() + datetime.timedelta(seconds=120)
+        result = self.driver.reserve(FakeContext('test_project', 'test_class'),
+                                     quota.QUOTAS._resources,
+                                     dict(instances=2), expire=expire)
+
+        self.assertEqual(self.calls, [
+                'get_project_quotas',
+                ('quota_reserve', expire, 500, 0),
+                ])
+        self.assertEqual(result, ['resv-1', 'resv-2', 'resv-3'])
+
+    def test_reserve_max_age(self):
+        self._stub_get_project_quotas()
+        self._stub_quota_reserve()
+        self.flags(max_age=86400)
+        expire = utils.utcnow() + datetime.timedelta(seconds=120)
+        result = self.driver.reserve(FakeContext('test_project', 'test_class'),
+                                     quota.QUOTAS._resources,
+                                     dict(instances=2), expire=expire)
+
+        self.assertEqual(self.calls, [
+                'get_project_quotas',
+                ('quota_reserve', expire, 0, 86400),
+                ])
+        self.assertEqual(result, ['resv-1', 'resv-2', 'resv-3'])
+
+
+class FakeSession(object):
+    def begin(self):
+        return self
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        return False
+
+
+class FakeUsage(sqa_models.QuotaUsage):
+    def save(self, *args, **kwargs):
+        pass
+
+
+class QuotaReserveSqlAlchemyTestCase(test.TestCase):
+    # nova.db.sqlalchemy.api.quota_reserve is so complex it needs its
+    # own test case, and since it's a quota manipulator, this is the
+    # best place to put it...
+
+    def setUp(self):
+        super(QuotaReserveSqlAlchemyTestCase, self).setUp()
+
+        self.sync_called = set()
+
+        def make_sync(res_name):
+            def sync(context, project_id, session):
+                self.sync_called.add(res_name)
+                if res_name in self.usages:
+                    return {res_name: self.usages[res_name].in_use - 1}
+                return {res_name: 0}
+            return sync
+
+        self.resources = {}
+        for res_name in ('instances', 'cores', 'ram'):
+            res = quota.ReservableResource(res_name, make_sync(res_name))
+            self.resources[res_name] = res
+
+        self.expire = utils.utcnow() + datetime.timedelta(seconds=3600)
+
+        self.usages = {}
+        self.usages_created = {}
+        self.reservations_created = {}
+
+        def fake_get_session():
+            return FakeSession()
+
+        def fake_get_quota_usages(context, session, keys):
+            return self.usages.copy()
+
+        def fake_quota_usage_create(context, project_id, resource, in_use,
+                                    reserved, until_refresh, session=None,
+                                    save=True):
+            quota_usage_ref = self._make_quota_usage(
+                project_id, resource, in_use, reserved, until_refresh,
+                utils.utcnow(), utils.utcnow())
+
+            self.usages_created[resource] = quota_usage_ref
+
+            return quota_usage_ref
+
+        def fake_reservation_create(context, uuid, usage_id, project_id,
+                                    resource, delta, expire, session=None):
+            reservation_ref = self._make_reservation(
+                uuid, usage_id, project_id, resource, delta, expire,
+                utils.utcnow(), utils.utcnow())
+
+            self.reservations_created[resource] = reservation_ref
+
+            return reservation_ref
+
+        self.stubs.Set(sqa_api, 'get_session', fake_get_session)
+        self.stubs.Set(sqa_api, '_get_quota_usages', fake_get_quota_usages)
+        self.stubs.Set(sqa_api, 'quota_usage_create', fake_quota_usage_create)
+        self.stubs.Set(sqa_api, 'reservation_create', fake_reservation_create)
+
+        utils.set_time_override()
+
+    def _make_quota_usage(self, project_id, resource, in_use, reserved,
+                          until_refresh, created_at, updated_at):
+        quota_usage_ref = FakeUsage()
+        quota_usage_ref.id = len(self.usages) + len(self.usages_created)
+        quota_usage_ref.project_id = project_id
+        quota_usage_ref.resource = resource
+        quota_usage_ref.in_use = in_use
+        quota_usage_ref.reserved = reserved
+        quota_usage_ref.until_refresh = until_refresh
+        quota_usage_ref.created_at = created_at
+        quota_usage_ref.updated_at = updated_at
+        quota_usage_ref.deleted_at = None
+        quota_usage_ref.deleted = False
+
+        return quota_usage_ref
+
+    def init_usage(self, project_id, resource, in_use, reserved,
+                   until_refresh=None, created_at=None, updated_at=None):
+        if created_at is None:
+            created_at = utils.utcnow()
+        if updated_at is None:
+            updated_at = utils.utcnow()
+
+        quota_usage_ref = self._make_quota_usage(project_id, resource, in_use,
+                                                 reserved, until_refresh,
+                                                 created_at, updated_at)
+
+        self.usages[resource] = quota_usage_ref
+
+    def compare_usage(self, usage_dict, expected):
+        for usage in expected:
+            resource = usage['resource']
+            for key, value in usage.items():
+                actual = getattr(usage_dict[resource], key)
+                self.assertEqual(actual, value,
+                                 "%s != %s on usage for resource %s" %
+                                 (actual, value, resource))
+
+    def _make_reservation(self, uuid, usage_id, project_id, resource,
+                          delta, expire, created_at, updated_at):
+        reservation_ref = sqa_models.Reservation()
+        reservation_ref.id = len(self.reservations_created)
+        reservation_ref.uuid = uuid
+        reservation_ref.usage_id = usage_id
+        reservation_ref.project_id = project_id
+        reservation_ref.resource = resource
+        reservation_ref.delta = delta
+        reservation_ref.expire = expire
+        reservation_ref.created_at = created_at
+        reservation_ref.updated_at = updated_at
+        reservation_ref.deleted_at = None
+        reservation_ref.deleted = False
+
+        return reservation_ref
+
+    def compare_reservation(self, reservations, expected):
+        reservations = set(reservations)
+        for resv in expected:
+            resource = resv['resource']
+            resv_obj = self.reservations_created[resource]
+
+            self.assertIn(resv_obj.uuid, reservations)
+            reservations.discard(resv_obj.uuid)
+
+            for key, value in resv.items():
+                actual = getattr(resv_obj, key)
+                self.assertEqual(actual, value,
+                                 "%s != %s on reservation for resource %s" %
+                                 (actual, value, resource))
+
+        self.assertEqual(len(reservations), 0)
+
+    def test_quota_reserve_create_usages(self):
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=2,
+            cores=4,
+            ram=2 * 1024,
+            )
+        result = sqa_api.quota_reserve(context, self.resources, quotas,
+                                       deltas, self.expire, 0, 0)
+
+        self.assertEqual(self.sync_called, set(['instances', 'cores', 'ram']))
+        self.compare_usage(self.usages_created, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=0,
+                     reserved=2,
+                     until_refresh=None),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=0,
+                     reserved=4,
+                     until_refresh=None),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=0,
+                     reserved=2 * 1024,
+                     until_refresh=None),
+                ])
+        self.compare_reservation(result, [
+                dict(resource='instances',
+                     usage_id=self.usages_created['instances'],
+                     project_id='test_project',
+                     delta=2),
+                dict(resource='cores',
+                     usage_id=self.usages_created['cores'],
+                     project_id='test_project',
+                     delta=4),
+                dict(resource='ram',
+                     usage_id=self.usages_created['ram'],
+                     delta=2 * 1024),
+                ])
+
+    def test_quota_reserve_until_refresh(self):
+        self.init_usage('test_project', 'instances', 3, 0, until_refresh=1)
+        self.init_usage('test_project', 'cores', 3, 0, until_refresh=1)
+        self.init_usage('test_project', 'ram', 3, 0, until_refresh=1)
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=2,
+            cores=4,
+            ram=2 * 1024,
+            )
+        result = sqa_api.quota_reserve(context, self.resources, quotas,
+                                       deltas, self.expire, 5, 0)
+
+        self.assertEqual(self.sync_called, set(['instances', 'cores', 'ram']))
+        self.compare_usage(self.usages, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=2,
+                     reserved=2,
+                     until_refresh=5),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=2,
+                     reserved=4,
+                     until_refresh=5),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=2,
+                     reserved=2 * 1024,
+                     until_refresh=5),
+                ])
+        self.assertEqual(self.usages_created, {})
+        self.compare_reservation(result, [
+                dict(resource='instances',
+                     usage_id=self.usages['instances'],
+                     project_id='test_project',
+                     delta=2),
+                dict(resource='cores',
+                     usage_id=self.usages['cores'],
+                     project_id='test_project',
+                     delta=4),
+                dict(resource='ram',
+                     usage_id=self.usages['ram'],
+                     delta=2 * 1024),
+                ])
+
+    def test_quota_reserve_max_age(self):
+        max_age = 3600
+        record_created = utils.utcnow() - datetime.timedelta(seconds=max_age)
+        self.init_usage('test_project', 'instances', 3, 0,
+                        created_at=record_created, updated_at=record_created)
+        self.init_usage('test_project', 'cores', 3, 0,
+                        created_at=record_created, updated_at=record_created)
+        self.init_usage('test_project', 'ram', 3, 0,
+                        created_at=record_created, updated_at=record_created)
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=2,
+            cores=4,
+            ram=2 * 1024,
+            )
+        result = sqa_api.quota_reserve(context, self.resources, quotas,
+                                       deltas, self.expire, 0, max_age)
+
+        self.assertEqual(self.sync_called, set(['instances', 'cores', 'ram']))
+        self.compare_usage(self.usages, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=2,
+                     reserved=2,
+                     until_refresh=None),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=2,
+                     reserved=4,
+                     until_refresh=None),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=2,
+                     reserved=2 * 1024,
+                     until_refresh=None),
+                ])
+        self.assertEqual(self.usages_created, {})
+        self.compare_reservation(result, [
+                dict(resource='instances',
+                     usage_id=self.usages['instances'],
+                     project_id='test_project',
+                     delta=2),
+                dict(resource='cores',
+                     usage_id=self.usages['cores'],
+                     project_id='test_project',
+                     delta=4),
+                dict(resource='ram',
+                     usage_id=self.usages['ram'],
+                     delta=2 * 1024),
+                ])
+
+    def test_quota_reserve_no_refresh(self):
+        self.init_usage('test_project', 'instances', 3, 0)
+        self.init_usage('test_project', 'cores', 3, 0)
+        self.init_usage('test_project', 'ram', 3, 0)
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=2,
+            cores=4,
+            ram=2 * 1024,
+            )
+        result = sqa_api.quota_reserve(context, self.resources, quotas,
+                                       deltas, self.expire, 0, 0)
+
+        self.assertEqual(self.sync_called, set([]))
+        self.compare_usage(self.usages, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=3,
+                     reserved=2,
+                     until_refresh=None),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=3,
+                     reserved=4,
+                     until_refresh=None),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=3,
+                     reserved=2 * 1024,
+                     until_refresh=None),
+                ])
+        self.assertEqual(self.usages_created, {})
+        self.compare_reservation(result, [
+                dict(resource='instances',
+                     usage_id=self.usages['instances'],
+                     project_id='test_project',
+                     delta=2),
+                dict(resource='cores',
+                     usage_id=self.usages['cores'],
+                     project_id='test_project',
+                     delta=4),
+                dict(resource='ram',
+                     usage_id=self.usages['ram'],
+                     delta=2 * 1024),
+                ])
+
+    def test_quota_reserve_unders(self):
+        self.init_usage('test_project', 'instances', 1, 0)
+        self.init_usage('test_project', 'cores', 3, 0)
+        self.init_usage('test_project', 'ram', 1 * 1024, 0)
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=-2,
+            cores=-4,
+            ram=-2 * 1024,
+            )
+        self.assertRaises(exception.InvalidQuotaValue,
+                          sqa_api.quota_reserve,
+                          context, self.resources, quotas,
+                          deltas, self.expire, 0, 0)
+
+        self.assertEqual(self.sync_called, set([]))
+        self.compare_usage(self.usages, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=1,
+                     reserved=0,
+                     until_refresh=None),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=3,
+                     reserved=0,
+                     until_refresh=None),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=1 * 1024,
+                     reserved=0,
+                     until_refresh=None),
+                ])
+        self.assertEqual(self.usages_created, {})
+        self.assertEqual(self.reservations_created, {})
+
+    def test_quota_reserve_overs(self):
+        self.init_usage('test_project', 'instances', 4, 0)
+        self.init_usage('test_project', 'cores', 8, 0)
+        self.init_usage('test_project', 'ram', 10 * 1024, 0)
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=2,
+            cores=4,
+            ram=2 * 1024,
+            )
+        self.assertRaises(exception.OverQuota,
+                          sqa_api.quota_reserve,
+                          context, self.resources, quotas,
+                          deltas, self.expire, 0, 0)
+
+        self.assertEqual(self.sync_called, set([]))
+        self.compare_usage(self.usages, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=4,
+                     reserved=0,
+                     until_refresh=None),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=8,
+                     reserved=0,
+                     until_refresh=None),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=10 * 1024,
+                     reserved=0,
+                     until_refresh=None),
+                ])
+        self.assertEqual(self.usages_created, {})
+        self.assertEqual(self.reservations_created, {})
+
+    def test_quota_reserve_reduction(self):
+        self.init_usage('test_project', 'instances', 10, 0)
+        self.init_usage('test_project', 'cores', 20, 0)
+        self.init_usage('test_project', 'ram', 20 * 1024, 0)
+        context = FakeContext('test_project', 'test_class')
+        quotas = dict(
+            instances=5,
+            cores=10,
+            ram=10 * 1024,
+            )
+        deltas = dict(
+            instances=-2,
+            cores=-4,
+            ram=-2 * 1024,
+            )
+        result = sqa_api.quota_reserve(context, self.resources, quotas,
+                                       deltas, self.expire, 0, 0)
+
+        self.assertEqual(self.sync_called, set([]))
+        self.compare_usage(self.usages, [
+                dict(resource='instances',
+                     project_id='test_project',
+                     in_use=10,
+                     reserved=0,
+                     until_refresh=None),
+                dict(resource='cores',
+                     project_id='test_project',
+                     in_use=20,
+                     reserved=0,
+                     until_refresh=None),
+                dict(resource='ram',
+                     project_id='test_project',
+                     in_use=20 * 1024,
+                     reserved=0,
+                     until_refresh=None),
+                ])
+        self.assertEqual(self.usages_created, {})
+        self.compare_reservation(result, [
+                dict(resource='instances',
+                     usage_id=self.usages['instances'],
+                     project_id='test_project',
+                     delta=-2),
+                dict(resource='cores',
+                     usage_id=self.usages['cores'],
+                     project_id='test_project',
+                     delta=-4),
+                dict(resource='ram',
+                     usage_id=self.usages['ram'],
+                     project_id='test_project',
+                     delta=-2 * 1024),
+                ])