diff mbox series

[layerindex-web,4/4] tests: add tests for the Celery tasks

Message ID 20261113232808.96792-5-tim.orling@konsulko.com
State Under Review
Delegated to: Tim Orling
Headers show
Series Dependency upgrades, and a fix for site notices | expand

Commit Message

Tim Orling Nov. 13, 2026, 11:28 p.m. UTC 
Run send_email and run_update_command in-process, so that Celery
upgrades are checked without a broker.

AI-Generated: Claude Opus 5.5
Signed-off-by: Tim Orling <tim.orling@konsulko.com>
---
 tests/test_tasks.py | 39 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 39 insertions(+)
 create mode 100644 tests/test_tasks.py
diff mbox series

Patch

diff --git a/tests/test_tasks.py b/tests/test_tasks.py
new file mode 100644
index 0000000..d4cde76
--- /dev/null
+++ b/tests/test_tasks.py
@@ -0,0 +1,39 @@ 
+# layerindex-web - tests for the Celery tasks in layerindex/tasks.py
+#
+# The tasks run in-process with apply(), so no broker is needed.
+#
+# Licensed under the MIT license, see COPYING.MIT for details
+#
+# SPDX-License-Identifier: MIT
+
+from datetime import datetime
+
+import pytest
+
+from layerindex import tasks
+from layerindex.models import Update
+
+
+def test_send_email(mailoutbox):
+    result = tasks.send_email.apply(('Subject', 'Body', 'from@example.com', ['to@example.com']))
+    assert result.successful()
+    assert len(mailoutbox) == 1
+    msg = mailoutbox[0]
+    assert (msg.subject, msg.body, msg.from_email, msg.to) == ('Subject', 'Body', 'from@example.com', ['to@example.com'])
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize('command, retcode, log', [
+    ('echo update %update% on %branch%', 0, 'update {id} on master'),
+    ('echo failed; exit 3', 3, 'failed'),
+])
+def test_run_update_command(settings, tmp_path, command, retcode, log):
+    settings.TASK_LOG_DIR = str(tmp_path / 'logs')
+    update = Update.objects.create(started=datetime.now(), task_id='test-task')
+    result = tasks.run_update_command.apply(('master', command), task_id='test-task')
+    assert result.get()['retcode'] == retcode
+    update.refresh_from_db()
+    assert update.retcode == retcode
+    assert update.log == log.format(id=update.id)
+    assert update.finished is not None
+    assert (tmp_path / 'logs' / 'task_test-task.log').exists()