openstack
/
openstacksdk
Code Issues Proposed changes

Merge "Cron Triggers proxy"

This commit is contained in:
Zuul 2022-11-21 20:25:52 +00:00 committed by Gerrit Code Review
parent 75904fbdb1 d9ac526933
commit 77c9e919a2
8 changed files with 276 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
doc/source/user/proxies/workflow.rst
View File

@ -25,3 +25,11 @@ Execution Operations
:noindex:
:members: create_execution, delete_execution, get_execution,
find_execution, executions
Cron Trigger Operations
^^^^^^^^^^^^^^^^^^^^^^^
.. autoclass:: openstack.workflow.v2._proxy.Proxy
:noindex:
:members: create_cron_trigger, delete_cron_trigger, get_cron_trigger,
find_cron_trigger, cron_triggers

1
doc/source/user/resources/workflow/index.rst
View File

@ -6,3 +6,4 @@ Workflow Resources
v2/execution
v2/workflow
v2/crontrigger

12
doc/source/user/resources/workflow/v2/crontrigger.rst Normal file
View File

@ -0,0 +1,12 @@
openstack.workflow.v2.cron_trigger
==================================
.. automodule:: openstack.workflow.v2.cron_trigger
The CronTrigger Class
---------------------
The ``CronTrigger`` class inherits from :class:`~openstack.resource.Resource`.
.. autoclass:: openstack.workflow.v2.cron_trigger.CronTrigger
:members:

88
openstack/tests/unit/workflow/test_cron_trigger.py Normal file
View File

@ -0,0 +1,88 @@
# 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.
from openstack.tests.unit import base
from openstack.workflow.v2 import cron_trigger
FAKE_INPUT = {
'cluster_id': '8c74607c-5a74-4490-9414-a3475b1926c2',
'node_id': 'fba2cc5d-706f-4631-9577-3956048d13a2',
'flavor_id': '1'
}
FAKE_PARAMS = {}
FAKE = {
'id': 'ffaed25e-46f5-4089-8e20-b3b4722fd597',
'pattern': '0 * * * *',
'remaining_executions': 14,
'first_execution_time': '1970-01-01T01:00:00.000000',
'next_execution_time': '1970-01-01T02:00:00.000000',
'workflow_name': 'cluster-coldmigration',
'workflow_id': '1995cf40-c22d-4968-b6e8-558942830642',
'workflow_input': FAKE_INPUT,
'workflow_params': FAKE_PARAMS,
}
class TestCronTrigger(base.TestCase):
def test_basic(self):
sot = cron_trigger.CronTrigger()
self.assertEqual('cron_trigger', sot.resource_key)
self.assertEqual('cron_triggers', sot.resources_key)
self.assertEqual('/cron_triggers', sot.base_path)
self.assertTrue(sot.allow_fetch)
self.assertTrue(sot.allow_list)
self.assertTrue(sot.allow_create)
self.assertTrue(sot.allow_delete)
self.assertDictEqual(
{
'marker': 'marker',
'limit': 'limit',
'sort_keys': 'sort_keys',
'sort_dirs': 'sort_dirs',
'fields': 'fields',
'name': 'name',
'workflow_name': 'workflow_name',
'workflow_id': 'workflow_id',
'workflow_input': 'workflow_input',
'workflow_params': 'workflow_params',
'scope': 'scope',
'pattern': 'pattern',
'remaining_executions': 'remaining_executions',
'project_id': 'project_id',
'first_execution_time': 'first_execution_time',
'next_execution_time': 'next_execution_time',
'created_at': 'created_at',
'updated_at': 'updated_at',
'all_projects': 'all_projects',
},
sot._query_mapping._mapping
)
def test_make_it(self):
sot = cron_trigger.CronTrigger(**FAKE)
self.assertEqual(FAKE['id'], sot.id)
self.assertEqual(FAKE['pattern'], sot.pattern)
self.assertEqual(FAKE['remaining_executions'],
sot.remaining_executions)
self.assertEqual(FAKE['first_execution_time'],
sot.first_execution_time)
self.assertEqual(FAKE['next_execution_time'],
sot.next_execution_time)
self.assertEqual(FAKE['workflow_name'], sot.workflow_name)
self.assertEqual(FAKE['workflow_id'], sot.workflow_id)
self.assertEqual(FAKE['workflow_input'], sot.workflow_input)
self.assertEqual(FAKE['workflow_params'], sot.workflow_params)

27
openstack/tests/unit/workflow/v2/test_proxy.py
View File

@ -12,6 +12,7 @@
from openstack.tests.unit import test_proxy_base
from openstack.workflow.v2 import _proxy
from openstack.workflow.v2 import cron_trigger
from openstack.workflow.v2 import execution
from openstack.workflow.v2 import workflow
@ -60,3 +61,29 @@ class TestWorkflowProxy(test_proxy_base.TestProxyBase):
def test_execution_find(self):
self.verify_find(self.proxy.find_execution,
execution.Execution)
class TestCronTriggerProxy(test_proxy_base.TestProxyBase):
def setUp(self):
super().setUp()
self.proxy = _proxy.Proxy(self.session)
def test_cron_triggers(self):
self.verify_list(self.proxy.cron_triggers,
cron_trigger.CronTrigger)
def test_cron_trigger_get(self):
self.verify_get(self.proxy.get_cron_trigger,
cron_trigger.CronTrigger)
def test_cron_trigger_create(self):
self.verify_create(self.proxy.create_cron_trigger,
cron_trigger.CronTrigger)
def test_cron_trigger_delete(self):
self.verify_delete(self.proxy.delete_cron_trigger,
cron_trigger.CronTrigger, True)
def test_cron_trigger_find(self):
self.verify_find(self.proxy.find_cron_trigger,
cron_trigger.CronTrigger)

