tower-cli/tower_cli/resources/schedule.py

# Copyright 2016, Ansible by Red Hat.
# Aaron Tan <sitan@redhat.com>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import click

from tower_cli import models, exceptions as exc
from tower_cli.cli import types


UNIFIED_JT = {
    'job_template': '/job_templates',
    'inventory_source': '/inventory_sources',
    'project': '/projects',
    'workflow': '/workflow_job_templates',
}
CLICK_ATTRS = ('__click_params__', '_cli_command', '_cli_command_attrs')


def jt_aggregate(func, is_create=False, has_pk=False):
    """Decorator to aggregate unified_jt-related fields.

    Args:
        func: The CURD method to be decorated.
        is_create: Boolean flag showing whether this method is create.
        has_pk: Boolean flag showing whether this method uses pk as argument.

    Returns:
        A function with necessary click-related attributes whose keyworded
        arguments are aggregated.

    Raises:
        exc.UsageError: Either more than one unified jt fields are
            provided, or none is provided when is_create flag is set.
    """
    def helper(kwargs, obj):
        """The helper function preceding actual function that aggregates
        unified jt fields.
        """
        unified_job_template = None
        for item in UNIFIED_JT:
            if kwargs.get(item, None) is not None:
                jt_id = kwargs.pop(item)
                if unified_job_template is None:
                    unified_job_template = (item, jt_id)
                else:
                    raise exc.UsageError(
                        'More than one unified job template fields provided, '
                        'please tighten your criteria.'
                    )
        if unified_job_template is not None:
            kwargs['unified_job_template'] = unified_job_template[1]
            obj.identity = tuple(list(obj.identity) + ['unified_job_template'])
            return '/'.join([UNIFIED_JT[unified_job_template[0]],
                             str(unified_job_template[1]), 'schedules/'])
        elif is_create:
            raise exc.UsageError('You must provide exactly one unified job'
                                 ' template field during creation.')

    def decorator_without_pk(obj, *args, **kwargs):
        old_endpoint = obj.endpoint
        new_endpoint = helper(kwargs, obj)
        if is_create:
            obj.endpoint = new_endpoint
        result = func(obj, *args, **kwargs)
        obj.endpoint = old_endpoint
        return result

    def decorator_with_pk(obj, pk=None, *args, **kwargs):
        old_endpoint = obj.endpoint
        new_endpoint = helper(kwargs, obj)
        if is_create:
            obj.endpoint = new_endpoint
        result = func(obj, pk=pk, *args, **kwargs)
        obj.endpoint = old_endpoint
        return result

    decorator = decorator_with_pk if has_pk else decorator_without_pk
    for item in CLICK_ATTRS:
        setattr(decorator, item, getattr(func, item, []))
    decorator.__doc__ = func.__doc__

    return decorator


class Resource(models.Resource):
    """A resource for schedules."""
    cli_help = 'Manage schedules within Ansible Tower.'
    endpoint = '/schedules/'

    # General fields.
    name = models.Field(unique=True)
    description = models.Field(required=False, display=False)

    # Unified jt fields. note these fields will only be used during creation.
    # Plus, one and only one field should be provided.
    job_template = models.Field(type=types.Related('job_template'),
                                required=False, display=False)
    inventory_source = models.Field(type=types.Related('inventory_source'),
                                    required=False, display=False)
    project = models.Field(type=types.Related('project'), required=False,
                           display=False)
    workflow = models.Field(
        type=types.Related('workflow'), required=False, display=False
    )

    # Schedule-specific fields.
    unified_job_template = models.Field(required=False, type=int,
                                        help_text='Integer used to display'
                                        ' unified job template in result, '
                                        'Do not use it for create/'
                                        'modify.')
    enabled = models.Field(required=False, type=click.BOOL, default=True,
                           help_text='Whether this schedule will be used.',
                           show_default=True)
    rrule = models.Field(required=False, display=False,
                         help_text='Schedule rules specifications which is'
                         ' less than 255 characters.')

    # Prompts
    extra_data = models.Field(type=types.Variables(), required=False,
                              display=False, help_text='Extra data for '
                              'schedule rules in the form of a .json file.')
    inventory = models.Field(
        type=types.Related('inventory'), required=False, display=False)
    credentials = models.ManyToManyField('credential')
    job_type = models.Field(required=False, display=False)
    job_tags = models.Field(required=False, display=False)
    skip_tags = models.Field(required=False, display=False)
    limit = models.Field(required=False, display=False)
    diff_mode = models.Field(type=bool, required=False, display=False)
    verbosity = models.Field(
        display=False,
        type=types.MappedChoice([
            (0, 'default'),
            (1, 'verbose'),
            (2, 'more_verbose'),
            (3, 'debug'),
            (4, 'connection'),
            (5, 'winrm'),
        ]),
        required=False,
    )

    def _get_patch_url(self, url, pk):
        urlTokens = url.split('/')
        if len(urlTokens) > 3:
            # reconstruct url to prevent a rare corner case where resources
            # cannot be constructed independently. Open to modification if
            # API convention changes.
            url = '/'.join(urlTokens[:1] + urlTokens[-2:])
        return super(Resource, self)._get_patch_url(url, pk)


Resource.create = jt_aggregate(Resource.create, is_create=True)
Resource.delete = jt_aggregate(Resource.delete, has_pk=True)
Resource.get = jt_aggregate(Resource.get, has_pk=True)
Resource.list = jt_aggregate(Resource.list)
Resource.modify = jt_aggregate(Resource.modify, has_pk=True)