1# coding: utf-8 2import gc 3import weakref 4from datetime import datetime, timedelta 5from functools import partial 6 7import pytest 8import six 9 10from apscheduler.job import Job 11from apscheduler.schedulers.base import BaseScheduler 12from apscheduler.triggers.date import DateTrigger 13 14try: 15 from unittest.mock import MagicMock, patch 16except ImportError: 17 from mock import MagicMock, patch 18 19 20def dummyfunc(): 21 pass 22 23 24@pytest.fixture 25def job(create_job): 26 return create_job(func=dummyfunc) 27 28 29@pytest.mark.parametrize('job_id', ['testid', None]) 30def test_constructor(job_id): 31 with patch('apscheduler.job.Job._modify') as _modify: 32 scheduler_mock = MagicMock(BaseScheduler) 33 job = Job(scheduler_mock, id=job_id) 34 assert job._scheduler is scheduler_mock 35 assert job._jobstore_alias is None 36 37 modify_kwargs = _modify.call_args[1] 38 if job_id is None: 39 assert len(modify_kwargs['id']) == 32 40 else: 41 assert modify_kwargs['id'] == job_id 42 43 44def test_modify(job): 45 job.modify(bah=1, foo='x') 46 job._scheduler.modify_job.assert_called_once_with(job.id, None, bah=1, foo='x') 47 48 49def test_reschedule(job): 50 job.reschedule('trigger', bah=1, foo='x') 51 job._scheduler.reschedule_job.assert_called_once_with(job.id, None, 'trigger', bah=1, foo='x') 52 53 54def test_pause(job): 55 job.pause() 56 job._scheduler.pause_job.assert_called_once_with(job.id, None) 57 58 59def test_resume(job): 60 job.resume() 61 job._scheduler.resume_job.assert_called_once_with(job.id, None) 62 63 64def test_remove(job): 65 job.remove() 66 job._scheduler.remove_job.assert_called_once_with(job.id, None) 67 68 69def test_weakref(create_job): 70 job = create_job(func=dummyfunc) 71 ref = weakref.ref(job) 72 del job 73 gc.collect() 74 assert ref() is None 75 76 77def test_pending(job): 78 """ 79 Tests that the "pending" property return True when _jobstore_alias is a string, ``False`` 80 otherwise. 81 82 """ 83 assert job.pending 84 85 job._jobstore_alias = 'test' 86 assert not job.pending 87 88 89def test_get_run_times(create_job, timezone): 90 run_time = timezone.localize(datetime(2010, 12, 13, 0, 8)) 91 expected_times = [run_time + timedelta(seconds=1), run_time + timedelta(seconds=2)] 92 job = create_job(trigger='interval', 93 trigger_args={'seconds': 1, 'timezone': timezone, 'start_date': run_time}, 94 next_run_time=expected_times[0], func=dummyfunc) 95 96 run_times = job._get_run_times(run_time) 97 assert run_times == [] 98 99 run_times = job._get_run_times(expected_times[0]) 100 assert run_times == [expected_times[0]] 101 102 run_times = job._get_run_times(expected_times[1]) 103 assert run_times == expected_times 104 105 106def test_private_modify_bad_id(job): 107 """Tests that only strings are accepted for job IDs.""" 108 del job.id 109 exc = pytest.raises(TypeError, job._modify, id=3) 110 assert str(exc.value) == 'id must be a nonempty string' 111 112 113def test_private_modify_id(job): 114 """Tests that the job ID can't be changed.""" 115 exc = pytest.raises(ValueError, job._modify, id='alternate') 116 assert str(exc.value) == 'The job ID may not be changed' 117 118 119def test_private_modify_bad_func(job): 120 """Tests that given a func of something else than a callable or string raises a TypeError.""" 121 exc = pytest.raises(TypeError, job._modify, func=1) 122 assert str(exc.value) == 'func must be a callable or a textual reference to one' 123 124 125def test_private_modify_func_ref(job): 126 """Tests that the target callable can be given as a textual reference.""" 127 job._modify(func='tests.test_job:dummyfunc') 128 assert job.func is dummyfunc 129 assert job.func_ref == 'tests.test_job:dummyfunc' 130 131 132def test_private_modify_unreachable_func(job): 133 """Tests that func_ref remains None if no reference to the target callable can be found.""" 134 func = partial(dummyfunc) 135 job._modify(func=func) 136 assert job.func is func 137 assert job.func_ref is None 138 139 140def test_private_modify_update_name(job): 141 """Tests that the name attribute defaults to the function name.""" 142 del job.name 143 job._modify(func=dummyfunc) 144 assert job.name == 'dummyfunc' 145 146 147def test_private_modify_bad_args(job): 148 """ Tests that passing an argument list of the wrong type raises a TypeError.""" 149 exc = pytest.raises(TypeError, job._modify, args=1) 150 assert str(exc.value) == 'args must be a non-string iterable' 151 152 153def test_private_modify_bad_kwargs(job): 154 """Tests that passing an argument list of the wrong type raises a TypeError.""" 155 exc = pytest.raises(TypeError, job._modify, kwargs=1) 156 assert str(exc.value) == 'kwargs must be a dict-like object' 157 158 159@pytest.mark.parametrize('value', [1, ''], ids=['integer', 'empty string']) 160def test_private_modify_bad_name(job, value): 161 """ 162 Tests that passing an empty name or a name of something else than a string raises a TypeError. 163 164 """ 165 exc = pytest.raises(TypeError, job._modify, name=value) 166 assert str(exc.value) == 'name must be a nonempty string' 167 168 169@pytest.mark.parametrize('value', ['foo', 0, -1], ids=['string', 'zero', 'negative']) 170def test_private_modify_bad_misfire_grace_time(job, value): 171 """Tests that passing a misfire_grace_time of the wrong type raises a TypeError.""" 172 exc = pytest.raises(TypeError, job._modify, misfire_grace_time=value) 173 assert str(exc.value) == 'misfire_grace_time must be either None or a positive integer' 174 175 176@pytest.mark.parametrize('value', [None, 'foo', 0, -1], ids=['None', 'string', 'zero', 'negative']) 177def test_private_modify_bad_max_instances(job, value): 178 """Tests that passing a max_instances of the wrong type raises a TypeError.""" 179 exc = pytest.raises(TypeError, job._modify, max_instances=value) 180 assert str(exc.value) == 'max_instances must be a positive integer' 181 182 183def test_private_modify_bad_trigger(job): 184 """Tests that passing a trigger of the wrong type raises a TypeError.""" 185 exc = pytest.raises(TypeError, job._modify, trigger='foo') 186 assert str(exc.value) == 'Expected a trigger instance, got str instead' 187 188 189def test_private_modify_bad_executor(job): 190 """Tests that passing an executor of the wrong type raises a TypeError.""" 191 exc = pytest.raises(TypeError, job._modify, executor=1) 192 assert str(exc.value) == 'executor must be a string' 193 194 195def test_private_modify_bad_next_run_time(job): 196 """Tests that passing a next_run_time of the wrong type raises a TypeError.""" 197 exc = pytest.raises(TypeError, job._modify, next_run_time=1) 198 assert str(exc.value) == 'Unsupported type for next_run_time: int' 199 200 201def test_private_modify_bad_argument(job): 202 """Tests that passing an unmodifiable argument type raises an AttributeError.""" 203 exc = pytest.raises(AttributeError, job._modify, scheduler=1) 204 assert str(exc.value) == 'The following are not modifiable attributes of Job: scheduler' 205 206 207def test_getstate(job): 208 state = job.__getstate__() 209 assert state == dict( 210 version=1, trigger=job.trigger, executor='default', func='tests.test_job:dummyfunc', 211 name=b'n\xc3\xa4m\xc3\xa9'.decode('utf-8'), args=(), kwargs={}, 212 id=b't\xc3\xa9st\xc3\xafd'.decode('utf-8'), misfire_grace_time=1, coalesce=False, 213 max_instances=1, next_run_time=None) 214 215 216def test_setstate(job, timezone): 217 trigger = DateTrigger('2010-12-14 13:05:00', timezone) 218 state = dict( 219 version=1, scheduler=MagicMock(), jobstore=MagicMock(), trigger=trigger, 220 executor='dummyexecutor', func='tests.test_job:dummyfunc', name='testjob.dummyfunc', 221 args=[], kwargs={}, id='other_id', misfire_grace_time=2, coalesce=True, max_instances=2, 222 next_run_time=None) 223 job.__setstate__(state) 224 assert job.id == 'other_id' 225 assert job.func == dummyfunc 226 assert job.func_ref == 'tests.test_job:dummyfunc' 227 assert job.trigger == trigger 228 assert job.executor == 'dummyexecutor' 229 assert job.args == [] 230 assert job.kwargs == {} 231 assert job.name == 'testjob.dummyfunc' 232 assert job.misfire_grace_time == 2 233 assert job.coalesce is True 234 assert job.max_instances == 2 235 assert job.next_run_time is None 236 237 238def test_setstate_bad_version(job): 239 """Tests that __setstate__ rejects state of higher version that it was designed to handle.""" 240 exc = pytest.raises(ValueError, job.__setstate__, {'version': 9999}) 241 assert 'Job has version 9999, but only version' in str(exc.value) 242 243 244def test_eq(create_job): 245 job = create_job(func=lambda: None, id='foo') 246 job2 = create_job(func=lambda: None, id='foo') 247 job3 = create_job(func=lambda: None, id='bar') 248 assert job == job2 249 assert not job == job3 250 assert not job == 'foo' 251 252 253def test_repr(job): 254 if six.PY2: 255 assert repr(job) == '<Job (id=t\\xe9st\\xefd name=n\\xe4m\\xe9)>' 256 else: 257 assert repr(job) == \ 258 b'<Job (id=t\xc3\xa9st\xc3\xafd name=n\xc3\xa4m\xc3\xa9)>'.decode('utf-8') 259 260 261@pytest.mark.parametrize('status, expected_status', [ 262 ('scheduled', 'next run at: 2011-04-03 18:40:00 CEST'), 263 ('paused', 'paused'), 264 ('pending', 'pending') 265], ids=['scheduled', 'paused', 'pending']) 266@pytest.mark.parametrize('unicode', [False, True], ids=['nativestr', 'unicode']) 267def test_str(create_job, status, unicode, expected_status): 268 job = create_job(func=dummyfunc) 269 if status == 'scheduled': 270 job.next_run_time = job.trigger.run_date 271 elif status == 'pending': 272 del job.next_run_time 273 274 if six.PY2 and not unicode: 275 expected = 'n\\xe4m\\xe9 (trigger: date[2011-04-03 18:40:00 CEST], %s)' % expected_status 276 else: 277 expected = b'n\xc3\xa4m\xc3\xa9 (trigger: date[2011-04-03 18:40:00 CEST], %s)'.\ 278 decode('utf-8') % expected_status 279 280 result = job.__unicode__() if unicode else job.__str__() 281 assert result == expected 282