feat(core): add core module with mixins, services, and background jobs
- Add Model Mixins: TimestampMixin, SoftDeleteMixin, AuditMixin, etc. - Add Base Services: BaseService, BulkOperationsMixin, QueryOptimizerMixin - Add Base ViewSets with bulk operations - Add BackgroundJob model for Celery task tracking - Add BaseAppCommand for management commands - Add permissions, pagination, filters, cache, logging - Migrate tests to factory_boy + faker - Add CHANGELOG.md - 297 tests passing
This commit is contained in:
204
tests/apps/core/test_signals.py
Normal file
204
tests/apps/core/test_signals.py
Normal file
@@ -0,0 +1,204 @@
|
||||
"""Tests for core signals utilities"""
|
||||
|
||||
|
||||
from apps.core.signals import (
|
||||
SignalDispatcher,
|
||||
emit_password_changed,
|
||||
emit_user_registered,
|
||||
emit_user_verified,
|
||||
on_post_save,
|
||||
on_pre_save,
|
||||
password_changed,
|
||||
user_registered,
|
||||
user_verified,
|
||||
)
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.db.models.signals import post_save, pre_save
|
||||
from django.test import TestCase
|
||||
|
||||
from tests.apps.user.factories import UserFactory
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class SignalDispatcherTest(TestCase):
|
||||
"""Tests for SignalDispatcher"""
|
||||
|
||||
def setUp(self):
|
||||
self.dispatcher = SignalDispatcher()
|
||||
|
||||
def test_register_handler(self):
|
||||
"""Test handler registration"""
|
||||
|
||||
def my_handler(sender, **kwargs):
|
||||
pass
|
||||
|
||||
self.dispatcher.register(
|
||||
signal=post_save,
|
||||
sender="user.User",
|
||||
handler=my_handler,
|
||||
description="Test handler",
|
||||
)
|
||||
|
||||
self.assertEqual(len(self.dispatcher._handlers), 1)
|
||||
self.assertEqual(self.dispatcher._handlers[0]["handler"], my_handler)
|
||||
|
||||
def test_list_handlers(self):
|
||||
"""Test listing registered handlers"""
|
||||
|
||||
def handler1(sender, **kwargs):
|
||||
pass
|
||||
|
||||
def handler2(sender, **kwargs):
|
||||
pass
|
||||
|
||||
self.dispatcher.register(
|
||||
signal=post_save,
|
||||
sender="user.User",
|
||||
handler=handler1,
|
||||
description="Handler 1",
|
||||
)
|
||||
self.dispatcher.register(
|
||||
signal=pre_save,
|
||||
sender="user.User",
|
||||
handler=handler2,
|
||||
description="Handler 2",
|
||||
)
|
||||
|
||||
handlers = self.dispatcher.list_handlers()
|
||||
|
||||
self.assertEqual(len(handlers), 2)
|
||||
self.assertEqual(handlers[0]["description"], "Handler 1")
|
||||
self.assertEqual(handlers[1]["description"], "Handler 2")
|
||||
|
||||
def test_connect_all(self):
|
||||
"""Test connecting all handlers"""
|
||||
handler_called = {"value": False}
|
||||
|
||||
def test_handler(sender, instance, created, **kwargs):
|
||||
handler_called["value"] = True
|
||||
|
||||
self.dispatcher.register(
|
||||
signal=post_save,
|
||||
sender=User,
|
||||
handler=test_handler,
|
||||
description="Test",
|
||||
)
|
||||
|
||||
self.dispatcher.connect_all()
|
||||
|
||||
# Create user to trigger signal
|
||||
user = UserFactory.create_user()
|
||||
|
||||
self.assertTrue(handler_called["value"])
|
||||
|
||||
# Cleanup
|
||||
self.dispatcher.disconnect_all()
|
||||
|
||||
def test_disconnect_all(self):
|
||||
"""Test disconnecting all handlers"""
|
||||
handler_called = {"value": False}
|
||||
|
||||
def test_handler(sender, instance, created, **kwargs):
|
||||
handler_called["value"] = True
|
||||
|
||||
self.dispatcher.register(
|
||||
signal=post_save,
|
||||
sender=User,
|
||||
handler=test_handler,
|
||||
description="Test",
|
||||
)
|
||||
|
||||
self.dispatcher.connect_all()
|
||||
self.dispatcher.disconnect_all()
|
||||
|
||||
# Create user - handler should not be called
|
||||
handler_called["value"] = False
|
||||
user = UserFactory.create_user()
|
||||
|
||||
self.assertFalse(handler_called["value"])
|
||||
|
||||
|
||||
class SignalDecoratorsTest(TestCase):
|
||||
"""Tests for signal decorators"""
|
||||
|
||||
def test_on_post_save_registers_handler(self):
|
||||
"""Test @on_post_save registers handler"""
|
||||
from apps.core.signals import signal_dispatcher
|
||||
|
||||
initial_count = len(signal_dispatcher._handlers)
|
||||
|
||||
@on_post_save("user.User", description="Test decorator")
|
||||
def my_handler(sender, **kwargs):
|
||||
pass
|
||||
|
||||
new_count = len(signal_dispatcher._handlers)
|
||||
self.assertEqual(new_count, initial_count + 1)
|
||||
|
||||
def test_on_pre_save_registers_handler(self):
|
||||
"""Test @on_pre_save registers handler"""
|
||||
from apps.core.signals import signal_dispatcher
|
||||
|
||||
initial_count = len(signal_dispatcher._handlers)
|
||||
|
||||
@on_pre_save("user.User", description="Test pre_save")
|
||||
def my_pre_handler(sender, **kwargs):
|
||||
pass
|
||||
|
||||
new_count = len(signal_dispatcher._handlers)
|
||||
self.assertEqual(new_count, initial_count + 1)
|
||||
|
||||
|
||||
class CustomSignalsTest(TestCase):
|
||||
"""Tests for custom signals"""
|
||||
|
||||
def setUp(self):
|
||||
self.user = UserFactory.create_user()
|
||||
|
||||
def test_emit_user_registered(self):
|
||||
"""Test user_registered signal emission"""
|
||||
handler_called = {"value": False, "user": None}
|
||||
|
||||
def handler(sender, user, **kwargs):
|
||||
handler_called["value"] = True
|
||||
handler_called["user"] = user
|
||||
|
||||
user_registered.connect(handler)
|
||||
|
||||
try:
|
||||
emit_user_registered(self.user)
|
||||
|
||||
self.assertTrue(handler_called["value"])
|
||||
self.assertEqual(handler_called["user"], self.user)
|
||||
finally:
|
||||
user_registered.disconnect(handler)
|
||||
|
||||
def test_emit_user_verified(self):
|
||||
"""Test user_verified signal emission"""
|
||||
handler_called = {"value": False}
|
||||
|
||||
def handler(sender, user, **kwargs):
|
||||
handler_called["value"] = True
|
||||
|
||||
user_verified.connect(handler)
|
||||
|
||||
try:
|
||||
emit_user_verified(self.user)
|
||||
self.assertTrue(handler_called["value"])
|
||||
finally:
|
||||
user_verified.disconnect(handler)
|
||||
|
||||
def test_emit_password_changed(self):
|
||||
"""Test password_changed signal emission"""
|
||||
handler_called = {"value": False}
|
||||
|
||||
def handler(sender, user, **kwargs):
|
||||
handler_called["value"] = True
|
||||
|
||||
password_changed.connect(handler)
|
||||
|
||||
try:
|
||||
emit_password_changed(self.user)
|
||||
self.assertTrue(handler_called["value"])
|
||||
finally:
|
||||
password_changed.disconnect(handler)
|
||||
Reference in New Issue
Block a user