diff --git a/manila/api/openstack/api_version_request.py b/manila/api/openstack/api_version_request.py
index 9f488f4feb..c013ed5453 100644
--- a/manila/api/openstack/api_version_request.py
+++ b/manila/api/openstack/api_version_request.py
@@ -110,13 +110,14 @@ REST_API_VERSION_HISTORY = """
     * 2.38 - Support IPv6 validation in allow_access API to enable IPv6 in
              manila.
     * 2.39 - Added share-type quotas.
+    * 2.40 - Added share group and share group snapshot quotas.
 """
 
 # The minimum and maximum versions of the API supported
 # The default api version request is defined to be the
 # minimum version of the API supported.
 _MIN_API_VERSION = "2.0"
-_MAX_API_VERSION = "2.39"
+_MAX_API_VERSION = "2.40"
 DEFAULT_API_VERSION = _MIN_API_VERSION
 
 
diff --git a/manila/api/openstack/rest_api_version_history.rst b/manila/api/openstack/rest_api_version_history.rst
index 78aeaca420..acbd143a26 100644
--- a/manila/api/openstack/rest_api_version_history.rst
+++ b/manila/api/openstack/rest_api_version_history.rst
@@ -222,3 +222,7 @@ user documentation.
 2.39
 ----
   Added share-type quotas.
+
+2.40
+----
+  Added share group and share group snapshot quotas.
diff --git a/manila/api/v2/quota_class_sets.py b/manila/api/v2/quota_class_sets.py
index 0466bc9bde..7a84bf2c20 100644
--- a/manila/api/v2/quota_class_sets.py
+++ b/manila/api/v2/quota_class_sets.py
@@ -44,7 +44,7 @@ class QuotaClassSetsMixin(object):
             raise webob.exc.HTTPForbidden()
 
         return self._view_builder.detail_list(
-            QUOTAS.get_class_quotas(context, id), id)
+            req, QUOTAS.get_class_quotas(context, id), id)
 
     @wsgi.Controller.authorize("update")
     def _update(self, req, id, body):
@@ -60,7 +60,7 @@ class QuotaClassSetsMixin(object):
                 except exception.AdminRequired:
                     raise webob.exc.HTTPForbidden()
         return self._view_builder.detail_list(
-            QUOTAS.get_class_quotas(context, quota_class))
+            req, QUOTAS.get_class_quotas(context, quota_class))
 
 
 class QuotaClassSetsControllerLegacy(QuotaClassSetsMixin, wsgi.Controller):
diff --git a/manila/api/v2/quota_sets.py b/manila/api/v2/quota_sets.py
index cdcd3e7c89..61c1e5a468 100644
--- a/manila/api/v2/quota_sets.py
+++ b/manila/api/v2/quota_sets.py
@@ -79,10 +79,20 @@ class QuotaSetsMixin(object):
         share_type = params.get('share_type', [None])[0]
         if share_type:
             msg = _("'share_type' key is not supported by this microversion. "
-                    "Use 2.38 or greater microversion to be able "
+                    "Use 2.39 or greater microversion to be able "
                     "to use 'share_type' quotas.")
             raise webob.exc.HTTPBadRequest(explanation=msg)
 
+    @staticmethod
+    def _ensure_share_group_related_args_are_absent(body):
+        body = body.get('quota_set', body)
+        for key in ('share_groups', 'share_group_snapshots'):
+            if body.get(key):
+                msg = _("'%(key)s' key is not supported by this microversion. "
+                        "Use 2.40 or greater microversion to be able "
+                        "to use '%(key)s' quotas.") % {"key": key}
+                raise webob.exc.HTTPBadRequest(explanation=msg)
+
     def _get_quotas(self, context, project_id, user_id=None,
                     share_type_id=None, usages=False):
         self._validate_user_id_and_share_type_args(user_id, share_type_id)
@@ -112,14 +122,16 @@ class QuotaSetsMixin(object):
             share_type_id = self._get_share_type_id(context, share_type)
             quotas = self._get_quotas(
                 context, id, user_id, share_type_id, usages=detail)
-            return self._view_builder.detail_list(quotas, id, share_type_id)
+            return self._view_builder.detail_list(
+                req, quotas, id, share_type_id)
         except exception.NotAuthorized:
             raise webob.exc.HTTPForbidden()
 
     @wsgi.Controller.authorize('show')
     def _defaults(self, req, id):
         context = req.environ['manila.context']
-        return self._view_builder.detail_list(QUOTAS.get_defaults(context), id)
+        return self._view_builder.detail_list(
+            req, QUOTAS.get_defaults(context), id)
 
     @wsgi.Controller.authorize("update")
     def _update(self, req, id, body):
@@ -132,6 +144,12 @@ class QuotaSetsMixin(object):
         share_type = params.get('share_type', [None])[0]
         self._validate_user_id_and_share_type_args(user_id, share_type)
         share_type_id = self._get_share_type_id(context, share_type)
