OM-239 Modify the method used to calculate the limit

worker_voucher/gql_queries.py

@@ -1,20 +1,20 @@
 import graphene
 from graphene_django import DjangoObjectType
 
-from core import ExtendedConnection, prefix_filterset
+from core import ExtendedConnection, prefix_filterset, datetime
 from insuree.gql_queries import InsureeGQLType, PhotoGQLType, GenderGQLType
 from insuree.models import Insuree
 from invoice.models import Bill
 from policyholder.gql import PolicyHolderGQLType
 from worker_voucher.models import WorkerVoucher
-from worker_voucher.services import get_worker_yearly_voucher_count
+from worker_voucher.services import get_worker_yearly_voucher_count_counts
 
 
 class WorkerGQLType(InsureeGQLType):
-    vouchers_this_year = graphene.Int()
+    vouchers_this_year = graphene.JSONString()
 
     def resolve_vouchers_this_year(self, info):
-        return get_worker_yearly_voucher_count(self.id)
+        return get_worker_yearly_voucher_count_counts(self.id, info.context.user, datetime.date.today().year)
 
     class Meta:
         model = Insuree

worker_voucher/services.py

@@ -4,7 +4,7 @@
 from uuid import uuid4
 
 from django.db import transaction
-from django.db.models import Q, QuerySet, UUIDField
+from django.db.models import Q, QuerySet, UUIDField, Count
 from django.db.models.functions import Cast
 from django.utils.translation import gettext as _
 
