J.A.R.V.I.S-py/tests/test_scheduler_store.py

"""Tests for scheduler_store: parser + add/remove + serde."""

import time
import pytest


def test_parse_daily(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = s.parse_schedule("daily 09:00")
    assert schedule == {'kind': 'daily', 'hour': 9, 'minute': 0}


def test_parse_at_today_or_tomorrow(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = s.parse_schedule("at 23:59")
    assert schedule['kind'] == 'once'
    assert schedule['at'] > time.time()


def test_parse_every_minutes(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = s.parse_schedule("every 30 minutes")
    assert schedule == {'kind': 'interval', 'seconds': 1800}


def test_parse_every_russian(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = s.parse_schedule("every 2 часа")
    assert schedule == {'kind': 'interval', 'seconds': 7200}


def test_parse_in_minutes_is_once(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = s.parse_schedule("in 5 minutes")
    assert schedule['kind'] == 'once'
    # Should be roughly 300 seconds in the future
    delta = schedule['at'] - time.time()
    assert 290 <= delta <= 310


def test_parse_bad_unit_raises(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    with pytest.raises(ValueError):
        s.parse_schedule("every 5 weeks")


def test_parse_bad_time_raises(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    with pytest.raises(ValueError):
        s.parse_schedule("daily 25:00")
    with pytest.raises(ValueError):
        s.parse_schedule("at 12:99")


def test_add_remove(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    tid = s.add({
        'name': 'Test',
        'schedule': 'in 1 hour',
        'action': {'type': 'speak', 'text': 'hello'},
    })
    assert tid
    tasks = s.list_all()
    assert len(tasks) == 1
    assert tasks[0]['name'] == 'Test'

    assert s.remove(tid) is True
    assert s.list_all() == []
    assert s.remove(tid) is False  # idempotent


def test_remove_by_text(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    s.add({
        'name': 'Water',
        'schedule': 'every 2 hours',
        'action': {'type': 'speak', 'text': 'Попей воды.'},
    })
    s.add({
        'name': 'Stretch',
        'schedule': 'every 50 minutes',
        'action': {'type': 'speak', 'text': 'Размяться.'},
    })

    removed = s.remove_by_text("воды")
    assert removed == 1
    assert len(s.list_all()) == 1


def test_clear(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    s.add({'name': 'a', 'schedule': 'in 1 hour', 'action': {'type': 'speak', 'text': 'x'}})
    s.add({'name': 'b', 'schedule': 'in 2 hours', 'action': {'type': 'speak', 'text': 'y'}})
    assert s.clear() == 2
    assert s.list_all() == []


def test_next_fire_once(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = {'kind': 'once', 'at': 500}
    assert s._next_fire(schedule, None, 1000) == 500
    # Already fired
    assert s._next_fire(schedule, 500, 1000) is None


def test_next_fire_interval(isolated_state):
    s = isolated_state('scheduler_store', ['_PATH'])
    schedule = {'kind': 'interval', 'seconds': 60}
    assert s._next_fire(schedule, 100, 1000) == 160