+        body = body.get('quota_set', {})
+        if share_type and body.get('share_groups',
+                                   body.get('share_group_snapshots')):
+            msg = _("Share type quotas handle only 'shares', 'gigabytes', "
+                    "'snapshots' and 'snapshot_gigabytes' quotas.")
+            raise webob.exc.HTTPBadRequest(explanation=msg)
 
         try:
             settable_quotas = QUOTAS.get_settable_quotas(
@@ -140,7 +158,7 @@ class QuotaSetsMixin(object):
         except exception.NotAuthorized:
             raise webob.exc.HTTPForbidden()
 
-        for key, value in body.get('quota_set', {}).items():
+        for key, value in body.items():
             if key == 'share_networks' and share_type_id:
                 msg = _("'share_networks' quota cannot be set for share type. "
                         "It can be set only for project or user.")
@@ -171,7 +189,7 @@ class QuotaSetsMixin(object):
         except exception.NotAuthorized:
             raise webob.exc.HTTPForbidden()
 
-        for key, value in body.get('quota_set', {}).items():
+        for key, value in body.items():
             if key in NON_QUOTA_KEYS or (not value and value != 0):
                 continue
             # validate whether already used and reserved exceeds the new
@@ -216,6 +234,7 @@ class QuotaSetsMixin(object):
             except exception.AdminRequired:
                 raise webob.exc.HTTPForbidden()
         return self._view_builder.detail_list(
+            req,
             self._get_quotas(
                 context, id, user_id=user_id, share_type_id=share_type_id),
             share_type=share_type_id,
@@ -262,6 +281,7 @@ class QuotaSetsControllerLegacy(QuotaSetsMixin, wsgi.Controller):
     @wsgi.Controller.api_version('1.0', '2.6')
     def update(self, req, id, body):
         self._ensure_share_type_arg_is_absent(req)
+        self._ensure_share_group_related_args_are_absent(body)
         return self._update(req, id, body)
 
     @wsgi.Controller.api_version('1.0', '2.6')
@@ -297,6 +317,8 @@ class QuotaSetsController(QuotaSetsMixin, wsgi.Controller):
     def update(self, req, id, body):
         if req.api_version_request < api_version.APIVersionRequest("2.39"):
             self._ensure_share_type_arg_is_absent(req)
+        elif req.api_version_request < api_version.APIVersionRequest("2.40"):
+            self._ensure_share_group_related_args_are_absent(body)
         return self._update(req, id, body)
 
     @wsgi.Controller.api_version('2.7')
diff --git a/manila/api/views/quota_class_sets.py b/manila/api/views/quota_class_sets.py
index 90d6578bc8..90600341ad 100644
--- a/manila/api/views/quota_class_sets.py
+++ b/manila/api/views/quota_class_sets.py
@@ -19,8 +19,11 @@ from manila.api import common
 class ViewBuilder(common.ViewBuilder):
 
     _collection_name = "quota_class_set"
+    _detail_version_modifiers = [
+        "add_share_group_quotas",
+    ]
 
-    def detail_list(self, quota_set, quota_class=None):
+    def detail_list(self, request, quota_class_set, quota_class=None):
         """Detailed view of quota class set."""
         keys = (
             'shares',
@@ -29,7 +32,17 @@ class ViewBuilder(common.ViewBuilder):
             'snapshot_gigabytes',
             'share_networks',
         )
-        view = {key: quota_set.get(key) for key in keys}
+        view = {key: quota_class_set.get(key) for key in keys}
         if quota_class:
             view['id'] = quota_class
+        self.update_versioned_resource_dict(request, view, quota_class_set)
         return {self._collection_name: view}
+
+    @common.ViewBuilder.versioned_method("2.40")
+    def add_share_group_quotas(self, context, view, quota_class_set):
+        share_groups = quota_class_set.get('share_groups')
+        share_group_snapshots = quota_class_set.get('share_group_snapshots')
+        if share_groups is not None:
+            view['share_groups'] = share_groups
+        if share_group_snapshots is not None:
+            view['share_group_snapshots'] = share_group_snapshots
diff --git a/manila/api/views/quota_sets.py b/manila/api/views/quota_sets.py
index 4d89bf78f7..b85b583eda 100644
--- a/manila/api/views/quota_sets.py
+++ b/manila/api/views/quota_sets.py
@@ -19,8 +19,12 @@ from manila.api import common
 class ViewBuilder(common.ViewBuilder):
 
     _collection_name = "quota_set"
+    _detail_version_modifiers = [
+        "add_share_group_quotas",
+    ]
 
-    def detail_list(self, quota_set, project_id=None, share_type=None):
+    def detail_list(self, request, quota_set, project_id=None,
+                    share_type=None):
         """Detailed view of quota set."""
         keys = (
             'shares',
@@ -31,6 +35,21 @@ class ViewBuilder(common.ViewBuilder):
         view = {key: quota_set.get(key) for key in keys}
         if project_id:
             view['id'] = project_id
-        if not share_type:
+        if share_type:
+            # NOTE(vponomaryov): remove share groups related data for quotas
+            # that are share-type based.
+            quota_set.pop('share_groups', None)
+            quota_set.pop('share_group_snapshots', None)
+        else:
             view['share_networks'] = quota_set.get('share_networks')
+        self.update_versioned_resource_dict(request, view, quota_set)
         return {self._collection_name: view}
+
+    @common.ViewBuilder.versioned_method("2.40")
+    def add_share_group_quotas(self, context, view, quota_set):
+        share_groups = quota_set.get('share_groups')
+        share_group_snapshots = quota_set.get('share_group_snapshots')
+        if share_groups is not None:
+            view['share_groups'] = share_groups
+        if share_group_snapshots is not None:
+            view['share_group_snapshots'] = share_group_snapshots
diff --git a/manila/db/sqlalchemy/api.py b/manila/db/sqlalchemy/api.py
index b4ed676600..2974d193b2 100644
--- a/manila/db/sqlalchemy/api.py
+++ b/manila/db/sqlalchemy/api.py
@@ -317,12 +317,30 @@ def _sync_share_networks(context, project_id, user_id, session,
     return {'share_networks': share_networks_count}
 
 
+def _sync_share_groups(context, project_id, user_id, session,
+                       share_type_id=None):
+    share_groups_count = count_share_groups(
+        context, project_id, user_id, share_type_id=share_type_id,
+        session=session)
+    return {'share_groups': share_groups_count}
+
+
+def _sync_share_group_snapshots(context, project_id, user_id, session,
+                                share_type_id=None):
+    share_group_snapshots_count = count_share_group_snapshots(
+        context, project_id, user_id, share_type_id=share_type_id,
+        session=session)
+    return {'share_group_snapshots': share_group_snapshots_count}
+
+
 QUOTA_SYNC_FUNCTIONS = {
     '_sync_shares': _sync_shares,
     '_sync_snapshots': _sync_snapshots,
     '_sync_gigabytes': _sync_gigabytes,
     '_sync_snapshot_gigabytes': _sync_snapshot_gigabytes,
     '_sync_share_networks': _sync_share_networks,
+    '_sync_share_groups': _sync_share_groups,
+    '_sync_share_group_snapshots': _sync_share_group_snapshots,
 }
 
 
@@ -4223,6 +4241,41 @@ def get_all_shares_by_share_group(context, share_group_id, session=None):
             all())
 
 
+@require_context
+def count_share_groups(context, project_id, user_id=None,
+                       share_type_id=None, session=None):
+    query = model_query(
+        context, models.ShareGroup,
+        func.count(models.ShareGroup.id),
+        read_deleted="no",
+        session=session).filter_by(project_id=project_id)
+    if share_type_id:
+        query = query.join("share_group_share_type_mappings").filter_by(
+            share_type_id=share_type_id)
+    elif user_id is not None:
+        query = query.filter_by(user_id=user_id)
+    return query.first()[0]
+
+
+@require_context
+def count_share_group_snapshots(context, project_id, user_id=None,
+                                share_type_id=None, session=None):
+    query = model_query(
+        context, models.ShareGroupSnapshot,
+        func.count(models.ShareGroupSnapshot.id),
+        read_deleted="no",
+        session=session).filter_by(project_id=project_id)
+    if share_type_id:
+        query = query.join(
+            "share_group"
+        ).join(
+            "share_group_share_type_mappings"
+        ).filter_by(share_type_id=share_type_id)
+    elif user_id is not None:
+        query = query.filter_by(user_id=user_id)
+    return query.first()[0]
+
+
 @require_context
 def count_share_group_snapshots_in_share_group(context, share_group_id,
                                                session=None):
diff --git a/manila/exception.py b/manila/exception.py
index dad14db505..2c0bdef5c1 100644
--- a/manila/exception.py
+++ b/manila/exception.py
@@ -428,6 +428,16 @@ class ShareNetworksLimitExceeded(QuotaError):
                 "allowed (%(allowed)d) exceeded.")
 
 
+class ShareGroupsLimitExceeded(QuotaError):
+    message = _(
+        "Maximum number of allowed share-groups is exceeded.")
+
+
+class ShareGroupSnapshotsLimitExceeded(QuotaError):
+    message = _(
+        "Maximum number of allowed share-group-snapshots is exceeded.")
+
+
 class GlusterfsException(ManilaException):
     message = _("Unknown Gluster exception.")
 
diff --git a/manila/quota.py b/manila/quota.py
index 934eb775a7..670f4aa5c8 100644
--- a/manila/quota.py
+++ b/manila/quota.py
@@ -45,6 +45,14 @@ quota_opts = [
     cfg.IntOpt('quota_share_networks',
                default=10,
                help='Number of share-networks allowed per project.'),
+
+    cfg.IntOpt('quota_share_groups',
+               default=50,
+               help='Number of share groups allowed.'),
+    cfg.IntOpt('quota_share_group_snapshots',
+               default=50,
+               help='Number of share group snapshots allowed.'),
+
     cfg.IntOpt('reservation_expire',
                default=86400,
                help='Number of seconds until a reservation expires.'),
@@ -1047,6 +1055,10 @@ resources = [
                        'quota_snapshot_gigabytes'),
     ReservableResource('share_networks', '_sync_share_networks',
                        'quota_share_networks'),
+    ReservableResource('share_groups', '_sync_share_groups',
+                       'quota_share_groups'),
+    ReservableResource('share_group_snapshots', '_sync_share_group_snapshots',
+                       'quota_share_group_snapshots'),
 ]
 
 
diff --git a/manila/share_group/api.py b/manila/share_group/api.py
index 74a38a9da1..a91b9f7a35 100644
--- a/manila/share_group/api.py
+++ b/manila/share_group/api.py
@@ -27,15 +27,15 @@ from manila.common import constants
 from manila.db import base
 from manila import exception
 from manila.i18n import _
+from manila import quota
 from manila.scheduler import rpcapi as scheduler_rpcapi
 from manila import share
 from manila.share import rpcapi as share_rpcapi
 from manila.share import share_types
 
-
 CONF = cfg.CONF
-
 LOG = log.getLogger(__name__)
+QUOTAS = quota.QUOTAS
 
 
 class API(base.Base):
@@ -138,6 +138,28 @@ class API(base.Base):
                     "types supported by the share group type.")
             raise exception.InvalidInput(reason=msg)
 
+        try:
+            reservations = QUOTAS.reserve(context, share_groups=1)
+        except exception.OverQuota as e:
+            overs = e.kwargs['overs']
+            usages = e.kwargs['usages']
+            quotas = e.kwargs['quotas']
+
+            def _consumed(name):
+                return (usages[name]['reserved'] + usages[name]['in_use'])
+
+            if 'share_groups' in overs:
+                msg = ("Quota exceeded for '%(s_uid)s' user in '%(s_pid)s' "
+                       "project. (%(d_consumed)d of "
+                       "%(d_quota)d already consumed).")
+                LOG.warning(msg, {
+                    's_pid': context.project_id,
+                    's_uid': context.user_id,
+                    'd_consumed': _consumed('share_groups'),
+                    'd_quota': quotas['share_groups'],
+                })
+            raise exception.ShareGroupsLimitExceeded()
+
         options = {
             'share_group_type_id': share_group_type_id,
             'source_share_group_snapshot_id': source_share_group_snapshot_id,
@@ -154,9 +176,9 @@ class API(base.Base):
         if original_share_group:
             options['host'] = original_share_group['host']
 
-        share_group = self.db.share_group_create(context, options)
-
+        share_group = None
         try:
+            share_group = self.db.share_group_create(context, options)
             if share_group_snapshot:
                 members = self.db.share_group_snapshot_members_get_all(
                     context, source_share_group_snapshot_id)
@@ -178,8 +200,16 @@ class API(base.Base):
                         share_network_id=share_network_id)
         except Exception:
             with excutils.save_and_reraise_exception():
-                self.db.share_group_destroy(
-                    context.elevated(), share_group['id'])
+                if share_group:
+                    self.db.share_group_destroy(
+                        context.elevated(), share_group['id'])
+                QUOTAS.rollback(context, reservations)
+
+        try:
+            QUOTAS.commit(context, reservations)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                QUOTAS.rollback(context, reservations)
 
         request_spec = {'share_group_id': share_group['id']}
         request_spec.update(options)
@@ -224,7 +254,30 @@ class API(base.Base):
         share_group = self.db.share_group_update(
             context, share_group_id, {'status': constants.STATUS_DELETING})
 
-        self.share_rpcapi.delete_share_group(context, share_group)
+        try:
+            reservations = QUOTAS.reserve(
+                context,
+                share_groups=-1,
+                project_id=share_group['project_id'],
+                user_id=share_group['user_id'],
+            )
+        except exception.OverQuota as e:
+            reservations = None
+            LOG.exception(
+                ("Failed to update quota for deleting share group: %s"), e)
+
+        try:
+            self.share_rpcapi.delete_share_group(context, share_group)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                QUOTAS.rollback(context, reservations)
+
+        if reservations:
+            QUOTAS.commit(
+                context, reservations,
+                project_id=share_group['project_id'],
+                user_id=share_group['user_id'],
+            )
 
     def update(self, context, group, fields):
         return self.db.share_group_update(context, group['id'], fields)
@@ -285,8 +338,31 @@ class API(base.Base):
                           "status": constants.STATUS_AVAILABLE})
                 raise exception.InvalidShareGroup(reason=msg)
 
-        snap = self.db.share_group_snapshot_create(context, options)
         try:
