Arnike/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge "Drops forbidden patch/delete/post action apis"

Browse Source
This commit is contained in:
Zuul
2025-07-08 18:38:40 +00:00
committed by Gerrit Code Review
parent 1ab5babbb6 c0a5abe29c
commit 98b56b66ac
3 changed files with 6 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
releasenotes/notes/drop-patch-delete-post-action-api-37fe4ce5be6500db.yaml Normal file
View File

@@ -0,0 +1,6 @@
---
other:
- |
The `DELETE`, `POST` and `Patch` REST methods for the `action` APIs
are forbidden and not implemented. They are now now removed from
the API controller.

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

@@ -55,15 +55,12 @@ possible to :ref:`develop new implementations <implement_action_plugin>` which
are dynamically loaded by Watcher at launch time.
"""
from http import HTTPStatus
from oslo_utils import timeutils
import pecan
from pecan import rest
import wsme
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
from watcher._i18n import _
from watcher.api.controllers import base
from watcher.api.controllers import link
from watcher.api.controllers.v1 import collection
@@ -362,78 +359,3 @@ class ActionsController(rest.RestController):
policy.enforce(context, 'action:get', action, action='action:get')
return Action.convert_with_links(action)
@wsme_pecan.wsexpose(Action, body=Action, status_code=HTTPStatus.CREATED)
def post(self, action):
"""Create a new action(forbidden).
:param action: a action within the request body.
"""
# FIXME: blueprint edit-action-plan-flow
raise exception.OperationNotPermitted(
_("Cannot create an action directly"))
if self.from_actions:
raise exception.OperationNotPermitted
action_dict = action.as_dict()
context = pecan.request.context
new_action = objects.Action(context, **action_dict)
new_action.create()
# Set the HTTP Location Header
pecan.response.location = link.build_url('actions', new_action.uuid)
return Action.convert_with_links(new_action)
@wsme.validate(types.uuid, [ActionPatchType])
@wsme_pecan.wsexpose(Action, types.uuid, body=[ActionPatchType])
def patch(self, action_uuid, patch):
"""Update an existing action(forbidden).
:param action_uuid: UUID of a action.
:param patch: a json PATCH document to apply to this action.
"""
# FIXME: blueprint edit-action-plan-flow
raise exception.OperationNotPermitted(
_("Cannot modify an action directly"))
if self.from_actions:
raise exception.OperationNotPermitted
action_to_update = objects.Action.get_by_uuid(pecan.request.context,
action_uuid)
try:
action_dict = action_to_update.as_dict()
action = Action(**api_utils.apply_jsonpatch(action_dict, patch))
except api_utils.JSONPATCH_EXCEPTIONS as e:
raise exception.PatchError(patch=patch, reason=e)
# Update only the fields that have changed
for field in objects.Action.fields:
try:
patch_val = getattr(action, field)
except AttributeError:
# Ignore fields that aren't exposed in the API
continue
if patch_val == wtypes.Unset:
patch_val = None
if action_to_update[field] != patch_val:
action_to_update[field] = patch_val
action_to_update.save()
return Action.convert_with_links(action_to_update)
@wsme_pecan.wsexpose(None, types.uuid, status_code=HTTPStatus.NO_CONTENT)
def delete(self, action_uuid):
"""Delete a action(forbidden).
:param action_uuid: UUID of a action.
"""
# FIXME: blueprint edit-action-plan-flow
raise exception.OperationNotPermitted(
_("Cannot delete an action directly"))
action_to_delete = objects.Action.get_by_uuid(
pecan.request.context,
action_uuid)
action_to_delete.soft_delete()

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

@@ -10,9 +10,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import datetime
import itertools
from unittest import mock
from http import HTTPStatus
from oslo_config import cfg
@@ -21,7 +19,6 @@ from wsme import types as wtypes
from watcher.api.controllers.v1 import action as api_action
from watcher.common import utils
from watcher.db import api as db_api
from watcher import objects
from watcher.tests.api import base as api_base
from watcher.tests.api import utils as api_utils
@@ -459,70 +456,6 @@ class TestListAction(api_base.FunctionalTest):
self.assertEqual(3, len(response['actions']))
class TestPatch(api_base.FunctionalTest):
def setUp(self):
super(TestPatch, self).setUp()
obj_utils.create_test_goal(self.context)
obj_utils.create_test_strategy(self.context)
obj_utils.create_test_audit(self.context)
obj_utils.create_test_action_plan(self.context)
self.action = obj_utils.create_test_action(self.context, parents=None)
p = mock.patch.object(db_api.BaseConnection, 'update_action')
self.mock_action_update = p.start()
self.mock_action_update.side_effect = self._simulate_rpc_action_update
self.addCleanup(p.stop)
def _simulate_rpc_action_update(self, action):
action.save()
return action
@mock.patch('oslo_utils.timeutils.utcnow')
def test_patch_not_allowed(self, mock_utcnow):
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
new_state = objects.audit.State.SUCCEEDED
response = self.get_json('/actions/%s' % self.action.uuid)
self.assertNotEqual(new_state, response['state'])
response = self.patch_json(
'/actions/%s' % self.action.uuid,
[{'path': '/state', 'value': new_state, 'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertTrue(response.json['error_message'])
class TestDelete(api_base.FunctionalTest):
def setUp(self):
super(TestDelete, self).setUp()
self.goal = obj_utils.create_test_goal(self.context)
self.strategy = obj_utils.create_test_strategy(self.context)
self.audit = obj_utils.create_test_audit(self.context)
self.action_plan = obj_utils.create_test_action_plan(self.context)
self.action = obj_utils.create_test_action(self.context, parents=None)
p = mock.patch.object(db_api.BaseConnection, 'update_action')
self.mock_action_update = p.start()
self.mock_action_update.side_effect = self._simulate_rpc_action_update
self.addCleanup(p.stop)
def _simulate_rpc_action_update(self, action):
action.save()
return action
@mock.patch('oslo_utils.timeutils.utcnow')
def test_delete_action_not_allowed(self, mock_utcnow):
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
response = self.delete('/actions/%s' % self.action.uuid,
expect_errors=True)
self.assertEqual(HTTPStatus.FORBIDDEN, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
class TestActionPolicyEnforcement(api_base.FunctionalTest):
def setUp(self):