aboutsummaryrefslogtreecommitdiff
path: root/tests/tasks/test_extension.py
diff options
context:
space:
mode:
authorDennis Fink2026-08-21 10:37:56 +0200
committerDennis Fink2026-08-21 10:37:56 +0200
commitc5492398f100ccb154fa557d657bae2322f95cf7 (patch)
tree21efa67e00c26ab0733a931e6a5810b4c933388b /tests/tasks/test_extension.py
parent5bfcc3c06ebe6d6694221c25f07a4901572f97f5 (diff)
downloadwebmentions-ssg-c5492398f100ccb154fa557d657bae2322f95cf7.tar.gz
webmentions-ssg-c5492398f100ccb154fa557d657bae2322f95cf7.zip
test(core): align tests with module ownership
Move direct validator, model, and Huey extension coverage into dedicated test modules and mock dependencies where behavior belongs to another module. Expand Huey extension typing and coverage while removing duplicate assertions from scanner, sender, form, and view tests.
Diffstat (limited to '')
-rw-r--r--tests/tasks/test_extension.py175
1 files changed, 175 insertions, 0 deletions
diff --git a/tests/tasks/test_extension.py b/tests/tasks/test_extension.py
new file mode 100644
index 0000000..5e9cdfa
--- /dev/null
+++ b/tests/tasks/test_extension.py
@@ -0,0 +1,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)