Merge pull request #35118 from dimagi/jls/clean-up-ent-perm-caching

Clean up caching for EnterprisePermissions

corehq/apps/case_importer/tracking/dbaccessors.py

@@ -29,9 +29,8 @@ def get_case_upload_record_count(domain, user):
 
 
 def get_case_ids_for_case_upload(case_upload):
-    enterprise_perms = EnterprisePermissions.get_by_domain(case_upload.domain)
     for form_record in case_upload.form_records.order_by('pk').all():
-        if enterprise_perms.is_enabled and enterprise_perms.source_domain == case_upload.domain:
+        if EnterprisePermissions.is_source_domain(case_upload.domain):
             form = XFormInstance.objects.get_form(form_record.form_id)
         else:
             form = XFormInstance.objects.get_form(form_record.form_id, case_upload.domain)

corehq/apps/enterprise/models.py

@@ -28,7 +28,6 @@ class EnterprisePermissions(models.Model):
     )
 
     @classmethod
-    @quickcache(['domain'], timeout=7 * 24 * 60 * 60)
     def get_by_domain(cls, domain):
         """
         Get or create the configuration associated with the given domain's account.
@@ -41,6 +40,18 @@ def get_by_domain(cls, domain):
         except cls.DoesNotExist:
             return cls(account=account)
 
+    @classmethod
+    @quickcache(['domain'], timeout=7 * 24 * 60 * 60)
+    def get_source_domain(cls, domain):
+        """
+        If the given domain is controlled by another domain via enterprise permissions,
+        returns that controlling domain. Otherwise, returns None.
+        """
+        config = EnterprisePermissions.get_by_domain(domain)
+        if config.is_enabled and domain in config.domains:
+            return config.source_domain
+        return None
+
     @classmethod
     @quickcache(['source_domain'], timeout=7 * 24 * 60 * 60)
     def get_domains(cls, source_domain):
@@ -53,11 +64,29 @@ def get_domains(cls, source_domain):
             return []
         return list(set(config.domains) - {config.source_domain})
 
+    @classmethod
+    @quickcache(['domain'], timeout=7 * 24 * 60 * 60)
+    def is_source_domain(cls, domain):
+        """
+        Returns true if given domain is the source domain for an enabled configuration.
+        """
+        try:
+            cls.objects.get(is_enabled=True, source_domain=domain)
+        except cls.DoesNotExist:
+            return False
+        return True
+
     def save(self, *args, **kwargs):
+        self.is_source_domain.clear(self.__class__, self.source_domain)
+        for domain in self.account.get_domains():
+            self.get_domains.clear(self.__class__, domain)
+            self.get_source_domain.clear(self.__class__, domain)
+
         super().save(*args, **kwargs)
+        self.is_source_domain.clear(self.__class__, self.source_domain)
         for domain in self.account.get_domains():
             self.get_domains.clear(self.__class__, domain)
-            self.get_by_domain.clear(self.__class__, domain)
+            self.get_source_domain.clear(self.__class__, domain)
 
 
 class EnterpriseMobileWorkerSettings(models.Model):

corehq/apps/enterprise/tests/test_permissions.py

@@ -7,6 +7,7 @@
 from .utils import create_enterprise_permissions
 from corehq.apps.domain.models import Domain
 from corehq.apps.domain.shortcuts import create_domain
+from corehq.apps.enterprise.models import EnterprisePermissions
 from corehq.apps.fixtures.resources.v0_1 import InternalFixtureResource
 from corehq.apps.users.models import (
     HQApiKey,
@@ -68,6 +69,36 @@ def tearDownClass(cls):
         Domain.get_by_name('staging').delete()
         super().tearDownClass()
 
+    def test_get_by_domain(self):
+        # Any domain in the account can be used to get the config
+        configs = [
+            EnterprisePermissions.get_by_domain('state'),
+            EnterprisePermissions.get_by_domain('county'),
+            EnterprisePermissions.get_by_domain('staging'),
+        ]
+        for config in configs:
+            self.assertTrue(config.is_enabled)
+            self.assertEqual(config.source_domain, 'state')
+            self.assertListEqual(config.domains, ['county'])
+
+        empty_config = EnterprisePermissions.get_by_domain('not-a-domain')
+        self.assertFalse(empty_config.is_enabled)
+
+    def test_get_source_domain(self):
+        self.assertEqual(EnterprisePermissions.get_source_domain('state'), None)
+        self.assertEqual(EnterprisePermissions.get_source_domain('county'), 'state')
+        self.assertEqual(EnterprisePermissions.get_source_domain('staging'), None)
+
+    def test_get_domains(self):
+        self.assertListEqual(EnterprisePermissions.get_domains('state'), ['county'])
+        self.assertListEqual(EnterprisePermissions.get_domains('county'), [])
+        self.assertListEqual(EnterprisePermissions.get_domains('staging'), [])
+
+    def test_is_source_domain(self):
+        self.assertTrue(EnterprisePermissions.is_source_domain('state'))
+        self.assertFalse(EnterprisePermissions.is_source_domain('county'))
+        self.assertFalse(EnterprisePermissions.is_source_domain('staging'))
+
     def test_permission_mirroring(self):
         for domain in ('state', 'county'):
             self.assertTrue(self.web_user_admin.is_domain_admin(domain))

corehq/apps/es/users.py

@@ -128,9 +128,9 @@ def domain(domain, allow_enterprise=False):
     domain_list = [domain]
     if allow_enterprise:
         from corehq.apps.enterprise.models import EnterprisePermissions
-        config = EnterprisePermissions.get_by_domain(domain)
-        if config.is_enabled and domain in config.domains:
-            domain_list.append(config.source_domain)
+        source_domain = EnterprisePermissions.get_source_domain(domain)
+        if source_domain:
+            domain_list.append(source_domain)
     return domains(domain_list)
 
 

corehq/apps/users/models.py

@@ -481,9 +481,9 @@ def _is_member_of(self, domain, allow_enterprise):
 
         if allow_enterprise:
             from corehq.apps.enterprise.models import EnterprisePermissions
-            config = EnterprisePermissions.get_by_domain(domain)
-            if config.is_enabled and domain in config.domains:
-                return self.is_member_of(config.source_domain, allow_enterprise=False)
+            source_domain = EnterprisePermissions.get_source_domain(domain)
+            if source_domain:
+                return self.is_member_of(source_domain, allow_enterprise=False)
 
         return False
 
@@ -520,9 +520,9 @@ def get_domain_membership(self, domain, allow_enterprise=True):
                 if domain in self.domains:
                     raise self.Inconsistent("Domain '%s' is in domain but not in domain_memberships" % domain)
                 from corehq.apps.enterprise.models import EnterprisePermissions
-                config = EnterprisePermissions.get_by_domain(domain)
-                if allow_enterprise and config.is_enabled and domain in config.domains:
-                    return self.get_domain_membership(config.source_domain, allow_enterprise=False)
+                source_domain = EnterprisePermissions.get_source_domain(domain)
+                if allow_enterprise and source_domain:
+                    return self.get_domain_membership(source_domain, allow_enterprise=False)
         except self.Inconsistent as e:
             logging.warning(e)
             self.domains = [d.domain for d in self.domain_memberships]

corehq/tabs/tabclasses.py

@@ -1838,7 +1838,7 @@ def sidebar_items(self):
         if self.couch_user.is_superuser:
             from corehq.apps.enterprise.models import EnterprisePermissions
             if toggles.DOMAIN_PERMISSIONS_MIRROR.enabled_for_request(self._request) \
-                    or EnterprisePermissions.get_by_domain(self.domain).is_enabled:
+                    or EnterprisePermissions.is_source_domain(self.domain):
                 enterprise_views.append({
                     'title': _("Enterprise Permissions"),
                     'url': reverse("enterprise_permissions", args=[self.domain]),