vorta/tests/test_scheduler.py

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