From 03e8a3599c09e1d3c41ed73607f113b785a54a5d Mon Sep 17 00:00:00 2001 From: ebuzerdrmz44 Date: Wed, 16 Sep 2026 17:41:34 +0300 Subject: [PATCH] Extract Execution from VortaScheduler and record scheduled runs through their full lifecycle --- src/vorta/borg/borg_job.py | 2 + src/vorta/scheduler/__init__.py | 172 +++----------------------- src/vorta/scheduler/execution.py | 205 +++++++++++++++++++++++++++++++ src/vorta/scheduler/state.py | 55 +++++++-- src/vorta/store/connection.py | 16 +++ tests/unit/conftest.py | 2 + tests/unit/test_schedule.py | 3 +- tests/unit/test_scheduler.py | 159 ++++++++++++++++++++++-- 8 files changed, 442 insertions(+), 172 deletions(-) create mode 100644 src/vorta/scheduler/execution.py diff --git a/src/vorta/borg/borg_job.py b/src/vorta/borg/borg_job.py index 840d5bbfb..d1c5ec037 100644 --- a/src/vorta/borg/borg_job.py +++ b/src/vorta/borg/borg_job.py @@ -342,6 +342,8 @@ def read_async(fd): log_entry.returncode = p.returncode log_entry.repo_url = self.params.get('repo_url', None) log_entry.end_time = dt.now() + result['log_entry_id'] = log_entry.id + with db_lock: log_entry.save() self.process_result(result) diff --git a/src/vorta/scheduler/__init__.py b/src/vorta/scheduler/__init__.py index ae2996a39..a0db8e7d4 100644 --- a/src/vorta/scheduler/__init__.py +++ b/src/vorta/scheduler/__init__.py @@ -2,22 +2,15 @@ import logging from datetime import datetime as dt -from datetime import timedelta from typing import Any -from packaging import version from PyQt6 import QtCore from PyQt6.QtCore import QTimer from PyQt6.QtWidgets import QApplication from vorta import application -from vorta.borg.check import BorgCheckJob -from vorta.borg.compact import BorgCompactJob -from vorta.borg.create import BorgCreateJob -from vorta.borg.list_repo import BorgListRepoJob -from vorta.borg.prune import BorgPruneJob from vorta.i18n import translate -from vorta.notifications import VortaNotifications +from vorta.scheduler.execution import SchedulerExecution from vorta.scheduler.scheduling import ( MAX_TIMER_MS, PENDING_STATUSES, @@ -30,8 +23,7 @@ arm_deadline_timer, ) from vorta.scheduler.state import WAKE_CHECK_INTERVAL_MS, WAKE_GAP_THRESHOLD, SchedulerState -from vorta.store.models import BackupProfileModel, EventLogModel, JobModel -from vorta.utils import borg_compat +from vorta.store.models import BackupProfileModel, JobModel logger = logging.getLogger(__name__) @@ -62,10 +54,9 @@ def __init__(self) -> None: self.app: application.VortaApp = QApplication.instance() - #: profiles being submitted, so a timer tick cannot submit one twice - self._submitting: set[int] = set() - - # Scheduling is built first: restoring the pauses writes a status into its timers. + # Execution first: it starts nothing, and the other two arm timers that route back into it. + self._execution = SchedulerExecution(self) + # Scheduling before State: restoring the pauses writes a status into its timers. self._timers = SchedulerTimers(self) self._state = SchedulerState(self) self._state.restore_pauses() @@ -147,6 +138,14 @@ def record_skip( ) -> None: self._state.record_skip(profile, trigger, reason, status=status, scheduled_at=scheduled_at) + def record_start(self, profile: BackupProfileModel, trigger: str) -> int | None: + return self._state.record_start(profile, trigger) + + def record_finish( + self, record_id: int, status: str, log_entry_id: int | None = None, reason: str | None = None + ) -> None: + self._state.record_finish(record_id, status, log_entry_id, reason) + def set_timer_for_profile(self, profile_id: int) -> None: """Set a timer for next scheduled backup run of this profile, and run a missed one.""" catch_up = self._timers.arm_profile(profile_id) @@ -172,149 +171,10 @@ def pending_jobs(self) -> list[PendingJob]: return self._timers.pending_jobs() def create_backup(self, profile_id: int, trigger: str = JobModel.Trigger.SCHEDULED.value) -> None: - notifier = VortaNotifications.pick() - profile = BackupProfileModel.get_or_none(id=profile_id) - - if profile is None: - logger.info('Profile not found. Maybe deleted?') - return - - if profile_id in self._submitting: - logger.debug('A run for profile %s is already being submitted.', profile_id) - return - - # Skip if a job for this profile (repo) is already in progress - if self.app.jobs_manager.is_worker_running(site=profile.repo.id): - logger.debug('A job for repo %s is already active.', profile.repo.id) - self.record_skip(profile, trigger, 'Repository is busy with another job.') - self.pause(profile_id) - return - - self._submitting.add(profile_id) - try: - logger.info('Starting background backup for %s', profile.name) - notifier.deliver( - self.tr('Vorta Backup'), - self.tr('Starting background backup for %s.') % profile.name, - level='info', - ) - msg = BorgCreateJob.prepare(profile) - if msg['ok']: - logger.info('Preparation for backup successful.') - msg['category'] = 'scheduled' - job = BorgCreateJob(msg['cmd'], msg, profile.repo.id) - job.result.connect(self.notify) - self.app.jobs_manager.add_job(job) - else: - # Default to 'error': unexpected failures notify. - # Expected skips (WiFi/metered) use 'info' to suppress. - level = msg.get('level', 'error') - if level == 'error': - logger.error('Conditions for backup not met. Aborting.') - logger.error(msg['message']) - notifier.deliver( - self.tr('Vorta Backup'), - translate('messages', msg['message']), - level='error', - ) - status = JobModel.Status.FAILED.value - else: - logger.info('Backup skipped: %s', msg['message']) - status = JobModel.Status.SKIPPED.value - self.record_skip(profile, trigger, msg['message'], status=status) - self.pause(profile_id) - finally: - self._submitting.discard(profile_id) + self._execution.create_backup(profile_id, trigger) def notify(self, result: dict[str, Any]) -> None: - notifier = VortaNotifications.pick() - profile_name = result['params']['profile_name'] - profile_id = result['params']['profile'].id - - if result['returncode'] in [0, 1]: - notifier.deliver( - self.tr('Vorta Backup'), - self.tr('Backup successful for %s.') % profile_name, - level='info', - ) - logger.info('Backup creation successful.') - # unpause scheduler - self.unpause(result['params']['profile_id']) - - self.post_backup_tasks(profile_id) - else: - notifier.deliver( - self.tr('Vorta Backup'), - self.tr('Error during backup creation for %s.') % profile_name, - level='error', - ) - logger.error('Error during backup creation.') - # pause scheduler - # if a scheduled backup fails the scheduler should pause - # temporarily. - self.pause(result['params']['profile_id']) - - self.set_timer_for_profile(profile_id) + self._execution.notify(result) def post_backup_tasks(self, profile_id: int) -> None: - """ - Pruning and checking after successful backup. - """ - profile = BackupProfileModel.get(id=profile_id) - notifier = VortaNotifications.pick() - logger.info('Doing post-backup jobs for %s', profile.name) - if profile.prune_on: - msg = BorgPruneJob.prepare(profile) - if msg['ok']: - job = BorgPruneJob(msg['cmd'], msg, profile.repo.id) - self.app.jobs_manager.add_job(job) - - # Refresh archives - msg = BorgListRepoJob.prepare(profile) - if msg['ok']: - job = BorgListRepoJob(msg['cmd'], msg, profile.repo.id) - self.app.jobs_manager.add_job(job) - - validation_cutoff = dt.now() - timedelta(days=7 * profile.validation_weeks) - recent_validations = ( - EventLogModel.select() - .where( - (EventLogModel.subcommand == 'check') - & (EventLogModel.start_time > validation_cutoff) - & (EventLogModel.repo_url == profile.repo.url) - ) - .count() - ) - if profile.validation_on and recent_validations == 0: - msg = BorgCheckJob.prepare(profile) - if msg['ok']: - job = BorgCheckJob(msg['cmd'], msg, profile.repo.id) - self.app.jobs_manager.add_job(job) - - compaction_cutoff = dt.now() - timedelta(days=7 * profile.compaction_weeks) - recent_compactions = ( - EventLogModel.select() - .where( - (EventLogModel.subcommand == '--info') - & (EventLogModel.start_time > compaction_cutoff) - & (EventLogModel.repo_url == profile.repo.url) - ) - .count() - ) - - if ( - profile.compaction_on - and recent_compactions == 0 - and version.parse(borg_compat.version) >= version.parse("1.2") - ): - msg = BorgCompactJob.prepare(profile) - if msg['ok']: - job = BorgCompactJob(msg['cmd'], msg, profile.repo.id) - self.app.jobs_manager.add_job(job) - - logger.info('Finished background task for profile %s', profile.name) - notifier.deliver( - self.tr('Vorta Backup'), - self.tr('Post Backup Tasks successful for %s' % profile.name), - level='info', - ) + self._execution.post_backup_tasks(profile_id) diff --git a/src/vorta/scheduler/execution.py b/src/vorta/scheduler/execution.py new file mode 100644 index 000000000..672a06b07 --- /dev/null +++ b/src/vorta/scheduler/execution.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import logging +from datetime import datetime as dt +from datetime import timedelta +from typing import TYPE_CHECKING, Any + +from packaging import version + +from vorta.borg.check import BorgCheckJob +from vorta.borg.compact import BorgCompactJob +from vorta.borg.create import BorgCreateJob +from vorta.borg.list_repo import BorgListRepoJob +from vorta.borg.prune import BorgPruneJob +from vorta.i18n import translate +from vorta.notifications import VortaNotifications +from vorta.store.models import BackupProfileModel, EventLogModel, JobModel +from vorta.utils import borg_compat + +if TYPE_CHECKING: + from vorta.scheduler import VortaScheduler + +logger = logging.getLogger(__name__) + + +def worst_error(errors: list[tuple[int, str]] | None) -> str | None: + """The most severe message Borg logged, to show as the reason a run failed.""" + if not errors: + return None + + worst = max(level for level, _ in errors) + return next(message for level, message in errors if level == worst) + + +class SchedulerExecution: + """Submitted backup runs, their results and the post-backup tasks.""" + + def __init__(self, scheduler: VortaScheduler) -> None: + self.scheduler = scheduler + + #: profiles being submitted, so a timer tick cannot submit one twice + self._submitting: set[int] = set() + + def create_backup(self, profile_id: int, trigger: str) -> None: + notifier = VortaNotifications.pick() + profile = BackupProfileModel.get_or_none(id=profile_id) + + if profile is None: + logger.info('Profile not found. Maybe deleted?') + return + + if profile_id in self._submitting: + logger.debug('A run for profile %s is already being submitted.', profile_id) + return + + # Skip if a job for this profile (repo) is already in progress + if self.scheduler.app.jobs_manager.is_worker_running(site=profile.repo.id): + logger.debug('A job for repo %s is already active.', profile.repo.id) + self.scheduler.record_skip(profile, trigger, 'Repository is busy with another job.') + self.scheduler.pause(profile_id) + return + + self._submitting.add(profile_id) + try: + logger.info('Starting background backup for %s', profile.name) + notifier.deliver( + self.scheduler.tr('Vorta Backup'), + self.scheduler.tr('Starting background backup for %s.') % profile.name, + level='info', + ) + msg = BorgCreateJob.prepare(profile) + if msg['ok']: + logger.info('Preparation for backup successful.') + msg['category'] = 'scheduled' + msg['job_record_id'] = self.scheduler.record_start(profile, trigger) + # The timer that fired still holds this run as pending; `notify` re-arms it afterwards. + self.scheduler.remove_job(profile_id) + self.scheduler.schedule_changed.emit() + job = BorgCreateJob(msg['cmd'], msg, profile.repo.id) + job.result.connect(self.scheduler.notify) + self.scheduler.app.jobs_manager.add_job(job) + else: + # Default to 'error': unexpected failures notify. + # Expected skips (WiFi/metered) use 'info' to suppress. + level = msg.get('level', 'error') + if level == 'error': + logger.error('Conditions for backup not met. Aborting.') + logger.error(msg['message']) + notifier.deliver( + self.scheduler.tr('Vorta Backup'), + translate('messages', msg['message']), + level='error', + ) + status = JobModel.Status.FAILED.value + else: + logger.info('Backup skipped: %s', msg['message']) + status = JobModel.Status.SKIPPED.value + self.scheduler.record_skip(profile, trigger, msg['message'], status=status) + self.scheduler.pause(profile_id) + finally: + self._submitting.discard(profile_id) + + def notify(self, result: dict[str, Any]) -> None: + notifier = VortaNotifications.pick() + profile_name = result['params']['profile_name'] + profile_id = result['params']['profile'].id + succeeded = result['returncode'] in [0, 1] + + record_id = result['params'].get('job_record_id') + if record_id is not None: + status = JobModel.Status.COMPLETED.value if succeeded else JobModel.Status.FAILED.value + reason = None if succeeded else worst_error(result.get('errors')) + self.scheduler.record_finish(record_id, status, result.get('log_entry_id'), reason) + + if succeeded: + notifier.deliver( + self.scheduler.tr('Vorta Backup'), + self.scheduler.tr('Backup successful for %s.') % profile_name, + level='info', + ) + logger.info('Backup creation successful.') + # unpause scheduler + self.scheduler.unpause(result['params']['profile_id']) + + self.scheduler.post_backup_tasks(profile_id) + else: + notifier.deliver( + self.scheduler.tr('Vorta Backup'), + self.scheduler.tr('Error during backup creation for %s.') % profile_name, + level='error', + ) + logger.error('Error during backup creation.') + # pause scheduler + # if a scheduled backup fails the scheduler should pause + # temporarily. + self.scheduler.pause(result['params']['profile_id']) + + self.scheduler.set_timer_for_profile(profile_id) + + def post_backup_tasks(self, profile_id: int) -> None: + """ + Pruning and checking after successful backup. + """ + profile = BackupProfileModel.get_or_none(id=profile_id) + if profile is None: + logger.info('Profile not found. Maybe deleted?') + return + + notifier = VortaNotifications.pick() + logger.info('Doing post-backup jobs for %s', profile.name) + if profile.prune_on: + msg = BorgPruneJob.prepare(profile) + if msg['ok']: + job = BorgPruneJob(msg['cmd'], msg, profile.repo.id) + self.scheduler.app.jobs_manager.add_job(job) + + # Refresh archives + msg = BorgListRepoJob.prepare(profile) + if msg['ok']: + job = BorgListRepoJob(msg['cmd'], msg, profile.repo.id) + self.scheduler.app.jobs_manager.add_job(job) + + validation_cutoff = dt.now() - timedelta(days=7 * profile.validation_weeks) + recent_validations = ( + EventLogModel.select() + .where( + (EventLogModel.subcommand == 'check') + & (EventLogModel.start_time > validation_cutoff) + & (EventLogModel.repo_url == profile.repo.url) + ) + .count() + ) + if profile.validation_on and recent_validations == 0: + msg = BorgCheckJob.prepare(profile) + if msg['ok']: + job = BorgCheckJob(msg['cmd'], msg, profile.repo.id) + self.scheduler.app.jobs_manager.add_job(job) + + compaction_cutoff = dt.now() - timedelta(days=7 * profile.compaction_weeks) + recent_compactions = ( + EventLogModel.select() + .where( + (EventLogModel.subcommand == '--info') + & (EventLogModel.start_time > compaction_cutoff) + & (EventLogModel.repo_url == profile.repo.url) + ) + .count() + ) + + if ( + profile.compaction_on + and recent_compactions == 0 + and version.parse(borg_compat.version) >= version.parse("1.2") + ): + msg = BorgCompactJob.prepare(profile) + if msg['ok']: + job = BorgCompactJob(msg['cmd'], msg, profile.repo.id) + self.scheduler.app.jobs_manager.add_job(job) + + logger.info('Finished background task for profile %s', profile.name) + notifier.deliver( + self.scheduler.tr('Vorta Backup'), + self.scheduler.tr('Post Backup Tasks successful for %s' % profile.name), + level='info', + ) diff --git a/src/vorta/scheduler/state.py b/src/vorta/scheduler/state.py index 2ec2e8b0a..4bec6eda8 100644 --- a/src/vorta/scheduler/state.py +++ b/src/vorta/scheduler/state.py @@ -224,12 +224,7 @@ def record_skip( 'status': status, 'scheduled_at': scheduled_at, } - details = { - 'profile_name': profile.name, - 'repo_url': profile.repo.url if profile.repo else None, - 'job_type': JobModel.Type.BACKUP.value, - 'reason': reason, - } + details = {**self._job_fields(profile), 'reason': reason} try: if scheduled_at is None: @@ -242,5 +237,51 @@ def record_skip( logger.warning('Could not record job for profile %s.', profile.id, exc_info=True) return - # `arm_profile` records under the scheduler's lock, and the jobs view reads the table here. + self._announce_job() + + def _job_fields(self, profile: BackupProfileModel) -> dict[str, str | None]: + return { + 'profile_name': profile.name, + 'repo_url': profile.repo.url if profile.repo else None, + 'job_type': JobModel.Type.BACKUP.value, + } + + def _announce_job(self) -> None: + # A caller can be holding the scheduler's lock, as `arm_profile` is, and the jobs view reads the table here. QTimer.singleShot(0, self.scheduler.jobs_changed.emit) + + def record_start(self, profile: BackupProfileModel, trigger: str) -> int | None: + """Record a submitted run, returning the row id its outcome will settle.""" + try: + job = JobModel.create( + profile=str(profile.id), + trigger=trigger, + status=JobModel.Status.RUNNING.value, + **self._job_fields(profile), + ) + except pw.PeeweeException: + logger.warning('Could not record run for profile %s.', profile.id, exc_info=True) + return None + + self._announce_job() + return job.id + + def record_finish( + self, record_id: int, status: str, log_entry_id: int | None = None, reason: str | None = None + ) -> None: + """Settle a recorded run with its outcome and the log entry it produced.""" + try: + settled = ( + JobModel.update(status=status, event_log=log_entry_id, reason=reason) + .where(JobModel.id == record_id) + .execute() + ) + except pw.PeeweeException: + logger.warning('Could not settle job record %s.', record_id, exc_info=True) + return + + if not settled: + logger.warning('Job record %s was gone before its outcome could be stored.', record_id) + return + + self._announce_job() diff --git a/src/vorta/store/connection.py b/src/vorta/store/connection.py index 86c966874..b90df743c 100644 --- a/src/vorta/store/connection.py +++ b/src/vorta/store/connection.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import os import shutil from datetime import datetime, timedelta @@ -29,6 +30,8 @@ ) from .settings import get_misc_settings +logger = logging.getLogger(__name__) + SCHEMA_VERSION = 23 @@ -44,6 +47,17 @@ def cleanup_db() -> None: DB.close() +def recover_interrupted_jobs() -> None: + """Runs still marked as running at startup belong to a process that died mid-backup.""" + try: + JobModel.update( + status=JobModel.Status.INTERRUPTED.value, + reason='Vorta stopped while this backup was running.', + ).where(JobModel.status == JobModel.Status.RUNNING.value).execute() + except pw.PeeweeException: + logger.warning('Could not recover interrupted jobs.', exc_info=True) + + def init_db(con: pw.SqliteDatabase | None = None) -> None: if con is not None: os.umask(0o0077) @@ -91,6 +105,8 @@ def init_db(con: pw.SqliteDatabase | None = None) -> None: # Delete old job records after 6 months. Nothing derives scheduling state from them. JobModel.delete().where(JobModel.created_at < six_months_ago).execute() + recover_interrupted_jobs() + # Migrations current_schema, created = SchemaVersion.get_or_create(id=1, defaults={'version': SCHEMA_VERSION}) current_schema.save() diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 7a4aea144..e96981a15 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -131,6 +131,7 @@ def init_db(qapp, qtbot, tmpdir_factory, request): # Using disconnect_all() instead of disconnect() to ensure ALL handlers are removed, # not just one (which can leave stale connections from previous tests) disconnect_all(qapp.scheduler.schedule_changed) + disconnect_all(qapp.scheduler.jobs_changed) # Reload the window to apply the mock data # If this test has the `window_load` fixture, @@ -158,6 +159,7 @@ def init_db(qapp, qtbot, tmpdir_factory, request): # Disconnect signals disconnect_all(qapp.backup_finished_event) disconnect_all(qapp.scheduler.schedule_changed) + disconnect_all(qapp.scheduler.jobs_changed) # Clear the workers dict to prevent accumulation of dead thread references qapp.jobs_manager.workers.clear() diff --git a/tests/unit/test_schedule.py b/tests/unit/test_schedule.py index e31660cdc..d9fe356a4 100644 --- a/tests/unit/test_schedule.py +++ b/tests/unit/test_schedule.py @@ -7,6 +7,7 @@ from PyQt6.QtWidgets import QWidget import vorta.scheduler +import vorta.scheduler.execution import vorta.scheduler.scheduling import vorta.scheduler.state from vorta.application import VortaApp @@ -20,7 +21,7 @@ @pytest.fixture def clockmock(monkeypatch): datetime_mock = MagicMock(wraps=dt) - for module in (vorta.scheduler, vorta.scheduler.scheduling, vorta.scheduler.state): + for module in (vorta.scheduler, vorta.scheduler.execution, vorta.scheduler.scheduling, vorta.scheduler.state): monkeypatch.setattr(module, "dt", datetime_mock) return datetime_mock diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index b9ad4c642..884108346 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -10,9 +10,11 @@ import vorta.borg import vorta.scheduler +import vorta.scheduler.execution import vorta.scheduler.scheduling import vorta.scheduler.state from vorta.scheduler import PendingJob, ScheduleStatus, ScheduleStatusType, VortaScheduler +from vorta.store.connection import recover_interrupted_jobs from vorta.store.models import BackupProfileModel, EventLogModel, JobModel, SchedulerPauseModel PROFILE_NAME = 'Default' @@ -24,7 +26,7 @@ @pytest.fixture def clockmock(monkeypatch): datetime_mock = MagicMock(wraps=dt) - for module in (vorta.scheduler, vorta.scheduler.scheduling, vorta.scheduler.state): + for module in (vorta.scheduler, vorta.scheduler.execution, vorta.scheduler.scheduling, vorta.scheduler.state): monkeypatch.setattr(module, "dt", datetime_mock) return datetime_mock @@ -59,7 +61,7 @@ def do(qapp, qtbot, mocker, borg_json_output): @prepare def test_scheduler_create_backup(qapp, qtbot, mocker, borg_json_output): - """Test running a backup with `create_backup`.""" + """A run submitted by the scheduler is recorded, then settled as completed against its log entry.""" events_before = EventLogModel.select().count() with qtbot.waitSignal(qapp.backup_finished_event, **pytest._wait_defaults): @@ -67,6 +69,15 @@ def test_scheduler_create_backup(qapp, qtbot, mocker, borg_json_output): assert EventLogModel.select().count() == events_before + 1 + # `backup_finished_event` is emitted just before the `result` signal that carries the run into `notify`. + job = JobModel.select().order_by(JobModel.id.desc()).get() + qtbot.waitUntil(lambda: JobModel.get_by_id(job.id).status == JobModel.Status.COMPLETED.value) + + job = JobModel.get_by_id(job.id) + assert job.trigger == JobModel.Trigger.SCHEDULED.value + # Post-backup tasks log runs of their own, so match the linked entry rather than the newest row. + assert job.event_log.subcommand == 'create' + def test_manual_mode(): """Test scheduling in manual mode.""" @@ -289,7 +300,7 @@ def test_deleting_a_paused_profile_clears_the_pause(qapp, qtbot, mocker): qapp.scheduler.pause(profile.id) assert SchedulerPauseModel.get_or_none(profile=profile.id) is not None - prepare_mock = mocker.patch('vorta.scheduler.BorgCreateJob.prepare') + prepare_mock = mocker.patch('vorta.scheduler.execution.BorgCreateJob.prepare') mocker.patch.object(QMessageBox, 'question', return_value=QMessageBox.StandardButton.Yes) mocker.patch.object(qapp.scheduler._timers, '_net_up', True) qtbot.mouseClick(main.profileDeleteButton, QtCore.Qt.MouseButton.LeftButton) @@ -614,7 +625,7 @@ def test_create_backup_no_error_notification_on_info_level(qapp, qtbot, mocker, """Test that notifier.deliver() is not called with level='error' when prepare() returns level='info' (e.g. WiFi disallowed or metered connection).""" mocker.patch( - 'vorta.scheduler.BorgCreateJob.prepare', + 'vorta.scheduler.execution.BorgCreateJob.prepare', return_value={ 'ok': False, 'message': 'Current Wifi is not allowed.', @@ -634,7 +645,7 @@ def test_create_backup_no_error_notification_on_info_level(qapp, qtbot, mocker, def test_create_backup_records_skip_reason(qapp, qtbot, mocker): """A skipped scheduled backup is recorded as a JobModel row with its reason.""" mocker.patch( - 'vorta.scheduler.BorgCreateJob.prepare', + 'vorta.scheduler.execution.BorgCreateJob.prepare', return_value={ 'ok': False, 'message': 'Current Wifi is not allowed.', @@ -664,7 +675,7 @@ def test_recording_a_skip_emits_jobs_changed(qapp, qtbot, mocker): def test_create_backup_records_failure_not_skip(qapp, qtbot, mocker): """An unexpected prepare() failure is recorded as failed, not skipped.""" mocker.patch( - 'vorta.scheduler.BorgCreateJob.prepare', + 'vorta.scheduler.execution.BorgCreateJob.prepare', return_value={ 'ok': False, 'message': 'Add a backup repository first.', @@ -693,6 +704,138 @@ def test_create_backup_records_skip_when_repo_busy(qapp, mocker): assert job.reason == 'Repository is busy with another job.' +@prepare +def test_create_backup_records_a_running_job(qapp, qtbot, mocker, borg_json_output): + """A submitted run is on the jobs table as running before its outcome is known.""" + started = [] + real_add_job = qapp.jobs_manager.add_job + + def capture(job): + record = JobModel.select().order_by(JobModel.id.desc()).get() + started.append((record.status, record.trigger, record.profile_name)) + return real_add_job(job) + + mocker.patch.object(qapp.jobs_manager, 'add_job', side_effect=capture) + + with qtbot.waitSignal(qapp.backup_finished_event, **pytest._wait_defaults): + qapp.scheduler.create_backup(1) + + assert started == [(JobModel.Status.RUNNING.value, JobModel.Trigger.SCHEDULED.value, PROFILE_NAME)] + + # `notify` arrives after `backup_finished_event`; let it settle here rather than in the next test. + record = JobModel.select().order_by(JobModel.id.desc()).get() + qtbot.waitUntil(lambda: JobModel.get_by_id(record.id).status == JobModel.Status.COMPLETED.value) + + +def test_a_failed_run_is_recorded_as_failed(qapp, qtbot, mocker, borg_json_output): + """A non-zero return code settles the recorded run as failed rather than skipped.""" + stdout, stderr = borg_json_output('create') + popen_result = mocker.MagicMock(stdout=stdout, stderr=stderr, returncode=2) + mocker.patch.object(vorta.borg.borg_job, 'Popen', return_value=popen_result) + + with qtbot.waitSignal(qapp.backup_finished_event, **pytest._wait_defaults): + qapp.scheduler.create_backup(1) + + job = JobModel.select().order_by(JobModel.id.desc()).get() + qtbot.waitUntil(lambda: JobModel.get_by_id(job.id).status == JobModel.Status.FAILED.value) + + +@prepare +def test_recording_a_run_emits_jobs_changed(qapp, qtbot, mocker, borg_json_output): + """The jobs view refreshes off this signal, so submitting a run has to announce itself.""" + # Silence the settling emit, so only the submission can satisfy the wait. + settle = mocker.patch.object(qapp.scheduler, 'record_finish') + + with qtbot.waitSignal(qapp.scheduler.jobs_changed, timeout=1000): + qapp.scheduler.create_backup(1) + + # The run is still in flight; drain it here so `notify` cannot land in the next test. + qtbot.waitUntil(lambda: settle.called) + + +@mark.parametrize( + 'errors, expected', + [ + ([], None), + (None, None), + ([(30, 'a warning')], 'a warning'), + ([(30, 'a warning'), (40, 'the real problem'), (40, 'a later one')], 'the real problem'), + ], +) +def test_worst_error_picks_the_most_severe_message(errors, expected): + """A failed run shows the worst thing Borg said, not the first or last thing.""" + assert vorta.scheduler.execution.worst_error(errors) == expected + + +@prepare +def test_a_submitted_run_stops_being_pending(qapp, qtbot, mocker, borg_json_output): + """The fired timer still holds the run, so without dropping it the page lists the run twice.""" + 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() + + EventLogModel.create( + subcommand='create', + profile=profile.id, + returncode=0, + category='scheduled', + start_time=dt.now(), + end_time=dt.now(), + ) + qapp.scheduler.set_timer_for_profile(profile.id) + assert len(qapp.scheduler.pending_jobs()) == 1 + + pending_during_run = [] + real_add_job = qapp.jobs_manager.add_job + + def capture(job): + pending_during_run.append(qapp.scheduler.pending_jobs()) + return real_add_job(job) + + mocker.patch.object(qapp.jobs_manager, 'add_job', side_effect=capture) + + with qtbot.waitSignal(qapp.backup_finished_event, **pytest._wait_defaults): + qapp.scheduler.create_backup(profile.id) + + assert pending_during_run == [[]] + + record = JobModel.select().order_by(JobModel.id.desc()).get() + qtbot.waitUntil(lambda: JobModel.get_by_id(record.id).status == JobModel.Status.COMPLETED.value) + + +def test_interrupted_runs_are_recovered_at_startup(qapp): + """A row still marked running belongs to a process that died mid-backup.""" + profile = BackupProfileModel.get(name=PROFILE_NAME) + stale = JobModel.create( + profile=str(profile.id), + profile_name=profile.name, + status=JobModel.Status.RUNNING.value, + trigger=JobModel.Trigger.SCHEDULED.value, + ) + + settled = JobModel.create( + profile=str(profile.id), + profile_name=profile.name, + status=JobModel.Status.COMPLETED.value, + trigger=JobModel.Trigger.SCHEDULED.value, + ) + + recover_interrupted_jobs() + + stale = JobModel.get_by_id(stale.id) + assert stale.status == JobModel.Status.INTERRUPTED.value + assert stale.reason == 'Vorta stopped while this backup was running.' + assert JobModel.get_by_id(settled.id).status == JobModel.Status.COMPLETED.value + + +def test_post_backup_tasks_ignores_a_deleted_profile(qapp): + """The profile can be deleted between the run finishing and its follow-up jobs starting.""" + qapp.scheduler.post_backup_tasks(-1) + + def test_create_backup_keeps_the_catchup_trigger(qapp, mocker): """A catch-up run that gets skipped is not recorded as an ordinary scheduled run.""" mocker.patch.object(qapp.jobs_manager, 'is_worker_running', return_value=True) @@ -791,7 +934,7 @@ def prepare(profile): held.append(scheduler.lock.locked()) return {'ok': False, 'message': 'Current Wifi is not allowed.', 'level': 'info'} - mocker.patch('vorta.scheduler.BorgCreateJob.prepare', side_effect=prepare) + mocker.patch('vorta.scheduler.execution.BorgCreateJob.prepare', side_effect=prepare) scheduler.create_backup(1) @@ -811,7 +954,7 @@ def prepare(profile): scheduler.create_backup(profile.id) return {'ok': False, 'message': 'Current Wifi is not allowed.', 'level': 'info'} - mocker.patch('vorta.scheduler.BorgCreateJob.prepare', side_effect=prepare) + mocker.patch('vorta.scheduler.execution.BorgCreateJob.prepare', side_effect=prepare) scheduler.create_backup(1)