Merge "Move policy enforcement into REST API layer for v2.1 api assisted_volume_snapshots"

nova/api/openstack/compute/plugins/v3/assisted_volume_snapshots.py

@@ -32,15 +32,14 @@ from nova.i18n import _LI
 
 LOG = logging.getLogger(__name__)
 ALIAS = 'os-assisted-volume-snapshots'
-authorize = extensions.extension_authorizer('compute',
-                                            'v3:' + ALIAS)
+authorize = extensions.os_compute_authorizer(ALIAS)
 
 
 class AssistedVolumeSnapshotsController(wsgi.Controller):
     """The Assisted volume snapshots API controller for the OpenStack API."""
 
     def __init__(self):
-        self.compute_api = compute.API()
+        self.compute_api = compute.API(skip_policy_check=True)
         super(AssistedVolumeSnapshotsController, self).__init__()
 
     @extensions.expected_errors(400)

nova/tests/unit/api/openstack/compute/contrib/test_volumes.py

@@ -821,3 +821,38 @@ class AssistedSnapshotDeleteTestCaseV2(AssistedSnapshotDeleteTestCaseV21):
 
     def _check_status(self, expected_status, res, controller_method):
         self.assertEqual(expected_status, res.status_int)
+
+
+class TestAssistedVolumeSnapshotsPolicyEnforcementV21(test.NoDBTestCase):
+
+    def setUp(self):
+        super(TestAssistedVolumeSnapshotsPolicyEnforcementV21, self).setUp()
+        self.controller = (
+            assisted_snaps_v21.AssistedVolumeSnapshotsController())
+        self.req = fakes.HTTPRequest.blank('')
+
+    def test_create_assisted_volumes_snapshots_policy_failed(self):
+        rule_name = "compute_extension:v3:os-assisted-volume-snapshots:create"
+        self.policy.set_rules({rule_name: "project:non_fake"})
+        body = {'snapshot':
+                   {'volume_id': '1',
+                    'create_info': {'type': 'qcow2',
+                                    'new_file': 'new_file',
+                                    'snapshot_id': 'snapshot_id'}}}
+        exc = self.assertRaises(
+            exception.PolicyNotAuthorized,
+            self.controller.create, self.req, body=body)
+        self.assertEqual(
+            "Policy doesn't allow %s to be performed." % rule_name,
+            exc.format_message())
+
+    def test_delete_assisted_volumes_snapshots_policy_failed(self):
+        rule_name = "compute_extension:v3:os-assisted-volume-snapshots:delete"
+        self.policy.set_rules({rule_name: "project:non_fake"})
+        exc = self.assertRaises(
+            exception.PolicyNotAuthorized,
+            self.controller.delete, self.req, '5')
+
+        self.assertEqual(
+            "Policy doesn't allow %s to be performed." % rule_name,
+            exc.format_message())