Merge "Import HTTPStatus instead of http_client (policy tests)"
This commit is contained in:
Zuul
Gerrit Code Review
commit
4f0ce33d0a
|
|
@ -13,7 +13,7 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from http import client as http_client
|
||||
from http import HTTPStatus
|
||||
from unittest import mock
|
||||
|
||||
from cinder.api import microversions as mv
|
||||
|
|
@ -69,7 +69,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_project_admin_can_set_default(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -84,7 +84,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_project_admin_cannot_set_default_for_other_project(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -99,7 +99,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(db, 'project_default_volume_type_get',
|
||||
return_value=FakeDefaultType())
|
||||
|
|
@ -112,7 +112,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_project_admin_can_get_default(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -133,7 +133,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_project_admin_cannot_get_default_for_other_project(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -144,7 +144,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_system_admin_can_get_all_default(self):
|
||||
system_admin_context = self.system_admin_context
|
||||
|
|
@ -155,7 +155,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_project_admin_cannot_get_all_default(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -166,7 +166,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_system_admin_can_unset_default(self):
|
||||
system_admin_context = self.system_admin_context
|
||||
|
|
@ -177,7 +177,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.NO_CONTENT, response.status_int)
|
||||
self.assertEqual(HTTPStatus.NO_CONTENT, response.status_int)
|
||||
|
||||
def test_project_admin_can_unset_default(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -188,7 +188,7 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.NO_CONTENT, response.status_int)
|
||||
self.assertEqual(HTTPStatus.NO_CONTENT, response.status_int)
|
||||
|
||||
def test_project_admin_cannot_unset_default_for_other_project(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -199,4 +199,4 @@ class DefaultVolumeTypesPolicyTests(test_base.CinderPolicyTests):
|
|||
microversion=
|
||||
mv.DEFAULT_TYPE_OVERRIDES)
|
||||
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from http import client as http_client
|
||||
from http import HTTPStatus
|
||||
from unittest import mock
|
||||
|
||||
from cinder.tests.unit import fake_constants
|
||||
|
|
@ -36,7 +36,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_nonadmin_user_can_create_volume(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -48,7 +48,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_admin_can_create_volume_from_image(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -60,7 +60,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_nonadmin_user_can_create_volume_from_image(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -72,7 +72,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get_volume')
|
||||
def test_admin_can_show_volumes(self, mock_volume):
|
||||
|
|
@ -87,7 +87,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(admin_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
self.assertEqual(response.json_body['volume']['id'], volume.id)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get_volume')
|
||||
|
|
@ -103,7 +103,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(user_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
self.assertEqual(response.json_body['volume']['id'], volume.id)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get_volume')
|
||||
|
|
@ -124,7 +124,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
# volume, because they didn't create it and it lives in a different
|
||||
# project. Does cinder return a 404 in cases like this? Or is a 403
|
||||
# expected?
|
||||
self.assertEqual(http_client.NOT_FOUND, response.status_int)
|
||||
self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
|
||||
|
||||
def test_admin_can_get_all_volumes_detail(self):
|
||||
# Make sure administrators are authorized to list volumes
|
||||
|
|
@ -137,7 +137,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(admin_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_vol = response.json_body['volumes'][0]
|
||||
|
||||
self.assertEqual(volume.id, res_vol['id'])
|
||||
|
|
@ -153,7 +153,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(user_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_vol = response.json_body['volumes'][0]
|
||||
|
||||
self.assertEqual(volume.id, res_vol['id'])
|
||||
|
|
@ -171,7 +171,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"volume": {"name": "update_name"}}
|
||||
response = self._get_request_response(admin_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_can_update_volumes(self, mock_volume):
|
||||
|
|
@ -186,7 +186,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"volume": {"name": "update_name"}}
|
||||
response = self._get_request_response(user_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_update_volumes_for_others(self, mock_volume):
|
||||
|
|
@ -203,7 +203,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"volume": {"name": "update_name"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_can_delete_volumes(self, mock_volume):
|
||||
|
|
@ -216,7 +216,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(user_context, path, 'DELETE')
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_admin_can_delete_volumes(self, mock_volume):
|
||||
|
|
@ -229,7 +229,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(admin_context, path, 'DELETE')
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_delete_volumes_for_others(self, mock_volume):
|
||||
|
|
@ -245,7 +245,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(non_owner_context, path,
|
||||
'DELETE')
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get_volume')
|
||||
def test_admin_can_show_tenant_id_in_volume(self, mock_volume):
|
||||
|
|
@ -260,7 +260,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(admin_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_vol = response.json_body['volume']
|
||||
self.assertEqual(admin_context.project_id,
|
||||
res_vol['os-vol-tenant-attr:tenant_id'])
|
||||
|
|
@ -278,7 +278,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(user_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_vol = response.json_body['volume']
|
||||
self.assertEqual(user_context.project_id,
|
||||
res_vol['os-vol-tenant-attr:tenant_id'])
|
||||
|
|
@ -294,7 +294,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(admin_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_vol = response.json_body['volumes'][0]
|
||||
# Make sure owners are authorized to show tenant_id
|
||||
self.assertEqual(admin_context.project_id,
|
||||
|
|
@ -311,7 +311,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
|
||||
response = self._get_request_response(user_context, path, 'GET')
|
||||
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_vol = response.json_body['volumes'][0]
|
||||
# Make sure owners are authorized to show tenant_id
|
||||
self.assertEqual(user_context.project_id,
|
||||
|
|
@ -328,7 +328,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k1": "v1"}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_admin_can_get_metadata(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -339,7 +339,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(admin_context, path, 'GET')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v', res_meta['k'])
|
||||
|
|
@ -355,7 +355,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k": "v2"}}
|
||||
response = self._get_request_response(admin_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v2', res_meta['k'])
|
||||
|
|
@ -370,7 +370,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
'key': 'k'
|
||||
}
|
||||
response = self._get_request_response(admin_context, path, 'DELETE')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_owner_can_create_metadata(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -383,7 +383,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k1": "v1"}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_owner_can_get_metadata(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -394,7 +394,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(user_context, path, 'GET')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v', res_meta['k'])
|
||||
|
|
@ -410,7 +410,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k": "v2"}}
|
||||
response = self._get_request_response(user_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v2', res_meta['k'])
|
||||
|
|
@ -425,7 +425,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
'key': 'k'
|
||||
}
|
||||
response = self._get_request_response(user_context, path, 'DELETE')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_create_metadata_for_others(self, mock_volume):
|
||||
|
|
@ -441,7 +441,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k1": "v1"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_get_metadata_for_others(self, mock_volume):
|
||||
|
|
@ -455,7 +455,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(non_owner_context, path, 'GET')
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_update_metadata_for_others(self, mock_volume):
|
||||
|
|
@ -471,7 +471,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k": "v2"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_delete_metadata_for_others(self, mock_volume):
|
||||
|
|
@ -487,4 +487,4 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
response = self._get_request_response(non_owner_context, path,
|
||||
'DELETE')
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from http import client as http_client
|
||||
from http import HTTPStatus
|
||||
from unittest import mock
|
||||
|
||||
from cinder.api import microversions as mv
|
||||
|
|
@ -40,7 +40,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-extend": {"new_size": "2"}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_owner_can_extend_volume(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -53,7 +53,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-extend": {"new_size": "2"}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_extend_volume_for_others(self, mock_volume):
|
||||
|
|
@ -69,7 +69,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-extend": {"new_size": "2"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_extend_attached_volume(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -83,7 +83,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(
|
||||
admin_context, path, 'POST', body=body,
|
||||
microversion=mv.VOLUME_EXTEND_INUSE)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_owner_can_extend_attached_volume(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -97,7 +97,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(
|
||||
user_context, path, 'POST', body=body,
|
||||
microversion=mv.VOLUME_EXTEND_INUSE)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_extend_attached_volume_for_others(self, mock_volume):
|
||||
|
|
@ -114,7 +114,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(
|
||||
non_owner_context, path, 'POST', body=body,
|
||||
microversion=mv.VOLUME_EXTEND_INUSE)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_retype_volume(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -129,7 +129,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"migration_policy": "never"}}
|
||||
response = self._get_request_response(
|
||||
admin_context, path, 'POST', body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_owner_can_retype_volume(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -144,7 +144,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"migration_policy": "never"}}
|
||||
response = self._get_request_response(
|
||||
user_context, path, 'POST', body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_retype_volume_for_others(self, mock_volume):
|
||||
|
|
@ -162,7 +162,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"migration_policy": "never"}}
|
||||
response = self._get_request_response(
|
||||
non_owner_context, path, 'POST', body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_update_readonly(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -177,7 +177,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-update_readonly_flag": {"readonly": "True"}}
|
||||
response = self._get_request_response(
|
||||
admin_context, path, 'POST', body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_owner_can_update_readonly(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -192,7 +192,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-update_readonly_flag": {"readonly": "True"}}
|
||||
response = self._get_request_response(
|
||||
user_context, path, 'POST', body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_update_readonly_for_others(self, mock_volume):
|
||||
|
|
@ -209,7 +209,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-update_readonly_flag": {"readonly": "True"}}
|
||||
response = self._get_request_response(
|
||||
non_owner_context, path, 'POST', body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get_volume')
|
||||
def test_admin_can_force_delete_volumes(self, mock_volume):
|
||||
|
|
@ -225,7 +225,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get_volume')
|
||||
def test_nonadmin_cannot_force_delete_volumes(self, mock_volume):
|
||||
|
|
@ -241,7 +241,7 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI, 'attach_volume')
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI, 'detach_volume')
|
||||
|
|
@ -257,12 +257,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"mountpoint": "/dev/vdc"}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
body = {"os-detach": {}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI, 'attach_volume')
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI, 'detach_volume')
|
||||
|
|
@ -278,12 +278,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"mountpoint": "/dev/vdc"}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
body = {"os-detach": {}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI, 'attach_volume')
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI, 'detach_volume')
|
||||
|
|
@ -304,12 +304,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"mountpoint": "/dev/vdc"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
body = {"os-detach": {}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_reserve_unreserve_volume(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -322,12 +322,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-reserve": {}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
body = {"os-unreserve": {}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_owner_can_reserve_unreserve_volume(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -340,12 +340,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-reserve": {}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
body = {"os-unreserve": {}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_reserve_unreserve_volume_for_others(self,
|
||||
|
|
@ -363,12 +363,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
"mountpoint": "/dev/vdc"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
body = {"os-detach": {}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI,
|
||||
'initialize_connection')
|
||||
|
|
@ -385,12 +385,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-initialize_connection": {'connector': {}}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
body = {"os-terminate_connection": {'connector': {}}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI,
|
||||
'initialize_connection')
|
||||
|
|
@ -407,12 +407,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-initialize_connection": {'connector': {}}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
body = {"os-terminate_connection": {'connector': {}}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.volume_rpcapi.VolumeAPI,
|
||||
'initialize_connection')
|
||||
|
|
@ -435,12 +435,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-initialize_connection": {'connector': {}}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
body = {"os-terminate_connection": {'connector': {}}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_begin_roll_detaching(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -454,12 +454,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-begin_detaching": {}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
body = {"os-roll_detaching": {}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
def test_owner_can_begin_roll_detaching(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -473,12 +473,12 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-begin_detaching": {}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
body = {"os-roll_detaching": {}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.ACCEPTED, response.status_int)
|
||||
self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_begin_roll_detaching_for_others(self, mock_volume):
|
||||
|
|
@ -495,9 +495,9 @@ class VolumeProtectionTests(test_base.CinderPolicyTests):
|
|||
body = {"os-begin_detaching": {}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
body = {"os-roll_detaching": {}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from http import client as http_client
|
||||
from http import HTTPStatus
|
||||
from unittest import mock
|
||||
|
||||
from cinder.tests.unit.policies import test_base
|
||||
|
|
@ -31,7 +31,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(admin_context, path, 'GET')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v', res_meta['k'])
|
||||
|
|
@ -45,7 +45,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(user_context, path, 'GET')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v', res_meta['k'])
|
||||
|
|
@ -62,7 +62,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
|
||||
response = self._get_request_response(non_owner_context, path, 'GET')
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_create_metadata(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -75,7 +75,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k1": "v1"}}
|
||||
response = self._get_request_response(admin_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_owner_can_create_metadata(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -88,7 +88,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k1": "v1"}}
|
||||
response = self._get_request_response(user_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_create_metadata_for_others(self, mock_volume):
|
||||
|
|
@ -104,7 +104,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k1": "v1"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'POST',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_delete_metadata(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -116,7 +116,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
'key': 'k'
|
||||
}
|
||||
response = self._get_request_response(admin_context, path, 'DELETE')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
def test_owner_can_delete_metadata(self):
|
||||
user_context = self.user_context
|
||||
|
|
@ -128,7 +128,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
'key': 'k'
|
||||
}
|
||||
response = self._get_request_response(user_context, path, 'DELETE')
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
|
||||
@mock.patch.object(volume_api.API, 'get')
|
||||
def test_owner_cannot_delete_metadata_for_others(self, mock_volume):
|
||||
|
|
@ -144,7 +144,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
}
|
||||
response = self._get_request_response(non_owner_context, path,
|
||||
'DELETE')
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
||||
def test_admin_can_update_metadata(self):
|
||||
admin_context = self.admin_context
|
||||
|
|
@ -157,7 +157,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k": "v2"}}
|
||||
response = self._get_request_response(admin_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v2', res_meta['k'])
|
||||
|
|
@ -173,7 +173,7 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k": "v2"}}
|
||||
response = self._get_request_response(user_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.OK, response.status_int)
|
||||
self.assertEqual(HTTPStatus.OK, response.status_int)
|
||||
res_meta = response.json_body['metadata']
|
||||
self.assertIn('k', res_meta)
|
||||
self.assertEqual('v2', res_meta['k'])
|
||||
|
|
@ -192,4 +192,4 @@ class VolumePolicyTests(test_base.CinderPolicyTests):
|
|||
body = {"metadata": {"k": "v2"}}
|
||||
response = self._get_request_response(non_owner_context, path, 'PUT',
|
||||
body=body)
|
||||
self.assertEqual(http_client.FORBIDDEN, response.status_int)
|
||||
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
|
||||
|
|
|
|||
Loading…
Reference in New Issue