Merge "Add policy checks to Volume.API"

nova/tests/policy.json

@@ -66,5 +66,27 @@
     "compute:delete": [],
     "compute:soft_delete": [],
     "compute:force_delete": [],
-    "compute:restore": []
+    "compute:restore": [],
+
+
+    "volume:create": [],
+    "volume:get": [],
+    "volume:get_all": [],
+    "volume:get_volume_metadata": [],
+    "volume:delete": [],
+    "volume:update": [],
+    "volume:delete_volume_metadata": [],
+    "volume:update_volume_metadata": [],
+
+    "volume:attach": [],
+    "volume:detach": [],
+    "volume:check_attach": [],
+    "volume:check_detach": [],
+    "volume:initialize_connection": [],
+    "volume:terminate_connection": [],
+
+    "volume:create_snapshot": [],
+    "volume:delete_snapshot": [],
+    "volume:get_snapshot": [],
+    "volume:get_all_snapshots": []
 }

nova/tests/test_volume.py

@@ -27,6 +27,7 @@ from nova import exception
 from nova import db
 from nova import flags
 from nova import log as logging
+import nova.policy
 from nova import rpc
 from nova import test
 from nova import utils
@@ -399,3 +400,47 @@ class ISCSITestCase(DriverTestCase):
         self.mox.UnsetStubs()
 
         self._detach_volume(volume_id_list)
+
+
+class VolumePolicyTestCase(test.TestCase):
+
+    def setUp(self):
+        super(VolumePolicyTestCase, self).setUp()
+
+        nova.policy.reset()
+        nova.policy.init()
+
+        self.context = context.get_admin_context()
+        self.volume_api = nova.volume.api.API()
+
+    def tearDown(self):
+        super(VolumePolicyTestCase, self).tearDown()
+        nova.policy.reset()
+
+    def _set_rules(self, rules):
+        nova.common.policy.set_brain(nova.common.policy.HttpBrain(rules))
+
+    def test_check_policy(self):
+        self.mox.StubOutWithMock(nova.policy, 'enforce')
+        target = {
+            'project_id': self.context.project_id,
+            'user_id': self.context.user_id,
+        }
+        nova.policy.enforce(self.context, 'volume:attach', target)
+        self.mox.ReplayAll()
+        nova.volume.api.check_policy(self.context, 'attach')
+        self.mox.UnsetStubs()
+        self.mox.VerifyAll()
+
+    def test_check_policy_with_target(self):
+        self.mox.StubOutWithMock(nova.policy, 'enforce')
+        target = {
+            'project_id': self.context.project_id,
+            'user_id': self.context.user_id,
+            'id': 2,
+        }
+        nova.policy.enforce(self.context, 'volume:attach', target)
+        self.mox.ReplayAll()
+        nova.volume.api.check_policy(self.context, 'attach', {'id': 2})
+        self.mox.UnsetStubs()
+        self.mox.VerifyAll()