opendev/storyboard
Code Issues Proposed changes

Replaced python-crontab with apscheduler

Browse Source 
python-crontab is GPL'd, so we can't use it. This replaces all
related functionality with apscheduler. Notable API changes: Since
the APScheduler guarantees single execution environments, we no
longer have to provide execution time windows for our plugins.

Note, pymysql 0.6.4 seems to have issues with some unicode characters.

Change-Id: Ie8b3473ac316c8c661f7ffe1cdf069e7e822f23b
This commit is contained in:
Michael Krotscheck 2015-02-24 12:23:31 -08:00
parent ee7a86c03f
commit 1d8bd9eb9e
19 changed files with 557 additions and 928 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
etc/storyboard.conf.sample
View File

@ -63,10 +63,10 @@ lock_path = $state_path/lock
# A list of valid client id's that may connect to StoryBoard.
# valid_oauth_clients = storyboard.openstack.org, localhost
[cron]
# Storyboard's cron management configuration
[scheduler]
# Storyboard's scheduled task management configuration
# Enable or disable cron (Default disabled)
# Enable or disable scheduling (Default disabled)
# enable = true
[cors]
@ -148,7 +148,7 @@ lock_path = $state_path/lock
# pool_timeout = 10
[plugin_token_cleaner]
# Enable/Disable the token cleaning cron plugin. This requires cron
# Enable/Disable the periodic token cleaner plugin. This requires scheduled
# management to be enabled.
# enable = True

2
requirements-py3.txt
View File

@ -22,7 +22,7 @@ sqlalchemy-migrate>=0.9.1,!=0.9.2
SQLAlchemy-FullText-Search
eventlet>=0.13.0
stevedore>=1.0.0
python-crontab>=1.8.1
tzlocal>=1.1.2
Jinja2>=2.7.3
PyMySQL>=0.6.2,!=0.6.4
apscheduler>=3.0.1

2
requirements.txt
View File

@ -22,8 +22,8 @@ sqlalchemy-migrate>=0.9.1,!=0.9.2
SQLAlchemy-FullText-Search
eventlet>=0.13.0
stevedore>=1.0.0
python-crontab>=1.8.1
tzlocal>=1.1.2
email>=4.0.2
Jinja2>=2.7.3
PyMySQL>=0.6.2,!=0.6.4
apscheduler>=3.0.1

3
setup.cfg
View File

@ -37,8 +37,7 @@ console_scripts =
storyboard.worker.task =
subscription = storyboard.worker.task.subscription:Subscription
storyboard.plugin.user_preferences =
storyboard.plugin.cron =
cron-management = storyboard.plugin.cron.manager:CronManager
storyboard.plugin.scheduler =
token-cleaner = storyboard.plugin.token_cleaner.cleaner:TokenCleaner
[build_sphinx]

6
storyboard/api/app.py
View File

@ -30,7 +30,7 @@ from storyboard.api.v1.search import impls as search_engine_impls
from storyboard.api.v1.search import search_engine
from storyboard.notifications.notification_hook import NotificationHook
from storyboard.openstack.common.gettextutils import _LI # noqa
from storyboard.plugin.cron import load_crontab
from storyboard.plugin.scheduler import initialize_scheduler
from storyboard.plugin.user_preferences import initialize_user_preferences
CONF = cfg.CONF
@ -97,8 +97,8 @@ def setup_app(pecan_config=None):
# Load user preference plugins
initialize_user_preferences()
# Initialize crontab
load_crontab()
# Initialize scheduler
initialize_scheduler()
# Setup notifier
if CONF.enable_notifications:

78
storyboard/plugin/cron/__init__.py
View File

@ -1,78 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 atexit
from oslo_log import log
from oslo.config import cfg
from storyboard.plugin.base import StoryboardPluginLoader
from storyboard.plugin.cron.manager import CronManager
LOG = log.getLogger(__name__)
CONF = cfg.CONF
CRON_OPTS = [
cfg.StrOpt("plugin",
default="storyboard.plugin.cron.manager:CronManager",
help="The name of the cron plugin to execute.")
]
def main():
"""Run a specific cron plugin from the commandline. Used by the system's
crontab to target different plugins on different execution intervals.
"""
CONF.register_cli_opts(CRON_OPTS)
try:
log.register_options(CONF)
except cfg.ArgsAlreadyParsedError:
pass
CONF(project='storyboard')
log.setup(CONF, 'storyboard')
loader = StoryboardPluginLoader(namespace="storyboard.plugin.cron")
if loader.extensions:
loader.map(execute_plugin, CONF.plugin)
def execute_plugin(ext, name):
"""Private handler method that checks individual loaded plugins.
"""
plugin_name = ext.obj.get_name()
if name == plugin_name:
LOG.info("Executing cron plugin: %s" % (plugin_name,))
ext.obj.execute()
def load_crontab():
"""Initialize all registered crontab plugins."""
# We cheat here - crontab plugin management is implemented as a crontab
# plugin itself, so we create a single instance to kick things off,
# which will then add itself to recheck periodically.
manager_plugin = CronManager(CONF)
if manager_plugin.enabled():
manager_plugin.execute()
atexit.register(unload_crontab, manager_plugin)
else:
unload_crontab(manager_plugin)
def unload_crontab(manager_plugin):
manager_plugin.remove()

128
storyboard/plugin/cron/base.py
View File

