-
Notifications
You must be signed in to change notification settings - Fork 115
Expand file tree
/
Copy pathtest_routes.py
More file actions
130 lines (83 loc) · 3.62 KB
/
Copy pathtest_routes.py
File metadata and controls
130 lines (83 loc) · 3.62 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
120
121
122
123
124
125
126
127
128
129
130
from builtins import str
from builtins import bool
import json
import pytest
def test_routes_taskexceptions(worker, api):
task_path = "tests.tasks.general.RaiseException"
worker.send_task(task_path, {
"message": "xyz"}, block=True, accept_statuses=["failed"])
data, _ = api.GET("/api/datatables/taskexceptions?sEcho=1")
assert len(data["aaData"]) == 1
assert data["aaData"][0]["_id"]["path"] == task_path
def test_routes_status(worker, api):
worker.send_task("tests.tasks.general.RaiseException", {
"message": "xyz"}, block=True, accept_statuses=["failed"])
data, _ = api.GET("/api/datatables/status?sEcho=1")
assert len(data["aaData"]) == 1
def test_routes_taskpaths(worker, api):
task_path = "tests.tasks.general.RaiseException"
worker.send_task(task_path, {
"message": "xyz"}, block=True, accept_statuses=["failed"])
data, _ = api.GET("/api/datatables/taskpaths?sEcho=1")
assert len(data["aaData"]) == 1
assert data["aaData"][0]["_id"] == task_path
assert data["aaData"][0]["jobs"] == 1
def test_routes_workers(worker, api):
worker.start()
data, _ = api.GET("/workers")
assert len(data) == 1
def test_routes_traceback(worker, api):
worker.send_task("tests.tasks.general.RaiseException", {
"message": "xyz"}, block=True, accept_statuses=["failed"])
job = worker.mongodb_jobs.mrq_jobs.find_one()
assert job
data, _ = api.GET("/api/job/%s/traceback" % job["_id"])
assert "xyz" in data["traceback"]
def test_routes_result(worker, api):
worker.send_task("tests.tasks.general.ReturnParams", {
"message": "xyz"}, block=True)
job = worker.mongodb_jobs.mrq_jobs.find_one()
assert job
data, _ = api.GET("/api/job/%s/result" % job["_id"])
assert data["result"]["message"] == "xyz"
def test_routes_jobaction(worker, api):
worker.send_task("tests.tasks.general.ReturnParams", {
"message": "xyz"}, block=False, queue="tmp")
job = worker.mongodb_jobs.mrq_jobs.find_one()
assert job
params = {
"action": "cancel",
"id": str(job["_id"])
}
data, _ = api.POST("/api/jobaction", data=json.dumps(params))
assert "job_id" in data
assert bool(data["job_id"])
def test_routes_datatables(worker, api):
worker.start(flags="--config tests/fixtures/config-raw1.py")
task_path = "tests.tasks.general.RaiseException"
worker.send_task(task_path, {
"message": "xyz"}, block=True, accept_statuses=["failed"], queue="low")
worker.send_task(task_path, {
"message": "xyz"}, block=True, accept_statuses=["failed"], queue="high")
unit = "workers"
data, _ = api.GET("/api/datatables/%s?sEcho=1" % unit)
assert len(data["aaData"]) == 1
unit = "jobs"
data, _ = api.GET("/api/datatables/%s?sEcho=1" % unit)
assert len(data["aaData"]) == 2
assert data["aaData"][0]["path"] == task_path
assert data["aaData"][0]["status"] == "failed"
assert "xyz" in data["aaData"][0]["traceback"]
unit = "queues"
data, _ = api.GET("/api/datatables/%s?sEcho=1" % unit)
assert len(data["aaData"]) == 2
# TODO: test unit = "scheduled_jobs"
def test_routes_logs(worker, api):
task_path = "tests.tasks.general.RaiseException"
worker.send_task(task_path, {
"message": "xyz"}, block=True, accept_statuses=["failed"])
job = worker.mongodb_jobs.mrq_jobs.find_one()
assert job
data, _ = api.GET("/api/logs?job=%s" % job["_id"])
assert bool(data["last_log_id"])
# TODO: test actual logs