larigira/larigira/tests/test_time_every.py

from datetime import timedelta, datetime
from pprint import pprint

import pytest

from larigira.timegen_every import FrequencyAlarm, SingleAlarm
from larigira.timegen import timegenerate


def eq_(a, b, reason=None):
    """migrating tests from nose"""
    if reason is not None:
        assert a == b, reason
    else:
        assert a == b


@pytest.fixture
def now():
    return datetime.now()


@pytest.fixture(params=["seconds", "human", "humanlong", "coloned"])
def onehour(now, request):
    """a FrequencyAlarm: every hour for one day"""
    intervals = dict(
        seconds=3600, human="1h", humanlong="30m 1800s", coloned="01:00:00"
    )
    return FrequencyAlarm(
        {
            "start": now - timedelta(days=1),
            "interval": intervals[request.param],
            "end": now + days(1),
        }
    )


@pytest.fixture(params=[1, "1"])
def onehour_monday(request):
    weekday = request.param
    yield FrequencyAlarm({"interval": 3600 * 12, "weekdays": [weekday], "start": 0})


@pytest.fixture(params=[7, "7"])
def onehour_sunday(request):
    weekday = request.param
    yield FrequencyAlarm({"interval": 3600 * 12, "weekdays": [weekday], "start": 0})


@pytest.fixture(params=[1, 2, 3, 4, 5, 6, 7])
def singledow(request):
    weekday = request.param
    yield FrequencyAlarm({"interval": 3600 * 24, "weekdays": [weekday], "start": 0})


@pytest.fixture(params=["seconds", "human", "coloned"])
def tenseconds(now, request):
    """a FrequencyAlarm: every 10 seconds for one day"""
    intervals = dict(seconds=10, human="10s", coloned="00:10")
    return FrequencyAlarm(
        {
            "start": now - timedelta(days=1),
            "interval": intervals[request.param],
            "end": now + days(1),
        }
    )


@pytest.fixture(params=[1, 2, 3, 4, 5, 6, 7, 8])
def manyweeks(request):
    yield FrequencyAlarm({"interval": "{}w".format(request.param), "start": 0})


def days(n):
    return timedelta(days=n)


def test_single_creations(now):
    return SingleAlarm({"timestamp": now})


def test_freq_creations(now):
    return FrequencyAlarm(
        {"start": now - timedelta(days=1), "interval": 3600, "end": now}
    )


@pytest.mark.timeout(1)
def test_single_ring(now):
    dt = now + days(1)
    s = SingleAlarm({"timestamp": dt})
    eq_(s.next_ring(), dt)
    eq_(s.next_ring(now), dt)
    assert s.next_ring(dt) is None, "%s - %s" % (str(s.next_ring(dt)), str(dt))
    assert s.next_ring(now + days(2)) is None
    assert s.has_ring(dt)
    assert not s.has_ring(now)
    assert not s.has_ring(now + days(2))


@pytest.mark.timeout(1)
def test_single_all(now):
    dt = now + timedelta(days=1)
    s = SingleAlarm({"timestamp": dt})
    eq_(list(s.all_rings()), [dt])
    eq_(list(s.all_rings(now)), [dt])
    eq_(list(s.all_rings(now + days(2))), [])


def test_freq_short(now, tenseconds):
    f = tenseconds
    assert now in f.all_rings(now - days(3))
    assert f.next_ring(now) is not None
    assert f.next_ring(now) != now
    assert f.next_ring(now) > now
    assert now not in f.all_rings(now)
    for r in f.all_rings(now):
        assert r > now


@pytest.mark.timeout(1)
def test_freq_ring(now, onehour):
    f = onehour
    assert now in f.all_rings(now - days(3))
    assert f.next_ring(now) is not None
    assert f.next_ring(now) != now
    assert f.next_ring(now) > now
    assert now not in f.all_rings(now)
    for r in f.all_rings(now):
        assert r > now
    allr = list(f.all_rings(now))
    eq_(len(allr), 24)

    eq_(len(tuple(f.all_rings(now + days(2)))), 0)

    allr = tuple(f.all_rings(now - days(20)))
    eq_(f.next_ring(now - days(20)), now - days(1))
    eq_(len(allr), 49, pprint(allr))


def test_weekday_skip(onehour_monday):
    t = datetime.fromtimestamp(0)
    for _ in range(20):  # 20 is an arbitrary number
        t = onehour_monday.next_ring(t)
        assert t.isoweekday() == 1  # monday; don't get confused by .weekday()


def test_weekday_skip_2(onehour_sunday):
    t = datetime.fromtimestamp(0)
    for _ in range(20):  # 20 is an arbitrary number
        t = onehour_sunday.next_ring(t)
        assert t.isoweekday() == 7  # monday; don't get confused by .weekday()


def test_sunday_is_not_0():
    with pytest.raises(ValueError) as excinfo:
        FrequencyAlarm({"interval": 3600 * 12, "weekdays": [0], "start": 0})
    assert "Not a valid weekday:" in excinfo.value.args[0]


def test_long_interval(manyweeks):
    t = datetime.fromtimestamp(1)
    expected = manyweeks.interval
    got = manyweeks.next_ring(t)
    assert got is not None
    assert int(got.strftime("%s")) == expected
    assert manyweeks.next_ring(got) is not None


def test_singledow(singledow):
    t = datetime.fromtimestamp(1)
    got = singledow.next_ring(t)
    wd = singledow.weekdays[0]
    assert got is not None
    assert got.isoweekday() == wd

    assert singledow.next_ring(got) is not None
    assert singledow.next_ring(got).isoweekday() == wd


def test_single_registered():
    timegenerate({"kind": "single", "timestamp": 1234567890})


def test_frequency_registered():
    timegenerate({"kind": "frequency", "start": 1234567890, "interval": 60 * 15})