@ -1,128 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 abc
import calendar
import datetime
import os
import pytz
import six
from oslo_log import log
from storyboard.common.working_dir import get_plugin_directory
import storyboard.plugin.base as plugin_base
LOG = log.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class CronPluginBase(plugin_base.PluginBase):
"""Base class for a plugin that executes business logic on a time
interval. In order to prevent processing overlap on long-running
processes that may exceed the tick interval, the plugin will be provided
with the time range for which it is responsible.
It is likely that multiple instances of a plugin may be running
simultaneously, as a previous execution may not have finished processing
by the time the next one is started. Please ensure that your plugin
operates in a time bounded, thread safe manner.
"""
@abc.abstractmethod
def run(self, start_time, end_time):
"""Execute a periodic task.
:param start_time: The last time the plugin was run.
:param end_time: The current timestamp.
:return: Nothing.
"""
def _get_file_mtime(self, path, date=None):
"""Retrieve the date of this plugin's last_run file. If a date is
provided, it will also update the file's date before returning that
date.
:param path: The path of the file to retreive.
:param date: A datetime to use to set as the mtime of the file.
:return: The mtime of the file.
"""
# Get our timezones.
utc_tz = pytz.utc
# If the file doesn't exist, create it with a sane base time.
if not os.path.exists(path):
base_time = datetime.datetime \
.utcfromtimestamp(0) \
.replace(tzinfo=utc_tz)
with open(path, 'a'):
base_timestamp = calendar.timegm(base_time.timetuple())
os.utime(path, (base_timestamp, base_timestamp))
# If a date was passed, use it to update the file.
if date:
# If the date does not have a timezone, throw an exception.
# That's bad practice and makes our date/time conversions
# impossible.
if not date.tzinfo:
raise TypeError("Please include a timezone when passing"
" datetime instances")
with open(path, 'a'):
mtimestamp = calendar.timegm(date
.astimezone(utc_tz)
.timetuple())
os.utime(path, (mtimestamp, mtimestamp))
# Retrieve the file's last mtime.
pid_info = os.stat(path)
return datetime.datetime \
.fromtimestamp(pid_info.st_mtime, utc_tz)
def execute(self):
"""Execute this cron plugin, first by determining its own working
directory, then calculating the appropriate runtime interval,
and finally executing the run() method. If the working directory is
not available, it will log an error and exit cleanly.
"""
plugin_name = self.get_name()
try:
cron_directory = get_plugin_directory('cron')
except IOError as e:
LOG.error('Cannot create cron run cache: %s' % (e,))
return
lr_file = os.path.join(cron_directory, plugin_name)
now = datetime.datetime.now(pytz.utc)
start_time = self._get_file_mtime(path=lr_file)
end_time = self._get_file_mtime(path=lr_file,
date=now)
self.run(start_time, end_time)
@abc.abstractmethod
def interval(self):
"""The plugin's cron interval, as a string.
:return: The cron interval. Example: "* * * * *"
"""
def get_name(self):
"""A simple name for this plugin."""
return self.__module__ + ":" + self.__class__.__name__

146
storyboard/plugin/cron/manager.py
View File

@ -1,146 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 crontab import CronTab
from oslo.config import cfg
from oslo_log import log
from storyboard.common.working_dir import get_working_directory
from storyboard.plugin.base import StoryboardPluginLoader
from storyboard.plugin.cron.base import CronPluginBase
LOG = log.getLogger(__name__)
CRON_MANAGEMENT_OPTS = [
cfg.BoolOpt('enable',
default=False,
help='Enable StoryBoard\'s Crontab management.')
]
CONF = cfg.CONF
CONF.register_opts(CRON_MANAGEMENT_OPTS, 'cron')
class CronManager(CronPluginBase):
"""A Cron Plugin serves both as the manager for other storyboard
cron tasks, and as an example plugin for storyboard. It checks every
5 minutes or so to see what storyboard cron plugins are registered vs.
running, and enables/disables them accordingly.
"""
def __init__(self, config, tabfile=None):
super(CronManager, self).__init__(config=config)
self.tabfile = tabfile
def enabled(self):
"""Indicate whether this plugin is enabled. This indicates whether
this plugin alone is runnable, as opposed to the entire cron system.
Note that this plugin cannot operate if the system cannot create a
working directory.
"""
try:
# This will raise an exception if the working directory cannot
# be created.
get_working_directory()
# Return the configured cron flag.
return self.config.cron.enable
except IOError as e:
LOG.error("Cannot enable crontab management: Working directory is"
" not available: %s" % (e,))
return False
def interval(self):
"""This plugin executes every 5 minutes.
:return: "*/5 * * * *"
"""
return "*/5 * * * *"
def run(self, start_time, end_time):
"""Execute a periodic task.
:param start_time: The last time the plugin was run.
:param end_time: The current timestamp.
"""
# First, go through the stevedore registration and update the plugins
# we know about.
loader = StoryboardPluginLoader(namespace="storyboard.plugin.cron")
handled_plugins = dict()
if loader.extensions:
loader.map(self._manage_plugins, handled_plugins)
# Now manually go through the cron list and remove anything that
# isn't registered.
cron = CronTab(tabfile=self.tabfile)
not_handled = lambda x: x.comment not in handled_plugins
jobs = filter(not_handled, cron.find_command('storyboard-cron'))
cron.remove(*jobs)
cron.write()
def _manage_plugins(self, ext, handled_plugins=dict()):
"""Adds a plugin instance to crontab."""
plugin = ext.obj
cron = CronTab(tabfile=self.tabfile)
plugin_name = plugin.get_name()
plugin_interval = plugin.interval()
command = "storyboard-cron --plugin %s" % (plugin_name,)
# Pull any existing jobs.
job = None
for item in cron.find_comment(plugin_name):
LOG.info("Found existing cron job: %s" % (plugin_name,))
job = item
job.set_command(command)
job.set_comment(plugin_name)
job.setall(plugin_interval)
break
if not job:
LOG.info("Adding cron job: %s" % (plugin_name,))
job = cron.new(command=command, comment=plugin_name)
job.setall(plugin_interval)
# Update everything.
job.set_command(command)
job.set_comment(plugin_name)
job.setall(plugin_interval)
# This code us usually not triggered, because the stevedore plugin
# loader harness already filters based on the results of the
# enabled() method, however we're keeping it in here since plugin
# loading and individual plugin functionality are independent, and may
# change independently.
if plugin.enabled():
job.enable()
else:
LOG.info("Disabled cron plugin: %s", (plugin_name,))
job.enable(False)
# Remember the state of this plugin
handled_plugins[plugin_name] = True
# Save it.
cron.write()
def remove(self):
"""Remove all storyboard cron extensions.
"""
# Flush all orphans
cron = CronTab(tabfile=self.tabfile)
cron.remove_all(command='storyboard-cron')
cron.write()

