➕ Unit testing for managers inheritance

Author: tarsil

saffier/db/queryset.py

@@ -440,6 +440,9 @@ async def get_or_none(self, **kwargs):
         return self.model_class._from_row(rows[0], select_related=self._select_related)
 
     async def all(self, **kwargs):
+        """
+        Returns the queryset records based on specific filters
+        """
         queryset = self._clone()
         if kwargs:
             return await queryset.filter(**kwargs).all()
@@ -452,6 +455,9 @@ async def all(self, **kwargs):
         ]
 
     async def get(self, **kwargs):
+        """
+        Returns a single record based on the given kwargs.
+        """
         if kwargs:
             return await self.filter(**kwargs).get()
 
@@ -465,6 +471,9 @@ async def get(self, **kwargs):
         return self.model_class._from_row(rows[0], select_related=self._select_related)
 
     async def first(self, **kwargs):
+        """
+        Returns the first record of a given queryset.
+        """
         queryset = self._clone()
         if kwargs:
             return await queryset.filter(**kwargs).order_by("id").get()
@@ -474,6 +483,9 @@ async def first(self, **kwargs):
             return rows[0]
 
     async def last(self, **kwargs):
+        """
+        Returns the last record of a given queryset.
+        """
         queryset = self._clone()
         if kwargs:
             return await queryset.filter(**kwargs).order_by("-id").get()
@@ -483,6 +495,9 @@ async def last(self, **kwargs):
             return rows[0]
 
     async def create(self, **kwargs):
+        """
+        Creates a record in a specific table.
+        """
         kwargs = self._validate_kwargs(**kwargs)
         instance = self.model_class(**kwargs)
         expression = self.table.insert().values(**kwargs)
@@ -495,6 +510,9 @@ async def create(self, **kwargs):
         return instance
 
     async def bulk_create(self, objs: typing.List[typing.Dict]) -> None:
+        """
+        Bulk creates records in a table
+        """
         new_objs = [self._validate_kwargs(**obj) for obj in objs]
 
         expression = self.table.insert().values(new_objs)
@@ -508,6 +526,9 @@ async def delete(self) -> None:
         await self.database.execute(expression)
 
     async def update(self, **kwargs) -> None:
+        """
+        Updates a record in a specific table with the given kwargs.
+        """
         fields = {
             key: field.validator for key, field in self.model_class.fields.items() if key in kwargs
         }
@@ -524,6 +545,9 @@ async def update(self, **kwargs) -> None:
     async def get_or_create(
         self, defaults: typing.Dict[str, typing.Any], **kwargs
     ) -> typing.Tuple[typing.Any, bool]:
+        """
+        Creates a record in a specific table or updates if already exists.
+        """
         try:
             instance = await self.get(**kwargs)
             return instance, False
