Merge pull request #34912 from dimagi/ad/location-has-users-select-wi…

…dget-filtering

Location select widget - filter by whether the location "has users"

corehq/apps/locations/forms.py

@@ -13,6 +13,7 @@
 
 from dimagi.utils.couch.database import iter_docs
 
+from corehq import toggles
 from corehq.apps.custom_data_fields.edit_entity import (
     CUSTOM_DATA_FIELD_PREFIX,
     CustomDataEditor,
@@ -43,13 +44,17 @@
 
 
 class LocationSelectWidget(forms.Widget):
-    def __init__(self, domain, attrs=None, id='supply-point', multiselect=False, placeholder=None):
+    def __init__(self, domain, attrs=None, id='supply-point', multiselect=False, placeholder=None,
+                 include_locations_with_no_users_allowed=True):
         super(LocationSelectWidget, self).__init__(attrs)
         self.domain = domain
         self.id = id
         self.multiselect = multiselect
         self.placeholder = placeholder
-        self.query_url = reverse('location_search', args=[self.domain])
+        url_name = 'location_search'
+        if not include_locations_with_no_users_allowed and toggles.LOCATION_HAS_USERS.enabled(self.domain):
+            url_name = 'location_search_has_users_only'
+        self.query_url = reverse(url_name, args=[self.domain])
         self.template = 'locations/manage/partials/autocomplete_select_widget.html'
 
     def render(self, name, value, attrs=None, renderer=None):

corehq/apps/locations/tests/test_views.py

@@ -13,10 +13,13 @@
 from corehq.apps.es.users import user_adapter
 from corehq.apps.locations.exceptions import LocationConsistencyError
 from corehq.apps.locations.models import LocationType
+from corehq.apps.locations.tests.util import make_loc
 from corehq.apps.locations.views import LocationTypesView, LocationImportView
+from corehq.apps.users.dbaccessors import delete_all_users
 from corehq.apps.users.models import WebUser, HQApiKey
-from corehq.util.workbook_json.excel import WorkbookJSONError
 from corehq.util.test_utils import flag_enabled
+from corehq.util.workbook_json.excel import WorkbookJSONError
+
 
 OTHER_DETAILS = {
     'expand_from': None,
@@ -157,6 +160,70 @@ def test_invalid_remove_has_users(self, _):
             self.send_request(data)
 
 
+class LocationsSearchViewTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super(LocationsSearchViewTest, cls).setUpClass()
+        cls.domain = "test-domain"
+        cls.project = create_domain(cls.domain)
+        cls.username = "request-er"
+        cls.password = "foobar"
+        cls.web_user = WebUser.create(cls.domain, cls.username, cls.password, None, None)
+        cls.web_user.add_domain_membership(cls.domain, is_admin=True)
+        cls.web_user.set_role(cls.domain, "admin")
+        cls.web_user.save()
+        cls.loc_type1 = LocationType(domain=cls.domain, name='type1', code='code1')
+        cls.loc_type1.save()
+        cls.loc1 = make_loc(
+            'loc_1', type=cls.loc_type1, domain=cls.domain
+        )
+        cls.loc2 = make_loc(
+            'loc_2', type=cls.loc_type1, domain=cls.domain
+        )
+
+    def setUp(self):
+        self.client.login(username=self.username, password=self.password)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.project.delete()
+        delete_all_users()
+        return super().tearDownClass()
+
+    @mock.patch('django_prbac.decorators.has_privilege', return_value=True)
+    def send_request(self, url, data, _):
+        return self.client.get(url, {'json': json.dumps(data)})
+
+    def test_search_view_basic(self):
+        url = reverse('location_search', args=[self.domain])
+        data = {'q': 'loc'}
+        response = self.send_request(url, data)
+        self.assertEqual(response.status_code, 200)
+        results = json.loads(response.content)['results']
+        self.assertEqual(results[0]['id'], self.loc1.location_id)
+        self.assertEqual(results[1]['id'], self.loc2.location_id)
+
+    @flag_enabled('LOCATION_HAS_USERS')
+    def test_search_view_has_users_only(self):
+        loc_type2 = LocationType(domain=self.domain, name='type2', code='code2')
+        loc_type2.has_users = False
+        loc_type2.save()
+        self.loc3 = make_loc(
+            'loc_3', type=loc_type2, domain=self.domain
+        )
+        self.loc3 = make_loc(
+            'loc_4', type=loc_type2, domain=self.domain
+        )
+        url = reverse('location_search_has_users_only', args=[self.domain])
+        data = {'q': 'loc'}
+        response = self.send_request(url, data)
+        self.assertEqual(response.status_code, 200)
+        results = json.loads(response.content)['results']
+        self.assertEqual(len(results), 2)
+        self.assertEqual(results[0]['id'], self.loc1.location_id)
+        self.assertEqual(results[1]['id'], self.loc2.location_id)
+
+
 class BulkLocationUploadAPITest(TestCase):
     @classmethod
     def setUpClass(cls):

corehq/apps/locations/urls.py

@@ -32,6 +32,8 @@
     url(r'^$', default, name='default_locations_view'),
     url(r'^list/$', LocationsListView.as_view(), name=LocationsListView.urlname),
     url(r'^location_search/$', LocationsSearchView.as_view(), name='location_search'),
+    url(r'^location_search_has_users_only/$', LocationsSearchView.as_view(
+        include_locations_with_no_users_allowed=False), name='location_search_has_users_only'),
     url(r'^location_types/$', LocationTypesView.as_view(), name=LocationTypesView.urlname),
     url(r'^import/$', waf_allow('XSS_BODY')(LocationImportView.as_view()), name=LocationImportView.urlname),
     url(r'^import/bulk_location_upload_api/$', bulk_location_upload_api, name='bulk_location_upload_api'),

corehq/apps/locations/views.py

@@ -266,6 +266,10 @@ class LocationOptionsController(EmwfOptionsController):
     namespace_locations = False
     case_sharing_only = False
 
+    def __init__(self, *args, include_locations_with_no_users_allowed=True):
+        self.include_locations_with_no_users_allowed = include_locations_with_no_users_allowed
+        super().__init__(*args)
+
     @property
     def data_sources(self):
         return [
@@ -276,11 +280,13 @@ def data_sources(self):
 @method_decorator(locations_access_required, name='dispatch')
 @location_safe
 class LocationsSearchView(EmwfOptionsView):
+    include_locations_with_no_users_allowed = True
 
     @property
     @memoized
     def options_controller(self):
-        return LocationOptionsController(self.request, self.domain, self.search)
+        return LocationOptionsController(self.request, self.domain, self.search,
+            include_locations_with_no_users_allowed=self.include_locations_with_no_users_allowed)
 
 
 @method_decorator(use_bootstrap5, name='dispatch')

corehq/apps/registration/forms.py

@@ -22,7 +22,7 @@
 from corehq.apps.domain.models import Domain
 from corehq.apps.hqwebapp import crispy as hqcrispy
 from corehq.apps.programs.models import Program
-from corehq.apps.users.forms import BaseLocationForm, BaseTableauUserForm
+from corehq.apps.users.forms import SelectUserLocationForm, BaseTableauUserForm
 from corehq.apps.users.models import CouchUser
 
 
@@ -483,7 +483,7 @@ def clean_email(self):
         return ""
 
 
-class AdminInvitesUserForm(BaseLocationForm):
+class AdminInvitesUserForm(SelectUserLocationForm):
     email = forms.EmailField(label="Email Address",
                              max_length=User._meta.get_field('email').max_length)
     role = forms.ChoiceField(choices=(), label="Project Role")

corehq/apps/reports/filters/controllers.py

@@ -1,5 +1,6 @@
 from memoized import memoized
 
+from corehq import toggles
 from corehq.apps.enterprise.models import EnterprisePermissions
 from corehq.apps.es import GroupES, UserES, groups
 from corehq.apps.locations.models import SQLLocation
@@ -96,6 +97,9 @@ def get_locations_query(self, query):
 
         if self.case_sharing_only:
             locations = locations.filter(location_type__shares_cases=True)
+        if (toggles.LOCATION_HAS_USERS.enabled(self.domain)
+                and not self.include_locations_with_no_users_allowed):
+            locations = locations.filter(location_type__has_users=True)
         return locations.accessible_to_user(self.domain, self.request.couch_user)
 
     def get_locations_size(self, query):

corehq/apps/users/forms.py

@@ -58,6 +58,7 @@
 from corehq.const import LOADTEST_HARD_LIMIT, USER_CHANGE_VIA_WEB
 from corehq.pillows.utils import MOBILE_USER_TYPE, WEB_USER_TYPE
 from corehq.toggles import (
+    LOCATION_HAS_USERS,
     TWO_STAGE_USER_PROVISIONING,
     TWO_STAGE_USER_PROVISIONING_BY_SMS,
 )
@@ -1141,7 +1142,7 @@ def render(self, name, value, attrs=None, renderer=None):
         })
 
 
-class BaseLocationForm(forms.Form):
+class SelectUserLocationForm(forms.Form):
     assigned_locations = forms.CharField(
         label=gettext_noop("Locations"),
         required=False,
@@ -1156,10 +1157,11 @@ class BaseLocationForm(forms.Form):
     def __init__(self, domain: str, *args, **kwargs):
         from corehq.apps.locations.forms import LocationSelectWidget
         self.request = kwargs.pop('request')
-        super(BaseLocationForm, self).__init__(*args, **kwargs)
+        super(SelectUserLocationForm, self).__init__(*args, **kwargs)
         self.domain = domain
         self.fields['assigned_locations'].widget = LocationSelectWidget(
-            self.domain, multiselect=True, id='id_assigned_locations'
+            self.domain, multiselect=True, id='id_assigned_locations',
+            include_locations_with_no_users_allowed=False
         )
         self.fields['assigned_locations'].help_text = ExpandedMobileWorkerFilter.location_search_help
         self.fields['primary_location'].widget = PrimaryLocationWidget(
@@ -1176,7 +1178,9 @@ def clean_assigned_locations(self):
             locations = get_locations_from_ids(location_ids, self.domain)
         except SQLLocation.DoesNotExist:
             raise forms.ValidationError(_('One or more of the locations was not found.'))
-
+        if LOCATION_HAS_USERS.enabled(self.domain) and locations.filter(location_type__has_users=False).exists():
+            raise forms.ValidationError(
+                _('One or more of the locations you specified cannot have users assigned.'))
         return [location.location_id for location in locations]
 
     def _user_has_permission_to_access_locations(self, new_location_ids):
@@ -1185,7 +1189,7 @@ def _user_has_permission_to_access_locations(self, new_location_ids):
             self.domain, self.request.couch_user))
 
     def clean(self):
-        self.cleaned_data = super(BaseLocationForm, self).clean()
+        self.cleaned_data = super(SelectUserLocationForm, self).clean()
 
         primary_location_id = self.cleaned_data['primary_location']
         assigned_location_ids = self.cleaned_data.get('assigned_locations', [])
@@ -1209,7 +1213,7 @@ def clean(self):
         return self.cleaned_data
 
 
-class CommtrackUserForm(BaseLocationForm):
+class CommtrackUserForm(SelectUserLocationForm):
     program_id = forms.ChoiceField(
         label=gettext_noop("Program"),
         choices=(),
@@ -1631,6 +1635,7 @@ def __init__(self, *args, **kwargs):
             id='id_location_id',
             placeholder=_("All Locations"),
             attrs={'data-bind': 'value: location_id'},
+            include_locations_with_no_users_allowed=False
         )
         self.fields['location_id'].widget.query_url = "{url}?show_all=true".format(
             url=self.fields['location_id'].widget.query_url

corehq/messaging/scheduling/async_handlers.py

@@ -159,18 +159,18 @@ def sms_case_registration_owner_id_response(self):
                     'id': u['id'],
                     'text': _("User: {}").format(u['text']),
                 } for u in users
-            ] +
-            [
+            ]
+            + [
                 {
                     'id': g['id'],
                     'text': _("User Group: {}").format(g['text']),
                 } for g in groups
-            ] +
-            [
+            ]
+            + [
                 {
-                    'id': l['id'],
-                    'text': _("Organization: {}").format(l['text']),
-                } for l in locations
+                    'id': loc['id'],
+                    'text': _("Organization: {}").format(loc['text']),
+                } for loc in locations
             ]
         )