state-corp-backend/tests/apps/core/test_signals.py

"""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
        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
        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)