diff --git a/license_manager/apps/subscriptions/admin.py b/license_manager/apps/subscriptions/admin.py index 2e8e6be9..cb88294a 100644 --- a/license_manager/apps/subscriptions/admin.py +++ b/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' ) diff --git a/license_manager/apps/subscriptions/tests/__init__.py b/license_manager/apps/subscriptions/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/license_manager/apps/subscriptions/tests/factories.py b/license_manager/apps/subscriptions/tests/factories.py index 7517fbaa..96db577c 100644 --- a/license_manager/apps/subscriptions/tests/factories.py +++ b/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)) diff --git a/license_manager/apps/subscriptions/tests/test_admin.py b/license_manager/apps/subscriptions/tests/test_admin.py index 99e9392b..e2bab224 100644 --- a/license_manager/apps/subscriptions/tests/test_admin.py +++ b/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}'].", + ) diff --git a/license_manager/apps/subscriptions/tests/test_api.py b/license_manager/apps/subscriptions/tests/test_api.py index 883d1d8e..8451ce6d 100644 --- a/license_manager/apps/subscriptions/tests/test_api.py +++ b/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']) diff --git a/license_manager/apps/subscriptions/tests/test_event_utils.py b/license_manager/apps/subscriptions/tests/test_event_utils.py index f951ce58..431696de 100644 --- a/license_manager/apps/subscriptions/tests/test_event_utils.py +++ b/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) diff --git a/license_manager/apps/subscriptions/tests/test_factories.py b/license_manager/apps/subscriptions/tests/test_factories.py index e1bd53a7..6eef8501 100644 --- a/license_manager/apps/subscriptions/tests/test_factories.py +++ b/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) diff --git a/license_manager/apps/subscriptions/tests/test_forms.py b/license_manager/apps/subscriptions/tests/test_forms.py index 62352ced..c7154ba8 100644 --- a/license_manager/apps/subscriptions/tests/test_forms.py +++ b/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), diff --git a/license_manager/apps/subscriptions/tests/test_models.py b/license_manager/apps/subscriptions/tests/test_models.py index ec5bda50..cc45c927 100644 --- a/license_manager/apps/subscriptions/tests/test_models.py +++ b/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) diff --git a/license_manager/apps/subscriptions/tests/test_tasks.py b/license_manager/apps/subscriptions/tests/test_tasks.py index 61a99e74..6d3bcc47 100644 --- a/license_manager/apps/subscriptions/tests/test_tasks.py +++ b/license_manager/apps/subscriptions/tests/test_tasks.py @@ -1,28 +1,17 @@ """ Tests for subscriptions app celery tasks """ -from datetime import datetime, timedelta from unittest import mock -from uuid import uuid4 import ddt -import freezegun -import pytest -from braze.exceptions import BrazeClientError -from django.conf import settings from django.test import TestCase -from django.test.utils import override_settings -from freezegun import freeze_time -from requests import models from license_manager.apps.api.utils import ( acquire_subscription_plan_lock, release_subscription_plan_lock, ) from license_manager.apps.subscriptions import tasks -from license_manager.apps.subscriptions.models import License, SubscriptionPlan from license_manager.apps.subscriptions.tests.factories import ( - LicenseFactory, SubscriptionPlanFactory, ) @@ -102,6 +91,7 @@ def test_provision_licenses_task(self, num_initial_licenses, desired_num_license self.subscription_plan.save() self.subscription_plan.increase_num_licenses(num_initial_licenses) + # pylint: disable=no-value-for-parameter tasks.provision_licenses_task(subscription_plan_uuid=self.subscription_plan.uuid) assert self.subscription_plan.num_licenses == expected_num_licenses @@ -116,6 +106,7 @@ def test_provision_licenses_task_locked(self): acquire_subscription_plan_lock(self.subscription_plan) with self.assertRaises(tasks.RequiredTaskUnreadyError): + # pylint: disable=no-value-for-parameter tasks.provision_licenses_task(subscription_plan_uuid=self.subscription_plan.uuid) assert self.subscription_plan.num_licenses == 0