From 4ebd2e383f0c09f229fb700723e897b11d3c1fbe Mon Sep 17 00:00:00 2001 From: Antoine Nguyen Date: Sun, 14 Dec 2025 18:41:08 +0100 Subject: [PATCH] Use SimplerWorker explicitely --- modoboa/contacts/tests.py | 12 +++- modoboa/rspamd/tests.py | 136 +++++++++++++++++++------------------- 2 files changed, 77 insertions(+), 71 deletions(-) diff --git a/modoboa/contacts/tests.py b/modoboa/contacts/tests.py index 89d2f967b..eb917b8a1 100644 --- a/modoboa/contacts/tests.py +++ b/modoboa/contacts/tests.py @@ -3,12 +3,13 @@ import os import httmock +from rq import SimpleWorker from django.core import management from django.urls import reverse from django.utils import timezone -from django_rq import get_worker +import django_rq from modoboa.admin import factories as admin_factories from modoboa.core import models as core_models @@ -137,11 +138,16 @@ class AddressBookViewSetTestCase(TestDataMixin, ModoAPITestCase): self.enable_cdav_sync() last_sync = timezone.now() self.user.addressbook_set.update(last_sync=last_sync) + addressbook = self.user.addressbook_set.first() + self.assertEqual(addressbook.sync_token, "") with httmock.HTTMock(mocks.options_mock, mocks.report_mock, mocks.get_mock): response = self.client.get(reverse("api:addressbook-sync-from-cdav")) - get_worker("modoboa").work(burst=True) + queue = django_rq.get_queue("modoboa") + worker = SimpleWorker([queue], connection=queue.connection) + worker.work(burst=True) self.assertEqual(response.status_code, 200) - self.assertIsNot(self.user.addressbook_set.first().sync_token, None) + addressbook.refresh_from_db() + self.assertNotEqual(addressbook.sync_token, "") class CategoryViewSetTestCase(TestDataMixin, ModoAPITestCase): diff --git a/modoboa/rspamd/tests.py b/modoboa/rspamd/tests.py index d5dcf9323..5594fb1ba 100644 --- a/modoboa/rspamd/tests.py +++ b/modoboa/rspamd/tests.py @@ -1,84 +1,84 @@ -# import os -# import shutil +import os +import shutil -# from django.core import management -# from django.test import modify_settings -# from django.urls import reverse +from django.core import management +from django.test import modify_settings +from django.urls import reverse -# from modoboa.admin import factories as admin_factories, models as admin_models -# from modoboa.lib.tests import ModoAPITestCase +from modoboa.admin import factories as admin_factories, models as admin_models +from modoboa.lib.tests import ModoAPITestCase -# @modify_settings( -# INSTALLED_APPS={ -# "append": "modoboa.rspamd", -# } -# ) -# class ManagementCommandTestCase(ModoAPITestCase): +@modify_settings( + INSTALLED_APPS={ + "append": "modoboa.rspamd", + } +) +class ManagementCommandTestCase(ModoAPITestCase): -# @classmethod -# def setUpTestData(cls): -# super().setUpTestData() -# admin_factories.populate_database() + @classmethod + def setUpTestData(cls): + super().setUpTestData() + admin_factories.populate_database() -# def tearDown(self): -# """Reset test env.""" -# shutil.rmtree(self.workdir) + def tearDown(self): + """Reset test env.""" + shutil.rmtree(self.workdir) -# def configure(self): -# self.key_map_path = f"{self.workdir}/keys.map" -# self.set_global_parameter("key_map_path", self.key_map_path) -# self.selector_map_path = f"{self.workdir}/selectors.map" -# self.set_global_parameter("selector_map_path", self.selector_map_path) + def configure(self): + self.key_map_path = f"{self.workdir}/keys.map" + self.set_global_parameter("key_map_path", self.key_map_path) + self.selector_map_path = f"{self.workdir}/selectors.map" + self.set_global_parameter("selector_map_path", self.selector_map_path) -# def unconfigure(self): -# self.set_global_parameter("key_map_path", "") -# self.set_global_parameter("selector_map_path", "") + def unconfigure(self): + self.set_global_parameter("key_map_path", "") + self.set_global_parameter("selector_map_path", "") -# def test_command(self): -# self.unconfigure() -# with self.assertRaises(management.CommandError): -# management.call_command("manage_rspamd_maps") + def test_command(self): + self.unconfigure() + with self.assertRaises(management.CommandError): + management.call_command("manage_rspamd_maps") -# self.configure() -# management.call_command("manage_rspamd_maps") -# self.assertFalse(os.path.exists(self.key_map_path)) -# self.assertFalse(os.path.exists(self.selector_map_path)) + self.configure() + management.call_command("manage_rspamd_maps") + self.assertFalse(os.path.exists(self.key_map_path)) + self.assertFalse(os.path.exists(self.selector_map_path)) -# domain = admin_models.Domain.objects.get(name="test.com") -# domain.enable_dkim = True -# domain.dkim_private_key_path = "xxx" -# domain.save() -# management.call_command("manage_rspamd_maps") -# self.assertTrue(os.path.exists(self.key_map_path)) -# self.assertTrue(os.path.exists(self.selector_map_path)) + domain = admin_models.Domain.objects.get(name="test.com") + domain.enable_dkim = True + domain.dkim_private_key_path = "xxx" + domain.save() + management.call_command("manage_rspamd_maps") + self.assertTrue(os.path.exists(self.key_map_path)) + self.assertTrue(os.path.exists(self.selector_map_path)) -# # Now, empty map files -# domain.enable_dkim = False -# domain.save() -# management.call_command("manage_rspamd_maps") -# with open(self.key_map_path) as fp: -# content = fp.read() -# self.assertNotIn(domain.name, content) + # Now, empty map files + domain.enable_dkim = False + domain.save() + management.call_command("manage_rspamd_maps") + with open(self.key_map_path) as fp: + content = fp.read() + self.assertNotIn(domain.name, content) -# def test_command_with_arg(self): -# self.configure() -# domain = admin_models.Domain.objects.get(name="test.com") -# domain.enable_dkim = True -# domain.save() -# management.call_command("manage_rspamd_maps", "--domain", "test.com") -# self.assertTrue(os.path.exists(self.key_map_path)) -# self.assertTrue(os.path.exists(self.selector_map_path)) + def test_command_with_arg(self): + self.configure() + domain = admin_models.Domain.objects.get(name="test.com") + domain.enable_dkim = True + domain.save() + management.call_command("manage_rspamd_maps", "--domain", "test.com") + self.assertTrue(os.path.exists(self.key_map_path)) + self.assertTrue(os.path.exists(self.selector_map_path)) -# class ParametersAPITestCase(ModoAPITestCase): +class ParametersAPITestCase(ModoAPITestCase): -# def test_update_settings(self): -# url = reverse("v2:parameter-global-detail", args=["rspamd"]) -# data = { -# "key_map_path": f"{self.workdir}/key.map", -# "selector_map_path": f"{self.workdir}/selector.map", -# "rspamd_dashboard_location": "/rspamd", -# } -# response = self.client.put(url, data, format="json") -# self.assertEqual(response.status_code, 200) + def test_update_settings(self): + url = reverse("v2:parameter-global-detail", args=["rspamd"]) + data = { + "key_map_path": f"{self.workdir}/key.map", + "selector_map_path": f"{self.workdir}/selector.map", + "rspamd_dashboard_location": "/rspamd", + } + response = self.client.put(url, data, format="json") + self.assertEqual(response.status_code, 200)