75
openstack/workflow/v2/_proxy.py
View File

@ -11,6 +11,7 @@
# under the License.
from openstack import proxy
from openstack.workflow.v2 import cron_trigger as _cron_trigger
from openstack.workflow.v2 import execution as _execution
from openstack.workflow.v2 import workflow as _workflow
@ -170,3 +171,77 @@ class Proxy(proxy.Proxy):
"""
return self._find(_execution.Execution, name_or_id,
ignore_missing=ignore_missing)
def create_cron_trigger(self, **attrs):
"""Create a new cron trigger from attributes
:param dict attrs: Keyword arguments which will be used to create
a :class:`~openstack.workflow.v2.cron_trigger.CronTrigger`,
comprised of the properties on the CronTrigger class.
:returns: The results of cron trigger creation
:rtype: :class:`~openstack.workflow.v2.cron_trigger.CronTrigger`
"""
return self._create(_cron_trigger.CronTrigger, **attrs)
def get_cron_trigger(self, cron_trigger):
"""Get a cron trigger
:param cron_trigger: The value can be the name of a cron_trigger or
:class:`~openstack.workflow.v2.cron_trigger.CronTrigger` instance.
:returns: One :class:`~openstack.workflow.v2.cron_trigger.CronTrigger`
:raises: :class:`~openstack.exceptions.ResourceNotFound` when no
cron triggers matching the criteria could be found.
"""
return self._get(_cron_trigger.CronTrigger, cron_trigger)
def cron_triggers(self, **query):
"""Retrieve a generator of cron triggers
:param kwargs query: Optional query parameters to be sent to
restrict the cron triggers to be returned. Available parameters
include:
* limit: Requests at most the specified number of items be
returned from the query.
* marker: Specifies the ID of the last-seen cron trigger. Use the
limit parameter to make an initial limited request and use
the ID of the last-seen cron trigger from the response as the
marker parameter value in a subsequent limited request.
:returns: A generator of CronTrigger instances.
"""
return self._list(_cron_trigger.CronTrigger, **query)
def delete_cron_trigger(self, value, ignore_missing=True):
"""Delete a cron trigger
:param value: The value can be either the name of a cron trigger or a
:class:`~openstack.workflow.v2.cron_trigger.CronTrigger`
instance.
:param bool ignore_missing: When set to ``False``
:class:`~openstack.exceptions.ResourceNotFound` will be
raised when the cron trigger does not exist.
When set to ``True``, no exception will be set when
attempting to delete a nonexistent cron trigger.
:returns: ``None``
"""
return self._delete(_cron_trigger.CronTrigger, value,
ignore_missing=ignore_missing)
def find_cron_trigger(self, name_or_id, ignore_missing=True):
"""Find a single cron trigger
:param name_or_id: The name or ID of a cron trigger.
:param bool ignore_missing: When set to ``False``
:class:`~openstack.exceptions.ResourceNotFound` will be
raised when the resource does not exist.
When set to ``True``, None will be returned when
attempting to find a nonexistent resource.
:returns: One :class:`~openstack.compute.v2.cron_trigger.CronTrigger`
or None
"""
return self._find(_cron_trigger.CronTrigger, name_or_id,
ignore_missing=ignore_missing)

61
openstack/workflow/v2/cron_trigger.py Normal file
View File

@ -0,0 +1,61 @@
# 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.
from openstack import resource
class CronTrigger(resource.Resource):
resource_key = 'cron_trigger'
resources_key = 'cron_triggers'
base_path = '/cron_triggers'
# capabilities
allow_create = True
allow_list = True
allow_fetch = True
allow_delete = True
_query_mapping = resource.QueryParameters(
'marker', 'limit', 'sort_keys', 'sort_dirs', 'fields', 'name',
'workflow_name', 'workflow_id', 'workflow_input', 'workflow_params',
'scope', 'pattern', 'remaining_executions', 'project_id',
'first_execution_time', 'next_execution_time', 'created_at',
'updated_at', 'all_projects')
#: The name of this Cron Trigger
name = resource.Body("name")
#: The pattern for this Cron Trigger
pattern = resource.Body("pattern")
#: Count of remaining exectuions
remaining_executions = resource.Body("remaining_executions")
#: Time of the first execution
first_execution_time = resource.Body("first_execution_time")
#: Time of the next execution
next_execution_time = resource.Body("next_execution_time")
#: Workflow name
workflow_name = resource.Body("workflow_name")
#: Workflow ID
workflow_id = resource.Body("workflow_id")
#: The inputs for Workflow
workflow_input = resource.Body("workflow_input")
#: Workflow params
workflow_params = resource.Body("workflow_params")
#: The ID of the associated project
project_id = resource.Body("project_id")
#: The time at which the cron trigger was created
created_at = resource.Body("created_at")
#: The time at which the cron trigger was created
updated_at = resource.Body("updated_at")
def create(self, session, base_path=None):
return super(CronTrigger, self).create(
session, prepend_key=False, base_path=base_path)

4
releasenotes/notes/cron_triggers_proxy-51aa89e91bbb9798.yaml Normal file
View File

@ -0,0 +1,4 @@
---
features:
- |
Add workflow CronTrigger resource and proxy methods.