@@ -535,6 +559,9 @@ async def get_or_create(
     async def update_or_create(
         self, defaults: typing.Dict[str, typing.Any], **kwargs
     ) -> typing.Tuple[typing.Any, bool]:
+        """
+        Updates a record in a specific table or creates a new one.
+        """
         try:
             instance = await self.get(**kwargs)
             await instance.update(**defaults)

saffier/metaclass.py

@@ -89,7 +89,7 @@ def _check_manager_for_bases(
             if isinstance(value, Manager) and key not in attrs:
                 attrs[key] = value.__class__()
     else:
-        if not getattr(meta, "abstract", False):
+        if not meta.abstract:
             for key, value in inspect.getmembers(base):
                 if isinstance(value, Manager) and key not in attrs:
                     attrs[key] = value.__class__()
@@ -123,7 +123,7 @@ def __search_for_fields(base: typing.Type, attrs: DictAny) -> None:
             meta: MetaInfo = getattr(base, "_meta", None)
             if not meta:
                 # Mixins and other classes
-                for key, value in base.__dict__.items():
+                for key, value in inspect.getmembers(base):
                     if isinstance(value, Field) and key not in attrs:
                         attrs[key] = value
 

tests/test_managers.py renamed to tests/managers/test_managers.py

@@ -40,7 +40,7 @@ class Meta:
 
 
 @pytest.fixture(autouse=True, scope="function")
-async def create_xtest_database():
+async def create_test_database():
     await models.create_all()
     yield
     await models.drop_all()
@@ -92,20 +92,3 @@ async def test_model_crud_different_manager():
     assert len(products) == 1
 
     assert products[0].pk == product.pk
-
-
-async def test_model_crud_different_manager_create():
-    products = await Product.active.all()
-    assert products == []
-
-    await Product.active.create(name="One", in_stock=True, is_active=False, rating=5)
-    await Product.active.create(name="Two", in_stock=True, is_active=False, rating=2)
-    product = await Product.query.create(name="Three", in_stock=True, is_active=True, rating=3)
-
-    products = await Product.query.all()
-    assert len(products) == 3
-
-    products = await Product.active.all()
-    assert len(products) == 1
-
-    assert products[0].pk == product.pk

tests/managers/test_managers_abstract.py

@@ -0,0 +1,76 @@
+import pytest
+from tests.settings import DATABASE_URL
+
+import saffier
+from saffier import Manager
+from saffier.db.connection import Database
+from saffier.db.queryset import QuerySet
+
+database = Database(url=DATABASE_URL)
+models = saffier.Registry(database=database)
+
+pytestmark = pytest.mark.anyio
+
+
+class ObjectsManager(Manager):
+    def get_queryset(self) -> QuerySet:
+        queryset = super().get_queryset().filter(is_active=True)
+        return queryset
+
+
+class LanguageManager(Manager):
+    def get_queryset(self) -> QuerySet:
+        queryset = super().get_queryset().filter(language="EN")
+        return queryset
+
+
+class BaseModel(saffier.Model):
+    query = ObjectsManager()
+
+    class Meta:
+        abstract = True
+        registry = models
+
+
+class HubUser(BaseModel):
+    name = saffier.CharField(max_length=100)
+    language = saffier.CharField(max_length=200, null=True)
+
+    languages = LanguageManager()
+
+    class Meta:
+        registry = models
+
+
+class HubProduct(BaseModel):
+    name = saffier.CharField(max_length=100)
+    rating = saffier.IntegerField(minimum=1, maximum=5)
+    in_stock = saffier.BooleanField(default=False)
+    is_active = saffier.BooleanField(default=False)
+
+
+@pytest.fixture(autouse=True, scope="function")
+async def create_test_database():
+    await models.create_all()
+    yield
+    await models.drop_all()
+
+
+@pytest.fixture(autouse=True)
+async def rollback_connections():
+    with database.force_rollback():
+        async with database:
+            yield
+
+
+async def test_inherited_abstract_base_model_managers():
+    await HubUser.query.create(name="test", language="EN")
+    await HubUser.query.create(name="test2", language="EN")
+    await HubUser.query.create(name="test3", language="PT")
+    await HubUser.query.create(name="test4", language="PT")
+
+    # users = await HubUser.query.all()
+    # assert len(users) == 4
+
+    users = await HubUser.languages.all()
+    assert len(users) == 2

tests/managers/test_managers_inherited.py

@@ -0,0 +1,96 @@
+import pytest
+from tests.settings import DATABASE_URL
+
+import saffier
+from saffier import Manager
+from saffier.db.connection import Database
+from saffier.db.queryset import QuerySet
+
+database = Database(url=DATABASE_URL)
+models = saffier.Registry(database=database)
+
+pytestmark = pytest.mark.anyio
+
+
+class ObjectsManager(Manager):
+    def get_queryset(self) -> QuerySet:
+        queryset = super().get_queryset().filter(is_active=True)
+        return queryset
+
+
+class LanguageManager(Manager):
+    def get_queryset(self) -> QuerySet:
+        queryset = super().get_queryset().filter(language="EN")
+        return queryset
+
+
+class RatingManager(Manager):
+    def get_queryset(self) -> QuerySet:
+        queryset = super().get_queryset().filter(rating__gte=3)
+        return queryset
+
+
+class BaseModel(saffier.Model):
+    query = ObjectsManager()
+    languages = LanguageManager()
+    ratings = RatingManager()
+
+    class Meta:
+        registry = models
+
+
+class User(BaseModel):
+    name = saffier.CharField(max_length=100)
+    language = saffier.CharField(max_length=200, null=True)
+
+    class Meta:
+        registry = models
+
+
+class Product(BaseModel):
+    name = saffier.CharField(max_length=100)
+    rating = saffier.IntegerField(minimum=1, maximum=5)
+    in_stock = saffier.BooleanField(default=False)
+    is_active = saffier.BooleanField(default=False)
+
+
+@pytest.fixture(autouse=True, scope="function")
+async def create_test_database():
+    await models.create_all()
+    yield
+    await models.drop_all()
+
+
+@pytest.fixture(autouse=True)
+async def rollback_connections():
+    with database.force_rollback():
+        async with database:
+            yield
+
+
+async def test_inherited_base_model_managers():
+    await User.query.create(name="test", language="EN")
+    await User.query.create(name="test2", language="EN")
+    await User.query.create(name="test3", language="PT")
+    await User.query.create(name="test4", language="PT")
+
+    users = await User.query.all()
+    assert len(users) == 4
+
+    users = await User.languages.all()
+    assert len(users) == 2
+
+
+async def test_inherited_base_model_managers_product():
+    await Product.query.create(name="test", rating=5)
+    await Product.query.create(name="test2", rating=4)
+    await Product.query.create(name="test3", rating=3)
+    await Product.query.create(name="test4", rating=2)
+    await Product.query.create(name="test5", rating=2)
+    await Product.query.create(name="test6", rating=1)
+
+    users = await Product.query.all()
+    assert len(users) == 6
+
+    users = await Product.ratings.all()
+    assert len(users) == 3