Location Type 'has users' util function check

corehq/apps/locations/tests/test_util.py

@@ -0,0 +1,43 @@
+from corehq.apps.domain.shortcuts import create_domain
+from corehq.apps.commtrack.tests.util import TEST_LOCATION_TYPE, make_loc
+from corehq.apps.users.models import CommCareUser, WebUser
+from corehq.apps.locations.util import does_location_type_have_users
+
+from django.test import TestCase
+from corehq.apps.users.dbaccessors import delete_all_users
+from corehq.apps.es.client import manager
+from corehq.apps.es.tests.utils import es_test
+from corehq.apps.es.users import user_adapter
+from corehq.util.es.testing import sync_users_to_es
+from corehq.apps.locations.models import LocationType
+
+
+@es_test(requires=[user_adapter])
+class UtilTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super(UtilTest, cls).setUpClass()
+        cls.domain = create_domain('locations-test')
+        cls.loc = make_loc('loc', type='outlet', domain=cls.domain.name)
+        cls.loc_type = LocationType.objects.get(name=TEST_LOCATION_TYPE)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.domain.delete()
+        super(UtilTest, cls).tearDownClass()
+
+    def tearDown(self):
+        delete_all_users()
+
+    @sync_users_to_es()
+    def test_loc_type_has_users(self):
+        delete_all_users()
+        user1 = WebUser.create(self.domain.name, '[email protected]', '123', None, None)
+        user2 = CommCareUser.create(self.domain.name, 'mobile-user', '123', None, None)
+        user1.set_location(self.domain.name, self.loc)
+        user2.set_location(self.loc)
+        manager.index_refresh(user_adapter.index_name)
+        self.assertTrue(does_location_type_have_users(self.loc_type))
+
+    def _loc_type_does_not_have_user(self):
+        self.assertFalse(does_location_type_have_users(self.loc_type))

corehq/apps/locations/util.py

@@ -20,6 +20,7 @@
     get_loaded_default_monthly_consumption,
 )
 from corehq.apps.domain.models import Domain
+from corehq.apps.es.users import UserES
 from corehq.apps.locations.const import (
     LOCATION_SHEET_HEADERS_BASE,
     LOCATION_SHEET_HEADERS_OPTIONAL,
@@ -429,3 +430,9 @@ def get_location_type(domain, location, parent, loc_type_string, exception, is_n
     return loc_type_obj
 
 # ---
+
+
+# Checks if a location type has any users assigned to it's related locations
+def does_location_type_have_users(loc_type):
+    location_ids = list(SQLLocation.objects.filter(location_type=loc_type).values_list('location_id', flat=True))
+    return bool(UserES().location(location_ids).count())

corehq/apps/locations/views.py

@@ -66,7 +66,7 @@
     user_can_edit_location_types,
 )
 from .tree_utils import assert_no_cycles
-from .util import load_locs_json, location_hierarchy_config
+from .util import does_location_type_have_users, load_locs_json, location_hierarchy_config
 from django.http import JsonResponse, HttpResponseBadRequest
 from corehq.apps.locations.dbaccessors import get_filtered_locations_count
 
@@ -389,6 +389,15 @@ def unique_name_and_code():
                         return False
             return True
 
+        def _verify_has_users_config(loc_type_payload, pk):
+            if not loc_type.get('has_users'):
+                if _is_fake_pk(pk):
+                    return
+                if does_location_type_have_users(LocationType.objects.get(pk=pk, domain=self.domain)):
+                    raise LocationConsistencyError(f"Locations of the organization level '{loc_type['name']}' "
+                                                "have users assigned to them. You can't uncheck the "
+                                                "'Has Users' setting for this level!")
+
         loc_types = payload['loc_types']
         pks = []
         payload_loc_type_name_by_pk = {}
@@ -405,6 +414,8 @@ def unique_name_and_code():
             payload_loc_type_name_by_pk[loc_type['pk']] = loc_type['name']
             if loc_type.get('code'):
                 payload_loc_type_code_by_pk[loc_type['pk']] = loc_type['code']
+            if toggles.LOCATION_HAS_USERS.enabled(self.domain):
+                _verify_has_users_config(loc_type, pk)
         names = list(payload_loc_type_name_by_pk.values())
         names_are_unique = len(names) == len(set(names))
         codes = list(payload_loc_type_code_by_pk.values())

corehq/toggles/__init__.py

@@ -2888,3 +2888,10 @@ def domain_has_privilege_from_toggle(privilege_slug, domain):
     tag=TAG_CUSTOM,
     namespaces=[NAMESPACE_DOMAIN],
 )
+
+LOCATION_HAS_USERS = StaticToggle(
+    slug='location_has_users',
+    label='USH Dev: Allows marking whether a location should have users assigned or not.',
+    tag=TAG_PRODUCT,
+    namespaces=[NAMESPACE_DOMAIN],
+)