Merge pull request #6083 from cfpb/prepaid_form

Use Django forms to sanitize and validate prepaid search parameters

cfgov/cfgov/settings/base.py

@@ -735,8 +735,6 @@
     "BETA_EXTERNAL_TESTING": [],
     # Used to hide new youth employment success pages prior to public launch
     "YOUTH_EMPLOYMENT_SUCCESS": [],
-    # Release of prepaid agreements database search
-    "PREPAID_AGREEMENTS_SEARCH": [],
     # Used to hide CCDB landing page updates prior to public launch
     "CCDB_CONTENT_UPDATES": [],
     # During a Salesforce system outage, the following flag should be enabled

cfgov/cfgov/urls.py

@@ -302,8 +302,7 @@ def empty_200_response(request, *args, **kwargs):
 
     re_path(r'^credit-cards/agreements/', include('agreements.urls')),
 
-    flagged_re_path(
-        'PREPAID_AGREEMENTS_SEARCH',
+    re_path(
         r'^data-research/prepaid-accounts/search-agreements/',
         include((
             'prepaid_agreements.urls',

cfgov/prepaid_agreements/forms.py

@@ -0,0 +1,41 @@
+from django import forms
+
+
+class SearchForm(forms.Form):
+    """Django form to validate the basic prepaid search fields"""
+    page = forms.IntegerField(required=False)
+    q = forms.CharField(required=False)
+    search_field = forms.ChoiceField(
+        choices=[
+            ('all', 'All fields'),
+            ('name', 'Product name'),
+            ('program_manager', 'Program manager'),
+            ('other_relevant_parties', 'Other relevant party')
+        ],
+        required=False,
+    )
+
+
+class FilterForm(forms.Form):
+    """Django form to validate the prepaid search filter fields"""
+    issuer_name = forms.MultipleChoiceField(
+        choices=[],
+        required=False
+    )
+    prepaid_type = forms.MultipleChoiceField(
+        choices=[],
+        required=False
+    )
+    status = forms.MultipleChoiceField(
+        choices=[],
+        required=False
+    )
+
+    def set_issuer_name_choices(self, choices):
+        self.fields['issuer_name'].choices = [(c, c) for c in choices]
+
+    def set_prepaid_type_choices(self, choices):
+        self.fields['prepaid_type'].choices = [(c, c) for c in choices]
+
+    def set_status_choices(self, choices):
+        self.fields['status'].choices = [(c, c) for c in choices]

cfgov/prepaid_agreements/tests/test_views.py

@@ -14,72 +14,66 @@
 
 class TestViews(TestCase):
 
+    def setUp(self):
+        self.product1 = PrepaidProduct(
+            issuer_name='Bank of CFPB',
+            prepaid_type='Tax',
+            status='Active',
+        )
+        self.product1.save()
+        self.product2 = PrepaidProduct(
+            program_manager='CFPB manager',
+            prepaid_type='Travel',
+            status='Withdrawn'
+        )
+        self.product2.save()
+        self.product3 = PrepaidProduct(
+            name='ABC Product',
+            issuer_name='ABC Bank',
+            prepaid_type='Payroll',
+            status='Active',
+        )
+        self.product3.save()
+
     def test_get_available_filters(self):
         products = PrepaidProduct.objects
-
         self.assertEqual(
-            get_available_filters(products),
+            get_available_filters(products.none()),
             {'prepaid_type': [], 'status': [], 'issuer_name': []}
         )
-
-        product1 = PrepaidProduct(
-            issuer_name='Bank of CFPB',
-            prepaid_type='Tax'
-        )
-        product1.save()
-        product2 = PrepaidProduct(prepaid_type='Travel')
-        product2.save()
         self.assertEqual(
             get_available_filters(products),
             {
-                'prepaid_type': ['Tax', 'Travel'],
-                'status': [],
-                'issuer_name': ['Bank of CFPB']
+                'prepaid_type': ['Tax', 'Travel', 'Payroll'],
+                'status': ['Active', 'Withdrawn'],
+                'issuer_name': ['Bank of CFPB', 'ABC Bank']
             }
         )
 
     @unittest.skipUnless(
         connection.vendor == 'postgresql', 'PostgreSQL-dependent')
     def test_search_products_issuer_name(self):
-        product1 = PrepaidProduct(issuer_name='Bank of CFPB')
-        product1.save()
-        product2 = PrepaidProduct(issuer_name='Bank of Foo Bar')
-        product2.save()
         results = search_products(
             search_term='cfpb',
             search_field=['issuer_name'],
             products=PrepaidProduct.objects.all()
         )
-        self.assertIn(product1, results)
-        self.assertEqual(results.count(), 1)
+        self.assertIn(self.product1, results)
+        self.assertEqual(results.count(), 2)
 
     @unittest.skipUnless(
         connection.vendor == 'postgresql', 'PostgreSQL-dependent')
     def test_search_products_all_fields(self):
-        product1 = PrepaidProduct(issuer_name='XYZ Bank')
-        product1.save()
-        product2 = PrepaidProduct(program_manager='xyz manager')
-        product2.save()
-        product3 = PrepaidProduct(name='Foo Bar Product')
-        product3.save()
         results = search_products(
-            search_term='Xyz',
+            search_term='cfpb',
             search_field=[],
             products=PrepaidProduct.objects.all()
         )
-        self.assertIn(product1, results)
-        self.assertIn(product2, results)
+        self.assertIn(self.product1, results)
+        self.assertIn(self.product2, results)
         self.assertEqual(results.count(), 2)
 
     def test_filter_products(self):
-        product1 = PrepaidProduct(status='Active', prepaid_type='Student')
-        product1.save()
-        product2 = PrepaidProduct(
-            status='Withdrawn', prepaid_type='Travel', issuer_name='XYZ Bank')
-        product2.save()
-        product3 = PrepaidProduct(
-            status='Active', prepaid_type='Payroll', issuer_name='ABC Bank')
-        product3.save()
         results = filter_products(
             filters={
                 'status': ['Active'],
@@ -88,7 +82,7 @@ def test_filter_products(self):
             },
             products=PrepaidProduct.objects.all()
         )
-        self.assertIn(product3, results)
+        self.assertIn(self.product3, results)
         self.assertEqual(results.count(), 1)
 
     def test_get_breadcrumb_if_referrer_is_search_page_with_query(self):
@@ -122,3 +116,19 @@ def test_get_breadcrumb_if_no_referrer(self):
         request = HttpRequest()
         search_path = reverse("prepaid_agreements:index")
         self.assertEqual(get_detail_page_breadcrumb(request), search_path)
+
+    def test_index_search(self):
+        response = self.client.get(
+            '/data-research/prepaid-accounts/search-agreements/',
+            {'q': 'cfpb', 'search_field': 'all', 'page': 1}
+        )
+        self.assertEqual(3, response.context_data['total_count'])
+        self.assertEqual(2, response.context_data['current_count'])
+
+    def test_index_filter(self):
+        response = self.client.get(
+            '/data-research/prepaid-accounts/search-agreements/',
+            {'status': 'Active'}
+        )
+        self.assertEqual(3, response.context_data['total_count'])
+        self.assertEqual(2, response.context_data['current_count'])

cfgov/prepaid_agreements/views.py

@@ -1,48 +1,34 @@
 from urllib.parse import urlparse
 
 from django.contrib.postgres.search import SearchVector
-from django.core.paginator import InvalidPage, Paginator
+from django.core.paginator import Paginator
 from django.db.models import Q
 from django.shortcuts import get_object_or_404, render
+from django.template.response import TemplateResponse
 from django.urls import reverse
 
+from prepaid_agreements.forms import FilterForm, SearchForm
 from prepaid_agreements.models import PrepaidProduct
 from v1.models.snippets import ReusableText
 
 
-def validate_page_number(request, paginator):
-    """
-    A utility for parsing a pagination request,
-    catching invalid page numbers and always returning
-    a valid page number, defaulting to 1.
-    TODO: can be replaced by Paginator.get_page
-    when/if upgraded to Django 2.0+
-    """
-    raw_page = request.GET.get('page', 1)
-    try:
-        page_number = int(raw_page)
-    except ValueError:
-        page_number = 1
-    try:
-        paginator.page(page_number)
-    except InvalidPage:
-        page_number = 1
-    return page_number
-
-
 def get_available_filters(products):
     available_filters = {'prepaid_type': [], 'status': [], 'issuer_name': []}
+
     for product in products.all():
         prepaid_type = product.prepaid_type
         if prepaid_type and prepaid_type != '':
             if prepaid_type not in available_filters['prepaid_type']:
                 available_filters['prepaid_type'].append(prepaid_type)
+
         status = product.status
         if status and status not in available_filters['status']:
             available_filters['status'].append(status)
+
         issuer_name = product.issuer_name
         if issuer_name and issuer_name not in available_filters['issuer_name']:
             available_filters['issuer_name'].append(issuer_name)
+
     return available_filters
 
 
@@ -95,47 +81,68 @@ def get_support_text():
 
 
 def index(request):
-    params = dict(request.GET.lists())
-    available_filters = {}
-    search_term = None
-    search_field = None
     products = PrepaidProduct.objects.valid()
     total_count = products.count()
     valid_filters = [
         'prepaid_type', 'status', 'issuer_name'
     ]
-    if params:
-        params.pop('page', None)
-        search_term = params.pop('q', None)
-        search_field = params.pop('search_field', None)
-        if search_field:
-            search_field = search_field[0]
-        if search_term:
-            search_term = search_term[0].strip()
-            if search_term != '':
-                products = search_products(
-                    search_term, search_field, products
-                )
-            available_filters = get_available_filters(products)
-        products = filter_products(params, products)
-
-    if not available_filters:
-        for filter_name in valid_filters:
-            available_filters[filter_name] = PrepaidProduct.objects.order_by(
-                filter_name).values_list(filter_name, flat=True).distinct()
+    available_filters = {}
+
+    # Provide valid initial values for the search form so that it is always
+    # valid. A blank search will return all products.
+    search_term = ''
+    search_field = 'all'
+    search_form = SearchForm(
+        request.GET,
+        initial={
+            'q': search_term,
+            'search_field': search_field,
+            'page': 1,
+        }
+    )
+    # Get our search results. If the form is not valid, our default search_term
+    # and search_field will be used below.
+    if search_form.is_valid():
+        search_field = search_form.cleaned_data['search_field']
+        search_term = search_form.cleaned_data['q'].strip()
+        if search_term != '':
+            products = search_products(
+                search_term, search_field, products
+            )
+
+    # Get the available filters for products in the search results and then set
+    # those filter choices on the filter form
+    filters = {}
+    available_filters = get_available_filters(products)
+    filter_form = FilterForm(request.GET)
+    filter_form.set_issuer_name_choices(available_filters['issuer_name'])
+    filter_form.set_prepaid_type_choices(available_filters['prepaid_type'])
+    filter_form.set_status_choices(available_filters['status'])
+
+    # Filter our products based on the sanitized values from the filter form.
+    # If the form is not valid, the products already selected based on the
+    # search form will be used below.
+    if filter_form.is_valid():
+        filters = {
+            k: filter_form.cleaned_data[k]
+            for k in valid_filters
+            if filter_form.cleaned_data[k]
+        }
+        products = filter_products(filters, products)
 
     current_count = products.count()
+
+    # Handle pagination
     paginator = Paginator(products.all(), 20)
-    page_number = validate_page_number(request, paginator)
-    page = paginator.page(page_number)
+    page = paginator.get_page(search_form.cleaned_data['page'])
 
-    return render(request, 'prepaid_agreements/index.html', {
-        'current_page': page_number,
+    return TemplateResponse(request, 'prepaid_agreements/index.html', {
+        'current_page': page.number,
         'results': page,
         'total_count': total_count,
         'paginator': paginator,
         'current_count': current_count,
-        'filters': params,
+        'filters': filters,
         'query': search_term or '',
         'active_filters': available_filters,
         'valid_filters': valid_filters,