diff options
author | Alex Grönholm <alex.gronholm@nextday.fi> | 2014-05-30 06:13:03 +0300 |
---|---|---|
committer | Alex Grönholm <alex.gronholm@nextday.fi> | 2014-05-30 10:35:01 +0300 |
commit | b56ba7f41b4c4f3784bc326ce2cc7ea7133ca263 (patch) | |
tree | ddc9378a128c175760c248acec16b4d70ae49d7a | |
parent | 223672e2e80029cf6aa14b7cfb2f0d500464dfdf (diff) | |
download | apscheduler-b56ba7f41b4c4f3784bc326ce2cc7ea7133ca263.tar.gz |
Overhauled the tests for Job class
-rw-r--r-- | apscheduler/job.py | 3 | ||||
-rw-r--r-- | tests/conftest.py | 3 | ||||
-rw-r--r-- | tests/test_job.py | 312 |
3 files changed, 229 insertions, 89 deletions
diff --git a/apscheduler/job.py b/apscheduler/job.py index 00ffd7a..b1f8d6e 100644 --- a/apscheduler/job.py +++ b/apscheduler/job.py @@ -4,6 +4,7 @@ from uuid import uuid4 import six +from apscheduler.triggers.base import BaseTrigger from apscheduler.util import ref_to_obj, obj_to_ref, datetime_repr, repr_escape, get_callable_name, check_callable_args @@ -177,7 +178,7 @@ class Job(object): if 'trigger' in changes: trigger = changes.pop('trigger') - if not callable(getattr(trigger, 'get_next_fire_time')): + if not isinstance(trigger, BaseTrigger): raise TypeError('Expected a trigger instance, got %s instead' % trigger.__class__.__name__) approved['trigger'] = trigger diff --git a/tests/conftest.py b/tests/conftest.py index 5054f32..4dd00cf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,6 +6,7 @@ import pytest import pytz from apscheduler.job import Job +from apscheduler.schedulers.base import BaseScheduler from apscheduler.schedulers.blocking import BlockingScheduler @@ -80,7 +81,7 @@ def job_defaults(timezone): @pytest.fixture(scope='session') def create_job(job_defaults): def create(**kwargs): - kwargs.setdefault('scheduler', Mock()) + kwargs.setdefault('scheduler', Mock(BaseScheduler)) job_kwargs = job_defaults.copy() job_kwargs.update(kwargs) job_kwargs['trigger'] = BlockingScheduler()._create_trigger(job_kwargs.pop('trigger'), diff --git a/tests/test_job.py b/tests/test_job.py index 250f665..9457269 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -1,17 +1,18 @@ # coding: utf-8 from datetime import datetime, timedelta -from functools import partial import pytest import six +from apscheduler.job import Job +from apscheduler.schedulers.base import BaseScheduler from apscheduler.triggers.date import DateTrigger from tests.conftest import maxpython try: - from unittest.mock import MagicMock + from unittest.mock import MagicMock, patch except ImportError: - from mock import MagicMock + from mock import MagicMock, patch def dummyfunc(): @@ -23,111 +24,248 @@ def job(create_job): return create_job(func=dummyfunc) -class TestJob(object): - def test_job_func(self, create_job): - """Tests that Job can accept a plain, direct function.""" +@pytest.mark.parametrize('job_id', ['testid', None]) +def test_constructor(job_id): + with patch('apscheduler.job.Job._modify') as _modify: + scheduler_mock = MagicMock(BaseScheduler) + job = Job(scheduler_mock, id=job_id) + assert job._scheduler is scheduler_mock + assert job._jobstore_alias is None - job = create_job(func=dummyfunc) - assert job.func is dummyfunc + modify_kwargs = _modify.call_args[1] + assert modify_kwargs['next_run_time'] is None - def test_job_non_inspectable_func(self, create_job): - """Tests that Job can accept a function that fails on inspect.getargspec().""" + if job_id is None: + assert len(modify_kwargs['id']) == 32 + else: + assert modify_kwargs['id'] == job_id - func = partial(dummyfunc) - job = create_job(func=func) - assert job.func is func - def test_job_func_ref(self, create_job): - """Tests that Job can accept a function by its textual reference.""" +def test_modify(job): + job.modify(bah=1, foo='x') + assert job._scheduler.modify_job.called_once_with(job.id, bah=1, foo='x') - job = create_job(func='%s:dummyfunc' % __name__) - assert job.func is dummyfunc - def test_job_bad_func(self, create_job): - exc = pytest.raises(TypeError, create_job, func=object()) - assert 'textual reference' in str(exc.value) +def test_reschedule(job): + job.reschedule('trigger', bah=1, foo='x') + assert job._scheduler.reschedule_job.called_once_with(job.id, 'trigger', bah=1, foo='x') - def test_job_invalid_version(self, job): - exc = pytest.raises(ValueError, job.__setstate__, {'version': 9999}) - assert 'version' in str(exc.value) - def test_remove(self, job): - assert job._scheduler.remove_job.called_once_with('testid', 'default') +def test_pause(job): + job.pause() + assert job._scheduler.pause_job.called_once_with(job.id) - def test_get_run_times(self, create_job, timezone): - run_time = timezone.localize(datetime(2010, 12, 13, 0, 8)) - expected_times = [run_time + timedelta(seconds=1), - run_time + timedelta(seconds=2)] - job = create_job(trigger='interval', trigger_args={'seconds': 1, 'timezone': timezone, 'start_date': run_time}, - next_run_time=expected_times[0], func=dummyfunc) - run_times = job._get_run_times(run_time) - assert run_times == [] - run_times = job._get_run_times(expected_times[0]) - assert run_times == [expected_times[0]] +def test_resume(job): + job.resume() + assert job._scheduler.resume_job.called_once_with(job.id) - run_times = job._get_run_times(expected_times[1]) - assert run_times == expected_times - def test_pending(self, job): - """Tests that the "pending" property return True when _jobstore is a string, False otherwise.""" +def test_remove(job): + job.remove() + assert job._scheduler.remove_job.called_once_with(job.id) - assert job.pending - job._jobstore_alias = 'test' - assert not job.pending +def test_pending(job): + """Tests that the "pending" property return True when _jobstore_alias is a string, False otherwise.""" - def test_getstate(self, job): - state = job.__getstate__() - expected = dict(version=1, trigger=job.trigger, executor='default', func='tests.test_job:dummyfunc', - name=b'n\xc3\xa4m\xc3\xa9'.decode('utf-8'), args=(), kwargs={}, - id=b't\xc3\xa9st\xc3\xafd'.decode('utf-8'), misfire_grace_time=1, coalesce=False, - max_instances=1, next_run_time=None) - assert state == expected + assert job.pending - def test_setstate(self, job, timezone): - trigger = DateTrigger('2010-12-14 13:05:00', timezone) - state = dict(version=1, scheduler=MagicMock(), jobstore=MagicMock(), trigger=trigger, executor='dummyexecutor', - func='tests.test_job:dummyfunc', name='testjob.dummyfunc', args=[], kwargs={}, id='other_id', - misfire_grace_time=2, coalesce=True, max_instances=2, next_run_time=None) - job.__setstate__(state) - assert job.id == 'other_id' - assert job.executor == 'dummyexecutor' - assert job.trigger == trigger - assert job.func == dummyfunc - assert job.coalesce is True - assert job.max_instances == 2 - assert job.next_run_time is None + job._jobstore_alias = 'test' + assert not job.pending - def test_eq(self, create_job, timezone): - job = create_job(func=dummyfunc) - assert job == job - job2 = create_job(trigger='date', id='otherid', trigger_args={'run_date': datetime.now(), 'timezone': timezone}, - func=dummyfunc) - assert not job == job2 +def test_get_run_times(create_job, timezone): + run_time = timezone.localize(datetime(2010, 12, 13, 0, 8)) + expected_times = [run_time + timedelta(seconds=1), run_time + timedelta(seconds=2)] + job = create_job(trigger='interval', trigger_args={'seconds': 1, 'timezone': timezone, 'start_date': run_time}, + next_run_time=expected_times[0], func=dummyfunc) - job2.id = job.id - assert job == job2 + run_times = job._get_run_times(run_time) + assert run_times == [] - assert not job == 'bleh' + run_times = job._get_run_times(expected_times[0]) + assert run_times == [expected_times[0]] - def test_repr(self, job): - if six.PY2: - assert repr(job) == '<Job (id=t\\xe9st\\xefd name=n\\xe4m\\xe9)>' - else: - assert repr(job) == b'<Job (id=t\xc3\xa9st\xc3\xafd name=n\xc3\xa4m\xc3\xa9)>'.decode('utf-8') + run_times = job._get_run_times(expected_times[1]) + assert run_times == expected_times - def test_str(self, job): - if six.PY2: - expected = 'n\\xe4m\\xe9 (trigger: date[2011-04-03 18:40:00 CEST], next run at: None)' - else: - expected = b'n\xc3\xa4m\xc3\xa9 (trigger: date[2011-04-03 18:40:00 CEST], next run at: None)'.\ - decode('utf-8') - assert str(job) == expected +def test_private_modify_bad_id(job): + """Tests that only strings are accepted for job IDs.""" + + del job.id + exc = pytest.raises(TypeError, job._modify, id=3) + assert str(exc.value) == 'id must be a nonempty string' + + +def test_private_modify_id(job): + """Tests that the job ID can't be changed.""" + + exc = pytest.raises(ValueError, job._modify, id='alternate') + assert str(exc.value) == 'The job ID may not be changed' + + +def test_private_modify_bad_func(job): + """Tests that given a func of something else than a callable or string raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, func=1) + assert str(exc.value) == 'func must be a callable or a textual reference to one' + + +def test_private_modify_func_ref(job): + """Tests that the target callable can be given as a textual reference.""" + + job._modify(func='tests.test_job:dummyfunc') + assert job.func is dummyfunc + assert job.func_ref == 'tests.test_job:dummyfunc' + + +def test_private_modify_unreachable_func(job): + """Tests that func_ref remains None if no reference to the target callable can be found.""" + + func = lambda: None + job._modify(func=func) + assert job.func is func + assert job.func_ref is None + + +def test_private_modify_update_name(job): + """Tests that the name attribute defaults to the function name.""" + + del job.name + job._modify(func=dummyfunc) + assert job.name == 'dummyfunc' + + +def test_private_modify_bad_args(job): + """Tests that passing an argument list of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, args=1) + assert str(exc.value) == 'args must be a non-string iterable' + + +def test_private_modify_bad_kwargs(job): + """Tests that passing an argument list of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, kwargs=1) + assert str(exc.value) == 'kwargs must be a dict-like object' + + +@pytest.mark.parametrize('value', [1, ''], ids=['integer', 'empty string']) +def test_private_modify_bad_name(job, value): + """Tests that passing an empty name or a name of something else than a string raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, name=value) + assert str(exc.value) == 'name must be a nonempty string' + + +@pytest.mark.parametrize('value', ['foo', 0, -1], ids=['string', 'zero', 'negative']) +def test_private_modify_bad_misfire_grace_time(job, value): + """Tests that passing a misfire_grace_time of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, misfire_grace_time=value) + assert str(exc.value) == 'misfire_grace_time must be either None or a positive integer' + + +@pytest.mark.parametrize('value', [None, 'foo', 0, -1], ids=['None', 'string', 'zero', 'negative']) +def test_private_modify_bad_max_instances(job, value): + """Tests that passing a max_instances of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, max_instances=value) + assert str(exc.value) == 'max_instances must be a positive integer' + + +def test_private_modify_bad_trigger(job): + """Tests that passing a trigger of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, trigger='foo') + assert str(exc.value) == 'Expected a trigger instance, got str instead' + + +def test_private_modify_bad_executor(job): + """Tests that passing an executor of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, executor=1) + assert str(exc.value) == 'executor must be a string' + + +def test_private_modify_bad_next_run_time(job): + """Tests that passing a next_run_time of the wrong type raises a TypeError.""" + + exc = pytest.raises(TypeError, job._modify, next_run_time=1) + assert str(exc.value) == 'next_run_time must be either None or a datetime instance' + + +def test_private_modify_bad_argument(job): + """Tests that passing an unmodifiable argument type raises an AttributeError.""" + + exc = pytest.raises(AttributeError, job._modify, scheduler=1) + assert str(exc.value) == 'The following are not modifiable attributes of Job: scheduler' + + +def test_getstate(job): + state = job.__getstate__() + assert state == dict(version=1, trigger=job.trigger, executor='default', func='tests.test_job:dummyfunc', + name=b'n\xc3\xa4m\xc3\xa9'.decode('utf-8'), args=(), kwargs={}, + id=b't\xc3\xa9st\xc3\xafd'.decode('utf-8'), misfire_grace_time=1, coalesce=False, + max_instances=1, next_run_time=None) + + +def test_setstate(job, timezone): + trigger = DateTrigger('2010-12-14 13:05:00', timezone) + state = dict(version=1, scheduler=MagicMock(), jobstore=MagicMock(), trigger=trigger, executor='dummyexecutor', + func='tests.test_job:dummyfunc', name='testjob.dummyfunc', args=[], kwargs={}, id='other_id', + misfire_grace_time=2, coalesce=True, max_instances=2, next_run_time=None) + job.__setstate__(state) + assert job.id == 'other_id' + assert job.func == dummyfunc + assert job.func_ref == 'tests.test_job:dummyfunc' + assert job.trigger == trigger + assert job.executor == 'dummyexecutor' + assert job.args == [] + assert job.kwargs == {} + assert job.name == 'testjob.dummyfunc' + assert job.misfire_grace_time == 2 + assert job.coalesce is True + assert job.max_instances == 2 + assert job.next_run_time is None + + +def test_setstate_bad_version(job): + """Tests that __setstate__ rejects state of higher version that it was designed to handle.""" + + exc = pytest.raises(ValueError, job.__setstate__, {'version': 9999}) + assert 'Job has version 9999, but only version' in str(exc.value) + + +def test_eq(create_job): + job = create_job(func=lambda: None, id='foo') + job2 = create_job(func=lambda: None, id='foo') + job3 = create_job(func=lambda: None, id='bar') + assert job == job2 + assert not job == job3 + assert not job == 'foo' + + +def test_repr(job): + if six.PY2: + assert repr(job) == '<Job (id=t\\xe9st\\xefd name=n\\xe4m\\xe9)>' + else: + assert repr(job) == b'<Job (id=t\xc3\xa9st\xc3\xafd name=n\xc3\xa4m\xc3\xa9)>'.decode('utf-8') + + +def test_str(job): + if six.PY2: + expected = 'n\\xe4m\\xe9 (trigger: date[2011-04-03 18:40:00 CEST], next run at: None)' + else: + expected = b'n\xc3\xa4m\xc3\xa9 (trigger: date[2011-04-03 18:40:00 CEST], next run at: None)'.\ + decode('utf-8') + + assert str(job) == expected + - @maxpython(3, 0) - def test_unicode(self, job): - assert job.__unicode__() == \ - b'n\xc3\xa4m\xc3\xa9 (trigger: date[2011-04-03 18:40:00 CEST], next run at: None)'.decode('utf-8') +@maxpython(3, 0) +def test_unicode(job): + assert job.__unicode__() == \ + b'n\xc3\xa4m\xc3\xa9 (trigger: date[2011-04-03 18:40:00 CEST], next run at: None)'.decode('utf-8') |