From d35a2945fc7b13455a37ad5ad873f96d3b74a4b1 Mon Sep 17 00:00:00 2001 From: imperosol Date: Wed, 13 May 2026 13:39:21 +0200 Subject: [PATCH] add field `Product.clic_limit` --- .../0040_remove_product_buying_groups.py | 15 ++++- counter/models.py | 47 +++++++++++++- counter/tests/test_product.py | 62 ++++++++++++++++++- 3 files changed, 119 insertions(+), 5 deletions(-) diff --git a/counter/migrations/0040_remove_product_buying_groups.py b/counter/migrations/0040_remove_product_buying_groups.py index faa4f215..42f2b3e9 100644 --- a/counter/migrations/0040_remove_product_buying_groups.py +++ b/counter/migrations/0040_remove_product_buying_groups.py @@ -1,6 +1,6 @@ # Generated by Django 5.2.13 on 2026-05-13 11:31 -from django.db import migrations +from django.db import migrations, models class Migration(migrations.Migration): @@ -8,4 +8,17 @@ class Migration(migrations.Migration): operations = [ migrations.RemoveField(model_name="product", name="buying_groups"), + migrations.AddField( + model_name="product", + name="clic_limit", + field=models.PositiveSmallIntegerField( + blank=True, + help_text=( + "If a limit is set, the product won't be purchasable " + "anymore once the latter is reached." + ), + null=True, + verbose_name="clic limit", + ), + ), ] diff --git a/counter/models.py b/counter/models.py index 899dd4cd..889a47e4 100644 --- a/counter/models.py +++ b/counter/models.py @@ -34,6 +34,7 @@ from django.forms import ValidationError from django.urls import reverse from django.utils import timezone from django.utils.functional import cached_property +from django.utils.timezone import now from django.utils.translation import gettext_lazy as _ from django_celery_beat.models import PeriodicTask from django_countries.fields import CountryField @@ -353,6 +354,38 @@ class ProductType(OrderedModel): return user.is_in_group(pk=settings.SITH_GROUP_ACCOUNTING_ADMIN_ID) +class ProductQuerySet(models.QuerySet): + def under_clic_limit(self) -> Self: + """Filter product which clic limit isn't reached yet. + + The clic limit is reached when the amount of sales + and of items in a basket for less than 15 minutes + is greater or equal than `Product.clic_limit`. + """ + # import here to avoid circular import + from eboutic.models import BasketItem + + nb_click_subquery = Subquery( + Selling.objects.filter(product_id=OuterRef("id")) + .values("product_id") + .annotate(res=Sum("quantity", default=0)) + .values("res")[:1] + ) + nb_basket_items_subquery = Subquery( + BasketItem.objects.filter( + product_id=OuterRef("id"), + basket__date__gt=now() - settings.SITH_EBOUTIC_BASKET_TIMEOUT, + ) + .values("product_id") + .annotate(res=Sum("quantity")) + .values("res")[:1] + ) + return self.annotate( + clicked=Coalesce(nb_click_subquery, 0), + reserved=Coalesce(nb_basket_items_subquery, 0), + ).filter(Q(clic_limit=None) | Q(clic_limit__gt=(F("clicked") + F("reserved")))) + + class Product(models.Model): """A product, with all its related information.""" @@ -370,8 +403,7 @@ class Product(models.Model): ) code = models.CharField(_("code"), max_length=16, blank=True) purchase_price = CurrencyField( - _("purchase price"), - help_text=_("Initial cost of purchasing the product"), + _("purchase price"), help_text=_("Initial cost of purchasing the product") ) icon = ResizedImageField( height=70, @@ -388,10 +420,21 @@ class Product(models.Model): tray = models.BooleanField( _("tray price"), help_text=_("Buy five, get the sixth free"), default=False ) + clic_limit = models.PositiveSmallIntegerField( + _("clic limit"), + help_text=_( + "If a limit is set, the product won't be purchasable " + "anymore on the eboutic once the latter is reached." + ), + null=True, + blank=True, + ) archived = models.BooleanField(_("archived"), default=False) created_at = models.DateTimeField(_("created at"), auto_now_add=True) updated_at = models.DateTimeField(_("updated at"), auto_now=True) + objects = ProductQuerySet.as_manager() + class Meta: verbose_name = _("product") diff --git a/counter/tests/test_product.py b/counter/tests/test_product.py index 5fb6e233..c856e266 100644 --- a/counter/tests/test_product.py +++ b/counter/tests/test_product.py @@ -1,3 +1,5 @@ +import itertools +from datetime import timedelta from io import BytesIO from typing import Callable from uuid import uuid4 @@ -8,6 +10,7 @@ from django.core.cache import cache from django.core.files.uploadedfile import SimpleUploadedFile from django.test import Client, TestCase from django.urls import reverse +from django.utils.timezone import now from model_bakery import baker from model_bakery.recipe import Recipe from PIL import Image @@ -16,9 +19,10 @@ from pytest_django.asserts import assertNumQueries, assertRedirects from club.models import Club from core.baker_recipes import board_user, subscriber_user from core.models import Group, User -from counter.baker_recipes import product_recipe +from counter.baker_recipes import product_recipe, sale_recipe from counter.forms import ProductForm, ProductPriceFormSet -from counter.models import Price, Product, ProductType +from counter.models import Price, Product, ProductType, Selling +from eboutic.models import Basket, BasketItem @pytest.mark.django_db @@ -222,3 +226,57 @@ def test_price_for_user(): assert list(qs.for_user(users[0])) == [prices[0], prices[1], prices[4]] assert list(qs.for_user(users[1])) == [prices[0], prices[4]] assert list(qs.for_user(users[2])) == [prices[0], prices[3]] + + +class TestProductClicLimit(TestCase): + @classmethod + def setUpTestData(cls): + cls.products = product_recipe.make( + clic_limit=itertools.chain([5, 10, 15], itertools.repeat(None)), + _quantity=6, + _bulk_create=True, + ) + cls.qs = Product.objects.filter(id__in=[p.id for p in cls.products]) + + def test_no_sales_or_basket(self): + """Test that it works if no sales has been made yet""" + assert list(self.qs.under_clic_limit()) == self.products + + def test_with_sales(self): + """Test that it works when there are existing sales""" + sales = sale_recipe.make( + product=itertools.cycle(self.products), + _quantity=len(self.products) * 5, + _bulk_create=True, + ) + Selling.objects.filter(id__in=[s.id for s in sales]).update(quantity=2) + assert list(self.qs.under_clic_limit()) == self.products[2:] + + def test_with_sales_and_basket(self): + """Test that it works when there are existing sales and basket items.""" + sales = sale_recipe.make( + product=itertools.cycle(self.products), + _quantity=len(self.products) * 5, + _bulk_create=True, + ) + Selling.objects.filter(id__in=[s.id for s in sales]).update(quantity=1) + basket = baker.make( + Basket, date=now() - settings.SITH_EBOUTIC_BASKET_TIMEOUT / 2 + ) + items = baker.make( + BasketItem, + product=itertools.cycle(self.products), + basket=basket, + _quantity=len(self.products) * 5, + ) + BasketItem.objects.filter(id__in=[i.id for i in items]).update(quantity=1) + assert list(self.qs.under_clic_limit()) == self.products[2:] + + # expired basket items shouldn't be accounted when computing clic limit + item = BasketItem.objects.filter(product=self.products[1])[0] + item.basket = baker.make( + Basket, + date=now() - settings.SITH_EBOUTIC_BASKET_TIMEOUT - timedelta(minutes=1), + ) + item.save() + assert list(self.qs.under_clic_limit()) == self.products[1:]