feat: can_allocate() implementation for assignment-based policies

enterprise_access/apps/content_assignments/api.py

@@ -0,0 +1,39 @@
+"""
+Primary Python API for interacting with Assignment
+records and business logic.
+"""
+from django.db.models import Sum
+
+from .constants import LearnerContentAssignmentStateChoices
+from .models import LearnerContentAssignment
+
+
+def get_assignments_for_policy(
+    subsidy_access_policy,
+    state=LearnerContentAssignmentStateChoices.ALLOCATED,
+):
+    """
+    Returns a queryset of all ``LearnerContentAssignment`` records
+    for the given policy, optionally filtered to only those
+    associated with the given ``learner_emails``.
+    """
+    queryset = LearnerContentAssignment.objects.select_related(
+        'assignment_policy',
+        'assignment_policy__subsidy_access_policy',
+    ).filter(
+        assignment_policy__subsidy_access_policy=subsidy_access_policy,
+        state=state,
+    )
+    return queryset
+
+
+def get_allocated_quantity_for_policy(subsidy_access_policy):
+    """
+    Returns a float representing the total quantity, in USD cents, currently allocated
+    via Assignments for the given policy.
+    """
+    assignments_queryset = get_assignments_for_policy(subsidy_access_policy)
+    aggregate = assignments_queryset.aggregate(
+        total_quantity=Sum('content_quantity'),
+    )
+    return aggregate['total_quantity']

enterprise_access/apps/content_assignments/models.py

@@ -7,8 +7,6 @@
 from django_extensions.db.models import TimeStampedModel
 from simple_history.models import HistoricalRecords
 
-from enterprise_access.apps.subsidy_access_policy.models import SubsidyAccessPolicy
-
 from .constants import LearnerContentAssignmentStateChoices
 
 
