Merge pull request #34885 from dimagi/em/form-submitting-trigger-invoice

Add form-submitting mobile worker option to Trigger Invoice testing tools

corehq/apps/accounting/forms.py

@@ -13,8 +13,8 @@
 from django.template.loader import render_to_string
 from django.urls import reverse
 from django.utils.dates import MONTHS
-from django.utils.safestring import mark_safe
 from django.utils.html import format_html, format_html_join
+from django.utils.safestring import mark_safe
 from django.utils.translation import gettext as _
 from django.utils.translation import gettext_lazy, gettext_noop
 
@@ -50,10 +50,12 @@
     CustomerBillingRecord,
     CustomerInvoice,
     DefaultProductPlan,
+    DomainUserHistory,
     EntryPoint,
     Feature,
     FeatureRate,
     FeatureType,
+    FormSubmittingMobileWorkerHistory,
     FundingSource,
     Invoice,
     InvoicingPlan,
@@ -68,7 +70,6 @@
     Subscription,
     SubscriptionType,
     WireBillingRecord,
-    DomainUserHistory,
 )
 from corehq.apps.accounting.tasks import send_subscription_reminder_emails
 from corehq.apps.accounting.utils import (
@@ -593,9 +594,12 @@ def __init__(self, subscription, account_id, web_user, *args, **kwargs):
         if is_existing:
             # circular import
             from corehq.apps.accounting.views import (
-                SoftwarePlanVersionView, ManageBillingAccountView
+                ManageBillingAccountView,
+                SoftwarePlanVersionView,
+            )
+            from corehq.apps.domain.views.settings import (
+                DefaultProjectSettingsView,
             )
-            from corehq.apps.domain.views.settings import DefaultProjectSettingsView
             self.fields['account'].initial = subscription.account.id
             account_field = hqcrispy.B3TextField(
                 'account',
@@ -846,7 +850,9 @@ def clean(self):
                     or not account.billingcontactinfo.email_list
                 )
             ):
