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

Drops forbidden patch/delete/post action apis

Browse Source 
These apis are not implemented with in the watcher code base and
was marked as a forbidden to use.

It does not make sense to keep these api as they are not implemented.
This cr drops the code around that to make the action apis cleaner.

Closes-Bug: #2110895

Change-Id: I0f465157e6cd481b27665ca6016db68c198cebeb
Signed-off-by: Chandan Kumar (raukadah) <chkumar@redhat.com>
This commit is contained in:
Chandan Kumar (raukadah)
2025-07-03 13:23:18 +05:30
parent bbe30f93f2
commit c0a5abe29c
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):