fix: validate subsidy uniqueness with clean() method.

Use a clean method to validate that non-internal-only subsidies
must be unique on (reference_id, reference_type), since mysql
does not support conditional unique constraints. ENT-7891

enterprise_subsidy/apps/api/v1/tests/test_views.py

@@ -156,7 +156,7 @@ def test_get_one_subsidy(self):
         expected_result = {
             "uuid": str(self.subsidy_1.uuid),
             "title": self.subsidy_1.title,
-            "enterprise_customer_uuid": self.subsidy_1.enterprise_customer_uuid,
+            "enterprise_customer_uuid": str(self.subsidy_1.enterprise_customer_uuid),
             "active_datetime": self.subsidy_1.active_datetime.strftime(SERIALIZED_DATE_PATTERN),
             "expiration_datetime": self.subsidy_1.expiration_datetime.strftime(SERIALIZED_DATE_PATTERN),
             "unit": self.subsidy_1.unit,

enterprise_subsidy/apps/subsidy/models.py

@@ -13,6 +13,7 @@
 from unittest import mock
 from uuid import uuid4
 
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils.functional import cached_property
 from edx_rbac.models import UserRole, UserRoleAssignment
@@ -210,6 +211,30 @@ class Meta:
     objects = ActiveSubsidyManager()
     all_objects = models.Manager()
 
+    def clean(self):
+        """
+        Ensures that non-internal-only subsidies are unique
+        on (reference_id, reference_type).  This is necessary
+        because MySQL does not support conditional unique constraints.
+        """
+        if not self.internal_only:
+            other_record = Subsidy.objects.filter(
+                reference_id=self.reference_id,
+                reference_type=self.reference_type,
+            ).exclude(uuid=self.uuid).first()
+            if other_record:
+                raise ValidationError(
+                    f'Subsidy {other_record.uuid} already exists with the same '
+                    f'reference_id {self.reference_id} and reference_type {self.reference_type}'
+                )
+
+    def save(self, *args, **kwargs):
+        """
+        Overrides default save() method to run full_clean.
+        """
+        self.full_clean()
+        super().save(*args, **kwargs)
+
     @property
     def is_active(self):
         """

enterprise_subsidy/apps/subsidy/tests/test_api.py

@@ -1,11 +1,13 @@
 """
 Tests for functions defined in the ``api.py`` module.
 """
+from datetime import timedelta
 from unittest import mock
 from uuid import uuid4
 
 import pytest
 from django.test import TestCase
+from django.utils import timezone
 from openedx_ledger.models import Reversal, TransactionStateChoices, UnitChoices
 from openedx_ledger.test_utils.factories import TransactionFactory
 
@@ -25,8 +27,8 @@ def learner_credit_fixture():
         default_enterprise_customer_uuid=uuid4(),
         default_unit=UnitChoices.USD_CENTS,
         default_starting_balance=1000000,
-        default_active_datetime=None,
-        default_expiration_datetime=None,
+        default_active_datetime=timezone.now() - timedelta(days=365),
+        default_expiration_datetime=timezone.now() + timedelta(days=365),
     )
     return subsidy
 
@@ -72,8 +74,8 @@ def test_create_internal_only_subsidy_record(learner_credit_fixture):  # pylint:
         default_enterprise_customer_uuid=other_customer_uuid,
         default_unit=UnitChoices.USD_CENTS,
         default_starting_balance=42,
-        default_active_datetime=None,
-        default_expiration_datetime=None,
+        default_active_datetime=timezone.now() - timedelta(days=365),
+        default_expiration_datetime=timezone.now() + timedelta(days=365),
         default_internal_only=True
     )
     assert created

enterprise_subsidy/apps/subsidy/tests/test_models.py

@@ -1,12 +1,14 @@
 """
 Tests for functionality provided in the ``models.py`` module.
 """
+import random
 from itertools import product
 from unittest import mock
 from uuid import uuid4
 
 import ddt
 import pytest
+from django.core.exceptions import ValidationError
 from django.test import TestCase
 from openedx_ledger.models import Transaction, TransactionStateChoices
 from openedx_ledger.test_utils.factories import (
@@ -99,6 +101,49 @@ def test_price_for_content_not_in_catalog(self):
         with self.assertRaises(ContentNotFoundForCustomerException):
             self.subsidy.price_for_content('some-content-key')
 
+    def test_reference_uniqueness(self):
+        """
+        Tests that not soft-deleted, non-internal-only subsidies
+        are validated to be unique on (reference_id, reference_type).
+        """
+        reference_id = random.randint(1, 10000000)
+        existing_record = SubsidyFactory.create(
+            enterprise_customer_uuid=self.enterprise_customer_uuid,
+            reference_id=reference_id,
+            internal_only=False,
+        )
+        existing_record.save()
+
+        with self.assertRaisesRegex(ValidationError, 'already exists with the same reference_id'):
+            new_record = SubsidyFactory.create(
+                enterprise_customer_uuid=self.enterprise_customer_uuid,
+                reference_id=reference_id,
+                internal_only=False,
+            )
+            new_record.save()
+
+    def test_reference_uniqueness_not_constrained_on_internal_only(self):
+        """
+        Tests that not soft-deleted, internal-only subsidies
+        are allowed to not be unique on (reference_id, reference_type).
+        """
+        reference_id = random.randint(1, 100000000)
+        existing_record = SubsidyFactory.create(
+            enterprise_customer_uuid=self.enterprise_customer_uuid,
+            reference_id=reference_id,
+            internal_only=True,
+        )
+        existing_record.save()
+
+        new_record = SubsidyFactory.create(
+            enterprise_customer_uuid=self.enterprise_customer_uuid,
+            reference_id=reference_id,
+            internal_only=True,
+        )
+        new_record.save()
+        self.assertEqual(existing_record.reference_id, new_record.reference_id)
+        self.assertEqual(existing_record.reference_type, new_record.reference_type)
+
     @ddt.data(True, False)
     def test_is_redeemable(self, expected_to_be_redeemable):
         """