+            reservations = QUOTAS.reserve(context, share_group_snapshots=1)
+        except exception.OverQuota as e:
+            overs = e.kwargs['overs']
+            usages = e.kwargs['usages']
+            quotas = e.kwargs['quotas']
+
+            def _consumed(name):
+                return (usages[name]['reserved'] + usages[name]['in_use'])
+
+            if 'share_group_snapshots' in overs:
+                msg = ("Quota exceeded for '%(s_uid)s' user in '%(s_pid)s' "
+                       "project. (%(d_consumed)d of "
+                       "%(d_quota)d already consumed).")
+                LOG.warning(msg, {
+                    's_pid': context.project_id,
+                    's_uid': context.user_id,
+                    'd_consumed': _consumed('share_group_snapshots'),
+                    'd_quota': quotas['share_group_snapshots'],
+                })
+            raise exception.ShareGroupSnapshotsLimitExceeded()
+
+        snap = None
+        try:
+            snap = self.db.share_group_snapshot_create(context, options)
             members = []
             for s in shares:
                 member_options = {
@@ -308,7 +384,15 @@ class API(base.Base):
         except Exception:
             with excutils.save_and_reraise_exception():
                 # This will delete the snapshot and all of it's members
-                self.db.share_group_snapshot_destroy(context, snap['id'])
+                if snap:
+                    self.db.share_group_snapshot_destroy(context, snap['id'])
+                QUOTAS.rollback(context, reservations)
+
+        try:
+            QUOTAS.commit(context, reservations)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                QUOTAS.rollback(context, reservations)
 
         return snap
 
@@ -325,10 +409,30 @@ class API(base.Base):
         self.db.share_group_snapshot_update(
             context, snap_id, {'status': constants.STATUS_DELETING})
 
+        try:
+            reservations = QUOTAS.reserve(
+                context,
+                share_group_snapshots=-1,
+                project_id=snap['project_id'],
+                user_id=snap['user_id'],
+            )
+        except exception.OverQuota as e:
+            reservations = None
+            LOG.exception(
+                ("Failed to update quota for deleting share group snapshot: "
+                 "%s"), e)
+
         # Cast to share manager
         self.share_rpcapi.delete_share_group_snapshot(
             context, snap, share_group['host'])
 
+        if reservations:
+            QUOTAS.commit(
+                context, reservations,
+                project_id=snap['project_id'],
+                user_id=snap['user_id'],
+            )
+
     def update_share_group_snapshot(self, context, share_group_snapshot,
                                     fields):
         return self.db.share_group_snapshot_update(
diff --git a/manila/tests/api/v2/test_quota_sets.py b/manila/tests/api/v2/test_quota_sets.py
index 6d5be719e1..99648a5fad 100644
--- a/manila/tests/api/v2/test_quota_sets.py
+++ b/manila/tests/api/v2/test_quota_sets.py
@@ -18,8 +18,6 @@
 Tests for manila.api.v2.quota_sets.py
 """
 
-import copy
-
 import ddt
 import mock
 from oslo_config import cfg
@@ -37,18 +35,18 @@ from manila import utils
 
 CONF = cfg.CONF
 
-REQ = mock.MagicMock(api_version_request=api_version.APIVersionRequest("2.39"))
-REQ.environ = {'manila.context': context.get_admin_context()}
-REQ.environ['manila.context'].is_admin = True
-REQ.environ['manila.context'].auth_token = 'foo_auth_token'
-REQ.environ['manila.context'].project_id = 'foo_project_id'
 
-REQ_WITH_USER = copy.deepcopy(REQ)
-REQ_WITH_USER.environ['manila.context'].user_id = 'foo_user_id'
-REQ_WITH_USER.environ['QUERY_STRING'] = 'user_id=foo_user_id'
-
-REQ_MEMBER = copy.deepcopy(REQ)
-REQ_MEMBER.environ['manila.context'].is_admin = False
+def _get_request(is_admin, user_in_url):
+    req = mock.MagicMock(
+        api_version_request=api_version.APIVersionRequest("2.40"))
+    req.environ = {'manila.context': context.get_admin_context()}
+    req.environ['manila.context'].is_admin = is_admin
+    req.environ['manila.context'].auth_token = 'foo_auth_token'
+    req.environ['manila.context'].project_id = 'foo_project_id'
+    if user_in_url:
+        req.environ['manila.context'].user_id = 'foo_user_id'
+        req.environ['QUERY_STRING'] = 'user_id=foo_user_id'
+    return req
 
 
 @ddt.ddt
@@ -72,8 +70,11 @@ class QuotaSetsControllerTest(test.TestCase):
         {"gigabytes": 7},
         {"snapshot_gigabytes": 10001},
         {"share_networks": 12345},
+        {"share_groups": 123456},
+        {"share_group_snapshots": 123456},
     )
     def test_defaults(self, quotas):
+        req = _get_request(True, False)
         for k, v in quotas.items():
             CONF.set_default('quota_' + k, v)
         expected = {
@@ -84,14 +85,17 @@ class QuotaSetsControllerTest(test.TestCase):
                 'snapshots': quotas.get('snapshots', 50),
                 'snapshot_gigabytes': quotas.get('snapshot_gigabytes', 1000),
                 'share_networks': quotas.get('share_networks', 10),
+                'share_groups': quotas.get('share_groups', 50),
+                'share_group_snapshots': quotas.get(
+                    'share_group_snapshots', 50),
             }
         }
 
-        result = self.controller.defaults(REQ, self.project_id)
+        result = self.controller.defaults(req, self.project_id)
 
         self.assertEqual(expected, result)
         self.mock_policy_check.assert_called_once_with(
-            REQ.environ['manila.context'], self.resource_name, 'show')
+            req.environ['manila.context'], self.resource_name, 'show')
 
     @ddt.data(
         ('os-', '1.0', quota_sets.QuotaSetsControllerLegacy, 'defaults'),
@@ -124,17 +128,18 @@ class QuotaSetsControllerTest(test.TestCase):
 
     @staticmethod
     def _get_share_type_request_object(microversion=None):
-        req = copy.deepcopy(REQ)
+        req = _get_request(True, False)
         req.environ['QUERY_STRING'] = 'share_type=fake_share_type_name_or_id'
         req.api_version_request = api_version.APIVersionRequest(
             microversion or '2.39')
         return req
 
-    def test_share_type_quota_detail(self):
+    @ddt.data('2.39', '2.40')
+    def test_share_type_quota_detail(self, microversion):
         self.mock_object(
             quota_sets.db, 'share_type_get_by_name_or_id',
             mock.Mock(return_value={'id': 'fake_st_id'}))
-        req = self._get_share_type_request_object('2.39')
+        req = self._get_share_type_request_object(microversion)
         quotas = {
             "shares": 23,
             "snapshots": 34,
@@ -176,11 +181,12 @@ class QuotaSetsControllerTest(test.TestCase):
         quota_sets.db.share_type_get_by_name_or_id.assert_called_once_with(
             req.environ['manila.context'], 'fake_share_type_name_or_id')
 
-    def test_show_share_type_quota(self):
+    @ddt.data('2.39', '2.40')
+    def test_show_share_type_quota(self, microversion):
         self.mock_object(
             quota_sets.db, 'share_type_get_by_name_or_id',
             mock.Mock(return_value={'id': 'fake_st_id'}))
-        req = self._get_share_type_request_object('2.39')
+        req = self._get_share_type_request_object(microversion)
         quotas = {
             "shares": 23,
             "snapshots": 34,
@@ -262,7 +268,30 @@ class QuotaSetsControllerTest(test.TestCase):
 
         self.assertIsNone(result)
 
-    @ddt.data(REQ, REQ_WITH_USER)
+    @ddt.data(
+        {},
+        {"quota_set": {}},
+        {"quota_set": {"foo": "bar"}},
+        {"foo": "bar"},
+    )
+    def test__ensure_share_group_related_args_are_absent_success(self, body):
+        result = self.controller._ensure_share_group_related_args_are_absent(
+            body)
+
+        self.assertIsNone(result)
+
+    @ddt.data(
+        {"share_groups": 5},
+        {"share_group_snapshots": 6},
+        {"quota_set": {"share_groups": 7}},
+        {"quota_set": {"share_group_snapshots": 8}},
+    )
+    def test__ensure_share_group_related_args_are_absent_error(self, body):
+        self.assertRaises(
+            webob.exc.HTTPBadRequest,
+            self.controller._ensure_share_group_related_args_are_absent, body)
+
+    @ddt.data(_get_request(True, True), _get_request(True, False))
     def test__ensure_share_type_arg_is_absent(self, req):
         result = self.controller._ensure_share_type_arg_is_absent(req)
 
@@ -276,7 +305,7 @@ class QuotaSetsControllerTest(test.TestCase):
             self.controller._ensure_share_type_arg_is_absent,
             req)
 
-    @ddt.data(REQ, REQ_WITH_USER)
+    @ddt.data(_get_request(True, True), _get_request(True, False))
     def test_quota_detail(self, request):
         request.api_version_request = api_version.APIVersionRequest('2.25')
         quotas = {
@@ -318,7 +347,7 @@ class QuotaSetsControllerTest(test.TestCase):
         self.mock_policy_check.assert_called_once_with(
             request.environ['manila.context'], self.resource_name, 'show')
 
-    @ddt.data(REQ, REQ_WITH_USER)
+    @ddt.data(_get_request(True, True), _get_request(True, False))
     def test_show_quota(self, request):
         quotas = {
             "shares": 23,
@@ -326,6 +355,8 @@ class QuotaSetsControllerTest(test.TestCase):
             "gigabytes": 45,
             "snapshot_gigabytes": 56,
             "share_networks": 67,
+            "share_groups": 53,
+            "share_group_snapshots": 57,
         }
         expected = {
             'quota_set': {
@@ -335,6 +366,9 @@ class QuotaSetsControllerTest(test.TestCase):
                 'snapshots': quotas.get('snapshots', 50),
                 'snapshot_gigabytes': quotas.get('snapshot_gigabytes', 1000),
                 'share_networks': quotas.get('share_networks', 10),
+                'share_groups': quotas.get('share_groups', 50),
+                'share_group_snapshots': quotas.get(
+                    'share_group_snapshots', 50),
             }
         }
         for k, v in quotas.items():
@@ -347,6 +381,7 @@ class QuotaSetsControllerTest(test.TestCase):
             request.environ['manila.context'], self.resource_name, 'show')
 
     def test_show_quota_not_authorized(self):
+        req = _get_request(True, False)
         self.mock_object(
             quota_sets.db,
             'authorize_project_context',
@@ -355,11 +390,11 @@ class QuotaSetsControllerTest(test.TestCase):
         self.assertRaises(
             webob.exc.HTTPForbidden,
             self.controller.show,
-            REQ, self.project_id)
+            req, self.project_id)
         self.mock_policy_check.assert_called_once_with(
-            REQ.environ['manila.context'], self.resource_name, 'show')
+            req.environ['manila.context'], self.resource_name, 'show')
 
-    @ddt.data(REQ, REQ_WITH_USER)
+    @ddt.data(_get_request(True, True), _get_request(True, False))
     def test_update_quota(self, request):
         self.mock_object(
             quota_sets.db, 'share_type_get_by_name_or_id',
@@ -374,6 +409,8 @@ class QuotaSetsControllerTest(test.TestCase):
                 'snapshots': 50,
                 'snapshot_gigabytes': 1000,
                 'share_networks': 10,
+                'share_groups': 50,
+                'share_group_snapshots': 50,
             }
         }
         mock_policy_update_check_call = mock.call(
@@ -394,12 +431,13 @@ class QuotaSetsControllerTest(test.TestCase):
             mock_policy_update_check_call, mock_policy_show_check_call])
         quota_sets.db.share_type_get_by_name_or_id.assert_not_called()
 
-    def test_update_share_type_quota(self):
+    @ddt.data('2.39', '2.40')
+    def test_update_share_type_quota(self, microversion):
         self.mock_object(
             quota_sets.db, 'share_type_get_by_name_or_id',
             mock.Mock(
                 return_value={'id': 'fake_st_id', 'name': 'fake_st_name'}))
-        req = self._get_share_type_request_object('2.39')
+        req = self._get_share_type_request_object(microversion)
 
         CONF.set_default('quota_shares', 789)
         body = {'quota_set': {'tenant_id': self.project_id, 'shares': 788}}
@@ -468,14 +506,15 @@ class QuotaSetsControllerTest(test.TestCase):
 
     @ddt.data(-2, 'foo', {1: 2}, [1])
     def test_update_quota_with_invalid_value(self, value):
+        req = _get_request(True, False)
         body = {'quota_set': {'tenant_id': self.project_id, 'shares': value}}
 
         self.assertRaises(
             webob.exc.HTTPBadRequest,
             self.controller.update,
-            REQ, self.project_id, body=body)
+            req, self.project_id, body=body)
         self.mock_policy_check.assert_called_once_with(
-            REQ.environ['manila.context'], self.resource_name, 'update')
+            req.environ['manila.context'], self.resource_name, 'update')
 
     def test_user_quota_can_not_be_bigger_than_tenant_quota(self):
         value = 777
@@ -486,14 +525,14 @@ class QuotaSetsControllerTest(test.TestCase):
                 'shares': value + 1,
             }
         }
+        req = _get_request(True, True)
 
         self.assertRaises(
             webob.exc.HTTPBadRequest,
             self.controller.update,
-            REQ_WITH_USER, self.project_id, body=body)
+            req, self.project_id, body=body)
         self.mock_policy_check.assert_called_once_with(
-            REQ_WITH_USER.environ['manila.context'], self.resource_name,
-            'update')
+            req.environ['manila.context'], self.resource_name, 'update')
 
     def test_update_inexistent_quota(self):
         body = {
@@ -502,23 +541,25 @@ class QuotaSetsControllerTest(test.TestCase):
                 'fake_quota': 13,
             }
         }
+        req = _get_request(True, False)
 
         self.assertRaises(
             webob.exc.HTTPBadRequest,
             self.controller.update,
-            REQ, self.project_id, body=body)
+            req, self.project_id, body=body)
         self.mock_policy_check.assert_called_once_with(
-            REQ.environ['manila.context'], self.resource_name, 'update')
+            req.environ['manila.context'], self.resource_name, 'update')
 
     def test_update_quota_not_authorized(self):
         body = {'quota_set': {'tenant_id': self.project_id, 'shares': 13}}
+        req = _get_request(False, False)
 
         self.assertRaises(
             webob.exc.HTTPForbidden,
             self.controller.update,
-            REQ_MEMBER, self.project_id, body=body)
+            req, self.project_id, body=body)
         self.mock_policy_check.assert_called_once_with(
-            REQ_MEMBER.environ['manila.context'], self.resource_name, 'update')
+            req.environ['manila.context'], self.resource_name, 'update')
 
     @ddt.data(
         ('os-quota-sets', '1.0', quota_sets.QuotaSetsControllerLegacy),
@@ -597,8 +638,9 @@ class QuotaSetsControllerTest(test.TestCase):
         project_id = 'foo_project_id'
         self.mock_object(quota_sets.QUOTAS, 'destroy_all_by_project_and_user')
         self.mock_object(quota_sets.QUOTAS, 'destroy_all_by_project')
+        req = _get_request(True, True)
 
-        result = self.controller.delete(REQ_WITH_USER, project_id)
+        result = self.controller.delete(req, project_id)
 
         self.assertTrue(
             utils.IsAMatcher(webob.response.Response) == result
@@ -607,13 +649,12 @@ class QuotaSetsControllerTest(test.TestCase):
         self.assertEqual(202, result.status_code)
         (quota_sets.QUOTAS.destroy_all_by_project_and_user.
             assert_called_once_with(
-                REQ_WITH_USER.environ['manila.context'],
+                req.environ['manila.context'],
                 project_id,
-                REQ_WITH_USER.environ['manila.context'].user_id))
+                req.environ['manila.context'].user_id))
         self.assertFalse(quota_sets.QUOTAS.destroy_all_by_project.called)
         self.mock_policy_check.assert_called_once_with(
-            REQ_WITH_USER.environ['manila.context'], self.resource_name,
-            'delete')
+            req.environ['manila.context'], self.resource_name, 'delete')
 
     def test_delete_share_type_quota(self):
         req = self._get_share_type_request_object('2.39')
@@ -656,12 +697,13 @@ class QuotaSetsControllerTest(test.TestCase):
         quota_sets.db.share_type_get_by_name_or_id.assert_not_called()
 
     def test_delete_not_authorized(self):
+        req = _get_request(False, False)
         self.assertRaises(
             webob.exc.HTTPForbidden,
             self.controller.delete,
-            REQ_MEMBER, self.project_id)
+            req, self.project_id)
         self.mock_policy_check.assert_called_once_with(
-            REQ_MEMBER.environ['manila.context'], self.resource_name, 'delete')
+            req.environ['manila.context'], self.resource_name, 'delete')
 
     @ddt.data(
         ('os-quota-sets', '2.7', quota_sets.QuotaSetsControllerLegacy),
diff --git a/manila/tests/api/views/test_quota_class_sets.py b/manila/tests/api/views/test_quota_class_sets.py
new file mode 100644
index 0000000000..630a8dfd3a
--- /dev/null
+++ b/manila/tests/api/views/test_quota_class_sets.py
@@ -0,0 +1,70 @@
+# Copyright (c) 2017 Mirantis, Inc.
+# All Rights Reserved.
+#
+#    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.
+
+import ddt
+
+from manila.api.openstack import api_version_request as api_version
+from manila.api.views import quota_class_sets
+from manila import test
+from manila.tests.api import fakes
+
+
+@ddt.ddt
+class ViewBuilderTestCase(test.TestCase):
+
+    def setUp(self):
+        super(ViewBuilderTestCase, self).setUp()
+        self.builder = quota_class_sets.ViewBuilder()
+
+    def test__collection_name(self):
+        self.assertEqual('quota_class_set', self.builder._collection_name)
+
+    @ddt.data(
+        ("fake_quota_class", "2.40"), (None, "2.40"),
+        ("fake_quota_class", "2.39"), (None, "2.39"),
+    )
+    @ddt.unpack
+    def test_detail_list_with_share_type(self, quota_class, microversion):
+        req = fakes.HTTPRequest.blank('/quota-sets', version=microversion)
+        quota_class_set = {
+            "shares": 13,
+            "gigabytes": 31,
+            "snapshots": 14,
+            "snapshot_gigabytes": 41,
+            "share_groups": 15,
+            "share_group_snapshots": 51,
+            "share_networks": 16,
+        }
+        expected = {self.builder._collection_name: {
+            "shares": quota_class_set["shares"],
+            "gigabytes": quota_class_set["gigabytes"],
+            "snapshots": quota_class_set["snapshots"],
+            "snapshot_gigabytes": quota_class_set["snapshot_gigabytes"],
+            "share_networks": quota_class_set["share_networks"],
+        }}
+        if quota_class:
+            expected[self.builder._collection_name]['id'] = quota_class
+        if (api_version.APIVersionRequest(microversion) >= (
+                api_version.APIVersionRequest("2.40"))):
+            expected[self.builder._collection_name][
+                "share_groups"] = quota_class_set["share_groups"]
+            expected[self.builder._collection_name][
+                "share_group_snapshots"] = quota_class_set[
+                    "share_group_snapshots"]
+
+        result = self.builder.detail_list(
+            req, quota_class_set, quota_class=quota_class)
+
+        self.assertEqual(expected, result)
diff --git a/manila/tests/api/views/test_quota_sets.py b/manila/tests/api/views/test_quota_sets.py
new file mode 100644
index 0000000000..683b4d46b3
--- /dev/null
+++ b/manila/tests/api/views/test_quota_sets.py
@@ -0,0 +1,79 @@
+# Copyright (c) 2017 Mirantis, Inc.
+# All Rights Reserved.
+#
+#    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.
+
+import ddt
+
+from manila.api.openstack import api_version_request as api_version
+from manila.api.views import quota_sets
+from manila import test
+from manila.tests.api import fakes
+
+
+@ddt.ddt
+class ViewBuilderTestCase(test.TestCase):
+
+    def setUp(self):
+        super(ViewBuilderTestCase, self).setUp()
+        self.builder = quota_sets.ViewBuilder()
+
+    def test__collection_name(self):
+        self.assertEqual('quota_set', self.builder._collection_name)
+
+    @ddt.data(
+        ('fake_project_id', 'fake_share_type_id', "2.40"),
+        (None, 'fake_share_type_id', "2.40"),
+        ('fake_project_id', None, "2.40"),
+        (None, None, "2.40"),
+        ('fake_project_id', 'fake_share_type_id', "2.39"),
+        (None, 'fake_share_type_id', "2.39"),
+        ('fake_project_id', None, "2.39"),
+        (None, None, "2.39"),
+    )
+    @ddt.unpack
+    def test_detail_list_with_share_type(self, project_id, share_type,
+                                         microversion):
+        req = fakes.HTTPRequest.blank('/quota-sets', version=microversion)
+        quota_set = {
+            "shares": 13,
+            "gigabytes": 31,
+            "snapshots": 14,
+            "snapshot_gigabytes": 41,
+            "share_groups": 15,
+            "share_group_snapshots": 51,
+            "share_networks": 16,
+        }
+        expected = {self.builder._collection_name: {
+            "shares": quota_set["shares"],
+            "gigabytes": quota_set["gigabytes"],
+            "snapshots": quota_set["snapshots"],
+            "snapshot_gigabytes": quota_set["snapshot_gigabytes"],
+        }}
+        if project_id:
+            expected[self.builder._collection_name]['id'] = project_id
+        if not share_type:
+            expected[self.builder._collection_name][
+                "share_networks"] = quota_set["share_networks"]
+            if (api_version.APIVersionRequest(microversion) >= (
+                    api_version.APIVersionRequest("2.40"))):
+                expected[self.builder._collection_name][
+                    "share_groups"] = quota_set["share_groups"]
+                expected[self.builder._collection_name][
+                    "share_group_snapshots"] = quota_set[
+                        "share_group_snapshots"]
+
+        result = self.builder.detail_list(
+            req, quota_set, project_id=project_id, share_type=share_type)
+
+        self.assertEqual(expected, result)
diff --git a/manila/tests/share_group/test_api.py b/manila/tests/share_group/test_api.py
index 950c85b787..66cfa1092a 100644
--- a/manila/tests/share_group/test_api.py
+++ b/manila/tests/share_group/test_api.py
@@ -79,7 +79,10 @@ def fake_share_group_snapshot(id, **kwargs):
 class ShareGroupsAPITestCase(test.TestCase):
     def setUp(self):
         super(ShareGroupsAPITestCase, self).setUp()
-        self.context = context.get_admin_context()
+        self.user_id = 'fake_user_id'
+        self.project_id = 'fake_project_id'
+        self.context = context.RequestContext(
+            user_id=self.user_id, project_id=self.project_id, is_admin=True)
         self.scheduler_rpcapi = mock.Mock()
         self.share_rpcapi = mock.Mock()
         self.share_api = mock.Mock()
@@ -108,8 +111,12 @@ class ShareGroupsAPITestCase(test.TestCase):
                 {'share_type_id': self.fake_share_type_2['id']},
             ]
         }
-        self.mock_object(db_driver, 'share_group_type_get',
-                         mock.Mock(return_value=self.fake_share_group_type))
+        self.mock_object(
+            db_driver, 'share_group_type_get',
+            mock.Mock(return_value=self.fake_share_group_type))
+        self.mock_object(share_group_api.QUOTAS, 'reserve')
+        self.mock_object(share_group_api.QUOTAS, 'commit')
+        self.mock_object(share_group_api.QUOTAS, 'rollback')
 
     def test_create_empty_request(self):
         share_group = fake_share_group(
@@ -126,6 +133,11 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_create.assert_called_once_with(
             self.context, expected_values)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_request_spec(self):
         """Ensure the correct values are sent to the scheduler."""
@@ -150,6 +162,11 @@ class ShareGroupsAPITestCase(test.TestCase):
         self.scheduler_rpcapi.create_share_group.assert_called_once_with(
             self.context, share_group_id=share_group['id'],
             request_spec=expected_request_spec, filter_properties={})
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_name(self):
         fake_name = 'fake_name'
@@ -172,6 +189,11 @@ class ShareGroupsAPITestCase(test.TestCase):
         self.scheduler_rpcapi.create_share_group.assert_called_once_with(
             self.context, share_group_id=share_group['id'],
             request_spec=mock.ANY, filter_properties={})
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_description(self):
         fake_desc = 'fake_desc'
@@ -190,6 +212,11 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_create.assert_called_once_with(
             self.context, expected_values)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_multiple_share_types(self):
         fake_share_types = [self.fake_share_type, self.fake_share_type_2]
@@ -213,6 +240,11 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_create.assert_called_once_with(
             self.context, expected_values)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_share_type_not_found(self):
         self.mock_object(share_types, 'get_share_type',
@@ -234,6 +266,31 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.api.create,
             self.context, share_type_ids=[self.fake_share_type['id']])
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
+    def test_create_with_error_on_quota_reserve(self):
+        overs = ["share_groups"]
+        usages = {"share_groups": {"reserved": 1, "in_use": 3, "limit": 4}}
+        quotas = {"share_groups": 5}
+        share_group_api.QUOTAS.reserve.side_effect = exception.OverQuota(
+            overs=overs,
+            usages=usages,
+            quotas=quotas,
+        )
+        self.mock_object(share_group_api.LOG, "warning")
+
+        self.assertRaises(
+            exception.ShareGroupsLimitExceeded,
+            self.api.create, self.context)
+
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+        share_group_api.LOG.warning.assert_called_once_with(mock.ANY, mock.ANY)
+
     def test_create_driver_handles_share_servers_is_false_with_net_id(self):
         fake_share_types = [self.fake_share_type]
         self.mock_object(share_types, 'get_share_type')
@@ -266,6 +323,10 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.api.create,
             self.context, share_type_ids=fake_share_type_ids)
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
     def test_create_with_conflicting_share_type_and_share_network(self):
         fake_share_type = {
             'name': 'default',
@@ -283,6 +344,10 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, share_type_ids=fake_share_types,
             share_network_id="fake_sn")
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
     def test_create_with_source_share_group_snapshot_id(self):
         snap = fake_share_group_snapshot(
             "fake_source_share_group_snapshot_id",
@@ -341,6 +406,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_values)
         self.share_rpcapi.create_share_group.assert_called_once_with(
             self.context, share_group, orig_share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_source_share_group_snapshot_id_with_member(self):
         snap = fake_share_group_snapshot(
@@ -403,6 +473,11 @@ class ShareGroupsAPITestCase(test.TestCase):
         self.assertTrue(self.share_api.create.called)
         self.share_rpcapi.create_share_group.assert_called_once_with(
             self.context, share_group, orig_share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_source_sg_snapshot_id_with_members_error(self):
         snap = fake_share_group_snapshot(
@@ -465,6 +540,12 @@ class ShareGroupsAPITestCase(test.TestCase):
         self.assertEqual(2, self.share_api.create.call_count)
         self.assertEqual(1, db_driver.share_group_destroy.call_count)
 
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+
     def test_create_with_source_sg_snapshot_id_error_snapshot_status(self):
         snap = fake_share_group_snapshot(
             "fake_source_share_group_snapshot_id",
@@ -478,6 +559,10 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.api.create,
             self.context, source_share_group_snapshot_id=snap['id'])
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
     def test_create_with_source_sg_snapshot_id_snap_not_found(self):
         snap = fake_share_group_snapshot(
             "fake_source_share_group_snapshot_id",
@@ -492,6 +577,10 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.api.create,
             self.context, source_share_group_snapshot_id=snap['id'])
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
     def test_create_with_multiple_fields(self):
         fake_desc = 'fake_desc'
         fake_name = 'fake_name'
@@ -512,6 +601,11 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_create.assert_called_once_with(
             self.context, expected_values)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_with_error_on_creation(self):
         share_group = fake_share_group(
@@ -528,11 +622,16 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_create.assert_called_once_with(
             self.context, expected_values)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=1)
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
 
     def test_delete_creating_no_host(self):
         share_group = fake_share_group(
-            'fakeid', user_id=self.context.user_id,
-            project_id=self.context.project_id,
+            'fakeid', user_id=self.user_id + '_different_user',
+            project_id=self.project_id + '_in_different_project',
             status=constants.STATUS_CREATING)
         self.mock_object(db_driver, 'share_group_destroy')
 
@@ -540,6 +639,9 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_destroy.assert_called_once_with(
             mock.ANY, share_group['id'])
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_delete_creating_with_host(self):
         share_group = fake_share_group(
@@ -553,8 +655,8 @@ class ShareGroupsAPITestCase(test.TestCase):
 
     def test_delete_available(self):
         share_group = fake_share_group(
-            'fakeid', user_id=self.context.user_id,
-            project_id=self.context.project_id,
+            'fakeid', user_id=self.user_id + '_different_user',
+            project_id=self.project_id + '_in_different_project',
             status=constants.STATUS_AVAILABLE, host="fake_host")
         deleted_share_group = copy.deepcopy(share_group)
         deleted_share_group['status'] = constants.STATUS_DELETING
@@ -570,6 +672,16 @@ class ShareGroupsAPITestCase(test.TestCase):
             {'status': constants.STATUS_DELETING})
         self.share_rpcapi.delete_share_group.assert_called_once_with(
             self.context, deleted_share_group)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=-1,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context,
+            share_group_api.QUOTAS.reserve.return_value,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_delete_error_with_host(self):
         share_group = fake_share_group(
@@ -591,6 +703,16 @@ class ShareGroupsAPITestCase(test.TestCase):
             {'status': constants.STATUS_DELETING})
         self.api.share_rpcapi.delete_share_group.assert_called_once_with(
             self.context, deleted_share_group)
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_groups=-1,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context,
+            share_group_api.QUOTAS.reserve.return_value,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_delete_error_without_host(self):
         share_group = fake_share_group(
@@ -603,6 +725,9 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_destroy.assert_called_once_with(
             mock.ANY, share_group['id'])
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_delete_with_shares(self):
         share_group = fake_share_group(
@@ -617,6 +742,10 @@ class ShareGroupsAPITestCase(test.TestCase):
             exception.InvalidShareGroup,
             self.api.delete, self.context, share_group)
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
     def test_delete_with_share_group_snapshots(self):
         share_group = fake_share_group(
             'fakeid', user_id=self.context.user_id,
@@ -630,6 +759,10 @@ class ShareGroupsAPITestCase(test.TestCase):
             exception.InvalidShareGroup,
             self.api.delete, self.context, share_group)
 
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
     @ddt.data({}, {"name": "fake_name"}, {"description": "fake_description"})
     def test_update(self, expected_values):
         share_group = fake_share_group(
@@ -739,6 +872,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_values)
         self.share_rpcapi.create_share_group_snapshot.assert_called_once_with(
             self.context, snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_sg_snapshot_minimal_request_no_members_with_name(self):
         fake_name = 'fake_name'
@@ -772,6 +910,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_values)
         self.share_rpcapi.create_share_group_snapshot.assert_called_once_with(
             self.context, snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_group_snapshot_minimal_request_no_members_with_desc(self):
         fake_description = 'fake_description'
@@ -807,6 +950,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_values)
         self.share_rpcapi.create_share_group_snapshot.assert_called_once_with(
             self.context, snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_share_group_snapshot_group_does_not_exist(self):
         share_group = fake_share_group(
@@ -837,6 +985,46 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_get.assert_called_once_with(
             self.context, share_group['id'])
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
+    def test_create_share_group_snapshot_failure_reserving_quota(self):
+        overs = ["share_group_snapshots"]
+        usages = {"share_group_snapshots": {
+            "reserved": 1,
+            "in_use": 3,
+            "limit": 4,
+        }}
+        quotas = {"share_group_snapshots": 5}
+        share_group = fake_share_group(
+            "fake_group_id", user_id=self.context.user_id,
+            project_id=self.context.project_id,
+            status=constants.STATUS_AVAILABLE)
+        self.mock_object(
+            db_driver, "share_group_get", mock.Mock(return_value=share_group))
+        self.mock_object(
+            db_driver, "share_get_all_by_share_group_id",
+            mock.Mock(return_value=[]))
+        share_group_api.QUOTAS.reserve.side_effect = exception.OverQuota(
+            overs=overs,
+            usages=usages,
+            quotas=quotas,
+        )
+        self.mock_object(share_group_api.LOG, "warning")
+
+        self.assertRaises(
+            exception.ShareGroupSnapshotsLimitExceeded,
+            self.api.create_share_group_snapshot,
+            self.context, share_group_id=share_group["id"])
+
+        db_driver.share_group_get.assert_called_once_with(
+            self.context, share_group["id"])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+        share_group_api.LOG.warning.assert_called_once_with(mock.ANY, mock.ANY)
 
     def test_create_share_group_snapshot_group_in_creating(self):
         self.mock_object(
@@ -851,6 +1039,9 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_get.assert_called_once_with(
             self.context, "fake_id")
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_share_group_snapshot_with_member(self):
         share_group = fake_share_group(
@@ -898,6 +1089,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_member_values)
         self.share_rpcapi.create_share_group_snapshot.assert_called_once_with(
             self.context, snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_share_group_snapshot_with_member_share_in_creating(self):
         share_group = fake_share_group(
@@ -919,6 +1115,9 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_get.assert_called_once_with(
             self.context, share_group['id'])
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_share_group_snapshot_with_two_members(self):
         share_group = fake_share_group(
@@ -979,6 +1178,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_member_2_values)
         self.share_rpcapi.create_share_group_snapshot.assert_called_once_with(
             self.context, snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_create_share_group_snapshot_error_creating_member(self):
         share_group = fake_share_group(
@@ -1031,6 +1235,11 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, expected_member_values)
         db_driver.share_group_snapshot_destroy.assert_called_once_with(
             self.context, snap['id'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=1)
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value)
 
     def test_delete_share_group_snapshot(self):
         share_group = fake_share_group('fake_id', host="fake_host")
@@ -1049,6 +1258,44 @@ class ShareGroupsAPITestCase(test.TestCase):
             self.context, sg_snap['id'], {'status': constants.STATUS_DELETING})
         self.share_rpcapi.delete_share_group_snapshot.assert_called_once_with(
             self.context, sg_snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=-1,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.commit.assert_called_once_with(
+            self.context, share_group_api.QUOTAS.reserve.return_value,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.rollback.assert_not_called()
+
+    def test_delete_share_group_snapshot_fail_on_quota_reserve(self):
+        share_group = fake_share_group('fake_id', host="fake_host")
+        sg_snap = fake_share_group_snapshot(
+            'fake_groupsnap_id', share_group_id='fake_id',
+            status=constants.STATUS_AVAILABLE)
+        self.mock_object(db_driver, 'share_group_get',
+                         mock.Mock(return_value=share_group))
+        self.mock_object(db_driver, 'share_group_snapshot_update')
+        share_group_api.QUOTAS.reserve.side_effect = exception.OverQuota(
+            'Failure')
+        self.mock_object(share_group_api.LOG, 'exception')
+
+        self.api.delete_share_group_snapshot(self.context, sg_snap)
+
+        db_driver.share_group_get.assert_called_once_with(
+            self.context, "fake_id")
+        db_driver.share_group_snapshot_update.assert_called_once_with(
+            self.context, sg_snap['id'], {'status': constants.STATUS_DELETING})
+        self.share_rpcapi.delete_share_group_snapshot.assert_called_once_with(
+            self.context, sg_snap, share_group['host'])
+        share_group_api.QUOTAS.reserve.assert_called_once_with(
+            self.context, share_group_snapshots=-1,
+            project_id=share_group['project_id'],
+            user_id=share_group['user_id'])
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
+        share_group_api.LOG.exception.assert_called_once_with(
+            mock.ANY, mock.ANY)
 
     def test_delete_share_group_snapshot_group_does_not_exist(self):
         snap = fake_share_group_snapshot(
@@ -1064,6 +1311,9 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_get.assert_called_once_with(
             self.context, "fake_id")
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     def test_delete_share_group_snapshot_creating_status(self):
         snap = fake_share_group_snapshot(
@@ -1077,6 +1327,9 @@ class ShareGroupsAPITestCase(test.TestCase):
 
         db_driver.share_group_get.assert_called_once_with(
             self.context, snap['share_group_id'])
+        share_group_api.QUOTAS.reserve.assert_not_called()
+        share_group_api.QUOTAS.commit.assert_not_called()
+        share_group_api.QUOTAS.rollback.assert_not_called()
 
     @ddt.data({}, {"name": "fake_name"})
     def test_update_share_group_snapshot_no_values(self, expected_values):
diff --git a/manila/tests/test_quota.py b/manila/tests/test_quota.py
index 4a68f493ad..f453c345b1 100644
--- a/manila/tests/test_quota.py
+++ b/manila/tests/test_quota.py
@@ -711,6 +711,6 @@ class QuotaEngineTestCase(test.TestCase):
 
     def test_current_common_resources(self):
         self.assertEqual(
-            ['gigabytes', 'share_networks', 'shares',
-             'snapshot_gigabytes', 'snapshots'],
+            ['gigabytes', 'share_group_snapshots', 'share_groups',
+             'share_networks', 'shares', 'snapshot_gigabytes', 'snapshots'],
             quota.QUOTAS.resources)
diff --git a/manila_tempest_tests/config.py b/manila_tempest_tests/config.py
index 8b27700d5d..9e4a16887b 100644
--- a/manila_tempest_tests/config.py
+++ b/manila_tempest_tests/config.py
@@ -30,7 +30,7 @@ ShareGroup = [
                help="The minimum api microversion is configured to be the "
                     "value of the minimum microversion supported by Manila."),
     cfg.StrOpt("max_api_microversion",
-               default="2.39",
+               default="2.40",
                help="The maximum api microversion is configured to be the "
                     "value of the latest microversion supported by Manila."),
     cfg.StrOpt("region",
diff --git a/manila_tempest_tests/services/share/v2/json/shares_client.py b/manila_tempest_tests/services/share/v2/json/shares_client.py
index 3631c68b41..94e1ac60de 100644
--- a/manila_tempest_tests/services/share/v2/json/shares_client.py
+++ b/manila_tempest_tests/services/share/v2/json/shares_client.py
@@ -909,7 +909,9 @@ class SharesV2Client(shares_client.SharesClient):
 
     def update_quotas(self, tenant_id, user_id=None, shares=None,
                       snapshots=None, gigabytes=None, snapshot_gigabytes=None,
-                      share_networks=None, force=True, share_type=None,
+                      share_networks=None,
+                      share_groups=None, share_group_snapshots=None,
+                      force=True, share_type=None,
                       url=None, version=LATEST_MICROVERSION):
         if url is None:
             url = self._get_quotas_url(version)
@@ -929,6 +931,10 @@ class SharesV2Client(shares_client.SharesClient):
             put_body["snapshot_gigabytes"] = snapshot_gigabytes
         if share_networks is not None:
             put_body["share_networks"] = share_networks
+        if share_groups is not None:
+            put_body["share_groups"] = share_groups
+        if share_group_snapshots is not None:
+            put_body["share_group_snapshots"] = share_group_snapshots
         put_body = json.dumps({"quota_set": put_body})
 
         resp, body = self.put(url, put_body, version=version)
diff --git a/manila_tempest_tests/tests/api/admin/test_quotas.py b/manila_tempest_tests/tests/api/admin/test_quotas.py
index 741a2bfb47..02d0a62d86 100644
--- a/manila_tempest_tests/tests/api/admin/test_quotas.py
+++ b/manila_tempest_tests/tests/api/admin/test_quotas.py
@@ -17,11 +17,15 @@ import ddt
 from tempest import config
 from tempest.lib.common.utils import data_utils
 from tempest.lib import exceptions as lib_exc
+import testtools
 from testtools import testcase as tc
 
 from manila_tempest_tests.tests.api import base
+from manila_tempest_tests import utils
 
 CONF = config.CONF
+PRE_SHARE_GROUPS_MICROVERSION = "2.39"
+SHARE_GROUPS_MICROVERSION = "2.40"
 
 
 @ddt.ddt
@@ -44,6 +48,9 @@ class SharesAdminQuotasTest(base.BaseSharesAdminTest):
         self.assertGreater(int(quotas["shares"]), -2)
         self.assertGreater(int(quotas["snapshots"]), -2)
         self.assertGreater(int(quotas["share_networks"]), -2)
+        if utils.is_microversion_supported(SHARE_GROUPS_MICROVERSION):
+            self.assertGreater(int(quotas["share_groups"]), -2)
+            self.assertGreater(int(quotas["share_group_snapshots"]), -2)
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_show_quotas(self):
@@ -53,6 +60,9 @@ class SharesAdminQuotasTest(base.BaseSharesAdminTest):
         self.assertGreater(int(quotas["shares"]), -2)
         self.assertGreater(int(quotas["snapshots"]), -2)
         self.assertGreater(int(quotas["share_networks"]), -2)
+        if utils.is_microversion_supported(SHARE_GROUPS_MICROVERSION):
+            self.assertGreater(int(quotas["share_groups"]), -2)
+            self.assertGreater(int(quotas["share_group_snapshots"]), -2)
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_show_quotas_for_user(self):
@@ -63,6 +73,28 @@ class SharesAdminQuotasTest(base.BaseSharesAdminTest):
         self.assertGreater(int(quotas["shares"]), -2)
         self.assertGreater(int(quotas["snapshots"]), -2)
         self.assertGreater(int(quotas["share_networks"]), -2)
+        if utils.is_microversion_supported(SHARE_GROUPS_MICROVERSION):
+            self.assertGreater(int(quotas["share_groups"]), -2)
+            self.assertGreater(int(quotas["share_group_snapshots"]), -2)
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API)
+    @base.skip_if_microversion_not_supported(PRE_SHARE_GROUPS_MICROVERSION)
+    def test_show_sg_quotas_using_too_old_microversion(self):
+        quotas = self.shares_v2_client.show_quotas(
+            self.tenant_id, version=PRE_SHARE_GROUPS_MICROVERSION)
+
+        for key in ('share_groups', 'share_group_snapshots'):
+            self.assertNotIn(key, quotas)
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API)
+    @base.skip_if_microversion_not_supported(PRE_SHARE_GROUPS_MICROVERSION)
+    def test_show_sg_quotas_for_user_using_too_old_microversion(self):
+        quotas = self.shares_v2_client.show_quotas(
+            self.tenant_id, self.user_id,
+            version=PRE_SHARE_GROUPS_MICROVERSION)
+
+        for key in ('share_groups', 'share_group_snapshots'):
+            self.assertNotIn(key, quotas)
 
     @ddt.data(
         ('id', True),
@@ -93,12 +125,16 @@ class SharesAdminQuotasTest(base.BaseSharesAdminTest):
         for key in ('shares', 'gigabytes', 'snapshots', 'snapshot_gigabytes'):
             self.assertEqual(st_quotas[key], p_quotas[key])
 
+        # Verify that we do not have share groups related quotas
+        # for share types.
+        for key in ('share_groups', 'share_group_snapshots'):
+            self.assertNotIn(key, st_quotas)
+
 
 @ddt.ddt
 class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
 
     force_tenant_isolation = True
-    client_version = '2'
 
     @classmethod
     def resource_setup(cls):
@@ -109,8 +145,7 @@ class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
 
     def setUp(self):
         super(self.__class__, self).setUp()
-        self.client = self.get_client_with_isolated_creds(
-            client_version=self.client_version)
+        self.client = self.get_client_with_isolated_creds(client_version='2')
         self.tenant_id = self.client.tenant_id
         self.user_id = self.client.user_id
 
@@ -124,6 +159,24 @@ class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
         updated = self.client.update_quotas(self.tenant_id, shares=new_quota)
         self.assertEqual(new_quota, int(updated["shares"]))
 
+    @ddt.data(
+        "share_groups",
+        "share_group_snapshots",
+    )
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API)
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_update_tenant_quota_share_groups(self, quota_key):
+        # Get current quotas
+        quotas = self.client.show_quotas(self.tenant_id)
+        new_quota = int(quotas[quota_key]) + 2
+
+        # Set new quota
+        updated = self.client.update_quotas(
+            self.tenant_id, **{quota_key: new_quota})
+        self.assertEqual(new_quota, int(updated[quota_key]))
+
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_update_user_quota_shares(self):
         # get current quotas
@@ -135,6 +188,24 @@ class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
             self.tenant_id, self.user_id, shares=new_quota)
         self.assertEqual(new_quota, int(updated["shares"]))
 
+    @ddt.data(
+        "share_groups",
+        "share_group_snapshots",
+    )
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API)
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_update_user_quota_share_groups(self, quota_key):
+        # Get current quotas
+        quotas = self.client.show_quotas(self.tenant_id, self.user_id)
+        new_quota = int(quotas[quota_key]) - 1
+
+        # Set new quota
+        updated = self.client.update_quotas(
+            self.tenant_id, self.user_id, **{quota_key: new_quota})
+        self.assertEqual(new_quota, int(updated[quota_key]))
+
     def _create_share_type(self):
         share_type = self.create_share_type(
             data_utils.rand_name("tempest-manila"),
@@ -280,44 +351,63 @@ class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_reset_tenant_quotas(self):
-        # get default_quotas
+        # Get default_quotas
         default = self.client.default_quotas(self.tenant_id)
 
-        # get current quotas
+        # Get current quotas
         custom = self.client.show_quotas(self.tenant_id)
 
-        # make quotas for update
-        shares = int(custom["shares"]) + 2
-        snapshots = int(custom["snapshots"]) + 2
-        gigabytes = int(custom["gigabytes"]) + 2
-        snapshot_gigabytes = int(custom["snapshot_gigabytes"]) + 2
-        share_networks = int(custom["share_networks"]) + 2
+        # Make quotas for update
+        data = {
+            "shares": int(custom["shares"]) + 2,
+            "snapshots": int(custom["snapshots"]) + 2,
+            "gigabytes": int(custom["gigabytes"]) + 2,
+            "snapshot_gigabytes": int(custom["snapshot_gigabytes"]) + 2,
+            "share_networks": int(custom["share_networks"]) + 2,
+        }
+        if (utils.is_microversion_supported(SHARE_GROUPS_MICROVERSION) and
+                CONF.share.run_share_group_tests):
+            data["share_groups"] = int(custom["share_groups"]) + 2
+            data["share_group_snapshots"] = (
+                int(custom["share_group_snapshots"]) + 2)
 
         # set new quota
-        updated = self.client.update_quotas(
-            self.tenant_id,
-            shares=shares,
-            snapshots=snapshots,
-            gigabytes=gigabytes,
-            snapshot_gigabytes=snapshot_gigabytes,
-            share_networks=share_networks)
-        self.assertEqual(shares, int(updated["shares"]))
-        self.assertEqual(snapshots, int(updated["snapshots"]))
-        self.assertEqual(gigabytes, int(updated["gigabytes"]))
-        self.assertEqual(snapshot_gigabytes,
-                         int(updated["snapshot_gigabytes"]))
-        self.assertEqual(share_networks, int(updated["share_networks"]))
+        updated = self.client.update_quotas(self.tenant_id, **data)
+        self.assertEqual(data["shares"], int(updated["shares"]))
+        self.assertEqual(data["snapshots"], int(updated["snapshots"]))
+        self.assertEqual(data["gigabytes"], int(updated["gigabytes"]))
+        self.assertEqual(
+            data["snapshot_gigabytes"], int(updated["snapshot_gigabytes"]))
+        self.assertEqual(
+            data["share_networks"], int(updated["share_networks"]))
+        if (utils.is_microversion_supported(SHARE_GROUPS_MICROVERSION) and
+                CONF.share.run_share_group_tests):
+            self.assertEqual(
+                data["share_groups"], int(updated["share_groups"]))
+            self.assertEqual(
+                data["share_group_snapshots"],
+                int(updated["share_group_snapshots"]))
 
-        # reset customized quotas
+        # Reset customized quotas
         self.client.reset_quotas(self.tenant_id)
 
-        # verify quotas
+        # Verify quotas
         reseted = self.client.show_quotas(self.tenant_id)
         self.assertEqual(int(default["shares"]), int(reseted["shares"]))
         self.assertEqual(int(default["snapshots"]), int(reseted["snapshots"]))
         self.assertEqual(int(default["gigabytes"]), int(reseted["gigabytes"]))
-        self.assertEqual(int(default["share_networks"]),
-                         int(reseted["share_networks"]))
+        self.assertEqual(
+            int(default["snapshot_gigabytes"]),
+            int(reseted["snapshot_gigabytes"]))
+        self.assertEqual(
+            int(default["share_networks"]), int(reseted["share_networks"]))
+        if (utils.is_microversion_supported(SHARE_GROUPS_MICROVERSION) and
+                CONF.share.run_share_group_tests):
+            self.assertEqual(
+                int(default["share_groups"]), int(reseted["share_groups"]))
+            self.assertEqual(
+                int(default["share_group_snapshots"]),
+                int(reseted["share_group_snapshots"]))
 
     @ddt.data(
         ('id', True),
@@ -450,6 +540,29 @@ class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
 
         self.assertEqual(-1, quotas.get('share_networks'))
 
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API)
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_unlimited_quota_for_share_groups(self):
+        self.client.update_quotas(self.tenant_id, share_groups=-1)
+
+        quotas = self.client.show_quotas(self.tenant_id)
+
+        self.assertEqual(-1, quotas.get('share_groups'))
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API)
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_unlimited_user_quota_for_share_group_snapshots(self):
+        self.client.update_quotas(
+            self.tenant_id, self.user_id, share_group_snapshots=-1)
+
+        quotas = self.client.show_quotas(self.tenant_id, self.user_id)
+
+        self.assertEqual(-1, quotas.get('share_group_snapshots'))
+
     @ddt.data(11, -1)
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_update_user_quotas_bigger_than_project_quota(self, user_quota):
@@ -541,3 +654,82 @@ class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
             for key in ('shares', 'gigabytes'):
                 self.assertEqual(0, quotas[key]['reserved'])
                 self.assertEqual(0, quotas[key]['in_use'])
+
+    def _check_sg_usages(self, quotas, in_use, limit):
+        """Helper method for 'test_share_group_quotas_usages' test."""
+        self.assertEqual(0, int(quotas['share_groups']['reserved']))
+        self.assertEqual(in_use, int(quotas['share_groups']['in_use']))
+        self.assertEqual(limit, int(quotas['share_groups']['limit']))
+
+    def _check_sgs_usages(self, quotas, in_use):
+        """Helper method for 'test_share_group_quotas_usages' test."""
+        self.assertEqual(0, int(quotas['share_group_snapshots']['reserved']))
+        self.assertEqual(
+            in_use, int(quotas['share_group_snapshots']['in_use']))
+        self.assertEqual(1, int(quotas['share_group_snapshots']['limit']))
+
+    def _check_usages(self, sg_in_use, sgs_in_use):
+        """Helper method for 'test_share_group_quotas_usages' test."""
+        p_quotas = self.client.detail_quotas(tenant_id=self.tenant_id)
+        u_quotas = self.client.detail_quotas(
+            tenant_id=self.tenant_id, user_id=self.user_id)
+        self._check_sg_usages(p_quotas, sg_in_use, 3)
+        self._check_sg_usages(u_quotas, sg_in_use, 2)
+        self._check_sgs_usages(p_quotas, sgs_in_use)
+        self._check_sgs_usages(u_quotas, sgs_in_use)
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @base.skip_if_microversion_lt(SHARE_GROUPS_MICROVERSION)
+    def test_share_group_quotas_usages(self):
+        # Set quotas for project (3 SG, 1 SGS) and user (2 SG, 1 SGS)
+        self.client.update_quotas(
+            self.tenant_id, share_groups=3, share_group_snapshots=1)
+        self.client.update_quotas(
+            self.tenant_id, user_id=self.user_id,
+            share_groups=2, share_group_snapshots=1)
+
+        # Check usages, they should be 0s
+        self._check_usages(0, 0)
+
+        # Create SG1 and check usages
+        share_group1 = self.create_share_group(
+            cleanup_in_class=False, client=self.client)
+        self._check_usages(1, 0)
+
+        # Create SGS1 and check usages
+        sg_snapshot = self.create_share_group_snapshot_wait_for_active(
+            share_group1['id'], cleanup_in_class=False, client=self.client)
+        self._check_usages(1, 1)
+
+        # Create SG2 from SGS1 and check usages
+        share_group2 = self.create_share_group(
+            cleanup_in_class=False, client=self.client,
+            source_share_group_snapshot_id=sg_snapshot['id'])
+        self._check_usages(2, 1)
+
+        # Try create SGS2, fail, then check usages
+        self.assertRaises(
+            lib_exc.OverLimit,
+            self.create_share_group,
+            client=self.client, cleanup_in_class=False)
+        self._check_usages(2, 1)
+
+        # Delete SG2 and check usages
+        self.client.delete_share_group(share_group2['id'])
+        self.client.wait_for_resource_deletion(
+            share_group_id=share_group2['id'])
+        self._check_usages(1, 1)
+
+        # Delete SGS1 and check usages
+        self.client.delete_share_group_snapshot(sg_snapshot['id'])
+        self.client.wait_for_resource_deletion(
+            share_group_snapshot_id=sg_snapshot['id'])
+        self._check_usages(1, 0)
+
+        # Delete SG1 and check usages
+        self.client.delete_share_group(share_group1['id'])
+        self.client.wait_for_resource_deletion(
+            share_group_id=share_group1['id'])
+        self._check_usages(0, 0)
diff --git a/manila_tempest_tests/tests/api/admin/test_quotas_negative.py b/manila_tempest_tests/tests/api/admin/test_quotas_negative.py
index fe4f562494..91b2507eb6 100644
--- a/manila_tempest_tests/tests/api/admin/test_quotas_negative.py
+++ b/manila_tempest_tests/tests/api/admin/test_quotas_negative.py
@@ -17,11 +17,15 @@ import ddt
 from tempest import config
 from tempest.lib.common.utils import data_utils
 from tempest.lib import exceptions as lib_exc
+import testtools
 from testtools import testcase as tc
 
 from manila_tempest_tests.tests.api import base
+from manila_tempest_tests import utils
 
 CONF = config.CONF
+PRE_SHARE_GROUPS_MICROVERSION = "2.39"
+SHARE_GROUPS_MICROVERSION = "2.40"
 
 
 @ddt.ddt
@@ -49,50 +53,35 @@ class SharesAdminQuotasNegativeTest(base.BaseSharesAdminTest):
         self.assertRaises(lib_exc.NotFound,
                           client.reset_quotas, "")
 
+    @ddt.data(
+        {"shares": -2},
+        {"snapshots": -2},
+        {"gigabytes": -2},
+        {"snapshot_gigabytes": -2},
+        {"share_networks": -2},
+    )
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
-    def test_update_shares_quota_with_wrong_data(self):
+    def test_update_quota_with_wrong_data(self, kwargs):
         # -1 is acceptable value as unlimited
         client = self.get_client_with_isolated_creds()
-        self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          shares=-2)
+        self.assertRaises(
+            lib_exc.BadRequest,
+            client.update_quotas, client.tenant_id, **kwargs)
 
+    @ddt.data(
+        {"share_groups": -2},
+        {"share_group_snapshots": -2},
+    )
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
-    def test_update_snapshots_quota_with_wrong_data(self):
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_update_sg_quota_with_wrong_data(self, kwargs):
         # -1 is acceptable value as unlimited
-        client = self.get_client_with_isolated_creds()
-        self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          snapshots=-2)
-
-    @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
-    def test_update_gigabytes_quota_with_wrong_data(self):
-        # -1 is acceptable value as unlimited
-        client = self.get_client_with_isolated_creds()
-        self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          gigabytes=-2)
-
-    @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
-    def test_update_snapshot_gigabytes_quota_with_wrong_data(self):
-        # -1 is acceptable value as unlimited
-        client = self.get_client_with_isolated_creds()
-        self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          snapshot_gigabytes=-2)
-
-    @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
-    def test_update_share_networks_quota_with_wrong_data(self):
-        # -1 is acceptable value as unlimited
-        client = self.get_client_with_isolated_creds()
-        self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          share_networks=-2)
+        client = self.get_client_with_isolated_creds(client_version='2')
+        self.assertRaises(
+            lib_exc.BadRequest,
+            client.update_quotas, client.tenant_id, **kwargs)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_create_share_with_size_bigger_than_quota(self):
@@ -105,6 +94,21 @@ class SharesAdminQuotasNegativeTest(base.BaseSharesAdminTest):
                           self.create_share,
                           size=overquota)
 
+    @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
+    @testtools.skipUnless(
+        CONF.share.run_share_group_tests, 'Share Group tests disabled.')
+    @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_create_share_group_with_exceeding_quota_limit(self):
+        client = self.get_client_with_isolated_creds(client_version='2')
+        client.update_quotas(client.tenant_id, share_groups=0)
+
+        # Try schedule share group creation
+        self.assertRaises(
+            lib_exc.OverLimit,
+            self.create_share_group,
+            client=client,
+            cleanup_in_class=False)
+
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_try_set_user_quota_shares_bigger_than_tenant_quota(self):
         client = self.get_client_with_isolated_creds()
@@ -267,6 +271,41 @@ class SharesAdminQuotasNegativeTest(base.BaseSharesAdminTest):
             share_networks=int(tenant_quotas["share_networks"]),
         )
 
+    @ddt.data('share_groups', 'share_group_snapshots')
+    @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
+    @base.skip_if_microversion_lt(SHARE_GROUPS_MICROVERSION)
+    def test_try_update_share_type_quota_for_share_groups(self, quota_name):
+        client = self.get_client_with_isolated_creds(client_version='2')
+        share_type = self._create_share_type()
+        tenant_quotas = client.show_quotas(client.tenant_id)
+
+        self.assertRaises(
+            lib_exc.BadRequest,
+            client.update_quotas,
+            client.tenant_id,
+            share_type=share_type["name"],
+            **{quota_name: int(tenant_quotas[quota_name])}
+        )
+
+    @ddt.data('share_groups', 'share_group_snapshots')
+    @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
+    @base.skip_if_microversion_not_supported(PRE_SHARE_GROUPS_MICROVERSION)
+    @base.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
+    def test_share_group_quotas_using_too_old_microversion(self, quota_key):
+        client = self.get_client_with_isolated_creds(client_version='2')
+        tenant_quotas = client.show_quotas(
+            client.tenant_id, version=SHARE_GROUPS_MICROVERSION)
+        kwargs = {
+            "version": PRE_SHARE_GROUPS_MICROVERSION,
+            quota_key: tenant_quotas[quota_key],
+        }
+
+        self.assertRaises(
+            lib_exc.BadRequest,
+            client.update_quotas,
+            client.tenant_id,
+            **kwargs)
+
     @ddt.data('show', 'reset', 'update')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.38")
diff --git a/releasenotes/notes/add-share-group-quotas-4e426907eed4c000.yaml b/releasenotes/notes/add-share-group-quotas-4e426907eed4c000.yaml
new file mode 100644
index 0000000000..a878beccac
--- /dev/null
+++ b/releasenotes/notes/add-share-group-quotas-4e426907eed4c000.yaml
@@ -0,0 +1,7 @@
+---
+features:
+  - Added quotas for amount of share groups and share group snapshots.
+upgrade:
+  - Two new config options are available for setting default quotas for share
+    groups and share group snapshots - 'quota_share_groups' and
+    'quota_share_group_snapshots'.