Repository: ansible/awx
File: awx/main/tests/unit/utils/test_schedule_fast_forward.py (Python)

Code before:
import pytest
import datetime

from django.utils.timezone import now

from awx.main.models.schedules import fast_forward_date
from dateutil.rrule import rrule, HOURLY, MINUTELY, MONTHLY


def test_fast_forward_date_all_hours():
    '''
    Assert that the resulting fast forwarded date is included in the original rrule
    occurrence list
    '''
    for interval in range(1, 24):
        dtstart = now() - datetime.timedelta(days=30)
        rule = rrule(freq=HOURLY, interval=interval, dtstart=dtstart)
        new_datetime = fast_forward_date(rule)

        # get occurrences of the rrule
        gen = rule.xafter(new_datetime - datetime.timedelta(days=1), count=100)
        found_matching_date = False
        for occurrence in gen:
            if occurrence == new_datetime:
                found_matching_date = True
                break

        assert found_matching_date


@pytest.mark.parametrize(
    'freq, interval',
    [
        (MINUTELY, 15),
        (MINUTELY, 120),
        (MINUTELY, 60 * 24 * 3),
        (HOURLY, 7),
        (HOURLY, 24 * 3),
    ],
)
def test_fast_forward_date(freq, interval):
    '''
    Assert that the resulting fast forwarded date is included in the original rrule
    occurrence list
    '''
    dtstart = now() - datetime.timedelta(days=30)
    rule = rrule(freq=freq, interval=interval, dtstart=dtstart)
    new_datetime = fast_forward_date(rule)

    # get occurrences of the rrule
    gen = rule.xafter(new_datetime - datetime.timedelta(days=1), count=100)
    found_matching_date = False
    for occurrence in gen:
        if occurrence == new_datetime:
            found_matching_date = True
            break

    assert found_matching_date


@pytest.mark.parametrize(
    'freq, interval, error',
    [
        (MINUTELY, 15.5555, "interval is a fraction of a second"),
        (MONTHLY, 1, "frequency must be HOURLY or MINUTELY"),
        (HOURLY, 24 * 30, "interval is greater than the fast forward amount"),
    ],
)
def test_error_fast_forward_date(freq, interval, error):
    dtstart = now() - datetime.timedelta(days=30)
    rule = rrule(freq=freq, interval=interval, dtstart=dtstart)
    if error:
        with pytest.raises(Exception) as e_info:
            fast_forward_date(rule)

        assert error in e_info.value.args[0]


def test_very_old_date():
    dtstart = now() - datetime.timedelta(days=365 * 10)
    rule = rrule(freq=HOURLY, interval=11, dtstart=dtstart)
    new_datetime = fast_forward_date(rule)

    # get occurrences of the rrule
    gen = rule.xafter(new_datetime - datetime.timedelta(days=1), count=100)
    found_matching_date = False
    for occurrence in gen:
        if occurrence == new_datetime:
            found_matching_date = True
            break

    assert found_matching_date


Review comment (suggestion):
It's best to avoid convoluted logic in tests. Error is always truthy, so the conditional is not really needed. Additionally, the right way of checking the exceptions raised is to narrow the expectation. `pytest.raises()` uses (and `pytest.warns()` or `pytest.deprecated()` for that matter) should always have a `match` argument to be as accurate as possible. If there's different expected exceptions, it's possible to bake them into parametrize. But let's start with the following:
```suggestion
@pytest.mark.parametrize(
    ('freq', 'interval', 'error'),
    (
        pytest.param(MINUTELY, 15.5555, r"^interval is a fraction of a second$", id='fraction-of-sec'),
        pytest.param(MONTHLY, 1, r"^frequency must be HOURLY or MINUTELY$", id='monthly'),
        pytest.param(HOURLY, 24 * 30, r"^interval is greater than the fast forward amount$", id='over-a-month'),
    ),
)
def test_error_fast_forward_date(freq, interval, error):
    dtstart = now() - datetime.timedelta(days=30)
    rule = rrule(freq=freq, interval=interval, dtstart=dtstart)

    with pytest.raises(ValueError, match=error):
        fast_forward_date(rule)
```

I'm also concerned by the use of `now()` as it may contribute to the flakiness of the tests. It's typical to freeze time in such cases.

Code after:
    rruleA: freq HOURLY interval 5, dtstart should be fast forwarded
    rruleB: freq HOURLY interval 7, dtstart should be fast forwarded
    rruleC: freq MONTHLY interval 1, dtstart should not be fast forwarded
    exruleA: freq HOURLY interval 5, dtstart should be fast forwarded
    '''
    rrulestr = '''DTSTART;TZID=America/New_York:20201118T200000
                RRULE:FREQ=HOURLY;INTERVAL=5
                RRULE:FREQ=HOURLY;INTERVAL=7
                RRULE:FREQ=MONTHLY
                EXRULE:FREQ=HOURLY;INTERVAL=5;BYDAY=MO,TU,WE'''
    rruleset = Schedule.rrulestr(rrulestr, ref_dt=REF_DT)

    rruleA, rruleB, rruleC = rruleset._rrule
    exruleA = rruleset._exrule[0]

    # assert that each rrule has its own dtstart
    assert rruleA._dtstart != rruleB._dtstart
    assert rruleA._dtstart != rruleC._dtstart

    assert exruleA._dtstart == rruleA._dtstart

    # the new dtstart should be within INTERVAL amount of hours from REF_DT
    assert (REF_DT - rruleA._dtstart) < datetime.timedelta(hours=6)
    assert (REF_DT - rruleB._dtstart) < datetime.timedelta(hours=8)
    assert (REF_DT - exruleA._dtstart) < datetime.timedelta(hours=6)

    # the freq=monthly rrule's dtstart should not have changed
    dateutil_rruleset = dateutil.rrule.rrulestr(rrulestr, forceset=True)
    assert rruleC._dtstart == dateutil_rruleset._rrule[2]._dtstart

    gen = rruleset.xafter(REF_DT, count=200)
    occurrences = [i for i in gen]

    orig_rruleset = dateutil.rrule.rrulestr(rrulestr, forceset=True)
    gen = orig_rruleset.xafter(REF_DT, count=200)
    orig_occurrences = [i for i in gen]

    assert occurrences == orig_occurrences


def test_future_date_does_not_fast_forward():
    dtstart = now() + datetime.timedelta(days=30)
    rrule = dateutil.rrule.rrule(freq=HOURLY, interval=7, dtstart=dtstart)
    new_rrule = _fast_forward_rrule(rrule, ref_dt=REF_DT)
    assert new_rrule == rrule


@pytest.mark.parametrize(
    ('freq', 'interval'),
    [
        pytest.param(MINUTELY, 15.5555, id="freq-MINUTELY-interval-15.5555"),