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

Code before:
    '''
    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"),

Review comment (suggestion):
(cosmetic changes)
```suggestion
    ('freq', 'interval'),
    (
        pytest.param(MINUTELY, 15, id='every-15-minutes-minutely'),
        pytest.param(MINUTELY, 120, id='every-2-hours-minutely'),
        pytest.param(MINUTELY, 60 * 24 * 3, id='every-3-days-minutely'),
        pytest.param(HOURLY, 7, id='every-7-hours-hourly'),
        pytest.param(HOURLY, 24 * 3, id='every-3-days-hourly'),
    ),
```

Code after:
    'rrulestr',
    [
        pytest.param('DTSTART;TZID=America/New_York:20201118T200000 RRULE:FREQ=MINUTELY;INTERVAL=5', id='every-5-min'),
        pytest.param('DTSTART;TZID=America/New_York:20201118T200000 RRULE:FREQ=HOURLY;INTERVAL=5', id='every-5-hours'),
        pytest.param('DTSTART;TZID=America/New_York:20201118T200000 RRULE:FREQ=YEARLY;INTERVAL=5', id='every-5-years'),
        pytest.param(
            'DTSTART;TZID=America/New_York:20201118T200000 RRULE:FREQ=MINUTELY;INTERVAL=5;WKST=SU;BYMONTH=2,3;BYMONTHDAY=18;BYHOUR=5;BYMINUTE=35;BYSECOND=0',
            id='every-5-minutes-at-5:35:00-am-on-the-18th-day-of-feb-or-march-with-week-starting-on-sundays',
        ),
        pytest.param(
            'DTSTART;TZID=America/New_York:20201118T200000 RRULE:FREQ=HOURLY;INTERVAL=5;WKST=SU;BYMONTH=2,3;BYHOUR=5',
            id='every-5-hours-at-5-am-in-feb-or-march-with-week-starting-on-sundays',
        ),
    ],
)
def test_fast_forwarded_rrule_matches_original_occurrence(rrulestr):
    '''
    Assert that the resulting fast forwarded date is included in the original rrule
    occurrence list
    '''
    rruleset = Schedule.rrulestr(rrulestr, ref_dt=REF_DT)

    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_fast_forward_rrule_hours():
    '''
    Generate an rrule for each hour of the day

    Assert that the resulting fast forwarded date is included in the original rrule
    occurrence list
    '''
    rrulestr_prefix = 'DTSTART;TZID=America/New_York:20201118T200000 RRULE:FREQ=HOURLY;'
    for interval in range(1, 24):
        rrulestr = f"{rrulestr_prefix}INTERVAL={interval}"
        rruleset = Schedule.rrulestr(rrulestr, ref_dt=REF_DT)

        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]