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