Method clone

st2api/st2api/controllers/v1/actions.py:293–431

Clone an action from source pack to destination pack. Handles requests: POST /actions/{ref_or_id}/clone

(self, dest_data, ref_or_id, requester_user)

        return Response(status=http_client.NO_CONTENT)

    def clone(self, dest_data, ref_or_id, requester_user):
        """
        Clone an action from source pack to destination pack.
        Handles requests:
            POST /actions/{ref_or_id}/clone
        """

        source_action_db = self._get_by_ref_or_id(ref_or_id=ref_or_id)
        if not source_action_db:
            msg = "The requested source for cloning operation doesn't exists"
            abort(http_client.BAD_REQUEST, six.text_type(msg))

        extra = {"action_db": source_action_db}
        LOG.audit(
            "Source action found. Action.id=%s" % (source_action_db.id), extra=extra
        )

        try:
            permission_type = PermissionType.ACTION_VIEW
            rbac_utils = get_rbac_backend().get_utils_class()
            rbac_utils.assert_user_has_resource_db_permission(
                user_db=requester_user,
                resource_db=source_action_db,
                permission_type=permission_type,
            )
        except ResourceAccessDeniedError as e:
            abort(http_client.UNAUTHORIZED, six.text_type(e))

        cloned_dest_action_db = clone_action_db(
            source_action_db=source_action_db,
            dest_pack=dest_data.dest_pack,
            dest_action=dest_data.dest_action,
        )

        cloned_action_api = ActionAPI.from_model(cloned_dest_action_db)

        try:
            permission_type = PermissionType.ACTION_CREATE
            rbac_utils.assert_user_has_resource_api_permission(
                user_db=requester_user,
                resource_api=cloned_action_api,
                permission_type=permission_type,
            )
        except ResourceAccessDeniedError as e:
            abort(http_client.UNAUTHORIZED, six.text_type(e))

        dest_pack_base_path = get_pack_base_path(pack_name=dest_data.dest_pack)

        if not os.path.isdir(dest_pack_base_path):
            msg = "Destination pack '%s' doesn't exist" % (dest_data.dest_pack)
            abort(http_client.BAD_REQUEST, six.text_type(msg))

        dest_pack_base_path = get_pack_base_path(pack_name=dest_data.dest_pack)
        dest_ref = ".".join([dest_data.dest_pack, dest_data.dest_action])
        dest_action_db = self._get_by_ref(resource_ref=dest_ref)

        try:
            validate_not_part_of_system_pack_by_name(dest_data.dest_pack)