tests for signals

Author: kalombos

peewee_async/signals.py

@@ -3,7 +3,7 @@
 from playhouse.signals import Signal
 
 class AioSignal(Signal):
-    async def send(self, instance, *args, **kwargs):
+    async def send(self, instance: "AioModel", *args: Any, **kwargs: Any) -> list[tuple[Any, Any]]:
         sender = type(instance)
         responses = []
         for n, r, s in self._receiver_list:

tests/models.py

@@ -1,73 +1,72 @@
+from email.policy import default
 import uuid
 
-import peewee
+import peewee as pw
 import peewee_async
 import peewee_async.signals
+import datetime as dt
 
 
 class TestModel(peewee_async.AioModel):
     __test__ = False  # disable pytest warnings
-    text = peewee.CharField(max_length=100, unique=True)
-    data = peewee.TextField(default='')
+    text = pw.CharField(max_length=100, unique=True)
+    data = pw.TextField(default='')
 
     def __str__(self) -> str:
         return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
 
 
 class TestModelAlpha(peewee_async.AioModel):
     __test__ = False
-    text = peewee.CharField()
+    text = pw.CharField()
 
     def __str__(self) -> str:
         return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
 
 
 class TestModelBeta(peewee_async.AioModel):
     __test__ = False
-    alpha = peewee.ForeignKeyField(TestModelAlpha, backref='betas')
-    text = peewee.CharField()
+    alpha = pw.ForeignKeyField(TestModelAlpha, backref='betas')
+    text = pw.CharField()
 
     def __str__(self) -> str:
         return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
 
 
 class TestModelGamma(peewee_async.AioModel):
     __test__ = False
-    text = peewee.CharField()
-    beta = peewee.ForeignKeyField(TestModelBeta, backref='gammas')
+    text = pw.CharField()
+    beta = pw.ForeignKeyField(TestModelBeta, backref='gammas')
 
     def __str__(self) -> str:
         return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
 
 
 class UUIDTestModel(peewee_async.AioModel):
-    id = peewee.UUIDField(primary_key=True, default=uuid.uuid4)
-    text = peewee.CharField()
+    id = pw.UUIDField(primary_key=True, default=uuid.uuid4)
+    text = pw.CharField()
 
     def __str__(self) -> str:
         return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
 
 
 class CompositeTestModel(peewee_async.AioModel):
     """A simple "through" table for many-to-many relationship."""
-    task_id = peewee.IntegerField()
-    product_type = peewee.CharField()
+    task_id = pw.IntegerField()
+    product_type = pw.CharField()
 
     class Meta:
-        primary_key = peewee.CompositeKey('task_id', 'product_type')
+        primary_key = pw.CompositeKey('task_id', 'product_type')
 
 
 class IntegerTestModel(peewee_async.AioModel):
     __test__ = False  # disable pytest warnings
-    num = peewee.IntegerField()
+    num = pw.IntegerField()
 
 
 class TestSignalModel(peewee_async.signals.AioModel):
     __test__ = False  # disable pytest warnings
-    text = peewee.CharField(max_length=100)
-
-    def __str__(self) -> str:
-        return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
+    text = pw.CharField(max_length=100)
 
 
 ALL_MODELS = (

tests/test_signals.py

@@ -1,19 +1,87 @@
-import uuid
+from contextlib import contextmanager
+from typing import Any, Callable, Coroutine, Iterator
 
 from peewee_async.databases import AioDatabase
-from tests.conftest import dbs_all, dbs_postgres
+from tests.conftest import dbs_all
 from tests.models import TestSignalModel
-from tests.utils import model_has_fields
-from peewee_async.signals import aio_pre_save
+from peewee_async.signals import AioModel, aio_pre_save, aio_post_save, aio_post_delete, aio_pre_delete, AioSignal, pre_init
 
 
 
 
+
+@contextmanager
+def _connect(signal: AioSignal , receiver: Callable[..., Coroutine[Any, Any, Any]], sender: type[AioModel]) -> Iterator[None]:
+    signal.connect(receiver=receiver, sender=sender)
+    yield
+    signal.disconnect(receiver=receiver, sender=sender)
+
+
 @dbs_all
 async def test_aio_pre_save(db: AioDatabase) -> None:
 
-    @aio_pre_save(sender=TestSignalModel)
-    async def on_save_handler(model_class, instance, created):
-        print(model_class, instance, created)
+    
+    async def on_save_handler(model_class: type[TestSignalModel], instance: TestSignalModel, created: bool) -> None:
+        assert await TestSignalModel.select().aio_exists() is False
+        assert model_class is TestSignalModel
+        assert isinstance(instance, TestSignalModel)
+        assert created
+
+    with _connect(aio_pre_save, receiver=on_save_handler, sender=TestSignalModel):
+        await TestSignalModel.aio_create(text="aio_create")
+
+
+@dbs_all
+async def test_aio_post_save(db: AioDatabase) -> None:
+    
+    
+    async def on_save_handler(model_class: type[TestSignalModel], instance: TestSignalModel, created: bool) -> None:
+        assert await TestSignalModel.select().aio_exists() is True
+        assert model_class is TestSignalModel
+        assert isinstance(instance, TestSignalModel)
+        assert created
+
+    with _connect(aio_post_save, receiver=on_save_handler, sender=TestSignalModel):
+        await TestSignalModel.aio_create(text="aio_create")
+
+
+@dbs_all
+async def test_aio_pre_delete(db: AioDatabase) -> None:
+    
+    t = await TestSignalModel.aio_create(text="aio_create")
+    
+    async def on_delete_handler(model_class: type[TestSignalModel], instance: TestSignalModel) -> None:
+        assert await TestSignalModel.select().aio_exists() is True
+        assert model_class is TestSignalModel
+        assert isinstance(instance, TestSignalModel)
+
+    with _connect(aio_pre_delete, receiver=on_delete_handler, sender=TestSignalModel):
+        await t.aio_delete_instance()
+
+
+@dbs_all
+async def test_aio_post_delete(db: AioDatabase) -> None:
+    
+    t = await TestSignalModel.aio_create(text="aio_create")
+    
+    async def on_delete_handler(model_class: type[TestSignalModel], instance: TestSignalModel) -> None:
+        assert await TestSignalModel.select().aio_exists() is False
+        assert model_class is TestSignalModel
+        assert isinstance(instance, TestSignalModel)
+
+    with _connect(aio_post_delete, receiver=on_delete_handler, sender=TestSignalModel):
+        await t.aio_delete_instance()
+
+
+@dbs_all
+def test_pre_init(db: AioDatabase) -> None:
+    
+    def on_init_handler(model_class: type[TestSignalModel], instance: TestSignalModel) -> None:
+        assert model_class is TestSignalModel
+        assert instance.text == "text"
+
+    pre_init.connect(receiver=on_init_handler, sender=TestSignalModel)
+
+    TestSignalModel(text="text")
 
-    await TestSignalModel.aio_create(text="text")
+    pre_init.disconnect(receiver=on_init_handler, sender=TestSignalModel)