Aleksandr Meshchriakov
a91ed1f1ae
Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 3m10s
CI/CD Pipeline / Run Tests (push) Successful in 3m35s
CI/CD Pipeline / Telegram Notify Success (push) Has been skipped
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 2m26s
CI/CD Pipeline / Run Tests (pull_request) Successful in 2m46s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""Tests for core signal dispatcher utilities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from apps.core.signals import (
|
|
SignalDispatcher,
|
|
on_post_delete,
|
|
on_post_save,
|
|
on_pre_delete,
|
|
on_pre_save,
|
|
signal_dispatcher,
|
|
)
|
|
from django.contrib.auth import get_user_model
|
|
from django.db.models.signals import post_save
|
|
from django.test import TestCase
|
|
|
|
from tests.utils.fixtures import fake
|
|
|
|
|
|
class SignalDispatcherTest(TestCase):
|
|
def test_register_connect_disconnect(self):
|
|
dispatcher = SignalDispatcher()
|
|
events: list[int] = []
|
|
|
|
def handler(sender, instance, **_kwargs):
|
|
events.append(instance.pk)
|
|
|
|
dispatcher.register(
|
|
signal=post_save,
|
|
sender="user.User",
|
|
handler=handler,
|
|
description="track user create",
|
|
)
|
|
dispatcher.connect_all()
|
|
|
|
user = get_user_model().objects.create_user(
|
|
email=fake.email(), username=fake.user_name(), password="pass"
|
|
)
|
|
|
|
self.assertIn(user.pk, events)
|
|
self.assertTrue(dispatcher.list_handlers())
|
|
|
|
dispatcher.disconnect_all()
|
|
self.assertFalse(dispatcher._connected)
|
|
|
|
def test_connect_all_idempotent(self):
|
|
dispatcher = SignalDispatcher()
|
|
dispatcher.register(
|
|
signal=post_save,
|
|
sender="user.User",
|
|
handler=lambda *_args, **_kwargs: None,
|
|
)
|
|
dispatcher.connect_all()
|
|
dispatcher.connect_all()
|
|
self.assertTrue(dispatcher._connected)
|
|
|
|
def test_decorator_registers_handler(self):
|
|
original_handlers = list(signal_dispatcher._handlers)
|
|
original_connected = signal_dispatcher._connected
|
|
events: list[int] = []
|
|
|
|
try:
|
|
signal_dispatcher._handlers = []
|
|
signal_dispatcher._connected = False
|
|
|
|
@on_post_save("user.User", description="decorator")
|
|
def _handler(sender, instance, **_kwargs):
|
|
events.append(instance.pk)
|
|
|
|
@on_pre_save("user.User")
|
|
def _pre_save(sender, instance, **_kwargs):
|
|
return None
|
|
|
|
@on_post_delete("user.User")
|
|
def _post_delete(sender, instance, **_kwargs):
|
|
return None
|
|
|
|
@on_pre_delete("user.User")
|
|
def _pre_delete(sender, instance, **_kwargs):
|
|
return None
|
|
|
|
signal_dispatcher.connect_all()
|
|
user = get_user_model().objects.create_user(
|
|
email=fake.email(), username=fake.user_name(), password="pass"
|
|
)
|
|
self.assertIn(user.pk, events)
|
|
self.assertTrue(signal_dispatcher.list_handlers())
|
|
finally:
|
|
signal_dispatcher.disconnect_all()
|
|
signal_dispatcher._handlers = original_handlers
|
|
signal_dispatcher._connected = original_connected
|