feat: add an action to delete revoked licenses without confirmation

We want a way to delete many (thousands) of revoked licenses from a plan
without having to wait for the confirmation page to load and possibly timeout.
Also, fixes many lint errors in the `subscriptions.tests` module.
ENT-8905

license_manager/apps/subscriptions/admin.py

@@ -292,6 +292,7 @@ def get_fields(self, request, obj=None):
     actions = [
         'process_unused_licenses_post_freeze',
         'create_actual_licenses_action',
+        'delete_all_revoked_licenses',
     ]
 
     def get_queryset(self, request):
@@ -345,6 +346,24 @@ def process_unused_licenses_post_freeze(self, request, queryset):
         except UnprocessableSubscriptionPlanFreezeError as exc:
             messages.add_message(request, messages.ERROR, exc)
 
+    @admin.action(
+        description='Delete all revoked licenses for the selected Subscription Plans'
+    )
+    def delete_all_revoked_licenses(self, request, queryset):
+        """
+        Delete all revoked licenses for the selected Subscription Plans. Good to use when
+        you want to delete a plan with thousands of revoked licenses and don't want to worry
+        about timeouts from the deletion confirmation page.
+        """
+        processed_plan_titles = []
+        with transaction.atomic():
+            for subscription_plan in queryset:
+                subscription_plan.revoked_licenses.delete()
+                processed_plan_titles.append(subscription_plan.title)
+        messages.add_message(
+            request, messages.SUCCESS, f'Successfully deleted revoked licenses for plans {processed_plan_titles}.',
+        )
+
     @admin.action(
         description='Create actual licenses to match desired number'
     )

license_manager/apps/subscriptions/tests/factories.py

@@ -31,7 +31,8 @@
 
 def get_random_salesforce_id():
     """
-    Returns a random alpha-numeric string of the correct length that starts with 00k for a salesforce opportunity line item.
+    Returns a random alpha-numeric string of the correct length that
+    starts with 00k for a salesforce opportunity line item.
     """
     return '00k' + ''.join(random.choices(string.ascii_uppercase + string.digits, k=SALESFORCE_ID_LENGTH - 3))
 

license_manager/apps/subscriptions/tests/test_admin.py

@@ -16,6 +16,7 @@
 )
 from license_manager.apps.subscriptions.tests.factories import (
     CustomerAgreementFactory,
+    LicenseFactory,
     SubscriptionPlanFactory,
     UserFactory,
 )
@@ -25,6 +26,8 @@
 )
 from license_manager.apps.subscriptions.utils import localized_utcnow
 
+from ..constants import REVOKED
+
 
 @pytest.mark.django_db
 def test_licenses_subscription_creation():