224
storyboard/plugin/scheduler/__init__.py Normal file
View File

@ -0,0 +1,224 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 atexit
import six
from oslo.config import cfg
from oslo_log import log
from pytz import utc
from apscheduler.executors.pool import ProcessPoolExecutor
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.base import BaseTrigger
from apscheduler.triggers.interval import IntervalTrigger
from storyboard.db.api.base import get_engine
from storyboard.plugin.base import StoryboardPluginLoader
LOG = log.getLogger(__name__)
CONF = cfg.CONF
SCHEDULER = None
SCHEDULE_MANAGER_ID = 'storyboard-scheduler-manager'
SCHEDULER_OPTS = [
cfg.BoolOpt("enable",
default=False,
help="Whether to enable the scheduler.")
]
CONF.register_opts(SCHEDULER_OPTS, 'scheduler')
def initialize_scheduler():
"""Initialize the task scheduler. This method configures the global
scheduler, checks the loaded tasks, and ensures they are all scheduled.
"""
global SCHEDULER
# If the scheduler is not enabled, clear it and exit. This prevents any
# unexpected database session issues.
if not CONF.scheduler.enable:
if SCHEDULER:
SCHEDULER.remove_all_jobs()
SCHEDULER = None
LOG.info("Scheduler is not enabled.")
return
# Use SQLAlchemy as a Job store.
jobstores = {
'default': SQLAlchemyJobStore(engine=get_engine())
}
# Two executors: The default is for all plugins, so that they load in a
# different process that does not impact the API. The second one is for
# the scheduler manager, which makes sure this scheduler instance is
# aware of all of our plugins.
executors = {
'default': ProcessPoolExecutor(10),
'manager': ThreadPoolExecutor(1),
}
# Allow executions to coalesce. See https://apscheduler.readthedocs.org/en
# /latest/userguide.html#missed-job-executions-and-coalescing
job_defaults = {
'coalesce': True,
'max_instances': 1,
'replace_existing': True
}
# This will automatically create the table.
SCHEDULER = BackgroundScheduler(jobstores=jobstores,
executors=executors,
job_defaults=job_defaults,
timezone=utc)
SCHEDULER.start()
atexit.register(shutdown_scheduler)
# Make sure we load in the update_scheduler job. If it exists,
# we remove/update it to make sure any code changes get propagated.
if SCHEDULER.get_job(SCHEDULE_MANAGER_ID):
SCHEDULER.remove_job(SCHEDULE_MANAGER_ID)
SCHEDULER.add_job(
update_scheduler,
id=SCHEDULE_MANAGER_ID,
trigger=IntervalTrigger(minutes=1),
executor='manager'
)
def shutdown_scheduler():
"""Shut down the scheduler. This method is registered using atexit,
and is run whenever the process in which initialize_scheduler() was
called ends.
"""
global SCHEDULER
if SCHEDULER:
LOG.info("Shutting down scheduler")
SCHEDULER.shutdown()
SCHEDULER = None
def update_scheduler():
"""Update the jobs loaded into the scheduler. This runs every minute to
keep track of anything that's since been loaded into our execution hooks.
"""
global SCHEDULER
if not SCHEDULER:
LOG.warning("Scheduler does not exist, cannot update it.")
return
# This may be running in a separate thread and/or process, so the log may
# not have been initialized.
try:
log.register_options(CONF)
except cfg.ArgsAlreadyParsedError:
pass
# Load all plugins that are registered and load them into the scheduler.
loader = StoryboardPluginLoader(namespace="storyboard.plugin.scheduler")
loaded_plugins = [SCHEDULE_MANAGER_ID]
if loader.extensions:
loader.map(add_plugins, loaded_plugins)
# Now manually go through the list of jobs in the scheduler and remove
# any that haven't been loaded, since some might have been uninstalled.
for job in SCHEDULER.get_jobs():
if job.id not in loaded_plugins:
LOG.info('Removing Job: %s' % (job.id,))
SCHEDULER.remove_job(job.id)
def add_plugins(ext, loaded_plugins=list()):
global SCHEDULER
if not SCHEDULER:
LOG.warn('Scheduler does not exist')
return
# Extract the plugin instance
plugin = ext.obj
# Get the plugin name
plugin_name = plugin.get_name()
# Plugin trigger object.
plugin_trigger = plugin.trigger()
# Plugin personal activation logic. This is duplicated from
# StoryboardPluginLoader, replicated here just in case that ever changes
# without us knowing about it.
plugin_enabled = plugin.enabled()
# Check to see if we have one of these loaded...
current_job = SCHEDULER.get_job(plugin_name)
# Assert that the trigger is of the correct type.
if not isinstance(plugin_trigger, BaseTrigger):
LOG.warn("Plugin does not provide BaseTrigger: %s" % (plugin_name,))
plugin_enabled = False
# If the plugin should be disabled, disable it, then exist.
if not plugin_enabled:
if current_job:
LOG.info("Disabling plugin: %s" % (plugin_name,))
SCHEDULER.remove_job(plugin_name)
return
# At this point we know it's loaded.
loaded_plugins.append(plugin_name)
# If it's already registered, check for a
if current_job:
# Reschedule if necessary. We're using a string comparison here
# because they're declarative for basic triggers, and because there's
# no other real good option.
if six.text_type(current_job.trigger) != six.text_type(plugin_trigger):
LOG.info('Rescheduling Job: %s' % (plugin_name,))
SCHEDULER.reschedule_job(plugin_name, trigger=plugin_trigger)
return
# At this point, load the plugin.
LOG.info('Adding job: %s' % (plugin_name,))
SCHEDULER.add_job(
execute_plugin,
args=[plugin.__class__],
id=plugin_name,
trigger=plugin_trigger,
executor='default'
)
def execute_plugin(plugin_class):
"""Run a specific cron plugin from the scheduler. Preloads our
environment, and then invokes the run method on the plugin. This will
only work properly if run in a separate process.
"""
try:
log.register_options(CONF)
except cfg.ArgsAlreadyParsedError:
pass
CONF(project='storyboard')
log.setup(CONF, 'storyboard')
plugin_instance = plugin_class(CONF)
LOG.info('Running plugin: %s' % (plugin_instance.get_name(),))
plugin_instance.run()
# This line is here for testability.
return plugin_instance