@@ -25,7 +23,7 @@ class AssignmentPolicy(TimeStampedModel):
         unique=True,
     )
     subsidy_access_policy = models.ForeignKey(
-        SubsidyAccessPolicy,
+        'subsidy_access_policy.SubsidyAccessPolicy',
         related_name="assignment_policy",
         on_delete=models.CASCADE,
         db_index=True,

enterprise_access/apps/content_assignments/tests/factories.py

@@ -0,0 +1,37 @@
+"""
+Factoryboy factories.
+"""
+
+from uuid import uuid4
+
+import factory
+from faker import Faker
+
+from ..models import LearnerContentAssignment
+
+FAKER = Faker()
+
+
+def random_content_key():
+    """
+    Helper to craft a random content key.
+    """
+    fake_words = [
+        FAKER.word() + str(FAKER.random_int())
+        for _ in range(3)
+    ]
+    return 'course-v1:{}+{}+{}'.format(*fake_words)
+
+
+class LearnerContentAssignmentFactory(factory.django.DjangoModelFactory):
+    """
+    Base Test factory for the ``LearnerContentAssisgnment`` model.
+    """
+    class Meta:
+        model = LearnerContentAssignment
+
+    uuid = factory.LazyFunction(uuid4)
+    learner_email = factory.LazyAttribute(lambda _: FAKER.email())
+    lms_user_id = factory.LazyAttribute(lambda _: FAKER.pyint())
+    content_key = factory.LazyAttribute(lambda _: random_content_key())
+    content_quantity = factory.LazyAttribute(lambda _: FAKER.pyint())

enterprise_access/apps/content_assignments/tests/test_api.py

@@ -0,0 +1,95 @@
+"""
+Tests for the ``api.py`` module of the content_assignments app.
+"""
+import uuid
+
+from django.test import TestCase
+
+from ...subsidy_access_policy.tests.factories import AssignedLearnerCreditAccessPolicyFactory
+from ..api import get_allocated_quantity_for_policy, get_assignments_for_policy
+from ..constants import LearnerContentAssignmentStateChoices
+from ..models import AssignmentPolicy
+from .factories import LearnerContentAssignmentFactory
+
+ACTIVE_ASSIGNED_LEARNER_CREDIT_POLICY_UUID = uuid.uuid4()
+
+
+class TestContentAssignmentApi(TestCase):
+    """
+    Tests functions of the ``content_assignment.api`` module.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.active_policy = AssignedLearnerCreditAccessPolicyFactory(
+            uuid=ACTIVE_ASSIGNED_LEARNER_CREDIT_POLICY_UUID,
+            spend_limit=10000,
+        )
+        cls.assignment_policy = AssignmentPolicy.objects.create(
+            subsidy_access_policy=cls.active_policy,
+        )
+
+    def test_get_assignments_for_policy(self):
+        """
+        Simple test to fetch assignment records related to a given policy.
+        """
+        expected_assignments = [
+            LearnerContentAssignmentFactory.create(
+                assignment_policy=self.assignment_policy,
+            ) for _ in range(10)
+        ]
+
+        with self.assertNumQueries(1):
+            actual_assignments = list(get_assignments_for_policy(self.active_policy))
+
+        self.assertEqual(
+            sorted(actual_assignments, key=lambda record: record.uuid),
+            sorted(expected_assignments, key=lambda record: record.uuid),
+        )
+
+    def test_get_assignments_for_policy_different_states(self):
+        """
+        Simple test to fetch assignment records related to a given policy,
+        filtered among different states
+        """
+        expected_assignments = {
+            LearnerContentAssignmentStateChoices.CANCELLED: [],
+            LearnerContentAssignmentStateChoices.ACCEPTED: [],
+        }
+        for index in range(10):
+            if index % 2:
+                state = LearnerContentAssignmentStateChoices.CANCELLED
+            else:
+                state = LearnerContentAssignmentStateChoices.ACCEPTED
+
+            expected_assignments[state].append(
+                LearnerContentAssignmentFactory.create(assignment_policy=self.assignment_policy, state=state)
+            )
+
+        for filter_state in (
+            LearnerContentAssignmentStateChoices.CANCELLED,
+            LearnerContentAssignmentStateChoices.ACCEPTED,
+        ):
+            with self.assertNumQueries(1):
+                actual_assignments = list(get_assignments_for_policy(self.active_policy, filter_state))
+
+            self.assertEqual(
+                sorted(actual_assignments, key=lambda record: record.uuid),
+                sorted(expected_assignments[filter_state], key=lambda record: record.uuid),
+            )
+
+    def test_get_allocated_quantity_for_policy(self):
+        """
+        Tests to verify that we can fetch the total allocated quantity across a set of assignments
+        related to some policy.
+        """
+        for amount in (1000, 2000, 3000):
+            LearnerContentAssignmentFactory.create(
+                assignment_policy=self.assignment_policy,
+                content_quantity=amount,
+            )
+
+        with self.assertNumQueries(1):
+            actual_amount = get_allocated_quantity_for_policy(self.active_policy)
+            self.assertEqual(actual_amount, 6000)

enterprise_access/apps/subsidy_access_policy/models.py

@@ -15,6 +15,7 @@
 from edx_django_utils.cache.utils import get_cache_key
 
 from enterprise_access.apps.api_client.lms_client import LmsApiClient
+from enterprise_access.apps.content_assignments import api as assignments_api
 
 from .constants import (
     CREDIT_POLICY_TYPE_PRIORITY,
@@ -809,3 +810,53 @@ def can_redeem(self, lms_user_id, content_key, skip_customer_user_check=False):
 
     def redeem(self, lms_user_id, content_key, all_transactions, metadata=None):
         raise NotImplementedError
+
+    def can_allocate(self, number_of_learners, content_key, content_price_cents):
+        """
+        Takes allocated LearnerContentAssignment records related to this policy
+        into account to determine if ``number_of_learners`` new assignment
+        records can be allocated in this policy for the given ``content_key``
+        and it's current ``content_price_cents``.
+        """
+        # inactive policy
+        if not self.active:
+            return (False, REASON_POLICY_EXPIRED)
+
+        # no content key in catalog
+        if not self.catalog_contains_content_key(content_key):
+            return (False, REASON_CONTENT_NOT_IN_CATALOG)
+
+        if not self.is_subsidy_active:
+            return (False, REASON_SUBSIDY_EXPIRED)
+
+        # Determine total cost, in cents, of content to potentially allocated
+        total_price_cents = number_of_learners * content_price_cents
+
+        # Determine total amount, in cents, already transacted via this policy.
+        # This is a number <= 0
+        spent_amount_cents = self.aggregates_for_policy().get('total_quantity') or 0
+
+        # Determine total amount, in cents, of assignments already
+        # allocated via this policy. This is a number <= 0
+        total_allocated_assignments_cents = assignments_api.get_allocated_quantity_for_policy(self)
+        total_allocated_and_spent_cents = spent_amount_cents + total_allocated_assignments_cents
+
+        # Use all of these pieces to ensure that the assignments to potentially
+        # allocate won't exceed the remaining balance of the related subsidy.
+        if self.content_would_exceed_limit(
+            total_allocated_and_spent_cents,
+            self.subsidy_balance(),
+            total_price_cents,
+        ):
+            return (False, REASON_NOT_ENOUGH_VALUE_IN_SUBSIDY)
+
+        # Lastly, use all of these pieces to ensure that the assignments to potentially
+        # allocate won't exceed the spend limit of this policy
+        if self.content_would_exceed_limit(
+            total_allocated_and_spent_cents,
+            self.spend_limit,
+            total_price_cents,
+        ):
+            return (False, REASON_POLICY_SPEND_LIMIT_REACHED)
+
+        return (True, None)

enterprise_access/apps/subsidy_access_policy/tests/factories.py

@@ -9,6 +9,7 @@
 
 from enterprise_access.apps.subsidy_access_policy.constants import AccessMethods
 from enterprise_access.apps.subsidy_access_policy.models import (
+    AssignedLearnerCreditAccessPolicy,
     PerLearnerEnrollmentCreditAccessPolicy,
     PerLearnerSpendCreditAccessPolicy
 )
@@ -47,3 +48,16 @@ class PerLearnerSpendCapLearnerCreditAccessPolicyFactory(SubsidyAccessPolicyFact
 
     class Meta:
         model = PerLearnerSpendCreditAccessPolicy
+
+
+class AssignedLearnerCreditAccessPolicyFactory(SubsidyAccessPolicyFactory):
+    """
+    Test factory for the `AssignedLearnerCreditAccessPolicy` model.
+    """
+
+    class Meta:
+        model = AssignedLearnerCreditAccessPolicy
+
+    access_method = AccessMethods.ASSIGNED
+    per_learner_spend_limit = None
+    per_learner_enrollment_limit = None