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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# SPDX-License-Identifier: BSD-3-Clause
from collections.abc import Callable
from typing import Any
import huey as huey_package
import pytest
from flask import Flask, has_app_context
from huey import crontab
from webmentions_ssg.tasks.extension import Huey
def test_init_without_app() -> None:
huey = Huey()
assert huey.app is None
with pytest.raises(RuntimeError, match="Huey has not been initialized"):
_ = huey.huey
def test_init_with_app(app: Flask) -> None:
huey = Huey(app)
assert huey.app is app
assert huey.huey.name == app.import_name
assert huey.huey.results is True
assert huey.huey.store_none is False
assert huey.huey.utc is True
assert huey.huey.immediate is True
assert app.extensions["huey"] is huey
def test_init_app_uses_huey_configuration(app: Flask) -> None:
app.config["HUEY"] = {
"url": "memory://",
"name": "custom-name",
"results": False,
"store_none": True,
"utc": False,
"immediate": False,
}
huey = Huey()
huey.init_app(app)
assert huey.huey.name == "custom-name"
assert huey.huey.results is False
assert huey.huey.store_none is True
assert huey.huey.utc is False
assert huey.huey.immediate is False
def test_huey_url_takes_precedence_over_huey_configuration(app: Flask) -> None:
app.config["HUEY"] = {"url": "blackhole://"}
app.config["HUEY_URL"] = "memory://"
huey = Huey(app)
assert type(huey.huey).__name__ == "MemoryHuey"
def task_decorator(huey: Huey, periodic: bool) -> Callable[[Callable[..., Any]], Any]:
if periodic:
return huey.periodic_task(crontab(minute="*"))
return huey.task()
@pytest.mark.parametrize("periodic", [False, True], ids=["task", "periodic-task"])
def test_task_runs_in_app_context(app: Flask, periodic: bool) -> None:
huey = Huey(app)
@task_decorator(huey, periodic)
def add(a: int, b: int) -> int:
assert has_app_context()
return a + b
assert add.call_local(2, 3) == 5
@pytest.mark.parametrize("periodic", [False, True], ids=["task", "periodic-task"])
def test_task_raises_without_app(app: Flask, periodic: bool) -> None:
huey = Huey(app)
@task_decorator(huey, periodic)
def task() -> None:
pass
huey.app = None
with pytest.raises(RuntimeError, match="Flask app is not available"):
task.call_local()
def test_getattr_delegates_to_huey(app: Flask) -> None:
huey = Huey(app)
assert huey.immediate is True
@pytest.mark.parametrize(
("url", "backend_name", "storage_kwargs"),
[
("redis://localhost:6379/0", "RedisHuey", {"url": "redis://localhost:6379/0"}),
(
"rediss://localhost:6379/0",
"RedisHuey",
{"url": "rediss://localhost:6379/0"},
),
(
"redis+priority://localhost:6379/0",
"PriorityRedisHuey",
{"url": "redis://localhost:6379/0"},
),
(
"redis+expire://localhost:6379/0",
"RedisExpireHuey",
{"url": "redis://localhost:6379/0"},
),
(
"redis+priority+expire://localhost:6379/0",
"PriorityRedisExpireHuey",
{"url": "redis://localhost:6379/0"},
),
("sqlite:///var/huey.db", "SqliteHuey", {"filename": "var/huey.db"}),
("file:///var/huey-queue", "FileHuey", {"path": "var/huey-queue"}),
(
"postgres://user:password@localhost/database",
"PostgresHuey",
{"dsn": "postgres://user:password@localhost/database"},
),
(
"postgresql://user:password@localhost/database",
"PostgresHuey",
{"dsn": "postgresql://user:password@localhost/database"},
),
("memory://", "MemoryHuey", {}),
("blackhole://", "BlackHoleHuey", {}),
],
)
def test_backend_from_url(
monkeypatch: pytest.MonkeyPatch,
url: str,
backend_name: str,
storage_kwargs: dict[str, str],
) -> None:
backend_class = type(f"Test{backend_name}", (), {})
monkeypatch.setattr(huey_package, backend_name, backend_class)
actual_class, actual_storage_kwargs = Huey.backend_from_url(url)
assert actual_class is backend_class
assert actual_storage_kwargs == storage_kwargs
@pytest.mark.parametrize(
("url", "message"),
[
(
"sqlite://var/huey.db",
"SQLite Huey URLs must look like sqlite:///var/huey.db",
),
("sqlite:///", "SQLite Huey URL must include a database path"),
(
"file://var/huey-queue",
"File Huey URLs must look like file:///var/huey-queue",
),
("file:///", "File Huey URL must include a directory path"),
("amqp://localhost", "Unsupported HUEY_URL scheme: 'amqp'"),
],
)
def test_backend_from_url_rejects_invalid_url(url: str, message: str) -> None:
with pytest.raises(RuntimeError, match=message):
Huey.backend_from_url(url)
|