52
storyboard/plugin/scheduler/base.py Normal file
View File

@ -0,0 +1,52 @@
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# 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 abc
import six
from oslo_log import log
import storyboard.plugin.base as plugin_base
LOG = log.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class SchedulerPluginBase(plugin_base.PluginBase):
"""Base class for a plugin that executes business logic on a schedule.
All plugins are loaded into the scheduler in such a way that long-running
plugins will not cause multiple 'overlapping' executions.
"""
@abc.abstractmethod
def run(self):
"""Execute a periodic task. It is guaranteed that no more than one of
these will be run on any one storyboard instance. If you are running
multiple instances, that is not the case.
"""
@abc.abstractmethod
def trigger(self):
"""The plugin's scheduler trigger. Must implement BaseTrigger from
the apscheduler package.
:return: A trigger that describes the interval under which this
plugin should execute.
"""
def get_name(self):
"""A simple name for this plugin."""
return self.__module__ + ":" + self.__class__.__name__

20
storyboard/plugin/token_cleaner/cleaner.py
View File

@ -16,12 +16,14 @@ from datetime import datetime
from datetime import timedelta
import pytz
from apscheduler.triggers.interval import IntervalTrigger
import storyboard.db.api.base as api_base
from storyboard.db.models import AccessToken
from storyboard.plugin.cron.base import CronPluginBase
from storyboard.plugin.scheduler.base import SchedulerPluginBase
class TokenCleaner(CronPluginBase):
class TokenCleaner(SchedulerPluginBase):
"""A Cron Plugin which checks periodically for expired auth tokens and
removes them from the database. By default it only cleans up expired
tokens that are more than a week old, to permit some historical debugging
@ -36,18 +38,12 @@ class TokenCleaner(CronPluginBase):
return self.config.plugin_token_cleaner.enable or False
return False
def interval(self):
"""This plugin executes on startup, and once every hour after that.
def trigger(self):
"""This plugin executes every hour."""
return IntervalTrigger(hours=1, timezone=pytz.utc)
:return: "? * * * *"
"""
return "? * * * *"
def run(self, start_time, end_time):
def run(self):
"""Remove all oauth tokens that are more than a week old.
:param start_time: The last time the plugin was run.
:param end_time: The current timestamp.
"""
# Calculate last week.
lastweek = datetime.now(pytz.utc) - timedelta(weeks=1)

56
storyboard/tests/plugin/cron/mock_plugin.py
View File

@ -1,56 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 oslo.config import cfg
import storyboard.plugin.cron.base as plugin_base
CONF = cfg.CONF
class MockPlugin(plugin_base.CronPluginBase):
"""A mock cron plugin for testing."""
def __init__(self, config, is_enabled=True,
plugin_interval="0 0 1 1 0"):
"""Create a new instance of the base plugin, with some sane defaults.
The default cron interval is '0:00 on January 1st if a Sunday', which
should ensure that the manipulation of the cron environment on the test
machine does not actually execute anything.
"""
super(MockPlugin, self).__init__(config)
self.is_enabled = is_enabled
self.plugin_interval = plugin_interval
def enabled(self):
"""Return our enabled value."""
return self.is_enabled
def run(self, start_time, end_time):
"""Stores the data to a global variable so we can test it.
:param working_dir: Path to a working directory your plugin can use.
:param start_time: The last time the plugin was run.
:param end_time: The current timestamp.
:return: Nothing.
"""
self.last_invocation_parameters = (start_time, end_time)
def interval(self):
"""The plugin's cron interval, as a string.
:return: The cron interval. Example: "* * * * *"
"""
return self.plugin_interval

112
storyboard/tests/plugin/cron/test_base.py
View File

@ -1,112 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 calendar
import datetime
import os
import pytz
import tzlocal
import storyboard.common.working_dir as w_dir
import storyboard.tests.base as base
from storyboard.tests.plugin.cron.mock_plugin import MockPlugin
class TestCronPluginBase(base.WorkingDirTestCase):
"""Test the abstract plugin core."""
def setUp(self):
super(TestCronPluginBase, self).setUp()
# Create the stamp directory
cron_directory = w_dir.get_plugin_directory('cron')
if not os.path.exists(cron_directory):
os.makedirs(cron_directory)
def test_get_name(self):
"""Test that the plugin can name itself."""
plugin = MockPlugin(dict())
self.assertEqual("storyboard.tests.plugin.cron.mock_plugin:MockPlugin",
plugin.get_name())
def test_mtime(self):
"""Assert that the mtime utility function always returns UTC dates,
yet correctly translates dates to systime.
"""
sys_tz = tzlocal.get_localzone()
# Generate the plugin and build our file path
plugin = MockPlugin(dict())
plugin_name = plugin.get_name()
cron_directory = w_dir.get_plugin_directory('cron')
last_run_path = os.path.join(cron_directory, plugin_name)
# Call the mtime method, ensuring that it is created.
self.assertFalse(os.path.exists(last_run_path))
creation_mtime = plugin._get_file_mtime(last_run_path)
self.assertTrue(os.path.exists(last_run_path))
# Assert that the returned timezone is UTC.
self.assertEquals(pytz.utc, creation_mtime.tzinfo)
# Assert that the creation time equals UTC 0.
creation_time = calendar.timegm(creation_mtime.timetuple())
self.assertEqual(0, creation_time.real)
# Assert that we can update the time.
updated_mtime = datetime.datetime(year=2000, month=1, day=1, hour=1,
minute=1, second=1, tzinfo=pytz.utc)
updated_result = plugin._get_file_mtime(last_run_path, updated_mtime)
self.assertEqual(updated_mtime, updated_result)
updated_stat = os.stat(last_run_path)
updated_time_from_file = datetime.datetime \
.fromtimestamp(updated_stat.st_mtime, tz=sys_tz)
self.assertEqual(updated_mtime, updated_time_from_file)
# Assert that passing a system timezone datetime is still applicable
# and comparable.
updated_sysmtime = datetime.datetime(year=2000, month=1, day=1, hour=1,
minute=1, second=1,
tzinfo=sys_tz)
updated_sysresult = plugin._get_file_mtime(last_run_path,
updated_sysmtime)
self.assertEqual(updated_sysmtime, updated_sysresult)
self.assertEqual(pytz.utc, updated_sysresult.tzinfo)
def test_execute(self):
"""Assert that the public execution method correctly builds the
plugin API's input parameters.
"""
# Generate the plugin and simulate a previous execution
plugin = MockPlugin(dict())
plugin_name = plugin.get_name()
cron_directory = w_dir.get_plugin_directory('cron')
last_run_path = os.path.join(cron_directory, plugin_name)
last_run_date = datetime.datetime(year=2000, month=1, day=1,
hour=12, minute=0, second=0,
microsecond=0, tzinfo=pytz.utc)
plugin._get_file_mtime(last_run_path, last_run_date)
# Execute the plugin
plugin.execute()
# Current timestamp, remove microseconds so that we don't run into
# execution time delay problems.
now = datetime.datetime.now(pytz.utc).replace(microsecond=0)
# Check the plugin's params.
self.assertEqual(last_run_date, plugin.last_invocation_parameters[0])
self.assertEqual(now, plugin.last_invocation_parameters[1])

381
storyboard/tests/plugin/cron/test_manager.py
View File

@ -1,381 +0,0 @@
# Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
#
# 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 os
import tempfile
import crontab
from oslo.config import cfg
from stevedore.extension import Extension
import storyboard.plugin.base as plugin_base
import storyboard.plugin.cron.manager as cronmanager
import storyboard.tests.base as base
from storyboard.tests.plugin.cron.mock_plugin import MockPlugin
CONF = cfg.CONF
class TestCronManager(base.WorkingDirTestCase):
def setUp(self):
super(TestCronManager, self).setUp()
(user, self.tabfile) = tempfile.mkstemp(prefix='cron_')
# Flush the crontab before test.
cron = crontab.CronTab(tabfile=self.tabfile)
cron.remove_all(command='storyboard-cron')
cron.write()
CONF.register_opts(cronmanager.CRON_MANAGEMENT_OPTS, 'cron')
CONF.set_override('enable', True, group='cron')
def tearDown(self):
super(TestCronManager, self).tearDown()
CONF.clear_override('enable', group='cron')
# Flush the crontab after test.
cron = crontab.CronTab(tabfile=self.tabfile)
cron.remove_all(command='storyboard-cron')
cron.write()
os.remove(self.tabfile)
def test_enabled(self):
"""This plugin must be enabled if the configuration tells it to be
enabled.
"""
enabled_plugin = cronmanager.CronManager(CONF, tabfile=self.tabfile)
self.assertTrue(enabled_plugin.enabled())
CONF.set_override('enable', False, group='cron')
enabled_plugin = cronmanager.CronManager(CONF)
self.assertFalse(enabled_plugin.enabled())
CONF.clear_override('enable', group='cron')
def test_interval(self):
"""Assert that the cron manager runs every 5 minutes."""
plugin = cronmanager.CronManager(CONF, tabfile=self.tabfile)
self.assertEqual("*/5 * * * *", plugin.interval())
def test_manage_plugins(self):
"""Assert that the cron manager adds plugins to crontab."""
mock_plugin = MockPlugin(dict())
mock_plugin_name = mock_plugin.get_name()
mock_extensions = [Extension('test_one', None, None, mock_plugin)]
loader = plugin_base.StoryboardPluginLoader.make_test_instance(
mock_extensions, namespace='storyboard.plugin.testing'
)
# Run the add_plugin routine.
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
loader.map(manager._manage_plugins)
# Manually test the crontab.
self.assertCronLength(1, command='storyboard-cron')
self.assertCronContains(
command='storyboard-cron --plugin %s' % (mock_plugin_name,),
comment=mock_plugin.get_name(),
interval=mock_plugin.interval(),
enabled=mock_plugin.enabled()
)
def test_manage_disabled_plugin(self):
"""Assert that a disabled plugin is added to the system crontab,
but disabled. While we don't anticipate this feature to ever be
triggered (since the plugin loader won't present disabled plugins),
it's still a good safety net.
"""
mock_plugin = MockPlugin(dict(), is_enabled=False)
mock_plugin_name = mock_plugin.get_name()
mock_extensions = [Extension('test_one', None, None, mock_plugin)]
loader = plugin_base.StoryboardPluginLoader.make_test_instance(
mock_extensions, namespace='storyboard.plugin.testing'
)
# Run the add_plugin routine.
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
loader.map(manager._manage_plugins)
# Manually test the crontab.
self.assertCronLength(1, command='storyboard-cron')
self.assertCronContains(
command='storyboard-cron --plugin %s' % (mock_plugin_name,),
comment=mock_plugin.get_name(),
interval=mock_plugin.interval(),
enabled=mock_plugin.enabled()
)
def test_manage_existing_update(self):
"""Assert that a plugin whose signature changes is appropriately
updated in the system crontab.
"""
mock_plugin = MockPlugin(dict(),
plugin_interval="*/10 * * * *",
is_enabled=False)
mock_plugin_name = mock_plugin.get_name()
mock_extensions = [Extension('test_one', None, None, mock_plugin)]
loader = plugin_base.StoryboardPluginLoader.make_test_instance(
mock_extensions, namespace='storyboard.plugin.testing'
)
# Run the add_plugin routine.
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
loader.map(manager._manage_plugins)
# Manually test the crontab.
self.assertCronLength(1, command='storyboard-cron')
self.assertCronContains(
command='storyboard-cron --plugin %s' % (mock_plugin_name,),
comment=mock_plugin.get_name(),
interval=mock_plugin.interval(),
enabled=mock_plugin.enabled()
)
# Update the plugin and re-run the loader
mock_plugin.plugin_interval = "*/5 * * * *"
loader.map(manager._manage_plugins)
# re-test the crontab.
self.assertCronLength(1, command='storyboard-cron')
self.assertCronContains(
command='storyboard-cron --plugin %s' % (mock_plugin_name,),
comment=mock_plugin.get_name(),
interval=mock_plugin.interval(),
enabled=mock_plugin.enabled()
)
def test_remove_plugin(self):
"""Assert that the remove() method on the manager removes plugins from
the crontab.
"""
mock_plugin = MockPlugin(dict(), is_enabled=False)
mock_plugin_name = mock_plugin.get_name()
mock_extensions = [Extension('test_one', None, None, mock_plugin)]
loader = plugin_base.StoryboardPluginLoader.make_test_instance(
mock_extensions, namespace='storyboard.plugin.testing'
)
# Run the add_plugin routine.
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
loader.map(manager._manage_plugins)
# Manually test the crontab.
self.assertCronLength(1, command='storyboard-cron')
self.assertCronContains(
command='storyboard-cron --plugin %s' % (mock_plugin_name,),
comment=mock_plugin.get_name(),
interval=mock_plugin.interval(),
enabled=mock_plugin.enabled()
)
# Now run the manager's remove method.
manager.remove()
# Make sure we don't leave anything behind.
self.assertCronLength(0, command='storyboard-cron')
def test_remove_only_storyboard(self):
"""Assert that the remove() method manager only removes storyboard
plugins, and not others.
"""
# Create a test job.
cron = crontab.CronTab(tabfile=self.tabfile)
job = cron.new(command='echo 1', comment='echo_test')
job.setall("0 0 */10 * *")
cron.write()
# Create a plugin and have the manager add it to cron.
mock_plugin = MockPlugin(dict(), is_enabled=False)
mock_extensions = [Extension('test_one', None, None, mock_plugin)]
loader = plugin_base.StoryboardPluginLoader.make_test_instance(
mock_extensions,
namespace='storyboard.plugin.testing'
)
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
loader.map(manager._manage_plugins)
# Assert that there's two jobs in our cron.
self.assertCronLength(1, command='storyboard-cron')
self.assertCronLength(1, comment='echo_test')
# Call manager remove.
manager.remove()
# Check crontab.
self.assertCronLength(0, command='storyboard-cron')
self.assertCronLength(1, comment='echo_test')
# Clean up after ourselves.
cron = crontab.CronTab(tabfile=self.tabfile)
cron.remove_all(comment='echo_test')
cron.write()
def test_remove_not_there(self):
"""Assert that the remove() method is idempotent and can happen if
we're already unregistered.
"""
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
manager.remove()
def test_execute(self):
"""Test that execute() method adds plugins."""
# Actually run the real cronmanager.
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
manager.execute()
# We're expecting 1 enabled in-branch plugins.
self.assertCronLength(1, command='storyboard-cron')
def test_execute_update(self):
"""Test that execute() method updates plugins."""
# Manually create an instance of a known plugin with a time interval
# that doesn't match what the plugin wants.
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
manager_name = manager.get_name()
manager_command = "storyboard-cron --plugin %s" % (manager_name,)
manager_comment = manager_name
manager_old_interval = "0 0 */2 * *"
cron = crontab.CronTab(tabfile=self.tabfile)
job = cron.new(
command=manager_command,
comment=manager_comment
)
job.enable(False)
job.setall(manager_old_interval)
cron.write()
# Run the manager
manager.execute()
# Check a new crontab to see what we find.
self.assertCronLength(1, command=manager_command)
cron = crontab.CronTab(tabfile=self.tabfile)
for job in cron.find_command(manager_command):
self.assertNotEqual(manager_old_interval, job.slices)
self.assertEqual(manager.interval(), job.slices)
self.assertTrue(job.enabled)
# Cleanup after ourselves.
manager.remove()
# Assert that things are gone.
self.assertCronLength(0, command='storyboard-cron')
def test_execute_remove_orphans(self):
"""Test that execute() method removes orphaned/deregistered plugins."""
# Manually create an instance of a plugin that's not in our default
# stevedore registration
plugin = MockPlugin(dict())
plugin_name = plugin.get_name()
plugin_command = "storyboard-cron --plugin %s" % (plugin_name,)
plugin_comment = plugin_name
plugin_interval = plugin.interval()
cron = crontab.CronTab(tabfile=self.tabfile)
job = cron.new(
command=plugin_command,
comment=plugin_comment
)
job.enable(False)
job.setall(plugin_interval)
cron.write()
# Run the manager
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
manager.execute()
# Check a new crontab to see what we find.
self.assertCronLength(0, command=plugin_command)
self.assertCronLength(1, command='storyboard-cron')
# Cleanup after ourselves.
manager.remove()
# Assert that things are gone.
self.assertCronLength(0, command='storyboard-cron')
def test_execute_add_new(self):
"""Test that execute() method adds newly registered plugins."""
# Manuall add the cron manager
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
manager_name = manager.get_name()
manager_command = "storyboard-cron --plugin %s" % (manager_name,)
manager_comment = manager_name
cron = crontab.CronTab(tabfile=self.tabfile)
job = cron.new(
command=manager_command,
comment=manager_comment
)
job.enable(manager.enabled())
job.setall(manager.interval())
cron.write()
# Run the manager
manager = cronmanager.CronManager(CONF, tabfile=self.tabfile)
manager.execute()
# Check a new crontab to see what we find.
self.assertCronLength(1, command='storyboard-cron')
# Cleanup after ourselves.
manager.remove()
# Assert that things are gone.
self.assertCronLength(0, command='storyboard-cron')
def assertCronLength(self, length=0, command=None, comment=None):
cron = crontab.CronTab(tabfile=self.tabfile)
if command:
self.assertEqual(length,
len(list(cron.find_command(command))))
elif comment:
self.assertEqual(length,
len(list(cron.find_comment(comment))))
else:
self.assertEqual(0, length)
def assertCronContains(self, command, comment, interval, enabled=True):
cron = crontab.CronTab(tabfile=self.tabfile)
found = False
for job in cron.find_comment(comment):
if job.command != command:
continue
elif job.comment != comment:
continue
elif job.enabled != enabled:
continue
elif str(job.slices) != interval:
continue
else:
found = True
break
self.assertTrue(found)

0
storyboard/tests/plugin/cron/__init__.py → storyboard/tests/plugin/scheduler/__init__.py
View File

51
storyboard/tests/plugin/scheduler/mock_plugin.py Normal file
View File

@ -0,0 +1,51 @@
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# 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 datetime
from apscheduler.triggers.date import DateTrigger
from oslo.config import cfg
import storyboard.plugin.scheduler.base as plugin_base
CONF = cfg.CONF
one_year_from_now = datetime.datetime.now() + datetime.timedelta(days=365)
test_trigger = DateTrigger(run_date=one_year_from_now)
class MockPlugin(plugin_base.SchedulerPluginBase):
"""A mock scheduler plugin for testing."""
def __init__(self, config, is_enabled=True,
trigger=test_trigger):
"""Create a new instance of the base plugin, with some sane defaults
and a time interval that will never execute.
"""
super(MockPlugin, self).__init__(config)
self.is_enabled = is_enabled
self.run_invoked = False
self._trigger = trigger
def enabled(self):
"""Return our enabled value."""
return self.is_enabled
def run(self):
"""Stores the data to a global variable so we can test it.
"""
self.run_invoked = True
def trigger(self):
"""The plugin's trigger."""
return self._trigger

39
storyboard/tests/plugin/scheduler/test_base.py Normal file
View File

@ -0,0 +1,39 @@
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# 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 storyboard.plugin.scheduler.base import SchedulerPluginBase
import storyboard.tests.base as base
class TestSchedulerBasePlugin(base.TestCase):
"""Test the scheduler app."""
def test_plugin_name(self):
plugin = TestPlugin(dict())
self.assertEqual("plugin.scheduler.test_base:TestPlugin",
plugin.get_name())
class TestPlugin(SchedulerPluginBase):
"""Test plugin to test the non-abstract methods."""
def enabled(self):
return True
def run(self):
pass
def trigger(self):
pass

167
storyboard/tests/plugin/scheduler/test_init.py Normal file
View File

@ -0,0 +1,167 @@
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# 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 datetime
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
from oslo.config import cfg
from stevedore.extension import Extension
from plugin.scheduler.mock_plugin import MockPlugin
from storyboard.plugin.base import StoryboardPluginLoader
import storyboard.plugin.scheduler as scheduler
import storyboard.tests.base as base
CONF = cfg.CONF
class TestSchedulerCoreMethods(base.DbTestCase):
"""Test methods defined in __init__.py."""
def setUp(self):
super(TestSchedulerCoreMethods, self).setUp()
self.addCleanup(self._remove_scheduler)
def _remove_scheduler(self):
if scheduler.SCHEDULER:
scheduler.shutdown_scheduler()
def test_disabled_initialize(self):
"""Test that the initialize method does nothing when disabled."""
CONF.set_override('enable', False, 'scheduler')
self.assertIsNone(scheduler.SCHEDULER)
scheduler.initialize_scheduler()
self.assertIsNone(scheduler.SCHEDULER)
CONF.clear_override('enable', 'scheduler')
def test_enabled_initialize(self):
"""Test that the initialize and shutdown methods work when enabled."""
CONF.set_override('enable', True, 'scheduler')
self.assertIsNone(scheduler.SCHEDULER)
scheduler.initialize_scheduler()
self.assertIsNotNone(scheduler.SCHEDULER)
scheduler.shutdown_scheduler()
self.assertIsNone(scheduler.SCHEDULER)
CONF.clear_override('enable', 'scheduler')
def test_intialize_with_manager(self):
"""Assert that the management plugin is loaded, and runs every
minute.
"""
CONF.set_override('enable', True, 'scheduler')
self.assertIsNone(scheduler.SCHEDULER)
scheduler.initialize_scheduler()
self.assertIsNotNone(scheduler.SCHEDULER)
manager_job = scheduler.SCHEDULER \
.get_job(scheduler.SCHEDULE_MANAGER_ID)
self.assertIsNotNone(manager_job)
trigger = manager_job.trigger
self.assertIsInstance(trigger, IntervalTrigger)
self.assertEqual(60, trigger.interval_length)
self.assertEqual(scheduler.SCHEDULE_MANAGER_ID, manager_job.id)
scheduler.shutdown_scheduler()
self.assertIsNone(scheduler.SCHEDULER)
CONF.clear_override('enable', 'scheduler')
def test_add_new_not_safe(self):
"""Try to add a plugin to a nonexistent scheduler."""
# Make sure that invoking without a scheduler is safe.
self.assertIsNone(scheduler.SCHEDULER)
scheduler.add_plugins(dict())
self.assertIsNone(scheduler.SCHEDULER)
def test_add_new(self):
"""Add a new plugin to the scheduler."""
CONF.set_override('enable', True, 'scheduler')
self.assertIsNone(scheduler.SCHEDULER)
scheduler.initialize_scheduler()
mock_plugin = MockPlugin(dict())
mock_plugin_name = mock_plugin.get_name()
mock_extensions = [
Extension(mock_plugin_name, None, None, mock_plugin)
]
loader = StoryboardPluginLoader.make_test_instance(
mock_extensions, namespace='storyboard.plugin.testing'
)
test_list = list()
loader.map(scheduler.add_plugins, test_list)
self.assertTrue(test_list.index(mock_plugin_name) == 0)
self.assertIsNotNone(scheduler.SCHEDULER.get_job(mock_plugin_name))
scheduler.shutdown_scheduler()
self.assertIsNone(scheduler.SCHEDULER)
CONF.clear_override('enable', 'scheduler')
def test_add_plugins_reschedule(self):
"""Assert that the test_add_plugins will reschedule existing plugins.
"""
CONF.set_override('enable', True, 'scheduler')
self.assertIsNone(scheduler.SCHEDULER)
scheduler.initialize_scheduler()
mock_plugin = MockPlugin(dict())
mock_plugin_name = mock_plugin.get_name()
mock_extensions = [
Extension(mock_plugin_name, None, None, mock_plugin)
]
loader = StoryboardPluginLoader.make_test_instance(
mock_extensions, namespace='storyboard.plugin.testing'
)
test_list = list()
loader.map(scheduler.add_plugins, test_list)
self.assertTrue(test_list.index(mock_plugin_name) == 0)
first_run_job = scheduler.SCHEDULER.get_job(mock_plugin_name)
first_run_trigger = first_run_job.trigger
self.assertEqual(mock_plugin._trigger.run_date,
first_run_trigger.run_date)
# Update the plugin's interval and re-run
new_date = datetime.datetime.now() + datetime.timedelta(days=2)
mock_plugin._trigger = DateTrigger(run_date=new_date)
test_list = list()
loader.map(scheduler.add_plugins, test_list)
# make sure the plugin is only loaded once.
self.assertTrue(test_list.index(mock_plugin_name) == 0)
self.assertEquals(len(test_list), 1)
# Get the job.
second_run_job = scheduler.SCHEDULER.get_job(mock_plugin_name)
second_run_trigger = second_run_job.trigger
self.assertNotEqual(second_run_trigger.run_date,
first_run_trigger.run_date)
scheduler.shutdown_scheduler()
self.assertIsNone(scheduler.SCHEDULER)
CONF.clear_override('enable', 'scheduler')

10
storyboard/tests/plugin/token_cleaner/test_cleaner.py
View File

@ -47,10 +47,12 @@ class TestTokenCleaner(db_base.BaseDbTestCase,
CONF.clear_override('enable', 'plugin_token_cleaner')
def test_interval(self):
"""Assert that the cron manager runs every 5 minutes."""
def test_trigger(self):
"""Assert that the this plugin runs every minute."""
plugin = TokenCleaner(CONF)
self.assertEqual("? * * * *", plugin.interval())
trigger = plugin.trigger()
self.assertEqual(3600, trigger.interval_length)
def test_token_removal(self):
"""Assert that the plugin deletes tokens whose expiration date passed
@ -100,7 +102,7 @@ class TestTokenCleaner(db_base.BaseDbTestCase,
# Run the plugin.
plugin = TokenCleaner(CONF)
plugin.execute()
plugin.run()
# Make sure we have 8 tokens left (since one plugin starts today).
self.assertEqual(8, db_api.model_query(AccessToken).count())