tasktiger/tests/test_task.py at eaef415eae0ae9bd54ade3ad90a10c56e811cb20 · closeio/tasktiger · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
import datetime

import pytest
from freezefrog import FreezeTime

from tasktiger import Task, TaskNotFound

from .tasks import simple_task
from .utils import get_tiger


@pytest.fixture
def tiger():
    return get_tiger()


class TestTaskFromId:
    @pytest.fixture
    def queued_task(self, tiger):
        return tiger.delay(simple_task)

    def test_task_found(self, tiger, queued_task):
        task = Task.from_id(tiger, "default", "queued", queued_task.id)
        assert queued_task.id == task.id

    def test_task_wrong_state(self, tiger, queued_task):
        with pytest.raises(TaskNotFound):
            Task.from_id(tiger, "default", "active", queued_task.id)

    def test_task_wrong_queue(self, tiger, queued_task):
        with pytest.raises(TaskNotFound):
            Task.from_id(tiger, "other", "active", queued_task.id)


class TestTaskMaxTrackedExecutions:
    def test_max_stored_executions_passed_to_tiger_delay(self, tiger):
        task = tiger.delay(simple_task, max_stored_executions=17)
        assert task.max_stored_executions == 17

    def test_max_stored_executions_passed_to_decorator(self, tiger):
        @tiger.task(max_stored_executions=17)
        def some_task():
            pass

        task = some_task.delay()
        assert task.max_stored_executions == 17

    def test_max_stored_executions_overridden_in_tiger_delay(self, tiger):
        @tiger.task(max_stored_executions=17)
        def some_task():
            pass

        task = tiger.delay(some_task, max_stored_executions=11)
        assert task.max_stored_executions == 11


class TestScheduledAt:
    FROZEN_NOW = datetime.datetime(2024, 1, 1, 12, 0, 0)

    def test_immediate_task_scheduled_at_equals_queue_time(self, tiger):
        with FreezeTime(self.FROZEN_NOW):
            task = tiger.delay(simple_task)
        assert task.scheduled_at == self.FROZEN_NOW

    def test_future_task_scheduled_at_equals_when(self, tiger):
        future = datetime.timedelta(minutes=5)
        with FreezeTime(self.FROZEN_NOW):
            task = tiger.delay(simple_task, when=future)
        assert task.scheduled_at == self.FROZEN_NOW + future

    def test_scheduled_at_survives_scheduled_to_queued_transition(self, tiger):
        future = datetime.timedelta(minutes=5)
        with FreezeTime(self.FROZEN_NOW):
            task = tiger.delay(simple_task, when=future)
        expected = self.FROZEN_NOW + future

        task._move(from_state="scheduled", to_state="queued")
        reloaded = Task.from_id(tiger, task.queue, "queued", task.id)
        assert reloaded.scheduled_at == expected

    def test_scheduled_at_persists_after_reload(self, tiger):
        with FreezeTime(self.FROZEN_NOW):
            task = tiger.delay(simple_task)
        reloaded = Task.from_id(tiger, task.queue, "queued", task.id)
        assert reloaded.scheduled_at == self.FROZEN_NOW

    def test_scheduled_at_none_for_unqueued_task(self, tiger):
        task = Task(tiger, simple_task)
        assert task.scheduled_at is None

    def test_update_scheduled_time_updates_scheduled_at(self, tiger):
        future = datetime.timedelta(minutes=5)
        later = datetime.timedelta(minutes=10)
        with FreezeTime(self.FROZEN_NOW):
            task = tiger.delay(simple_task, when=future)
        assert task.scheduled_at == self.FROZEN_NOW + future

        new_when = self.FROZEN_NOW + later
        task.update_scheduled_time(when=new_when)

        reloaded = Task.from_id(tiger, task.queue, "scheduled", task.id)
        assert reloaded.scheduled_at == new_when