@@ -127,3 +130,39 @@ def test_select_subscription_for_auto_applied_licenses(mock_toggle_auto_apply_li
     args = mock_toggle_auto_apply_licenses.call_args[0]
     assert args[0] is customer_agreement_uuid
     assert args[1] is subscription_uuid
+
+
+@pytest.mark.django_db
+@mock.patch('license_manager.apps.subscriptions.admin.messages.add_message')
+def test_delete_all_revoked_licenses(mock_add_message):
+    """
+    Verify that creating a SubscriptionPlan creates its associated Licenses after it is created.
+    """
+    subscription_admin = SubscriptionPlanAdmin(SubscriptionPlan, AdminSite())
+    request = RequestFactory()
+    request.user = UserFactory()
+
+    # Setup an existing plan with revoked licenses
+    customer_agreement = CustomerAgreementFactory()
+    subscription_plan = SubscriptionPlanFactory.create(
+        customer_agreement=customer_agreement,
+    )
+    LicenseFactory.create_batch(
+        10,
+        subscription_plan=subscription_plan,
+        status=REVOKED,
+    )
+    subscription_plan.refresh_from_db()
+    assert subscription_plan.revoked_licenses.count() == 10
+
+    # Now use the admin action to delete all revoked licenses for the plan
+    subscription_admin.delete_all_revoked_licenses(
+        request,
+        SubscriptionPlan.objects.filter(uuid=subscription_plan.uuid),
+    )
+    subscription_plan.refresh_from_db()
+    assert subscription_plan.revoked_licenses.count() == 0
+
+    mock_add_message.assert_called_once_with(
+        request, messages.SUCCESS, f"Successfully deleted revoked licenses for plans ['{subscription_plan.title}'].",
+    )

license_manager/apps/subscriptions/tests/test_api.py

@@ -97,21 +97,20 @@ def _assert_all_licenses_renewed(self, future_plan):
 
     def test_renewal_processed_with_no_existing_future_plan(self):
         prior_plan = SubscriptionPlanFactory()
-        original_activated_licenses = [
+        _ = [
             LicenseFactory.create(
                 subscription_plan=prior_plan,
                 status=constants.ACTIVATED,
                 user_email='activated_user_{}@example.com'.format(i)
             ) for i in range(5)
         ]
-        original_assigned_licenses = [
+        _ = [
             LicenseFactory.create(
                 subscription_plan=prior_plan,
                 status=constants.ASSIGNED,
                 user_email='assigned_user_{}@example.com'.format(i)
             ) for i in range(5)
         ]
-        original_licenses = original_activated_licenses + original_assigned_licenses
 
         renewal = SubscriptionPlanRenewalFactory(
             prior_subscription_plan=prior_plan,
@@ -133,7 +132,7 @@ def test_renewal_processed_with_no_existing_future_plan(self):
 
     def test_renewal_processed_with_existing_future_plan(self):
         prior_plan = SubscriptionPlanFactory()
-        original_licenses = [
+        _ = [
             LicenseFactory.create(
                 subscription_plan=prior_plan,
                 status=constants.ACTIVATED,
@@ -224,11 +223,11 @@ def test_renewal_disable_auto_apply_licenses(self):
     @mock.patch('license_manager.apps.subscriptions.event_utils.track_event')
     def test_renewal_processed_segment_events(self, mock_track_event):
         prior_plan = SubscriptionPlanFactory()
-        [LicenseFactory.create(
+        LicenseFactory.create(
             subscription_plan=prior_plan,
             status=constants.ACTIVATED,
             user_email='activated_user_{}@example.com'
-        )]
+        )
 
         renewal = SubscriptionPlanRenewalFactory(
             prior_subscription_plan=prior_plan,
@@ -237,18 +236,18 @@ def test_renewal_processed_segment_events(self, mock_track_event):
         )
         api.renew_subscription(renewal)
         assert mock_track_event.call_count == 2
-        assert (mock_track_event.call_args_list[0].args[1] == constants.SegmentEvents.LICENSE_CREATED)
-        assert (mock_track_event.call_args_list[1].args[1] == constants.SegmentEvents.LICENSE_RENEWED)
+        assert mock_track_event.call_args_list[0].args[1] == constants.SegmentEvents.LICENSE_CREATED
+        assert mock_track_event.call_args_list[1].args[1] == constants.SegmentEvents.LICENSE_RENEWED
         self.assertFalse(mock_track_event.call_args_list[1].args[2]['is_auto_renewed'])
 
     @mock.patch('license_manager.apps.subscriptions.event_utils.track_event')
     def test_renewal_processed_segment_events_is_auto_renewed(self, mock_track_event):
         prior_plan = SubscriptionPlanFactory()
-        [LicenseFactory.create(
+        LicenseFactory.create(
             subscription_plan=prior_plan,
             status=constants.ACTIVATED,
             user_email='activated_user_{}@example.com'
-        )]
+        )
 
         renewal = SubscriptionPlanRenewalFactory(
             prior_subscription_plan=prior_plan,
@@ -257,8 +256,8 @@ def test_renewal_processed_segment_events_is_auto_renewed(self, mock_track_event
         )
         api.renew_subscription(renewal, is_auto_renewed=True)
         assert mock_track_event.call_count == 2
-        assert (mock_track_event.call_args_list[0].args[1] == constants.SegmentEvents.LICENSE_CREATED)
-        assert (mock_track_event.call_args_list[1].args[1] == constants.SegmentEvents.LICENSE_RENEWED)
+        assert mock_track_event.call_args_list[0].args[1] == constants.SegmentEvents.LICENSE_CREATED
+        assert mock_track_event.call_args_list[1].args[1] == constants.SegmentEvents.LICENSE_RENEWED
         self.assertTrue(mock_track_event.call_args_list[1].args[2]['is_auto_renewed'])
 
 

license_manager/apps/subscriptions/tests/test_event_utils.py

@@ -3,7 +3,6 @@
 """
 from unittest import mock
 
-from django.test.utils import override_settings
 from pytest import mark
 
 from license_manager.apps.subscriptions.constants import (
@@ -56,7 +55,7 @@ def test_get_license_tracking_properties():
     # Check that all the data is a basic type that can be serialized so it will be clean in segment:
     for k, v in flat_data.items():
         assert isinstance(k, str)
-        assert (isinstance(v, str) or isinstance(v, int) or isinstance(v, bool))
+        assert isinstance(v, (str, int, bool))
 
 
 @mark.django_db
@@ -70,7 +69,9 @@ def test_track_license_changes(mock_track_event, _):
 
 
 @mark.django_db
-@mock.patch('license_manager.apps.subscriptions.event_utils.get_license_tracking_properties', return_value={'counter': 1})
+@mock.patch(
+    'license_manager.apps.subscriptions.event_utils.get_license_tracking_properties', return_value={'counter': 1},
+)
 @mock.patch('license_manager.apps.subscriptions.event_utils.track_event')
 def test_track_license_changes_with_properties(mock_track_event, _):
     licenses = LicenseFactory.create_batch(5)

license_manager/apps/subscriptions/tests/test_factories.py

@@ -1,5 +1,3 @@
-import pytest
-from django.db import IntegrityError
 from django.test import TestCase
 
 from license_manager.apps.subscriptions.constants import UNASSIGNED
@@ -20,17 +18,17 @@ def test_license_factory(self):
         """
         Verify an unassigned license is created and associated with a subscription.
         """
-        license = LicenseFactory()
-        self.assertEqual(license.status, UNASSIGNED)
-        subscription_licenses = [license.uuid for license in license.subscription_plan.licenses.all()]
-        self.assertIn(license.uuid, subscription_licenses)
+        _license = LicenseFactory()
+        self.assertEqual(_license.status, UNASSIGNED)
+        subscription_licenses = [_license.uuid for _license in _license.subscription_plan.licenses.all()]
+        self.assertIn(_license.uuid, subscription_licenses)
 
     def test_subscription_factory(self):
         """
         Verify an unexpired subscription plan is created by default.
         """
         subscription = SubscriptionPlanFactory()
-        self.assertTrue(subscription.start_date < subscription.expiration_date)
+        self.assertTrue(subscription.start_date < subscription.expiration_date)  # pylint: disable=wrong-assert-type
 
     def test_subscription_factory_licenses(self):
         """
@@ -40,8 +38,8 @@ def test_subscription_factory_licenses(self):
         licenses = LicenseFactory.create_batch(5)
         subscription.licenses.set(licenses)
         # Verify the subscription plan uuid is correctly set on the licenses
-        license = subscription.licenses.first()
-        self.assertEqual(subscription.uuid, license.subscription_plan.uuid)
+        _license = subscription.licenses.first()
+        self.assertEqual(subscription.uuid, _license.subscription_plan.uuid)
 
     def test_customer_agreement_factory(self):
         """
@@ -60,4 +58,5 @@ def test_subscription_plan_renewal_factory(self):
         Verify an unexpired subscription plan renewal is created by default.
         """
         subscription_renewal = SubscriptionPlanRenewalFactory()
+        # pylint: disable=wrong-assert-type
         self.assertTrue(subscription_renewal.effective_date < subscription_renewal.renewed_expiration_date)

license_manager/apps/subscriptions/tests/test_forms.py

@@ -97,9 +97,8 @@ def test_validate_catalog_uuid(
             error.response.status_code = http_status_code
             mock_catalog_api_client().get_enterprise_catalog.side_effect = error
         else:
-            mock_catalog_api_client().get_enterprise_catalog.return_value = {
-                'enterprise_customer': str(enterprise_customer_uuid) if catalog_enterprise_customer_matches else str(uuid4())
-            }
+            value = str(enterprise_customer_uuid) if catalog_enterprise_customer_matches else str(uuid4())
+            mock_catalog_api_client().get_enterprise_catalog.return_value = {'enterprise_customer': value}
 
         form = make_bound_subscription_form(
             enterprise_customer_uuid=enterprise_customer_uuid,
@@ -147,7 +146,9 @@ def test_salesforce_opportunity_line_item(self, salesforce_id, expected_value):
         """
         Verify subscription plan form is invalid if salesforce_opportunity_id is None and the product requires it.
         """
-        invalid_form = make_bound_subscription_form(is_sf_id_required=True, salesforce_opportunity_line_item=salesforce_id)
+        invalid_form = make_bound_subscription_form(
+            is_sf_id_required=True, salesforce_opportunity_line_item=salesforce_id,
+        )
         assert invalid_form.is_valid() is expected_value
 
 
@@ -262,8 +263,14 @@ def test_populate_subscription_for_auto_applied_licenses_choices_initial_choice(
         choices = field.choices
         self.assertEqual(len(choices), 2)
         self.assertEqual(choices[0], ('', '------'))
-        self.assertEqual(choices[1], (current_sub_for_auto_applied_licenses.uuid, current_sub_for_auto_applied_licenses.title))
-        self.assertEqual(field.initial, (current_sub_for_auto_applied_licenses.uuid, current_sub_for_auto_applied_licenses.title))
+        self.assertEqual(
+            choices[1],
+            (current_sub_for_auto_applied_licenses.uuid, current_sub_for_auto_applied_licenses.title),
+        )
+        self.assertEqual(
+            field.initial,
+            (current_sub_for_auto_applied_licenses.uuid, current_sub_for_auto_applied_licenses.title),
+        )
 
     def test_populate_subscription_for_auto_applied_licenses_plans_outside_agreement_not_included(self):
         customer_agreement_1 = CustomerAgreementFactory()
@@ -369,9 +376,8 @@ def test_validate_catalog_uuid(
             error.response.status_code = http_status_code
             mock_catalog_api_client().get_enterprise_catalog.side_effect = error
         else:
-            mock_catalog_api_client().get_enterprise_catalog.return_value = {
-                'enterprise_customer': str(enterprise_customer_uuid) if catalog_enterprise_customer_matches else str(uuid4())
-            }
+            value = str(enterprise_customer_uuid) if catalog_enterprise_customer_matches else str(uuid4())
+            mock_catalog_api_client().get_enterprise_catalog.return_value = {'enterprise_customer': value}
 
         form = make_bound_customer_agreement_form(
             customer_agreement=CustomerAgreementFactory(enterprise_customer_uuid=enterprise_customer_uuid),

license_manager/apps/subscriptions/tests/test_models.py

@@ -8,7 +8,6 @@
 from django.core.cache import cache
 from django.forms import ValidationError
 from django.test import TestCase
-from requests.exceptions import HTTPError
 
 from license_manager.apps.subscriptions.constants import (
     ACTIVATED,
@@ -17,7 +16,6 @@
     UNASSIGNED,
     SegmentEvents,
 )
-from license_manager.apps.subscriptions.exceptions import CustomerAgreementError
 from license_manager.apps.subscriptions.models import (
     License,
     LicenseTransferJob,
@@ -92,7 +90,9 @@ def test_is_locked_for_renewal_processing(self, is_locked_for_renewal_processing
             if is_locked_for_renewal_processing:
                 renewal_kwargs.update({'effective_date': renewed_subscription_plan.expiration_date})
             SubscriptionPlanRenewalFactory.create(**renewal_kwargs)
-            self.assertEqual(renewed_subscription_plan.is_locked_for_renewal_processing, is_locked_for_renewal_processing)
+            self.assertEqual(
+                renewed_subscription_plan.is_locked_for_renewal_processing, is_locked_for_renewal_processing,
+            )
 
     def test_auto_apply_licenses_turned_on_at(self):
         """
@@ -116,13 +116,19 @@ def test_auto_applied_licenses_count_since(self):
         """
         subscription_plan = SubscriptionPlanFactory.create(should_auto_apply_licenses=True)
         timestamp_1 = localized_utcnow()
-        LicenseFactory.create_batch(1, subscription_plan=subscription_plan, auto_applied=True, activation_date=timestamp_1)
-        LicenseFactory.create_batch(3, subscription_plan=subscription_plan, auto_applied=False, activation_date=timestamp_1)
+        LicenseFactory.create_batch(
+            1, subscription_plan=subscription_plan, auto_applied=True, activation_date=timestamp_1,
+        )
+        LicenseFactory.create_batch(
+            3, subscription_plan=subscription_plan, auto_applied=False, activation_date=timestamp_1,
+        )
 
         self.assertEqual(subscription_plan.auto_applied_licenses_count_since(), 1)
         timestamp_2 = timestamp_1 + timedelta(seconds=1)
         self.assertEqual(subscription_plan.auto_applied_licenses_count_since(timestamp_2), 0)
-        LicenseFactory.create_batch(5, subscription_plan=subscription_plan, auto_applied=True, activation_date=timestamp_2)
+        LicenseFactory.create_batch(
+            5, subscription_plan=subscription_plan, auto_applied=True, activation_date=timestamp_2,
+        )
         self.assertEqual(subscription_plan.auto_applied_licenses_count_since(timestamp_2), 5)