forked from inventree/InvenTree
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_tasks.py
More file actions
119 lines (84 loc) · 4.11 KB
/
test_tasks.py
File metadata and controls
119 lines (84 loc) · 4.11 KB
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"""Unit tests for task management."""
from datetime import timedelta
from django.test import TestCase
from django.utils import timezone
from django_q.models import Schedule
from error_report.models import Error
import InvenTree.tasks
from common.models import InvenTreeSetting
threshold = timezone.now() - timedelta(days=30)
threshold_low = threshold - timedelta(days=1)
class ScheduledTaskTests(TestCase):
"""Unit tests for scheduled tasks."""
def get_tasks(self, name):
"""Helper function to get a Schedule object."""
return Schedule.objects.filter(func=name)
def test_add_task(self):
"""Ensure that duplicate tasks cannot be added."""
task = 'InvenTree.tasks.heartbeat'
self.assertEqual(self.get_tasks(task).count(), 0)
InvenTree.tasks.schedule_task(task, schedule_type=Schedule.MINUTES, minutes=10)
self.assertEqual(self.get_tasks(task).count(), 1)
t = Schedule.objects.get(func=task)
self.assertEqual(t.minutes, 10)
# Attempt to schedule the same task again
InvenTree.tasks.schedule_task(task, schedule_type=Schedule.MINUTES, minutes=5)
self.assertEqual(self.get_tasks(task).count(), 1)
# But the 'minutes' should have been updated
t = Schedule.objects.get(func=task)
self.assertEqual(t.minutes, 5)
def get_result():
"""Demo function for test_offloading."""
return 'abc'
class InvenTreeTaskTests(TestCase):
"""Unit tests for tasks."""
def test_offloading(self):
"""Test task offloading."""
# Run with function ref
InvenTree.tasks.offload_task(get_result)
# Run with string ref
InvenTree.tasks.offload_task('InvenTree.test_tasks.get_result')
# Error runs
# Malformed taskname
with self.assertWarnsMessage(UserWarning, "WARNING: 'InvenTree' not started - Malformed function path"):
InvenTree.tasks.offload_task('InvenTree')
# Non exsistent app
with self.assertWarnsMessage(UserWarning, "WARNING: 'InvenTreeABC.test_tasks.doesnotmatter' not started - No module named 'InvenTreeABC.test_tasks'"):
InvenTree.tasks.offload_task('InvenTreeABC.test_tasks.doesnotmatter')
# Non exsistent function
with self.assertWarnsMessage(UserWarning, "WARNING: 'InvenTree.test_tasks.doesnotexsist' not started - No function named 'doesnotexsist'"):
InvenTree.tasks.offload_task('InvenTree.test_tasks.doesnotexsist')
def test_task_hearbeat(self):
"""Test the task heartbeat."""
InvenTree.tasks.offload_task(InvenTree.tasks.heartbeat)
def test_task_delete_successful_tasks(self):
"""Test the task delete_successful_tasks."""
from django_q.models import Success
Success.objects.create(name='abc', func='abc', stopped=threshold, started=threshold_low)
InvenTree.tasks.offload_task(InvenTree.tasks.delete_successful_tasks)
results = Success.objects.filter(started__lte=threshold)
self.assertEqual(len(results), 0)
def test_task_delete_old_error_logs(self):
"""Test the task delete_old_error_logs."""
# Create error
error_obj = Error.objects.create()
error_obj.when = threshold_low
error_obj.save()
# Check that it is not empty
errors = Error.objects.filter(when__lte=threshold,)
self.assertNotEqual(len(errors), 0)
# Run action
InvenTree.tasks.offload_task(InvenTree.tasks.delete_old_error_logs)
# Check that it is empty again
errors = Error.objects.filter(when__lte=threshold,)
self.assertEqual(len(errors), 0)
def test_task_check_for_updates(self):
"""Test the task check_for_updates."""
# Check that setting should be empty
self.assertEqual(InvenTreeSetting.get_setting('INVENTREE_LATEST_VERSION'), '')
# Get new version
InvenTree.tasks.offload_task(InvenTree.tasks.check_for_updates)
# Check that setting is not empty
response = InvenTreeSetting.get_setting('INVENTREE_LATEST_VERSION')
self.assertNotEqual(response, '')
self.assertTrue(bool(response))