Update attach user endpoint to check redemptions more closely (#3066)

Author: jkachel

b2b/migrations/0012_fix_contract_attachment_related_names.py

@@ -0,0 +1,46 @@
+# Generated by Django 4.2.25 on 2025-11-03 21:18
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("ecommerce", "0036_alter_order_state"),
+        ("b2b", "0011_sync_contract_type_fields"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="discountcontractattachmentredemption",
+            name="contract",
+            field=models.ForeignKey(
+                help_text="The contract that the user was attached to.",
+                on_delete=django.db.models.deletion.DO_NOTHING,
+                related_name="code_redemptions",
+                to="b2b.contractpage",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="discountcontractattachmentredemption",
+            name="discount",
+            field=models.ForeignKey(
+                help_text="The discount that was redemeed.",
+                on_delete=django.db.models.deletion.DO_NOTHING,
+                related_name="contract_redemptions",
+                to="ecommerce.discount",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="discountcontractattachmentredemption",
+            name="user",
+            field=models.ForeignKey(
+                help_text="The user that redeemed the discount.",
+                on_delete=django.db.models.deletion.DO_NOTHING,
+                related_name="+",
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]

b2b/models.py

@@ -432,16 +432,19 @@ class DiscountContractAttachmentRedemption(TimestampedModel):
         "ecommerce.Discount",
         on_delete=models.DO_NOTHING,
         help_text="The discount that was redemeed.",
+        related_name="contract_redemptions",
     )
     user = models.ForeignKey(
         settings.AUTH_USER_MODEL,
         on_delete=models.DO_NOTHING,
         help_text="The user that redeemed the discount.",
+        related_name="+",
     )
     contract = models.ForeignKey(
         ContractPage,
         on_delete=models.DO_NOTHING,
         help_text="The contract that the user was attached to.",
+        related_name="code_redemptions",
     )
 
 

b2b/views/v0/__init__.py

@@ -1,7 +1,7 @@
 """Views for the B2B API (v0)."""
 
 from django.contrib.contenttypes.models import ContentType
-from django.db.models import Q
+from django.db.models import Count, Q
 from django.views.decorators.csrf import csrf_exempt
 from drf_spectacular.utils import extend_schema
 from mitol.common.utils.datetime import now_in_utc
@@ -23,6 +23,7 @@
     OrganizationPageSerializer,
 )
 from courses.models import CourseRun
+from ecommerce.constants import REDEMPTION_TYPE_UNLIMITED
 from ecommerce.models import Discount, Product
 from main.authentication import CsrfExemptSessionAuthentication
 from main.constants import USER_MSG_TYPE_B2B_ENROLL_SUCCESS
@@ -107,7 +108,8 @@ def post(self, request, enrollment_code: str, format=None):  # noqa: A002, ARG00
 
         This will respect the activation and expiration dates (of both the contract
         and the discount), and will make sure there's sufficient available seats
-        in the contract.
+        in the contract. It will also make sure the code hasn't been used for
+        attachment purposes before.
 
         If the user is already in the contract, then we skip it.
 
@@ -118,10 +120,13 @@ def post(self, request, enrollment_code: str, format=None):  # noqa: A002, ARG00
         now = now_in_utc()
         try:
             code = (
-                Discount.objects.filter(
-                    Q(activation_date__isnull=True) | Q(activation_date__lte=now)
-                )
+                Discount.objects.annotate(Count("contract_redemptions"))
+                .filter(Q(activation_date__isnull=True) | Q(activation_date__lte=now))
                 .filter(Q(expiration_date__isnull=True) | Q(expiration_date__gte=now))
+                .filter(
+                    Q(redemption_type=REDEMPTION_TYPE_UNLIMITED)
+                    | Q(contract_redemptions__count__lt=1)
+                )
                 .get(discount_code=enrollment_code)
             )
         except Discount.DoesNotExist:

b2b/views/v0/views_test.py

@@ -39,17 +39,33 @@ def test_b2b_contract_attachment_bad_code(user):
     assert user.b2b_contracts.count() == 0
 
 
-def test_b2b_contract_attachment(mocker, user):
+@pytest.mark.parametrize(
+    "max_learners",
+    [
+        10,
+        None,
+    ],
+)
+@pytest.mark.parametrize(
+    "code_used",
+    [
+        True,
+        False,
+    ],
+)
+def test_b2b_contract_attachment(mocker, max_learners, code_used):
     """Ensure a supplied code results in attachment for the user."""
 
     mocked_attach_user = mocker.patch(
         "b2b.models.OrganizationPage.attach_user", return_value=True
     )
 
+    user = UserFactory.create()
+
     contract = ContractPageFactory.create(
         membership_type=CONTRACT_MEMBERSHIP_NONSSO,
         integration_type=CONTRACT_MEMBERSHIP_NONSSO,
-        max_learners=10,
+        max_learners=max_learners,
     )
 
     courserun = CourseRunFactory.create(b2b_contract=contract)
@@ -58,6 +74,14 @@ def test_b2b_contract_attachment(mocker, user):
     ensure_enrollment_codes_exist(contract)
     contract_codes = contract.get_discounts().all()
 
+    if code_used:
+        other_user = UserFactory.create()
+        DiscountContractAttachmentRedemption.objects.create(
+            discount=contract_codes[0],
+            user=other_user,
+            contract=contract,
+        )
+
     client = APIClient()
     client.force_login(user)
 
@@ -67,15 +91,25 @@ def test_b2b_contract_attachment(mocker, user):
     resp = client.post(url)
 
     assert resp.status_code == 200
-    mocked_attach_user.assert_called()
 
     user.refresh_from_db()
-    assert user.b2b_organizations.filter(pk=contract.organization.id).exists()
-    assert user.b2b_contracts.filter(pk=contract.id).exists()
 
-    assert DiscountContractAttachmentRedemption.objects.filter(
-        contract=contract, user=user, discount=contract_codes[0]
-    ).exists()
+    if code_used and max_learners:
+        assert not user.b2b_organizations.filter(pk=contract.organization.id).exists()
+        assert not user.b2b_contracts.filter(pk=contract.id).exists()
+
+        assert not DiscountContractAttachmentRedemption.objects.filter(
+            contract=contract, user=user, discount=contract_codes[0]
+        ).exists()
+    else:
+        mocked_attach_user.assert_called()
+
+        assert user.b2b_organizations.filter(pk=contract.organization.id).exists()
+        assert user.b2b_contracts.filter(pk=contract.id).exists()
+
+        assert DiscountContractAttachmentRedemption.objects.filter(
+            contract=contract, user=user, discount=contract_codes[0]
+        ).exists()
 
 
 @pytest.mark.parametrize(

courses/admin.py

@@ -129,6 +129,7 @@ class CourseRunAdmin(TimestampedModelAdmin):
     )
     list_filter = [
         "live",
+        "is_source_run",
         "course",
         "b2b_contract",
     ]