-                from corehq.apps.accounting.views import ManageBillingAccountView
+                from corehq.apps.accounting.views import (
+                    ManageBillingAccountView,
+                )
                 raise forms.ValidationError(format_html(_(
                     "Please update 'Client Contact Emails' "
                     '<strong><a href={link} target="_blank">here</a></strong> '
@@ -2005,7 +2011,10 @@ def __init__(self, domain, web_user, on_annual_plan, data=None, *args, **kwargs)
         self.domain = domain
         self.web_user = web_user
         super(AnnualPlanContactForm, self).__init__(data, *args, **kwargs)
-        from corehq.apps.domain.views.accounting import SelectPlanView, DomainSubscriptionView
+        from corehq.apps.domain.views.accounting import (
+            DomainSubscriptionView,
+            SelectPlanView,
+        )
         self.helper = FormHelper()
         self.helper.label_class = 'col-sm-3 col-md-2'
         self.helper.field_class = 'col-sm-9 col-md-8 col-lg-6'
@@ -2061,12 +2070,17 @@ class TriggerInvoiceForm(forms.Form):
     month = forms.ChoiceField(label="Statement Period Month")
     year = forms.ChoiceField(label="Statement Period Year")
     domain = forms.CharField(label="Project Space", widget=forms.Select(choices=[]))
-    num_users = forms.IntegerField(
-        label="Number of Users",
+    num_mobile_workers = forms.IntegerField(
+        label="Number of Mobile Workers",
+        required=False,
+        help_text="This will overwrite the DomainUserHistory recorded for this month. "
+                  "Please leave this blank to use what is already in the system."
+    )
+    num_form_submitting_workers = forms.IntegerField(
+        label="Number of Form-Submitting Mobile Workers",
         required=False,
-        help_text="This is part of accounting tests and overwrites the "
-                  "DomainUserHistory recorded for this month. Please leave "
-                  "this blank to use what is already in the system."
+        help_text="This will overwrite the FormSubmittingMobileWorkerHistory recorded for this month. "
+                  "Please leave this blank to use what is already in the system."
     )
 
     def __init__(self, *args, **kwargs):
@@ -2097,13 +2111,24 @@ def __init__(self, *args, **kwargs):
                 placeholder="Search for Project"
             )
         ]
+        self.helper.layout = crispy.Layout(
+            crispy.Fieldset(*details)
+        )
+
         if self.show_testing_options:
-            details.append(crispy.Field('num_users', css_class='input_large'))
+            testing_options = [
+                'Testing Options',
+                crispy.Field('num_mobile_workers', css_class='input_large'),
+                crispy.Field('num_form_submitting_workers', css_class='input_large'),
+            ]
+            self.helper.layout.append(
+                crispy.Fieldset(*testing_options)
+            )
         else:
-            del self.fields['num_users']
+            del self.fields['num_mobile_workers']
+            del self.fields['num_form_submitting_workers']
 
-        self.helper.layout = crispy.Layout(
-            crispy.Fieldset(*details),
+        self.helper.layout.append(
             hqcrispy.FormActions(
                 StrictButton(
                     "Trigger Invoice",
@@ -2122,20 +2147,18 @@ def trigger_invoice(self):
 
         self.clean_previous_invoices(invoice_start, invoice_end, domain_obj.name)
 
-        if self.show_testing_options and self.cleaned_data['num_users']:
-            num_users = int(self.cleaned_data['num_users'])
-            existing_histories = DomainUserHistory.objects.filter(
-                domain=domain_obj.name,
-                record_date__gte=invoice_start,
-                record_date__lte=invoice_end,
-            )
-            if existing_histories.exists():
-                existing_histories.all().delete()
-            DomainUserHistory.objects.create(
-                domain=domain_obj.name,
-                record_date=invoice_end,
-                num_users=num_users
-            )
+        if self.show_testing_options:
+            if self.cleaned_data['num_mobile_workers']:
+                num_users = int(self.cleaned_data['num_mobile_workers'])
+                self._overwrite_user_history(
+                    DomainUserHistory, domain_obj.name, num_users, invoice_start, invoice_end
+                )
+            if self.cleaned_data['num_form_submitting_workers']:
+                num_users = int(self.cleaned_data['num_form_submitting_workers'])
+                self._overwrite_user_history(
+                    FormSubmittingMobileWorkerHistory,
+                    domain_obj.name, num_users, invoice_start, invoice_end
+                )
 
         invoice_factory = DomainInvoiceFactory(
             invoice_start, invoice_end, domain_obj, recipients=[settings.ACCOUNTS_EMAIL]
@@ -2163,6 +2186,21 @@ def clean_previous_invoices(invoice_start, invoice_end, domain_name):
                 )
             )
 
+    @staticmethod
+    def _overwrite_user_history(history_cls, domain, num_users, invoice_start, invoice_end):
+        existing_histories = history_cls.objects.filter(
+            domain=domain,
+            record_date__gte=invoice_start,
+            record_date__lte=invoice_end,
+        )
+        if existing_histories.exists():
+            existing_histories.all().delete()
+        history_cls.objects.create(
+            domain=domain,
+            record_date=invoice_end,
+            num_users=num_users
+        )
+
     def clean(self):
         today = datetime.date.today()
         year = int(self.cleaned_data['year'])

corehq/apps/accounting/tests/test_forms.py

@@ -5,13 +5,20 @@
 
 from dateutil.relativedelta import relativedelta
 
-from corehq.apps.accounting.forms import AdjustBalanceForm, SubscriptionForm
+from corehq.apps.accounting.exceptions import InvoiceError
+from corehq.apps.accounting.forms import (
+    AdjustBalanceForm,
+    SubscriptionForm,
+    TriggerInvoiceForm,
+)
 from corehq.apps.accounting.models import (
     BillingAccount,
     CreditAdjustmentReason,
     CreditLine,
     CustomerInvoice,
     DefaultProductPlan,
+    DomainUserHistory,
+    FormSubmittingMobileWorkerHistory,
     Invoice,
     SoftwarePlanEdition,
     Subscription,
@@ -25,6 +32,7 @@
 from corehq.apps.accounting.tests.test_invoicing import BaseInvoiceTestCase
 from corehq.apps.domain.models import Domain
 from corehq.apps.users.models import WebUser
+from corehq.util.dates import get_first_last_days
 
 
 class TestAdjustBalanceForm(BaseInvoiceTestCase):
@@ -311,3 +319,88 @@ def test_customer_plan_not_added_to_regular_account(self):
         }
 
         self.assertRaises(ValidationError, lambda: subscription_form.clean_active_accounts())
+
+
+class TestTriggerInvoiceForm(BaseInvoiceTestCase):
+
+    def setUp(self):
+        super().setUp()
+        statement_period = self.subscription.date_start + relativedelta(months=1)
+        self.statement_start, self.statement_end = get_first_last_days(
+            statement_period.year, statement_period.month
+        )
+        calculate_users_in_all_domains(self.statement_end + datetime.timedelta(days=1))
+
+    def init_form(self, form_data, show_testing_options=False):
+        self.form = TriggerInvoiceForm(
+            data=form_data,
+            show_testing_options=show_testing_options
+        )
+
+    def form_data(self, **kwargs):
+        form_data = {
+            'month': str(self.statement_start.month),
+            'year': str(self.statement_start.year),
+            'domain': self.domain.name,
+        }
+        form_data.update({k: str(v) for k, v in kwargs.items()})
+        return form_data
+
+    def test_trigger_invoice(self):
+        self.init_form(self.form_data())
+        self.form.full_clean()
+        self.form.trigger_invoice()
+
+        invoice = self.subscription.invoice_set.latest('date_created')
+        self.assertEqual(invoice.date_start, self.statement_start)
+        self.assertEqual(invoice.date_end, self.statement_end)
+
+    def test_clean_previous_invoices(self):
+        prev_invoice = Invoice.objects.create(
+            date_start=self.statement_start,
+            date_end=self.statement_end,
+            subscription=self.subscription
+        )
+        self.init_form(self.form_data())
+        self.form.full_clean()
+
+        with self.assertRaises(InvoiceError) as e:
+            self.form.clean_previous_invoices(self.statement_start, self.statement_end, self.domain.name)
+        self.assertIn(prev_invoice.invoice_number, str(e.exception))
+
+    def test_show_testing_options(self):
+        self.init_form(self.form_data(), show_testing_options=False)
+        self.assertNotIn('num_mobile_workers', self.form.fields)
+        self.assertNotIn('num_form_submitting_workers', self.form.fields)
+
+        self.init_form(self.form_data(), show_testing_options=True)
+        self.assertIn('num_mobile_workers', self.form.fields)
+        self.assertIn('num_form_submitting_workers', self.form.fields)
+
+    def test_num_mobile_workers(self):
+        num_users = 10
+        self.init_form(
+            self.form_data(num_mobile_workers=num_users),
+            show_testing_options=True
+        )
+        self.form.full_clean()
+        self.form.trigger_invoice()
+
+        user_history = DomainUserHistory.objects.get(
+            domain=self.domain.name, record_date=self.statement_end
+        )
+        self.assertEqual(user_history.num_users, num_users)
+
+    def test_num_form_submitting_mobile_workers(self):
+        num_users = 5
+        self.init_form(
+            self.form_data(num_form_submitting_workers=num_users),
+            show_testing_options=True
+        )
+        self.form.full_clean()
+        self.form.trigger_invoice()
+
+        user_history = FormSubmittingMobileWorkerHistory.objects.get(
+            domain=self.domain.name, record_date=self.statement_end
+        )
+        self.assertEqual(user_history.num_users, num_users)