openstack
/
watcher
Code Issues Proposed changes

use HTTPStatus instead of direct code 

Python introduced http.HTTPStatus since version 3.5,
and Wallaby has targeted a minimum version of python 3.6.

Change-Id: I45f732f0f59b8fae831bb6c07f4fdd98cdd7409a
This commit is contained in:
sue 2021-03-25 15:49:10 +08:00 committed by Dantali0n
parent 546b730c9b
commit c28756c48b
25 changed files with 215 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
watcher/api/controllers/v1/action.py
View File

@ -57,6 +57,7 @@ are dynamically loaded by Watcher at launch time.
import datetime import datetime
from http import HTTPStatus
import pecan import pecan
from pecan import rest from pecan import rest
import wsme import wsme
@ -362,7 +363,7 @@ class ActionsController(rest.RestController):
return Action.convert_with_links(action) return Action.convert_with_links(action)
@wsme_pecan.wsexpose(Action, body=Action, status_code=201) @wsme_pecan.wsexpose(Action, body=Action, status_code=HTTPStatus.CREATED)
def post(self, action): def post(self, action):
"""Create a new action(forbidden). """Create a new action(forbidden).
@ -422,7 +423,7 @@ class ActionsController(rest.RestController):
action_to_update.save() action_to_update.save()
return Action.convert_with_links(action_to_update) return Action.convert_with_links(action_to_update)
@wsme_pecan.wsexpose(None, types.uuid, status_code=204) @wsme_pecan.wsexpose(None, types.uuid, status_code=HTTPStatus.NO_CONTENT)
def delete(self, action_uuid): def delete(self, action_uuid):
"""Delete a action(forbidden). """Delete a action(forbidden).

3
watcher/api/controllers/v1/action_plan.py
View File

