openstack/heat
Code Issues Proposed changes

Merge "Add remaining_executions attribute to OS::Mistral::CronTrigger"

Browse Source
This commit is contained in:
Jenkins 2015-05-12 11:28:09 +00:00 committed by Gerrit Code Review
commit 25659b9d11
2 changed files with 15 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
contrib/heat_mistral/heat_mistral/resources/cron_trigger.py
View File

@ -34,9 +34,9 @@ class CronTrigger(resource.Resource):
)
ATTRIBUTES = (
NEXT_EXECUTION_TIME,
NEXT_EXECUTION_TIME, REMAINING_EXECUTIONS
) = (
'next_execution_time',
'next_execution_time', 'remaining_executions'
)
properties_schema = {
@ -75,8 +75,11 @@ class CronTrigger(resource.Resource):
attributes_schema = {
NEXT_EXECUTION_TIME: attributes.Schema(
_('Time of the next execution in format "YYYY-MM-DD HH:MM".')
_('Time of the next execution in format "YYYY-MM-DD HH:MM:SS".')
),
REMAINING_EXECUTIONS: attributes.Schema(
_('Number of remaining executions.')
)
}
default_client_name = 'mistral'
@ -108,9 +111,15 @@ class CronTrigger(resource.Resource):
self.client_plugin().ignore_not_found(ex)
def _resolve_attribute(self, name):
if name == self.NEXT_EXECUTION_TIME:
try:
trigger = self.client().cron_triggers.get(self.resource_id)
except Exception as ex:
self.client_plugin().ignore_not_found(ex)
return ''
if name == self.NEXT_EXECUTION_TIME:
return trigger.next_execution_time
elif name == self.REMAINING_EXECUTIONS:
return trigger.remaining_executions
def resource_mapping():

2
contrib/heat_mistral/heat_mistral/tests/test_cron_trigger.py
View File

@ -40,6 +40,7 @@ class FakeCronTrigger(object):
def __init__(self, name):
self.name = name
self.next_execution_time = '2015-03-01 00:00:00'
self.remaining_executions = 3
class CronTriggerTest(common.HeatTestCase):
@ -89,6 +90,7 @@ class CronTriggerTest(common.HeatTestCase):
ct = self._create_resource('trigger', self.rsrc_defn, self.stack)
self.assertEqual('2015-03-01 00:00:00',
ct.FnGetAtt('next_execution_time'))
self.assertEqual(3, ct.FnGetAtt('remaining_executions'))
def test_delete(self):
ct = self._create_resource('trigger', self.rsrc_defn, self.stack)