1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
|
from datetime import datetime as dt
from datetime import timedelta as td
from functools import wraps
from unittest.mock import MagicMock
import pytest
from pytest import mark
import vorta.borg
import vorta.scheduler
from vorta.scheduler import ScheduleStatus, ScheduleStatusType, VortaScheduler
from vorta.store.models import BackupProfileModel, EventLogModel
PROFILE_NAME = 'Default'
FIXED_SCHEDULE = 'fixed'
INTERVAL_SCHEDULE = 'interval'
MANUAL_SCHEDULE = 'off'
@pytest.fixture
def clockmock(monkeypatch):
datetime_mock = MagicMock(wraps=dt)
monkeypatch.setattr(vorta.scheduler, "dt", datetime_mock)
return datetime_mock
def prepare(func):
"""Decorator adding common preparation steps."""
@wraps(func)
def do(qapp, qtbot, mocker, borg_json_output):
stdout, stderr = borg_json_output('create')
popen_result = mocker.MagicMock(stdout=stdout, stderr=stderr, returncode=0)
mocker.patch.object(vorta.borg.borg_job, 'Popen', return_value=popen_result)
return func(qapp, qtbot, mocker, borg_json_output)
return do
@prepare
def test_scheduler_create_backup(qapp, qtbot, mocker, borg_json_output):
"""Test running a backup with `create_backup`."""
events_before = EventLogModel.select().count()
with qtbot.waitSignal(qapp.backup_finished_event, **pytest._wait_defaults):
qapp.scheduler.create_backup(1)
assert EventLogModel.select().count() == events_before + 1
def test_manual_mode():
"""Test scheduling in manual mode."""
scheduler = VortaScheduler()
# setup model
profile = BackupProfileModel.get(name=PROFILE_NAME)
profile.schedule_make_up_missed = False
profile.schedule_mode = MANUAL_SCHEDULE
profile.save()
# test
scheduler.set_timer_for_profile(profile.id)
assert len(scheduler.timers) == 0
def test_simple_schedule(clockmock):
"""Test a simple scheduling including `next_job` and `remove_job`."""
scheduler = VortaScheduler()
# setup
time = dt(2020, 5, 6, 4, 30)
clockmock.now.return_value = time
profile = BackupProfileModel.get(name=PROFILE_NAME)
profile.schedule_make_up_missed = False
profile.schedule_mode = INTERVAL_SCHEDULE
profile.schedule_interval_unit = 'hours'
profile.schedule_interval_count = 3
profile.save()
event = EventLogModel(
subcommand='create', profile=profile.id, returncode=0, category='scheduled', start_time=time, end_time=time
)
event.save()
# test set timer and next_job
scheduler.set_timer_for_profile(profile.id)
assert len(scheduler.timers) == 1
assert scheduler.next_job() == '07:30 ({})'.format(PROFILE_NAME)
assert scheduler.next_job_for_profile(profile.id) == ScheduleStatus(
ScheduleStatusType.SCHEDULED, dt(2020, 5, 6, 7, 30)
)
# test remove_job and next_job
scheduler.remove_job(profile.id)
assert len(scheduler.timers) == 0
assert scheduler.next_job() == 'None scheduled'
assert scheduler.next_job_for_profile(profile.id) == ScheduleStatus(ScheduleStatusType.UNSCHEDULED)
@mark.parametrize("scheduled", [True, False])
@mark.parametrize(
"passed_time, now, unit, count, added_time",
[
# simple
(td(), td(hours=4, minutes=30), 'hours', 3, td(hours=3)),
# next day
(td(), td(hours=4, minutes=30), 'hours', 20, td(hours=20)),
# passed by less than interval
(td(hours=2), td(hours=4, minutes=30), 'hours', 3, td(hours=1)),
# passed by exactly interval
(td(hours=3), td(hours=4, minutes=30), 'hours', 3, td(hours=3)),
# passed by multiple times the interval
(td(hours=7), td(hours=4, minutes=30), 'hours', 3, td(hours=2)),
],
)
def test_interval(clockmock, passed_time, scheduled, now, unit, count, added_time):
"""Test scheduling in interval mode."""
# setup
scheduler = VortaScheduler()
time = dt(2020, 5, 4, 0, 0) + now
clockmock.now.return_value = time
profile = BackupProfileModel.get(name=PROFILE_NAME)
profile.schedule_make_up_missed = False
profile.schedule_mode = INTERVAL_SCHEDULE
profile.schedule_interval_unit = unit
profile.schedule_interval_count = count
profile.save()
event = EventLogModel(
subcommand='create',
profile=profile.id,
returncode=0,
category='scheduled' if scheduled else '',
start_time=time - passed_time,
end_time=time - passed_time,
)
event.save()
# run test
scheduler.set_timer_for_profile(profile.id)
assert scheduler.timers[profile.id]['dt'] == time + added_time
@mark.parametrize("scheduled", [True, False])
@mark.parametrize("passed_time", [td(hours=0), td(hours=5), td(hours=14), td(hours=27)])
@mark.parametrize(
"now, hour, minute",
[
# same day
(td(hours=4, minutes=30), 15, 00),
# next day
(td(hours=4, minutes=30), 3, 30),
],
)
def test_fixed(clockmock, passed_time, scheduled, now, hour, minute):
"""Test scheduling in fixed mode."""
# setup
scheduler = VortaScheduler()
time = dt(2020, 5, 4, 0, 0) + now
clockmock.now.return_value = time
profile = BackupProfileModel.get(name=PROFILE_NAME)
profile.schedule_make_up_missed = False
profile.schedule_mode = FIXED_SCHEDULE
profile.schedule_fixed_hour = hour
profile.schedule_fixed_minute = minute
profile.save()
last_time = time - passed_time
event = EventLogModel(
subcommand='create',
profile=profile.id,
returncode=0,
category='scheduled' if scheduled else '',
start_time=last_time,
end_time=last_time,
)
event.save()
# run test
expected = time.replace(hour=hour, minute=minute)
if time >= expected or last_time.date() == expected.date():
expected += td(days=1)
scheduler.set_timer_for_profile(profile.id)
assert scheduler.timers[profile.id]['dt'] == expected
@mark.parametrize(
"now, hour, minute, time_since_last_run, expect_catchup",
[
(td(hours=9), 18, 00, td(hours=12), False),
(td(hours=9), 18, 00, td(hours=24), False),
(td(hours=9), 18, 00, td(hours=36), True),
(td(hours=20), 18, 00, td(hours=2), False),
(td(hours=20), 18, 00, td(hours=24), True),
],
)
def test_missed_startup(qapp, qtbot, window_load, clockmock, now, hour, minute, time_since_last_run, expect_catchup):
time = dt(2020, 5, 4, 0, 0) + now
clockmock.now.return_value = time
profile = BackupProfileModel.get(name=PROFILE_NAME)
profile.schedule_make_up_missed = True
profile.schedule_mode = FIXED_SCHEDULE
profile.schedule_fixed_hour = hour
profile.schedule_fixed_minute = minute
profile.save()
last_time = time - time_since_last_run
event = EventLogModel(
subcommand='create',
profile=profile.id,
returncode=0,
category='scheduled',
start_time=last_time,
end_time=last_time,
)
event.save()
# We have to replace the scheduler because of shared state (namely, pauses)
# We also reload because app init does that (via `set_borg_details_result`)
qapp.scheduler = VortaScheduler()
qapp.scheduler.reload_all_timers()
window_load()
qtbot.waitSignal(qapp.main_window.loaded, **pytest._wait_defaults)
event_times = [log.start_time for log in EventLogModel.select()]
if expect_catchup:
assert len(event_times) == 2
else:
assert len(event_times) == 1
|