@ -56,6 +56,7 @@ state machine <action_plan_state_machine>`.
import datetime import datetime
from http import HTTPStatus
from oslo_log import log from oslo_log import log
import pecan import pecan
from pecan import rest from pecan import rest
@ -460,7 +461,7 @@ class ActionPlansController(rest.RestController):
return ActionPlan.convert_with_links(action_plan) return ActionPlan.convert_with_links(action_plan)
@wsme_pecan.wsexpose(None, types.uuid, status_code=204) @wsme_pecan.wsexpose(None, types.uuid, status_code=HTTPStatus.NO_CONTENT)
def delete(self, action_plan_uuid): def delete(self, action_plan_uuid):
"""Delete an action plan. """Delete an action plan.

6
watcher/api/controllers/v1/audit.py
View File

@ -32,6 +32,7 @@ states, visit :ref:`the Audit State machine <audit_state_machine>`.
import datetime import datetime
from dateutil import tz from dateutil import tz
from http import HTTPStatus
import pecan import pecan
from pecan import rest from pecan import rest
import wsme import wsme
@ -595,7 +596,8 @@ class AuditsController(rest.RestController):
return Audit.convert_with_links(rpc_audit) return Audit.convert_with_links(rpc_audit)
@wsme_pecan.wsexpose(Audit, body=AuditPostType, status_code=201) @wsme_pecan.wsexpose(Audit, body=AuditPostType,
status_code=HTTPStatus.CREATED)
def post(self, audit_p): def post(self, audit_p):
"""Create a new audit. """Create a new audit.
@ -717,7 +719,7 @@ class AuditsController(rest.RestController):
audit_to_update.save() audit_to_update.save()
return Audit.convert_with_links(audit_to_update) return Audit.convert_with_links(audit_to_update)
@wsme_pecan.wsexpose(None, wtypes.text, status_code=204) @wsme_pecan.wsexpose(None, wtypes.text, status_code=HTTPStatus.NO_CONTENT)
def delete(self, audit): def delete(self, audit):
"""Delete an audit. """Delete an audit.

5
watcher/api/controllers/v1/audit_template.py
View File

@ -45,6 +45,7 @@ will be launched automatically or will need a manual confirmation from the
import datetime import datetime
from http import HTTPStatus
import pecan import pecan
from pecan import rest from pecan import rest
import wsme import wsme
@ -618,7 +619,7 @@ class AuditTemplatesController(rest.RestController):
@wsme.validate(types.uuid, AuditTemplatePostType) @wsme.validate(types.uuid, AuditTemplatePostType)
@wsme_pecan.wsexpose(AuditTemplate, body=AuditTemplatePostType, @wsme_pecan.wsexpose(AuditTemplate, body=AuditTemplatePostType,
status_code=201) status_code=HTTPStatus.CREATED)
def post(self, audit_template_postdata): def post(self, audit_template_postdata):
"""Create a new audit template. """Create a new audit template.
@ -694,7 +695,7 @@ class AuditTemplatesController(rest.RestController):
audit_template_to_update.save() audit_template_to_update.save()
return AuditTemplate.convert_with_links(audit_template_to_update) return AuditTemplate.convert_with_links(audit_template_to_update)
@wsme_pecan.wsexpose(None, wtypes.text, status_code=204) @wsme_pecan.wsexpose(None, wtypes.text, status_code=HTTPStatus.NO_CONTENT)
def delete(self, audit_template): def delete(self, audit_template):
"""Delete a audit template. """Delete a audit template.

3
watcher/api/controllers/v1/webhooks.py
View File

@ -14,6 +14,7 @@
Webhook endpoint for Watcher v1 REST API. Webhook endpoint for Watcher v1 REST API.
""" """
from http import HTTPStatus
from oslo_log import log from oslo_log import log
import pecan import pecan
from pecan import rest from pecan import rest
@ -36,7 +37,7 @@ class WebhookController(rest.RestController):
self.dc_client = rpcapi.DecisionEngineAPI() self.dc_client = rpcapi.DecisionEngineAPI()
@wsme_pecan.wsexpose(None, wtypes.text, body=types.jsontype, @wsme_pecan.wsexpose(None, wtypes.text, body=types.jsontype,
status_code=202) status_code=HTTPStatus.ACCEPTED)
def post(self, audit_ident, body): def post(self, audit_ident, body):
"""Trigger the given audit. """Trigger the given audit.

6
watcher/api/hooks.py
View File

@ -15,7 +15,7 @@
# under the License. # under the License.
from http import client as http_client from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from pecan import hooks from pecan import hooks
@ -91,8 +91,8 @@ class NoExceptionTracebackHook(hooks.PecanHook):
# Do nothing if there is no error. # Do nothing if there is no error.
# Status codes in the range 200 (OK) to 399 (400 = BAD_REQUEST) are not # Status codes in the range 200 (OK) to 399 (400 = BAD_REQUEST) are not
# an error. # an error.
if (http_client.OK <= state.response.status_int < if (HTTPStatus.OK <= state.response.status_int <
http_client.BAD_REQUEST): HTTPStatus.BAD_REQUEST):
return return
json_body = state.response.json json_body = state.response.json

13
watcher/common/exception.py
View File

@ -25,6 +25,7 @@ SHOULD include dedicated exception logging.
import functools import functools
import sys import sys
from http import HTTPStatus
from keystoneclient import exceptions as keystone_exceptions from keystoneclient import exceptions as keystone_exceptions
from oslo_config import cfg from oslo_config import cfg
from oslo_log import log from oslo_log import log
@ -62,7 +63,7 @@ class WatcherException(Exception):
""" """
msg_fmt = _("An unknown exception occurred") msg_fmt = _("An unknown exception occurred")
code = 500 code = HTTPStatus.INTERNAL_SERVER_ERROR
headers = {} headers = {}
safe = False safe = False
@ -114,12 +115,12 @@ class UnsupportedError(WatcherException):
class NotAuthorized(WatcherException): class NotAuthorized(WatcherException):
msg_fmt = _("Not authorized") msg_fmt = _("Not authorized")
code = 403 code = HTTPStatus.FORBIDDEN
class NotAcceptable(WatcherException): class NotAcceptable(WatcherException):
msg_fmt = _("Request not acceptable.") msg_fmt = _("Request not acceptable.")
code = 406 code = HTTPStatus.NOT_ACCEPTABLE
class PolicyNotAuthorized(NotAuthorized): class PolicyNotAuthorized(NotAuthorized):
@ -132,7 +133,7 @@ class OperationNotPermitted(NotAuthorized):
class Invalid(WatcherException, ValueError): class Invalid(WatcherException, ValueError):
msg_fmt = _("Unacceptable parameters") msg_fmt = _("Unacceptable parameters")
code = 400 code = HTTPStatus.BAD_REQUEST
class ObjectNotFound(WatcherException): class ObjectNotFound(WatcherException):
@ -141,12 +142,12 @@ class ObjectNotFound(WatcherException):
class Conflict(WatcherException): class Conflict(WatcherException):
msg_fmt = _('Conflict') msg_fmt = _('Conflict')
code = 409 code = HTTPStatus.CONFLICT
class ResourceNotFound(ObjectNotFound): class ResourceNotFound(ObjectNotFound):
msg_fmt = _("The %(name)s resource %(id)s could not be found") msg_fmt = _("The %(name)s resource %(id)s could not be found")
code = 404 code = HTTPStatus.NOT_FOUND
class InvalidParameter(Invalid): class InvalidParameter(Invalid):

13
watcher/common/placement_helper.py
View File

@ -11,6 +11,7 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_log import log as logging from oslo_log import log as logging
@ -53,7 +54,7 @@ class PlacementHelper(object):
if rp_name: if rp_name:
url += '?name=%s' % rp_name url += '?name=%s' % rp_name
resp = self.get(url) resp = self.get(url)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
json_resp = resp.json() json_resp = resp.json()
return json_resp['resource_providers'] return json_resp['resource_providers']
@ -77,7 +78,7 @@ class PlacementHelper(object):
""" """
url = '/resource_providers/%s/inventories' % rp_uuid url = '/resource_providers/%s/inventories' % rp_uuid
resp = self.get(url) resp = self.get(url)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
json = resp.json() json = resp.json()
return json['inventories'] return json['inventories']
msg = ("Failed to get resource provider %(rp_uuid)s inventories. " msg = ("Failed to get resource provider %(rp_uuid)s inventories. "
@ -97,7 +98,7 @@ class PlacementHelper(object):
""" """
resp = self.get("/resource_providers/%s/traits" % rp_uuid) resp = self.get("/resource_providers/%s/traits" % rp_uuid)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
json = resp.json() json = resp.json()
return json['traits'] return json['traits']
msg = ("Failed to get resource provider %(rp_uuid)s traits. " msg = ("Failed to get resource provider %(rp_uuid)s traits. "
@ -118,7 +119,7 @@ class PlacementHelper(object):
""" """
url = '/allocations/%s' % consumer_uuid url = '/allocations/%s' % consumer_uuid
resp = self.get(url) resp = self.get(url)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
json = resp.json() json = resp.json()
return json['allocations'] return json['allocations']
msg = ("Failed to get allocations for consumer %(c_uuid). " msg = ("Failed to get allocations for consumer %(c_uuid). "
@ -139,7 +140,7 @@ class PlacementHelper(object):
""" """
url = '/resource_providers/%s/usages' % rp_uuid url = '/resource_providers/%s/usages' % rp_uuid
resp = self.get(url) resp = self.get(url)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
json = resp.json() json = resp.json()
return json['usages'] return json['usages']
msg = ("Failed to get resource provider %(rp_uuid)s usages. " msg = ("Failed to get resource provider %(rp_uuid)s usages. "
@ -164,7 +165,7 @@ class PlacementHelper(object):
""" """
url = "/allocation_candidates?%s" % resources url = "/allocation_candidates?%s" % resources
resp = self.get(url) resp = self.get(url)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
data = resp.json() data = resp.json()
return data['provider_summaries'] return data['provider_summaries']

7
watcher/decision_engine/datasources/grafana.py
View File

@ -18,6 +18,7 @@
from urllib import parse as urlparse from urllib import parse as urlparse
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_log import log from oslo_log import log
@ -138,11 +139,11 @@ class GrafanaHelper(base.DataSourceBase):
resp = requests.get(self._base_url + str(project_id) + '/query', resp = requests.get(self._base_url + str(project_id) + '/query',
params=params, headers=self._headers) params=params, headers=self._headers)
if resp.status_code == 200: if resp.status_code == HTTPStatus.OK:
return resp return resp
elif resp.status_code == 400: elif resp.status_code == HTTPStatus.BAD_REQUEST:
LOG.error("Query for metric is invalid") LOG.error("Query for metric is invalid")
elif resp.status_code == 401: elif resp.status_code == HTTPStatus.UNAUTHORIZED:
LOG.error("Authorization token is invalid") LOG.error("Authorization token is invalid")
raise exception.DataSourceNotAvailable(self.NAME) raise exception.DataSourceNotAvailable(self.NAME)

4
watcher/tests/api/test_base.py
View File

@ -13,6 +13,8 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from http import HTTPStatus
from watcher.tests.api import base from watcher.tests.api import base
@ -25,6 +27,6 @@ class TestBase(base.FunctionalTest):
response = self.get_json('/bad/path', response = self.get_json('/bad/path',
expect_errors=True, expect_errors=True,
headers={"Accept": "application/json"}) headers={"Accept": "application/json"})
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual("application/json", response.content_type) self.assertEqual("application/json", response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])

15
watcher/tests/api/v1/test_actions.py
View File

@ -14,6 +14,7 @@ import datetime
import itertools import itertools
from unittest import mock from unittest import mock
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from wsme import types as wtypes from wsme import types as wtypes
@ -102,7 +103,7 @@ class TestListAction(api_base.FunctionalTest):
response = self.get_json('/actions/%s' % action['uuid'], response = self.get_json('/actions/%s' % action['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
action = obj_utils.create_test_action(self.context, parents=None) action = obj_utils.create_test_action(self.context, parents=None)
@ -125,7 +126,7 @@ class TestListAction(api_base.FunctionalTest):
action = obj_utils.create_test_action(self.context, parents=None) action = obj_utils.create_test_action(self.context, parents=None)
response = self.get_json('/actions/%s/detail' % action['uuid'], response = self.get_json('/actions/%s/detail' % action['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
action_list = [] action_list = []
@ -266,7 +267,7 @@ class TestListAction(api_base.FunctionalTest):
url = '/actions?action_plan_uuid=%s&audit_uuid=%s' % ( url = '/actions?action_plan_uuid=%s&audit_uuid=%s' % (
action_plan.uuid, self.audit.uuid) action_plan.uuid, self.audit.uuid)
response = self.get_json(url, expect_errors=True) response = self.get_json(url, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
def test_many_with_sort_key_uuid(self): def test_many_with_sort_key_uuid(self):
action_plan = obj_utils.create_test_action_plan( action_plan = obj_utils.create_test_action_plan(
@ -327,7 +328,7 @@ class TestListAction(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/actions?sort_key=%s' % 'bad_name', '/actions?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
def test_many_with_soft_deleted_action_plan_uuid(self): def test_many_with_soft_deleted_action_plan_uuid(self):
action_plan1 = obj_utils.create_test_action_plan( action_plan1 = obj_utils.create_test_action_plan(
@ -488,7 +489,7 @@ class TestPatch(api_base.FunctionalTest):
[{'path': '/state', 'value': new_state, 'op': 'replace'}], [{'path': '/state', 'value': new_state, 'op': 'replace'}],
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -516,7 +517,7 @@ class TestDelete(api_base.FunctionalTest):
mock_utcnow.return_value = test_time mock_utcnow.return_value = test_time
response = self.delete('/actions/%s' % self.action.uuid, response = self.delete('/actions/%s' % self.action.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -536,7 +537,7 @@ class TestActionPolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:defaut"}) rule: "rule:defaut"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

43
watcher/tests/api/v1/test_actions_plans.py
View File

@ -14,6 +14,7 @@ import datetime
import itertools import itertools
from unittest import mock from unittest import mock
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
@ -87,7 +88,7 @@ class TestListActionPlan(api_base.FunctionalTest):
response = self.get_json('/action_plans/%s' % action_plan['uuid'], response = self.get_json('/action_plans/%s' % action_plan['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
action_plan = obj_utils.create_test_action_plan(self.context) action_plan = obj_utils.create_test_action_plan(self.context)
@ -113,7 +114,7 @@ class TestListActionPlan(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/action_plan/%s/detail' % action_plan['uuid'], '/action_plan/%s/detail' % action_plan['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
action_plan_list = [] action_plan_list = []
@ -260,7 +261,7 @@ class TestListActionPlan(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/action_plans?sort_key=%s' % 'bad_name', '/action_plans?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
def test_links(self): def test_links(self):
uuid = utils.generate_uuid() uuid = utils.generate_uuid()
@ -317,7 +318,7 @@ class TestDelete(api_base.FunctionalTest):
def test_delete_action_plan_without_action(self): def test_delete_action_plan_without_action(self):
response = self.delete('/action_plans/%s' % self.action_plan.uuid, response = self.delete('/action_plans/%s' % self.action_plan.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
self.action_plan.state = objects.action_plan.State.SUCCEEDED self.action_plan.state = objects.action_plan.State.SUCCEEDED
@ -325,7 +326,7 @@ class TestDelete(api_base.FunctionalTest):
self.delete('/action_plans/%s' % self.action_plan.uuid) self.delete('/action_plans/%s' % self.action_plan.uuid)
response = self.get_json('/action_plans/%s' % self.action_plan.uuid, response = self.get_json('/action_plans/%s' % self.action_plan.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -345,20 +346,20 @@ class TestDelete(api_base.FunctionalTest):
expect_errors=True) expect_errors=True)
# The action plan does not exist anymore # The action plan does not exist anymore
self.assertEqual(404, ap_response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, ap_response.status_int)
self.assertEqual('application/json', ap_response.content_type) self.assertEqual('application/json', ap_response.content_type)
self.assertTrue(ap_response.json['error_message']) self.assertTrue(ap_response.json['error_message'])
# Nor does the action # Nor does the action
self.assertEqual(0, len(acts_response['actions'])) self.assertEqual(0, len(acts_response['actions']))
self.assertEqual(404, act_response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, act_response.status_int)
self.assertEqual('application/json', act_response.content_type) self.assertEqual('application/json', act_response.content_type)
self.assertTrue(act_response.json['error_message']) self.assertTrue(act_response.json['error_message'])
def test_delete_action_plan_not_found(self): def test_delete_action_plan_not_found(self):
uuid = utils.generate_uuid() uuid = utils.generate_uuid()
response = self.delete('/action_plans/%s' % uuid, expect_errors=True) response = self.delete('/action_plans/%s' % uuid, expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -388,7 +389,7 @@ class TestStart(api_base.FunctionalTest):
uuid = utils.generate_uuid() uuid = utils.generate_uuid()
response = self.post('/v1/action_plans/%s/%s' % response = self.post('/v1/action_plans/%s/%s' %
(uuid, 'start'), expect_errors=True) (uuid, 'start'), expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -401,11 +402,11 @@ class TestStart(api_base.FunctionalTest):
response = self.post('/v1/action_plans/%s/%s/' response = self.post('/v1/action_plans/%s/%s/'
% (self.action_plan.uuid, 'start'), % (self.action_plan.uuid, 'start'),
expect_errors=True) expect_errors=True)
self.assertEqual(200, response.status_int) self.assertEqual(HTTPStatus.OK, response.status_int)
act_response = self.get_json( act_response = self.get_json(
'/actions/%s' % action.uuid, '/actions/%s' % action.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(200, act_response.status_int) self.assertEqual(HTTPStatus.OK, act_response.status_int)
self.assertEqual('PENDING', act_response.json['state']) self.assertEqual('PENDING', act_response.json['state'])
self.assertEqual('application/json', act_response.content_type) self.assertEqual('application/json', act_response.content_type)
@ -445,7 +446,7 @@ class TestPatch(api_base.FunctionalTest):
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
def test_replace_non_existent_action_plan_denied(self): def test_replace_non_existent_action_plan_denied(self):
@ -455,7 +456,7 @@ class TestPatch(api_base.FunctionalTest):
'value': objects.action_plan.State.PENDING, 'value': objects.action_plan.State.PENDING,
'op': 'replace'}], 'op': 'replace'}],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -465,7 +466,7 @@ class TestPatch(api_base.FunctionalTest):
[{'path': '/foo', 'value': 'bar', 'op': 'add'}], [{'path': '/foo', 'value': 'bar', 'op': 'add'}],
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
def test_remove_denied(self): def test_remove_denied(self):
@ -480,7 +481,7 @@ class TestPatch(api_base.FunctionalTest):
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
def test_remove_uuid_denied(self): def test_remove_uuid_denied(self):
@ -488,7 +489,7 @@ class TestPatch(api_base.FunctionalTest):
'/action_plans/%s' % self.action_plan.uuid, '/action_plans/%s' % self.action_plan.uuid,
[{'path': '/uuid', 'op': 'remove'}], [{'path': '/uuid', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -497,7 +498,7 @@ class TestPatch(api_base.FunctionalTest):
'/action_plans/%s' % self.action_plan.uuid, '/action_plans/%s' % self.action_plan.uuid,
[{'path': '/non-existent', 'op': 'remove'}], [{'path': '/non-existent', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_code) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_code)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -512,7 +513,7 @@ class TestPatch(api_base.FunctionalTest):
[{'path': '/state', 'value': new_state, [{'path': '/state', 'value': new_state,
'op': 'replace'}]) 'op': 'replace'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
applier_mock.assert_called_once_with(mock.ANY, applier_mock.assert_called_once_with(mock.ANY,
self.action_plan.uuid) self.action_plan.uuid)
@ -579,7 +580,7 @@ class TestPatchStateTransitionDenied(api_base.FunctionalTest):
self.assertNotEqual(self.new_state, initial_ap['state']) self.assertNotEqual(self.new_state, initial_ap['state'])
self.assertEqual(self.original_state, updated_ap['state']) self.assertEqual(self.original_state, updated_ap['state'])
self.assertEqual(400, response.status_code) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_code)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -618,7 +619,7 @@ class TestPatchStateTransitionOk(api_base.FunctionalTest):
self.assertNotEqual(self.new_state, initial_ap['state']) self.assertNotEqual(self.new_state, initial_ap['state'])
self.assertEqual(self.new_state, updated_ap['state']) self.assertEqual(self.new_state, updated_ap['state'])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
class TestActionPlanPolicyEnforcement(api_base.FunctionalTest): class TestActionPlanPolicyEnforcement(api_base.FunctionalTest):
@ -635,7 +636,7 @@ class TestActionPlanPolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:defaut"}) rule: "rule:defaut"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

60
watcher/tests/api/v1/test_audit_templates.py
View File

@ -15,6 +15,7 @@ import itertools
from unittest import mock from unittest import mock
from urllib import parse as urlparse from urllib import parse as urlparse
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from oslo_utils import timeutils from oslo_utils import timeutils
@ -126,7 +127,7 @@ class TestListAuditTemplate(FunctionalTestWithSetup):
response = self.get_json( response = self.get_json(
'/audit_templates/%s' % audit_template['uuid'], '/audit_templates/%s' % audit_template['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
audit_template = obj_utils.create_test_audit_template(self.context) audit_template = obj_utils.create_test_audit_template(self.context)
@ -152,7 +153,7 @@ class TestListAuditTemplate(FunctionalTestWithSetup):
response = self.get_json( response = self.get_json(
'/audit_templates/%s/detail' % audit_template['uuid'], '/audit_templates/%s/detail' % audit_template['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
audit_template_list = [] audit_template_list = []
@ -336,7 +337,7 @@ class TestListAuditTemplate(FunctionalTestWithSetup):
response = self.get_json( response = self.get_json(
'/audit_templates?sort_key=%s' % 'goal_bad_name', '/audit_templates?sort_key=%s' % 'goal_bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
class TestPatch(FunctionalTestWithSetup): class TestPatch(FunctionalTestWithSetup):
@ -362,7 +363,7 @@ class TestPatch(FunctionalTestWithSetup):
[{'path': '/goal', 'value': new_goal_uuid, [{'path': '/goal', 'value': new_goal_uuid,
'op': 'replace'}]) 'op': 'replace'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
response = self.get_json( response = self.get_json(
'/audit_templates/%s' % self.audit_template.uuid) '/audit_templates/%s' % self.audit_template.uuid)
@ -386,7 +387,7 @@ class TestPatch(FunctionalTestWithSetup):
[{'path': '/goal', 'value': new_goal_uuid, [{'path': '/goal', 'value': new_goal_uuid,
'op': 'replace'}]) 'op': 'replace'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
response = self.get_json( response = self.get_json(
'/audit_templates/%s' % self.audit_template.name) '/audit_templates/%s' % self.audit_template.name)
@ -401,7 +402,7 @@ class TestPatch(FunctionalTestWithSetup):
[{'path': '/goal', 'value': self.fake_goal1.uuid, [{'path': '/goal', 'value': self.fake_goal1.uuid,
'op': 'replace'}], 'op': 'replace'}],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -416,7 +417,7 @@ class TestPatch(FunctionalTestWithSetup):
[{'path': '/goal', 'value': utils.generate_uuid(), [{'path': '/goal', 'value': utils.generate_uuid(),
'op': 'replace'}], 'op': 'replace'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
assert not cn_mock.called assert not cn_mock.called
def test_add_goal_uuid(self): def test_add_goal_uuid(self):
@ -426,7 +427,7 @@ class TestPatch(FunctionalTestWithSetup):
'value': self.fake_goal2.uuid, 'value': self.fake_goal2.uuid,
'op': 'add'}]) 'op': 'add'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_int) self.assertEqual(HTTPStatus.OK, response.status_int)
response = self.get_json( response = self.get_json(
'/audit_templates/%s' % self.audit_template.uuid) '/audit_templates/%s' % self.audit_template.uuid)
@ -439,7 +440,7 @@ class TestPatch(FunctionalTestWithSetup):
'value': self.fake_strategy1.uuid, 'value': self.fake_strategy1.uuid,
'op': 'add'}]) 'op': 'add'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_int) self.assertEqual(HTTPStatus.OK, response.status_int)
response = self.get_json( response = self.get_json(
'/audit_templates/%s' % self.audit_template.uuid) '/audit_templates/%s' % self.audit_template.uuid)
@ -452,7 +453,7 @@ class TestPatch(FunctionalTestWithSetup):
'value': self.fake_strategy2['uuid'], 'value': self.fake_strategy2['uuid'],
'op': 'replace'}]) 'op': 'replace'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_int) self.assertEqual(HTTPStatus.OK, response.status_int)
response = self.get_json( response = self.get_json(
'/audit_templates/%s' % self.audit_template.uuid) '/audit_templates/%s' % self.audit_template.uuid)
@ -466,7 +467,7 @@ class TestPatch(FunctionalTestWithSetup):
'value': utils.generate_uuid(), # Does not exist 'value': utils.generate_uuid(), # Does not exist
'op': 'replace'}], expect_errors=True) 'op': 'replace'}], expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
def test_add_non_existent_property(self): def test_add_non_existent_property(self):
@ -475,7 +476,7 @@ class TestPatch(FunctionalTestWithSetup):
[{'path': '/foo', 'value': 'bar', 'op': 'add'}], [{'path': '/foo', 'value': 'bar', 'op': 'add'}],
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
def test_remove_strategy(self): def test_remove_strategy(self):
@ -492,7 +493,7 @@ class TestPatch(FunctionalTestWithSetup):
'/audit_templates/%s' % self.audit_template.uuid, '/audit_templates/%s' % self.audit_template.uuid,
[{'path': '/strategy', 'op': 'remove'}]) [{'path': '/strategy', 'op': 'remove'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
def test_remove_goal(self): def test_remove_goal(self):
response = self.get_json( response = self.get_json(
@ -503,7 +504,7 @@ class TestPatch(FunctionalTestWithSetup):
'/audit_templates/%s' % self.audit_template.uuid, '/audit_templates/%s' % self.audit_template.uuid,
[{'path': '/goal', 'op': 'remove'}], [{'path': '/goal', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(403, response.status_code) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_code)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -512,7 +513,7 @@ class TestPatch(FunctionalTestWithSetup):
'/audit_templates/%s' % self.audit_template.uuid, '/audit_templates/%s' % self.audit_template.uuid,
[{'path': '/uuid', 'op': 'remove'}], [{'path': '/uuid', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -521,7 +522,7 @@ class TestPatch(FunctionalTestWithSetup):
'/audit_templates/%s' % self.audit_template.uuid, '/audit_templates/%s' % self.audit_template.uuid,
[{'path': '/non-existent', 'op': 'remove'}], [{'path': '/non-existent', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_code) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_code)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -538,7 +539,7 @@ class TestPost(FunctionalTestWithSetup):
response = self.post_json('/audit_templates', audit_template_dict) response = self.post_json('/audit_templates', audit_template_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
# Check location header # Check location header
self.assertIsNotNone(response.location) self.assertIsNotNone(response.location)
expected_location = \ expected_location = \
@ -565,7 +566,7 @@ class TestPost(FunctionalTestWithSetup):
response = self.post_json('/audit_templates', audit_template_dict) response = self.post_json('/audit_templates', audit_template_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
# Check location header # Check location header
self.assertIsNotNone(response.location) self.assertIsNotNone(response.location)
expected_location = \ expected_location = \
@ -613,7 +614,8 @@ class TestPost(FunctionalTestWithSetup):
strategy=self.fake_strategy1.uuid, scope=scope) strategy=self.fake_strategy1.uuid, scope=scope)
response = self.post_json('/audit_templates', response = self.post_json('/audit_templates',
audit_template_dict, expect_errors=True) audit_template_dict, expect_errors=True)
self.assertEqual(500, response.status_int) self.assertEqual(HTTPStatus.INTERNAL_SERVER_ERROR,
response.status_int)
def test_create_audit_template_does_autogenerate_id(self): def test_create_audit_template_does_autogenerate_id(self):
audit_template_dict = post_get_test_audit_template( audit_template_dict = post_get_test_audit_template(
@ -635,7 +637,7 @@ class TestPost(FunctionalTestWithSetup):
response = self.post_json('/audit_templates', audit_template_dict) response = self.post_json('/audit_templates', audit_template_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertTrue(utils.is_uuid_like(response.json['uuid'])) self.assertTrue(utils.is_uuid_like(response.json['uuid']))
def test_create_audit_template_with_invalid_goal(self): def test_create_audit_template_with_invalid_goal(self):
@ -648,7 +650,7 @@ class TestPost(FunctionalTestWithSetup):
goal_uuid=utils.generate_uuid()) goal_uuid=utils.generate_uuid())
response = self.post_json('/audit_templates', response = self.post_json('/audit_templates',
audit_template_dict, expect_errors=True) audit_template_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
assert not cn_mock.called assert not cn_mock.called
def test_create_audit_template_with_invalid_strategy(self): def test_create_audit_template_with_invalid_strategy(self):
@ -662,7 +664,7 @@ class TestPost(FunctionalTestWithSetup):
strategy_uuid=utils.generate_uuid()) strategy_uuid=utils.generate_uuid())
response = self.post_json('/audit_templates', response = self.post_json('/audit_templates',
audit_template_dict, expect_errors=True) audit_template_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
assert not cn_mock.called assert not cn_mock.called
def test_create_audit_template_with_unrelated_strategy(self): def test_create_audit_template_with_unrelated_strategy(self):
@ -676,7 +678,7 @@ class TestPost(FunctionalTestWithSetup):
strategy=self.fake_strategy2['uuid']) strategy=self.fake_strategy2['uuid'])
response = self.post_json('/audit_templates', response = self.post_json('/audit_templates',
audit_template_dict, expect_errors=True) audit_template_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
assert not cn_mock.called assert not cn_mock.called
def test_create_audit_template_with_uuid(self): def test_create_audit_template_with_uuid(self):
@ -689,7 +691,7 @@ class TestPost(FunctionalTestWithSetup):
response = self.post_json('/audit_templates', audit_template_dict, response = self.post_json('/audit_templates', audit_template_dict,
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
assert not cn_mock.called assert not cn_mock.called
def test_create_audit_template_with_old_scope(self): def test_create_audit_template_with_old_scope(self):
@ -710,7 +712,7 @@ class TestPost(FunctionalTestWithSetup):
strategy=self.fake_strategy1.uuid, scope=scope) strategy=self.fake_strategy1.uuid, scope=scope)
response = self.post_json('/audit_templates', response = self.post_json('/audit_templates',
audit_template_dict) audit_template_dict)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
class TestDelete(api_base.FunctionalTest): class TestDelete(api_base.FunctionalTest):
@ -730,7 +732,7 @@ class TestDelete(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
urlparse.quote('/audit_templates/%s' % self.audit_template.uuid), urlparse.quote('/audit_templates/%s' % self.audit_template.uuid),
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -753,7 +755,7 @@ class TestDelete(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
urlparse.quote('/audit_templates/%s' % self.audit_template.name), urlparse.quote('/audit_templates/%s' % self.audit_template.name),
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -771,7 +773,7 @@ class TestDelete(api_base.FunctionalTest):
uuid = utils.generate_uuid() uuid = utils.generate_uuid()
response = self.delete( response = self.delete(
'/audit_templates/%s' % uuid, expect_errors=True) '/audit_templates/%s' % uuid, expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -784,7 +786,7 @@ class TestAuditTemplatePolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:defaut"}) rule: "rule:defaut"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

79
watcher/tests/api/v1/test_audits.py
View File

@ -16,6 +16,7 @@ import itertools
from unittest import mock from unittest import mock
from urllib import parse as urlparse from urllib import parse as urlparse
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from oslo_utils import timeutils from oslo_utils import timeutils
@ -130,7 +131,7 @@ class TestListAudit(api_base.FunctionalTest):
response = self.get_json('/audits/%s' % audit['uuid'], response = self.get_json('/audits/%s' % audit['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
audit = obj_utils.create_test_audit(self.context) audit = obj_utils.create_test_audit(self.context)
@ -153,7 +154,7 @@ class TestListAudit(api_base.FunctionalTest):
audit = obj_utils.create_test_audit(self.context) audit = obj_utils.create_test_audit(self.context)
response = self.get_json('/audits/%s/detail' % audit['uuid'], response = self.get_json('/audits/%s/detail' % audit['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
audit_list = [] audit_list = []
@ -225,7 +226,7 @@ class TestListAudit(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/audits?sort_key=%s' % 'bad_name', '/audits?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
def test_links(self): def test_links(self):
uuid = utils.generate_uuid() uuid = utils.generate_uuid()
@ -295,7 +296,7 @@ class TestPatch(api_base.FunctionalTest):
[{'path': '/state', 'value': new_state, [{'path': '/state', 'value': new_state,
'op': 'replace'}]) 'op': 'replace'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
response = self.get_json('/audits/%s' % self.audit.uuid) response = self.get_json('/audits/%s' % self.audit.uuid)
self.assertEqual(new_state, response['state']) self.assertEqual(new_state, response['state'])
@ -308,7 +309,7 @@ class TestPatch(api_base.FunctionalTest):
'/audits/%s' % utils.generate_uuid(), '/audits/%s' % utils.generate_uuid(),
[{'path': '/state', 'value': objects.audit.State.SUCCEEDED, [{'path': '/state', 'value': objects.audit.State.SUCCEEDED,
'op': 'replace'}], expect_errors=True) 'op': 'replace'}], expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -318,7 +319,7 @@ class TestPatch(api_base.FunctionalTest):
'/audits/%s' % self.audit.uuid, '/audits/%s' % self.audit.uuid,
[{'path': '/state', 'value': new_state, 'op': 'add'}]) [{'path': '/state', 'value': new_state, 'op': 'add'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_int) self.assertEqual(HTTPStatus.OK, response.status_int)
response = self.get_json('/audits/%s' % self.audit.uuid) response = self.get_json('/audits/%s' % self.audit.uuid)
self.assertEqual(new_state, response['state']) self.assertEqual(new_state, response['state'])
@ -329,7 +330,7 @@ class TestPatch(api_base.FunctionalTest):
[{'path': '/foo', 'value': 'bar', 'op': 'add'}], [{'path': '/foo', 'value': 'bar', 'op': 'add'}],
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
def test_remove_ok(self): def test_remove_ok(self):
@ -339,7 +340,7 @@ class TestPatch(api_base.FunctionalTest):
response = self.patch_json('/audits/%s' % self.audit.uuid, response = self.patch_json('/audits/%s' % self.audit.uuid,
[{'path': '/interval', 'op': 'remove'}]) [{'path': '/interval', 'op': 'remove'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
response = self.get_json('/audits/%s' % self.audit.uuid) response = self.get_json('/audits/%s' % self.audit.uuid)
self.assertIsNone(response['interval']) self.assertIsNone(response['interval'])
@ -348,7 +349,7 @@ class TestPatch(api_base.FunctionalTest):
response = self.patch_json('/audits/%s' % self.audit.uuid, response = self.patch_json('/audits/%s' % self.audit.uuid,
[{'path': '/uuid', 'op': 'remove'}], [{'path': '/uuid', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -357,7 +358,7 @@ class TestPatch(api_base.FunctionalTest):
'/audits/%s' % self.audit.uuid, '/audits/%s' % self.audit.uuid,
[{'path': '/non-existent', 'op': 'remove'}], [{'path': '/non-existent', 'op': 'remove'}],
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_code) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_code)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -415,7 +416,7 @@ class TestPatchStateTransitionDenied(api_base.FunctionalTest):
'op': 'replace'}], 'op': 'replace'}],
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_code) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_code)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
response = self.get_json('/audits/%s' % self.audit.uuid) response = self.get_json('/audits/%s' % self.audit.uuid)
@ -462,7 +463,7 @@ class TestPatchStateTransitionOk(api_base.FunctionalTest):
[{'path': '/state', 'value': self.new_state, [{'path': '/state', 'value': self.new_state,
'op': 'replace'}]) 'op': 'replace'}])
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code) self.assertEqual(HTTPStatus.OK, response.status_code)
response = self.get_json('/audits/%s' % self.audit.uuid) response = self.get_json('/audits/%s' % self.audit.uuid)
self.assertEqual(self.new_state, response['state']) self.assertEqual(self.new_state, response['state'])
@ -502,7 +503,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
# Check location header # Check location header
self.assertIsNotNone(response.location) self.assertIsNotNone(response.location)
expected_location = '/v1/audits/%s' % response.json['uuid'] expected_location = '/v1/audits/%s' % response.json['uuid']
@ -527,7 +528,7 @@ class TestPost(api_base.FunctionalTest):
audit_dict = post_get_test_audit(state=objects.audit.State.SUCCEEDED) audit_dict = post_get_test_audit(state=objects.audit.State.SUCCEEDED)
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -542,7 +543,7 @@ class TestPost(api_base.FunctionalTest):
'next_run_time', 'hostname']) 'next_run_time', 'hostname'])
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -557,7 +558,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertTrue(utils.is_uuid_like(response.json['uuid'])) self.assertTrue(utils.is_uuid_like(response.json['uuid']))
@ -573,7 +574,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertTrue(utils.is_uuid_like(response.json['uuid'])) self.assertTrue(utils.is_uuid_like(response.json['uuid']))
@ -590,7 +591,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertTrue(utils.is_uuid_like(response.json['uuid'])) self.assertTrue(utils.is_uuid_like(response.json['uuid']))
@ -608,7 +609,7 @@ class TestPost(api_base.FunctionalTest):
'01234567-8910-1112-1314-151617181920') '01234567-8910-1112-1314-151617181920')
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual("application/json", response.content_type) self.assertEqual("application/json", response.content_type)
expected_error_msg = ('The audit template UUID or name specified is ' expected_error_msg = ('The audit template UUID or name specified is '
'invalid') 'invalid')
@ -643,7 +644,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertTrue(utils.is_uuid_like(response.json['uuid'])) self.assertTrue(utils.is_uuid_like(response.json['uuid']))
@ -660,7 +661,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertEqual(audit_dict['interval'], response.json['interval']) self.assertEqual(audit_dict['interval'], response.json['interval'])
@ -679,7 +680,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertEqual(audit_dict['interval'], response.json['interval']) self.assertEqual(audit_dict['interval'], response.json['interval'])
@ -698,7 +699,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(500, response.status_int) self.assertEqual(HTTPStatus.INTERNAL_SERVER_ERROR, response.status_int)
expected_error_msg = ('Exactly 5 or 6 columns has to be ' expected_error_msg = ('Exactly 5 or 6 columns has to be '
'specified for iterator expression.') 'specified for iterator expression.')
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -714,7 +715,7 @@ class TestPost(api_base.FunctionalTest):
audit_dict['audit_type'] = objects.audit.AuditType.CONTINUOUS.value audit_dict['audit_type'] = objects.audit.AuditType.CONTINUOUS.value
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
expected_error_msg = ('Interval of audit must be specified ' expected_error_msg = ('Interval of audit must be specified '
'for CONTINUOUS.') 'for CONTINUOUS.')
@ -731,7 +732,7 @@ class TestPost(api_base.FunctionalTest):
audit_dict['audit_type'] = objects.audit.AuditType.ONESHOT.value audit_dict['audit_type'] = objects.audit.AuditType.ONESHOT.value
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
expected_error_msg = 'Interval of audit must not be set for ONESHOT.' expected_error_msg = 'Interval of audit must not be set for ONESHOT.'
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -755,7 +756,7 @@ class TestPost(api_base.FunctionalTest):
del audit_dict['scope'] del audit_dict['scope']
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
assert not mock_trigger_audit.called assert not mock_trigger_audit.called
@mock.patch.object(deapi.DecisionEngineAPI, 'trigger_audit') @mock.patch.object(deapi.DecisionEngineAPI, 'trigger_audit')
@ -769,7 +770,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
expected_error_msg = ('Specify parameters but no predefined ' expected_error_msg = ('Specify parameters but no predefined '
'strategy for audit, or no ' 'strategy for audit, or no '
'parameter spec in predefined strategy') 'parameter spec in predefined strategy')
@ -792,7 +793,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
expected_error_msg = ('Specify parameters but no predefined ' expected_error_msg = ('Specify parameters but no predefined '
'strategy for audit, or no ' 'strategy for audit, or no '
'parameter spec in predefined strategy') 'parameter spec in predefined strategy')
@ -816,7 +817,7 @@ class TestPost(api_base.FunctionalTest):
del audit_dict[k] del audit_dict[k]
response = self.post_json('/audits', audit_dict, expect_errors=True) response = self.post_json('/audits', audit_dict, expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual("application/json", response.content_type) self.assertEqual("application/json", response.content_type)
expected_error_msg = 'Audit parameter fake2 are not allowed' expected_error_msg = 'Audit parameter fake2 are not allowed'
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -875,13 +876,13 @@ class TestPost(api_base.FunctionalTest):
audit_dict['name'] = normal_name audit_dict['name'] = normal_name
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(normal_name, response.json['name']) self.assertEqual(normal_name, response.json['name'])
audit_dict['name'] = long_name audit_dict['name'] = long_name
response = self.post_json('/audits', audit_dict) response = self.post_json('/audits', audit_dict)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertNotEqual(long_name, response.json['name']) self.assertNotEqual(long_name, response.json['name'])
@mock.patch.object(deapi.DecisionEngineAPI, 'trigger_audit') @mock.patch.object(deapi.DecisionEngineAPI, 'trigger_audit')
@ -905,7 +906,7 @@ class TestPost(api_base.FunctionalTest):
audit_dict, audit_dict,
headers={'OpenStack-API-Version': 'infra-optim 1.1'}) headers={'OpenStack-API-Version': 'infra-optim 1.1'})
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertEqual(objects.audit.State.PENDING, self.assertEqual(objects.audit.State.PENDING,
response.json['state']) response.json['state'])
self.assertEqual(audit_dict['interval'], response.json['interval']) self.assertEqual(audit_dict['interval'], response.json['interval'])
@ -944,7 +945,7 @@ class TestPost(api_base.FunctionalTest):
headers={'OpenStack-API-Version': 'infra-optim 1.0'}, headers={'OpenStack-API-Version': 'infra-optim 1.0'},
expect_errors=True) expect_errors=True)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(406, response.status_int) self.assertEqual(HTTPStatus.NOT_ACCEPTABLE, response.status_int)
expected_error_msg = 'Request not acceptable.' expected_error_msg = 'Request not acceptable.'
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
self.assertIn(expected_error_msg, response.json['error_message']) self.assertIn(expected_error_msg, response.json['error_message'])
@ -963,7 +964,7 @@ class TestPost(api_base.FunctionalTest):
audit_dict, audit_dict,
headers={'OpenStack-API-Version': 'infra-optim 1.2'}) headers={'OpenStack-API-Version': 'infra-optim 1.2'})
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertFalse(response.json['force']) self.assertFalse(response.json['force'])
@mock.patch.object(deapi.DecisionEngineAPI, 'trigger_audit') @mock.patch.object(deapi.DecisionEngineAPI, 'trigger_audit')
@ -980,7 +981,7 @@ class TestPost(api_base.FunctionalTest):
audit_dict, audit_dict,
headers={'OpenStack-API-Version': 'infra-optim 1.2'}) headers={'OpenStack-API-Version': 'infra-optim 1.2'})
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int) self.assertEqual(HTTPStatus.CREATED, response.status_int)
self.assertTrue(response.json['force']) self.assertTrue(response.json['force'])
@ -1013,7 +1014,7 @@ class TestDelete(api_base.FunctionalTest):
'op': 'replace'}]) 'op': 'replace'}])
response = self.delete('/audits/%s' % self.audit.uuid, response = self.delete('/audits/%s' % self.audit.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -1025,7 +1026,7 @@ class TestDelete(api_base.FunctionalTest):
self.delete('/audits/%s' % self.audit.uuid) self.delete('/audits/%s' % self.audit.uuid)
response = self.get_json('/audits/%s' % self.audit.uuid, response = self.get_json('/audits/%s' % self.audit.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -1041,7 +1042,7 @@ class TestDelete(api_base.FunctionalTest):
def test_delete_audit_not_found(self): def test_delete_audit_not_found(self):
uuid = utils.generate_uuid() uuid = utils.generate_uuid()
response = self.delete('/audits/%s' % uuid, expect_errors=True) response = self.delete('/audits/%s' % uuid, expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -1058,7 +1059,7 @@ class TestAuditPolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:defaut"}) rule: "rule:defaut"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

5
watcher/tests/api/v1/test_data_model.py
View File

@ -15,6 +15,7 @@
from unittest import mock from unittest import mock
from http import HTTPStatus
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from watcher.decision_engine import rpcapi as deapi from watcher.decision_engine import rpcapi as deapi
@ -42,7 +43,7 @@ class TestListDataModel(api_base.FunctionalTest):
'/data_model/?data_model_type=compute', '/data_model/?data_model_type=compute',
headers={'OpenStack-API-Version': 'infra-optim 1.2'}, headers={'OpenStack-API-Version': 'infra-optim 1.2'},
expect_errors=True) expect_errors=True)
self.assertEqual(406, response.status_int) self.assertEqual(HTTPStatus.NOT_ACCEPTABLE, response.status_int)
class TestDataModelPolicyEnforcement(api_base.FunctionalTest): class TestDataModelPolicyEnforcement(api_base.FunctionalTest):
@ -59,7 +60,7 @@ class TestDataModelPolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:defaut"}) rule: "rule:defaut"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

9
watcher/tests/api/v1/test_goals.py
View File

@ -10,6 +10,7 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from urllib import parse as urlparse from urllib import parse as urlparse
@ -59,7 +60,7 @@ class TestListGoal(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/goals/%s' % goal['uuid'], '/goals/%s' % goal['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
goal = obj_utils.create_test_goal(self.context) goal = obj_utils.create_test_goal(self.context)
@ -71,7 +72,7 @@ class TestListGoal(api_base.FunctionalTest):
goal = obj_utils.create_test_goal(self.context) goal = obj_utils.create_test_goal(self.context)
response = self.get_json('/goals/%s/detail' % goal.uuid, response = self.get_json('/goals/%s/detail' % goal.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
goal_list = [] goal_list = []
@ -139,7 +140,7 @@ class TestListGoal(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/goals?sort_key=%s' % 'bad_name', '/goals?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
class TestGoalPolicyEnforcement(api_base.FunctionalTest): class TestGoalPolicyEnforcement(api_base.FunctionalTest):
@ -150,7 +151,7 @@ class TestGoalPolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:default"}) rule: "rule:default"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

6
watcher/tests/api/v1/test_microversions.py
View File

@ -10,6 +10,8 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from http import HTTPStatus
from watcher.api.controllers.v1 import versions from watcher.api.controllers.v1 import versions
from watcher.tests.api import base as api_base from watcher.tests.api import base as api_base
@ -38,7 +40,7 @@ class TestMicroversions(api_base.FunctionalTest):
'10'])}, '10'])},
expect_errors=True, return_json=False) expect_errors=True, return_json=False)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertEqual(406, response.status_int) self.assertEqual(HTTPStatus.NOT_ACCEPTABLE, response.status_int)
expected_error_msg = ('Invalid value for' expected_error_msg = ('Invalid value for'
' OpenStack-API-Version header') ' OpenStack-API-Version header')
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -98,7 +100,7 @@ class TestMicroversions(api_base.FunctionalTest):
headers={'OpenStack-API-Version': ' '.join([SERVICE_TYPE, headers={'OpenStack-API-Version': ' '.join([SERVICE_TYPE,
'1.999'])}, '1.999'])},
expect_errors=True) expect_errors=True)
self.assertEqual(406, response.status_int) self.assertEqual(HTTPStatus.NOT_ACCEPTABLE, response.status_int)
self.assertEqual(response.headers[H_MIN_VER], MIN_VER) self.assertEqual(response.headers[H_MIN_VER], MIN_VER)
self.assertEqual(response.headers[H_MAX_VER], MAX_VER) self.assertEqual(response.headers[H_MAX_VER], MAX_VER)
expected_error_msg = ('Version 1.999 was requested but the minor ' expected_error_msg = ('Version 1.999 was requested but the minor '

9
watcher/tests/api/v1/test_scoring_engines.py
View File

@ -10,6 +10,7 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from watcher.common import utils from watcher.common import utils
@ -44,7 +45,7 @@ class TestListScoringEngine(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/scoring_engines/%s' % scoring_engine['name'], '/scoring_engines/%s' % scoring_engine['name'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
obj_utils.create_test_goal(self.context) obj_utils.create_test_goal(self.context)
@ -63,7 +64,7 @@ class TestListScoringEngine(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/scoring_engines/%s/detail' % scoring_engine.id, '/scoring_engines/%s/detail' % scoring_engine.id,
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
scoring_engine_list = [] scoring_engine_list = []
@ -131,7 +132,7 @@ class TestListScoringEngine(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/goals?sort_key=%s' % 'bad_name', '/goals?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
class TestScoringEnginePolicyEnforcement(api_base.FunctionalTest): class TestScoringEnginePolicyEnforcement(api_base.FunctionalTest):
@ -142,7 +143,7 @@ class TestScoringEnginePolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:default"}) rule: "rule:default"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

9
watcher/tests/api/v1/test_services.py
View File

@ -10,6 +10,7 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from urllib import parse as urlparse from urllib import parse as urlparse
@ -57,7 +58,7 @@ class TestListService(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/services/%s' % service['id'], '/services/%s' % service['id'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
service = obj_utils.create_test_service(self.context) service = obj_utils.create_test_service(self.context)
@ -74,7 +75,7 @@ class TestListService(api_base.FunctionalTest):
service = obj_utils.create_test_service(self.context) service = obj_utils.create_test_service(self.context)
response = self.get_json('/services/%s/detail' % service.id, response = self.get_json('/services/%s/detail' % service.id,
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
service_list = [] service_list = []
@ -149,7 +150,7 @@ class TestListService(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/services?sort_key=%s' % 'bad_name', '/services?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
class TestServicePolicyEnforcement(api_base.FunctionalTest): class TestServicePolicyEnforcement(api_base.FunctionalTest):
@ -160,7 +161,7 @@ class TestServicePolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:default"}) rule: "rule:default"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

9
watcher/tests/api/v1/test_strategies.py
View File

@ -13,6 +13,7 @@
from unittest import mock from unittest import mock
from urllib import parse as urlparse from urllib import parse as urlparse
from http import HTTPStatus
from oslo_config import cfg from oslo_config import cfg
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
@ -88,7 +89,7 @@ class TestListStrategy(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/strategies/%s' % strategy['uuid'], '/strategies/%s' % strategy['uuid'],
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_detail(self): def test_detail(self):
strategy = obj_utils.create_test_strategy(self.context) strategy = obj_utils.create_test_strategy(self.context)
@ -104,7 +105,7 @@ class TestListStrategy(api_base.FunctionalTest):
strategy = obj_utils.create_test_strategy(self.context) strategy = obj_utils.create_test_strategy(self.context)
response = self.get_json('/strategies/%s/detail' % strategy.uuid, response = self.get_json('/strategies/%s/detail' % strategy.uuid,
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
def test_many(self): def test_many(self):
strategy_list = [] strategy_list = []
@ -240,7 +241,7 @@ class TestListStrategy(api_base.FunctionalTest):
response = self.get_json( response = self.get_json(
'/strategies?sort_key=%s' % 'bad_name', '/strategies?sort_key=%s' % 'bad_name',
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
class TestStrategyPolicyEnforcement(api_base.FunctionalTest): class TestStrategyPolicyEnforcement(api_base.FunctionalTest):
@ -256,7 +257,7 @@ class TestStrategyPolicyEnforcement(api_base.FunctionalTest):
"default": "rule:admin_api", "default": "rule:admin_api",
rule: "rule:defaut"}) rule: "rule:defaut"})
response = func(*arg, **kwarg) response = func(*arg, **kwarg)
self.assertEqual(403, response.status_int) self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue( self.assertTrue(
"Policy doesn't allow %s to be performed." % rule, "Policy doesn't allow %s to be performed." % rule,

24
watcher/tests/api/v1/test_types.py
View File

@ -18,6 +18,8 @@ import webtest
import wsme import wsme
from wsme import types as wtypes from wsme import types as wtypes
from http import HTTPStatus
from watcher.api.controllers.v1 import types from watcher.api.controllers.v1 import types
from watcher.common import exception from watcher.common import exception
from watcher.common import utils from watcher.common import utils
@ -120,68 +122,68 @@ class TestJsonPatchType(base.TestCase):
{'path': '/dict', 'op': 'add', {'path': '/dict', 'op': 'add',
'value': {'cat': 'meow'}}] 'value': {'cat': 'meow'}}]
ret = self._patch_json(valid_patches, False) ret = self._patch_json(valid_patches, False)
self.assertEqual(200, ret.status_int) self.assertEqual(HTTPStatus.OK, ret.status_int)
self.assertEqual(valid_patches, ret.json) self.assertEqual(valid_patches, ret.json)
def test_cannot_update_internal_attr(self): def test_cannot_update_internal_attr(self):
patch = [{'path': '/internal', 'op': 'replace', 'value': 'foo'}] patch = [{'path': '/internal', 'op': 'replace', 'value': 'foo'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_cannot_update_internal_dict_attr(self): def test_cannot_update_internal_dict_attr(self):
patch = [{'path': '/internal', 'op': 'replace', patch = [{'path': '/internal', 'op': 'replace',
'value': 'foo'}] 'value': 'foo'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_mandatory_attr(self): def test_mandatory_attr(self):
patch = [{'op': 'replace', 'path': '/mandatory', 'value': 'foo'}] patch = [{'op': 'replace', 'path': '/mandatory', 'value': 'foo'}]
ret = self._patch_json(patch, False) ret = self._patch_json(patch, False)
self.assertEqual(200, ret.status_int) self.assertEqual(HTTPStatus.OK, ret.status_int)
self.assertEqual(patch, ret.json) self.assertEqual(patch, ret.json)
def test_cannot_remove_mandatory_attr(self): def test_cannot_remove_mandatory_attr(self):
patch = [{'op': 'remove', 'path': '/mandatory'}] patch = [{'op': 'remove', 'path': '/mandatory'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_missing_required_fields_path(self): def test_missing_required_fields_path(self):
missing_path = [{'op': 'remove'}] missing_path = [{'op': 'remove'}]
ret = self._patch_json(missing_path, True) ret = self._patch_json(missing_path, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_missing_required_fields_op(self): def test_missing_required_fields_op(self):
missing_op = [{'path': '/foo'}] missing_op = [{'path': '/foo'}]
ret = self._patch_json(missing_op, True) ret = self._patch_json(missing_op, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_invalid_op(self): def test_invalid_op(self):
patch = [{'path': '/foo', 'op': 'invalid'}] patch = [{'path': '/foo', 'op': 'invalid'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_invalid_path(self): def test_invalid_path(self):
patch = [{'path': 'invalid-path', 'op': 'remove'}] patch = [{'path': 'invalid-path', 'op': 'remove'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_cannot_add_with_no_value(self): def test_cannot_add_with_no_value(self):
patch = [{'path': '/extra/foo', 'op': 'add'}] patch = [{'path': '/extra/foo', 'op': 'add'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])
def test_cannot_replace_with_no_value(self): def test_cannot_replace_with_no_value(self):
patch = [{'path': '/foo', 'op': 'replace'}] patch = [{'path': '/foo', 'op': 'replace'}]
ret = self._patch_json(patch, True) ret = self._patch_json(patch, True)
self.assertEqual(400, ret.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, ret.status_int)
self.assertTrue(ret.json['faultstring']) self.assertTrue(ret.json['faultstring'])

10
watcher/tests/api/v1/test_webhooks.py
View File

@ -12,6 +12,8 @@
from unittest import mock from unittest import mock
from http import HTTPStatus
from watcher.decision_engine import rpcapi as deapi from watcher.decision_engine import rpcapi as deapi
from watcher import objects from watcher import objects
from watcher.tests.api import base as api_base from watcher.tests.api import base as api_base
@ -34,7 +36,7 @@ class TestPost(api_base.FunctionalTest):
response = self.post_json( response = self.post_json(
'/webhooks/%s' % audit['uuid'], {}, '/webhooks/%s' % audit['uuid'], {},
headers={'OpenStack-API-Version': 'infra-optim 1.4'}) headers={'OpenStack-API-Version': 'infra-optim 1.4'})
self.assertEqual(202, response.status_int) self.assertEqual(HTTPStatus.ACCEPTED, response.status_int)
mock_trigger_audit.assert_called_once_with( mock_trigger_audit.assert_called_once_with(
mock.ANY, audit['uuid']) mock.ANY, audit['uuid'])
@ -43,7 +45,7 @@ class TestPost(api_base.FunctionalTest):
'/webhooks/no-audit', {}, '/webhooks/no-audit', {},
headers={'OpenStack-API-Version': 'infra-optim 1.4'}, headers={'OpenStack-API-Version': 'infra-optim 1.4'},
expect_errors=True) expect_errors=True)
self.assertEqual(404, response.status_int) self.assertEqual(HTTPStatus.NOT_FOUND, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -53,7 +55,7 @@ class TestPost(api_base.FunctionalTest):
'/webhooks/%s' % audit['uuid'], {}, '/webhooks/%s' % audit['uuid'], {},
headers={'OpenStack-API-Version': 'infra-optim 1.4'}, headers={'OpenStack-API-Version': 'infra-optim 1.4'},
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])
@ -66,6 +68,6 @@ class TestPost(api_base.FunctionalTest):
'/webhooks/%s' % audit['uuid'], {}, '/webhooks/%s' % audit['uuid'], {},
headers={'OpenStack-API-Version': 'infra-optim 1.4'}, headers={'OpenStack-API-Version': 'infra-optim 1.4'},
expect_errors=True) expect_errors=True)
self.assertEqual(400, response.status_int) self.assertEqual(HTTPStatus.BAD_REQUEST, response.status_int)
self.assertEqual('application/json', response.content_type) self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message']) self.assertTrue(response.json['error_message'])

9
watcher/tests/common/test_cinder_helper.py
View File

@ -13,9 +13,11 @@
# limitations under the License. # limitations under the License.
# #
import time
from unittest import mock from unittest import mock
from http import HTTPStatus
import time
from cinderclient import exceptions as cinder_exception from cinderclient import exceptions as cinder_exception
from watcher.common import cinder_helper from watcher.common import cinder_helper
@ -291,7 +293,8 @@ class TestCinderHelper(base.TestCase):
volume = self.fake_volume() volume = self.fake_volume()
cinder_util.get_volume = mock.MagicMock() cinder_util.get_volume = mock.MagicMock()
cinder_util.get_volume.side_effect = cinder_exception.NotFound(404) cinder_util.get_volume.side_effect =\
cinder_exception.NotFound(HTTPStatus.NOT_FOUND)
result = cinder_util._can_get_volume(volume.id) result = cinder_util._can_get_volume(volume.id)
self.assertFalse(result) self.assertFalse(result)
@ -339,7 +342,7 @@ class TestCinderHelper(base.TestCase):
cinder_util = cinder_helper.CinderHelper() cinder_util = cinder_helper.CinderHelper()
volume = self.fake_volume() volume = self.fake_volume()
side_effect = cinder_exception.NotFound(404) side_effect = cinder_exception.NotFound(HTTPStatus.NOT_FOUND)
cinder_util.cinder.volumes.get.side_effect = side_effect cinder_util.cinder.volumes.get.side_effect = side_effect
cinder_util.cinder.volumes.find.return_value = False cinder_util.cinder.volumes.find.return_value = False
result = cinder_util.get_volume(volume) result = cinder_util.get_volume(volume)

37
watcher/tests/common/test_placement_helper.py
View File

@ -11,6 +11,7 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from http import HTTPStatus
from unittest import mock from unittest import mock
from watcher.common import placement_helper from watcher.common import placement_helper
@ -55,10 +56,10 @@ class TestPlacementHelper(base.TestCase):
kss_req.assert_called_once_with(url, method, **kwargs) kss_req.assert_called_once_with(url, method, **kwargs)
def test_get(self, kss_req): def test_get(self, kss_req):
kss_req.return_value = fake_requests.FakeResponse(200) kss_req.return_value = fake_requests.FakeResponse(HTTPStatus.OK)
url = '/resource_providers' url = '/resource_providers'
resp = self.client.get(url) resp = self.client.get(url)
self.assertEqual(200, resp.status_code) self.assertEqual(HTTPStatus.OK, resp.status_code)
self._assert_keystone_called_once(kss_req, url, 'GET') self._assert_keystone_called_once(kss_req, url, 'GET')
def test_get_resource_providers_OK(self, kss_req): def test_get_resource_providers_OK(self, kss_req):
@ -76,7 +77,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_resource_providers(rp_name) result = self.client.get_resource_providers(rp_name)
@ -99,7 +100,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_resource_providers(rp_name) result = self.client.get_resource_providers(rp_name)
@ -110,7 +111,8 @@ class TestPlacementHelper(base.TestCase):
def test_get_resource_providers_fail(self, kss_req): def test_get_resource_providers_fail(self, kss_req):
rp_name = 'compute' rp_name = 'compute'
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
400, content=jsonutils.dump_as_bytes(self.fake_err_msg)) HTTPStatus.BAD_REQUEST,
content=jsonutils.dump_as_bytes(self.fake_err_msg))
result = self.client.get_resource_providers(rp_name) result = self.client.get_resource_providers(rp_name)
self.assertIsNone(result) self.assertIsNone(result)
@ -149,7 +151,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_inventories(rp_uuid) result = self.client.get_inventories(rp_uuid)
@ -160,7 +162,8 @@ class TestPlacementHelper(base.TestCase):
def test_get_inventories_fail(self, kss_req): def test_get_inventories_fail(self, kss_req):
rp_uuid = uuidutils.generate_uuid() rp_uuid = uuidutils.generate_uuid()
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
404, content=jsonutils.dump_as_bytes(self.fake_err_msg)) HTTPStatus.NOT_FOUND,
content=jsonutils.dump_as_bytes(self.fake_err_msg))
result = self.client.get_inventories(rp_uuid) result = self.client.get_inventories(rp_uuid)
self.assertIsNone(result) self.assertIsNone(result)
@ -175,7 +178,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_provider_traits(rp_uuid) result = self.client.get_provider_traits(rp_uuid)
@ -186,7 +189,8 @@ class TestPlacementHelper(base.TestCase):
def test_get_provider_traits_fail(self, kss_req): def test_get_provider_traits_fail(self, kss_req):
rp_uuid = uuidutils.generate_uuid() rp_uuid = uuidutils.generate_uuid()
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
404, content=jsonutils.dump_as_bytes(self.fake_err_msg)) HTTPStatus.NOT_FOUND,
content=jsonutils.dump_as_bytes(self.fake_err_msg))
result = self.client.get_provider_traits(rp_uuid) result = self.client.get_provider_traits(rp_uuid)
self.assertIsNone(result) self.assertIsNone(result)
@ -216,7 +220,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_allocations_for_consumer(c_uuid) result = self.client.get_allocations_for_consumer(c_uuid)
@ -227,7 +231,8 @@ class TestPlacementHelper(base.TestCase):
def test_get_allocations_for_consumer_fail(self, kss_req): def test_get_allocations_for_consumer_fail(self, kss_req):
c_uuid = uuidutils.generate_uuid() c_uuid = uuidutils.generate_uuid()
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
404, content=jsonutils.dump_as_bytes(self.fake_err_msg)) HTTPStatus.NOT_FOUND,
content=jsonutils.dump_as_bytes(self.fake_err_msg))
result = self.client.get_allocations_for_consumer(c_uuid) result = self.client.get_allocations_for_consumer(c_uuid)
self.assertIsNone(result) self.assertIsNone(result)
@ -245,7 +250,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_usages_for_resource_provider(rp_uuid) result = self.client.get_usages_for_resource_provider(rp_uuid)
@ -256,7 +261,8 @@ class TestPlacementHelper(base.TestCase):
def test_get_usages_for_resource_provider_fail(self, kss_req): def test_get_usages_for_resource_provider_fail(self, kss_req):
rp_uuid = uuidutils.generate_uuid() rp_uuid = uuidutils.generate_uuid()
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
404, content=jsonutils.dump_as_bytes(self.fake_err_msg)) HTTPStatus.NOT_FOUND,
content=jsonutils.dump_as_bytes(self.fake_err_msg))
result = self.client.get_usages_for_resource_provider(rp_uuid) result = self.client.get_usages_for_resource_provider(rp_uuid)
self.assertIsNone(result) self.assertIsNone(result)
@ -296,7 +302,7 @@ class TestPlacementHelper(base.TestCase):
} }
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
200, content=jsonutils.dump_as_bytes(mock_json_data)) HTTPStatus.OK, content=jsonutils.dump_as_bytes(mock_json_data))
result = self.client.get_candidate_providers(resources) result = self.client.get_candidate_providers(resources)
@ -307,6 +313,7 @@ class TestPlacementHelper(base.TestCase):
def test_get_candidate_providers_fail(self, kss_req): def test_get_candidate_providers_fail(self, kss_req):
rp_uuid = uuidutils.generate_uuid() rp_uuid = uuidutils.generate_uuid()
kss_req.return_value = fake_requests.FakeResponse( kss_req.return_value = fake_requests.FakeResponse(
404, content=jsonutils.dump_as_bytes(self.fake_err_msg)) HTTPStatus.NOT_FOUND,
content=jsonutils.dump_as_bytes(self.fake_err_msg))
result = self.client.get_candidate_providers(rp_uuid) result = self.client.get_candidate_providers(rp_uuid)
self.assertIsNone(result) self.assertIsNone(result)

3
watcher/tests/decision_engine/datasources/test_grafana_helper.py
View File

@ -26,6 +26,7 @@ from watcher.common import exception
from watcher.decision_engine.datasources import grafana from watcher.decision_engine.datasources import grafana
from watcher.tests import base from watcher.tests import base
from http import HTTPStatus
import requests import requests
CONF = cfg.CONF CONF = cfg.CONF
@ -128,7 +129,7 @@ class TestGrafana(base.BaseTestCase):
problems. problems.
""" """
m_request.return_value = mock.Mock(status_code=404) m_request.return_value = mock.Mock(status_code=HTTPStatus.NOT_FOUND)
t_grafana = grafana.GrafanaHelper(osc=mock.Mock()) t_grafana = grafana.GrafanaHelper(osc=mock.Mock())