@@ -105,7 +105,10 @@ def validate_acquire_assigned_vouchers(user: User, eu_code: str, workers: List[s
         vouchers_per_insuree_count = len(dates)
         check_existing_active_vouchers(ph, insurees, dates)
         for insuree in insurees:
-            _check_voucher_limit(insuree, vouchers_per_insuree_count)
+            years = {date.year for date in dates}
+            for year in years:
+                count = sum(1 for d in dates if d.year == year)
+                _check_voucher_limit(insuree, user, ph, year, count)
         count = insurees_count * vouchers_per_insuree_count
         return {
             "success": True,
@@ -130,7 +133,10 @@ def validate_assign_vouchers(user: User, eu_code: str, workers: List[str], date_
         dates = _check_dates(date_ranges)
         vouchers_per_insuree_count = len(dates)
         for insuree in insurees:
-            _check_voucher_limit(insuree, vouchers_per_insuree_count)
+            years = {date.year for date in dates}
+            for year in years:
+                count = sum(1 for d in dates if d.year == year)
+                _check_voucher_limit(insuree, user, ph, year, count)
         check_existing_active_vouchers(ph, insurees, dates)
         count = insurees_count * vouchers_per_insuree_count
         unassigned_vouchers = _check_unassigned_vouchers(ph, dates, count)
@@ -179,8 +185,10 @@ def _check_insurees(workers: List[str], eu_code: str, user: User):
     return insurees
 
 
-def _check_voucher_limit(insuree, count=1):
-    if get_worker_yearly_voucher_count(insuree.id) + count > WorkerVoucherConfig.yearly_worker_voucher_limit:
+def _check_voucher_limit(insuree, user, policyholder, year, count=1):
+    voucher_counts = get_worker_yearly_voucher_count_counts(insuree, user, year)
+
+    if voucher_counts.get(policyholder.code, 0) + count > WorkerVoucherConfig.yearly_worker_voucher_limit:
         raise VoucherException(_(f"Worker {insuree.chf_id} reached yearly voucher limit"))
 
 
@@ -235,13 +243,16 @@ def _check_unassigned_vouchers(ph, dates, count):
     return unassigned_vouchers
 
 
-def get_worker_yearly_voucher_count(insuree_id):
-    return WorkerVoucher.objects.filter(
+def get_worker_yearly_voucher_count_counts(insuree: Insuree, user: User, year):
+    res = WorkerVoucher.objects.filter(
+        economic_unit_user_filter(user, prefix="policyholder__"),
         is_deleted=False,
         status__in=(WorkerVoucher.Status.ASSIGNED, WorkerVoucher.Status.AWAITING_PAYMENT),
-        insuree_id=insuree_id,
-        assigned_date__year=datetime.datetime.now().year
-    ).count()
+        insuree=insuree,
+        assigned_date__year=year
+    ).values("policyholder__code").annotate(count=Count("id"))
+
+    return {row["policyholder__code"]: row["count"] for row in res}
 
 
 def create_assigned_voucher(user, date, insuree_id, policyholder_id):
@@ -252,11 +263,10 @@ def create_assigned_voucher(user, date, insuree_id, policyholder_id):
         expiry_date = datetime.datetime(current_date.year, 12, 31, 23, 59, 59)
     elif expiry_type == "fixed_period":
         expiry_period = WorkerVoucherConfig.voucher_expiry_period
-        expiry_date = datetime.datetime.now() + datetime.timedelta(**expiry_period)
+        expiry_date = datetime.datetime.now() + datetime.datetimedelta(**expiry_period)
     else:
         raise ValueError(f"Unknown expiry type: {expiry_type}")
 
-    expiry_period = WorkerVoucherConfig.voucher_expiry_period
     voucher_service = WorkerVoucherService(user)
     service_result = voucher_service.create({
         "policyholder_id": policyholder_id,
@@ -279,7 +289,7 @@ def create_unassigned_voucher(user, policyholder_id):
         expiry_date = datetime.datetime(current_date.year, 12, 31, 23, 59, 59)
     elif expiry_type == "fixed_period":
         expiry_period = WorkerVoucherConfig.voucher_expiry_period
-        expiry_date = datetime.datetime.now() + datetime.timedelta(**expiry_period)
+        expiry_date = datetime.datetime.now() + datetime.datetimedelta(**expiry_period)
     else:
         raise ValueError(f"Unknown expiry type: {expiry_type}")
 

worker_voucher/tests/test_validate_acquire_assigned.py

@@ -1,11 +1,13 @@
+from dateutils import years
 from django.test import TestCase
 
 from core import datetime
 from core.models import Role
-
 from core.test_helpers import create_test_interactive_user
-from worker_voucher.services import validate_acquire_assigned_vouchers
-from worker_voucher.tests.util import create_test_eu_for_user, create_test_worker_for_eu
+from worker_voucher.apps import WorkerVoucherConfig
+from worker_voucher.services import validate_acquire_assigned_vouchers, create_assigned_voucher
+from worker_voucher.tests.util import create_test_eu_for_user, create_test_worker_for_eu, \
+    OverrideAppConfigContextManager as override_config
 
 
 class ValidateAcquireAssignedTestCase(TestCase):
@@ -80,3 +82,51 @@ def test_validate_dates_overlap(self):
         res = validate_acquire_assigned_vouchers(*payload)
 
         self.assertFalse(res['success'])
+
+    @override_config(WorkerVoucherConfig, {"yearly_worker_voucher_limit": 3,
+                                           "voucher_expiry_type": "fixed_period",
+                                           "voucher_expiry_period": {"years": 2}})
+    def test_validate_worker_voucher_limit_reached(self):
+        voucher_limit = WorkerVoucherConfig.yearly_worker_voucher_limit
+        date_start = datetime.date(datetime.date.today().year + 1, 1, 1)
+        self._acquire_vouchers(date_start, voucher_limit)
+
+        date_test = date_start + datetime.datetimedelta(days=voucher_limit)
+
+        payload = (
+            self.user,
+            self.eu.code,
+            (self.worker.chf_id,),
+            ([{'start_date': date_test, 'end_date': date_test}])
+        )
+
+        res = validate_acquire_assigned_vouchers(*payload)
+
+        self.assertFalse(res['success'])
+
+    @override_config(WorkerVoucherConfig, {"yearly_worker_voucher_limit": 3,
+                                           "voucher_expiry_type": "fixed_period",
+                                           "voucher_expiry_period": {"years": 2}})
+    def test_validate_worker_voucher_limit_next_year(self):
+        voucher_limit = WorkerVoucherConfig.yearly_worker_voucher_limit
+        date_start = datetime.date(datetime.date.today().year + 1, 1, 1)
+        self._acquire_vouchers(date_start, voucher_limit)
+
+        date_test = date_start + datetime.datetimedelta(years=1)
+
+        payload = (
+            self.user,
+            self.eu.code,
+            (self.worker.chf_id,),
+            ([{'start_date': date_test, 'end_date': date_test}])
+        )
+
+        res = validate_acquire_assigned_vouchers(*payload)
+
+        self.assertTrue(res['success'])
+
+    def _acquire_vouchers(self, date_start, amount):
+        dates = [date_start + datetime.datetimedelta(days=i) for i in range(amount)]
+
+        for date in dates:
+            create_assigned_voucher(self.user, date, self.worker.id, self.eu.id)

worker_voucher/tests/util.py

@@ -1,4 +1,7 @@
 import random
+from contextlib import ContextDecorator
+
+from django.apps import AppConfig
 
 from insuree.models import Insuree
 from policyholder.models import PolicyHolder, PolicyHolderUser, PolicyHolderInsuree
@@ -12,6 +15,7 @@ def create_test_eu(user, code='test_eu'):
     eu.save(user=user)
     return eu
 
+
 def create_test_phu(user, eu):
     phu = PolicyHolderUser(
         policy_holder=eu,
@@ -26,6 +30,7 @@ def create_test_eu_for_user(user, code='test_eu'):
     _ = create_test_phu(user, eu)
     return eu
 
+
 def create_test_worker(user, chf_id="2675135421017"):
     worker = Insuree(
         other_names="Test",
@@ -36,6 +41,7 @@ def create_test_worker(user, chf_id="2675135421017"):
     worker.save()
     return worker
 
+
 def create_test_phi(user, eu, worker):
     phi = PolicyHolderInsuree(
         policy_holder=eu,
@@ -50,6 +56,7 @@ def create_test_worker_for_eu(user, eu, chf_id="2675135421017"):
     _ = create_test_phi(user, eu, worker)
     return worker
 
+
 def get_idnp_crc(idnp_first_12_digits):
     assert len(idnp_first_12_digits) == 12
 
@@ -64,4 +71,27 @@ def get_idnp_crc(idnp_first_12_digits):
 
 def generate_idnp():
     idnp_first_12_digits = random.randint(200000000000, 299999999999)
-    return str(idnp_first_12_digits) + str(get_idnp_crc(str(idnp_first_12_digits)))
+    return str(idnp_first_12_digits) + str(get_idnp_crc(str(idnp_first_12_digits)))
+
+
+class OverrideAppConfigContextManager(ContextDecorator):
+    """
+    Context manager/decorator for overriding default config for tests
+    """
+    config_class: AppConfig
+    temp_config: dict
+    original_config: dict
+
+    def __init__(self, config_class, config):
+        self.config_class = config_class
+        self.temp_config = config
+        self.original_config = dict()
+
+    def __enter__(self):
+        for key in self.temp_config:
+            self.original_config[key] = getattr(self.config_class, key)
+            setattr(self.config_class, key, self.temp_config[key])
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        for key in self.original_config:
+            setattr(self.config_class, key, self.original_config[key])