Merge pull request #79 from dimagi/due-updates

Post PR updates

hq_superset/const.py

@@ -27,12 +27,6 @@
     CAN_READ_PERMISSION,
 ]
 
-CAN_WRITE_PERMISSION = "can_write"
-CAN_EDIT_PERMISSION = "can_edit"
-CAN_ADD_PERMISSION = "can_add"
-CAN_DELETE_PERMISSIONS = "can_delete"
-
-
 READ_ONLY_MENU_PERMISSIONS = {
     "Chart": READ_ONLY_PERMISSIONS,
     "Dataset": READ_ONLY_PERMISSIONS,

hq_superset/hq_url.py

@@ -33,4 +33,4 @@ def datasource_unsubscribe(domain, datasource_id):
 
 
 def user_domain_roles(domain):
-    return f"a/{domain}/api/analytics-roles/v1"
+    return f"a/{domain}/api/analytics-roles/v1/"

hq_superset/tests/test_utils.py

@@ -7,6 +7,7 @@
 from hq_superset.tests.base_test import LoginUserTestMixin, SupersetTestCase
 from hq_superset.tests.const import TEST_DATASOURCE
 from hq_superset.utils import DomainSyncUtil, get_column_dtypes
+from hq_superset.hq_requests import HQRequest
 
 
 def test_get_column_dtypes():
@@ -30,9 +31,41 @@ def test_doctests():
     assert results.failed == 0
 
 
+class DomainAccessMockResponse:
+    status_code: int = 200
+
+    def __init__(self, status_code=None):
+        if status_code:
+            self.status_code = status_code
+
+    def json(self):
+        return {
+            'permissions': {'can_edit': True, 'can_view': True},
+            'roles': ['sql_lab']
+        }
+
+
 class TestDomainSyncUtil(LoginUserTestMixin, SupersetTestCase):
     PLATFORM_ROLE_NAMES = ["Gamma", "sql_lab", "dataset_editor"]
 
+    @patch.object(HQRequest, "get")
+    def test_domain_access_success_response(self, get_mock):
+        get_mock.return_value = DomainAccessMockResponse()
+
+        security_manager = self.app.appbuilder.sm
+        response = DomainSyncUtil(security_manager)._get_domain_access("test-domain")
+
+        assert response == (True, True, ["sql_lab"])
+
+    @patch.object(HQRequest, "get")
+    def test_domain_access_faulty_response(self, get_mock):
+        get_mock.return_value = DomainAccessMockResponse(status_code=400)
+
+        security_manager = self.app.appbuilder.sm
+        response = DomainSyncUtil(security_manager)._get_domain_access("test-domain")
+
+        assert response == (False, False, [])
+
     @patch.object(DomainSyncUtil, "_get_domain_access")
     def test_gamma_role_assigned_for_edit_permissions(self, get_domain_access_mock):
         security_manager = self.app.appbuilder.sm
@@ -122,7 +155,4 @@ def _ensure_platform_roles_exist(self, sm):
 
     @staticmethod
     def _to_permissions_response(can_write, can_read, roles):
-        return {
-            "can_write": can_write,
-            "can_read": can_read,
-        }, roles
+        return can_read, can_write, roles

hq_superset/tests/test_views.py

@@ -85,7 +85,7 @@ def test_redirects_to_domain_select_after_login(self):
             )
             self.logout(client)
 
-    @patch.object(DomainSyncUtil, "_get_domain_access", return_value=({"can_write": True, "can_read": True}, []))
+    @patch.object(DomainSyncUtil, "_get_domain_access", return_value=(True, True, []))
     def test_domain_select_works(self, *args):
         client = self.app.test_client()
         self.login(client)
@@ -135,7 +135,7 @@ def _do_assert(datasources):
         _do_assert(self.oauth_mock.test2_datasources)
         self.logout(client)
 
-    @patch.object(DomainSyncUtil, "_get_domain_access", return_value=({"can_write": True, "can_read": True}, []))
+    @patch.object(DomainSyncUtil, "_get_domain_access", return_value=(True, True, []))
     def test_datasource_upload(self, *args):
         client = self.app.test_client()
         self.login(client)

hq_superset/tests/utils.py

@@ -65,8 +65,8 @@ def get(self, url, token):
             'a/test1/configurable_reports/data_sources/export/test1_ucr1/?format=csv': MockResponse(
                 TEST_UCR_CSV_V1, 200
             ),
-            'a/test1/api/analytics-roles/v1': MockResponse(self.user_domain_roles, 200),
-            'a/test2/api/analytics-roles/v1': MockResponse(self.user_domain_roles, 200),
+            'a/test1/api/analytics-roles/v1/': MockResponse(self.user_domain_roles, 200),
+            'a/test2/api/analytics-roles/v1/': MockResponse(self.user_domain_roles, 200),
         }[url]
 
 

hq_superset/utils.py

@@ -18,8 +18,6 @@
 from superset.utils.database import get_or_create_db
 
 from hq_superset.const import (
-    CAN_READ_PERMISSION,
-    CAN_WRITE_PERMISSION,
     DOMAIN_PREFIX,
     GAMMA_ROLE_NAME,
     HQ_DATABASE_NAME,
@@ -135,10 +133,7 @@ def sync_domain_role(self, domain):
         The user gets assigned at least 3 roles in order to function on any domain:
         1. hq_user_role: gives access to superset platform
         2. domain_schema_role: restricts user access to specific domain schema
-        3. domain_user_role: restricts access for particular user on domain in accordance with how the permissions
-        are defined on CommCare HQ.
-
-        Any additional roles defined on CommCare HQ will also be assigned to the user.
+        3. Gamma role for users with write access or READ_ONLY_ROLE_NAME for users with read access only
         """
         hq_user_role = self._ensure_hq_user_role()
         domain_schema_role = self._create_domain_role(domain)
@@ -202,19 +197,19 @@ def _ensure_domain_role_created(self, domain):
         return self.sm.add_role(get_role_name_for_domain(domain))
 
     def _get_additional_user_roles(self, domain):
-        domain_permissions, roles_names = self._get_domain_access(domain)
-        if self._user_has_no_access(domain_permissions):
+        can_read, can_write, platform_roles_names = self._get_domain_access(domain)
+        if not (can_read or can_write):
             return []
 
-        if domain_permissions[CAN_WRITE_PERMISSION]:
-            user_role = GAMMA_ROLE_NAME
+        if can_write:
+            user_role_name = GAMMA_ROLE_NAME
         else:
             self._ensure_read_only_role_exists()
-            user_role = READ_ONLY_ROLE_NAME
-
-        roles_names.append(user_role)
+            user_role_name = READ_ONLY_ROLE_NAME
 
-        return self._get_platform_roles(roles_names)
+        return self._get_platform_roles(
+            platform_roles_names + [user_role_name]
+        )
 
     @staticmethod
     def _get_domain_access(domain):
@@ -225,23 +220,13 @@ def _get_domain_access(domain):
         response = hq_request.get()
 
         if response.status_code != 200:
-            return {}, []
+            return False, False, []
 
         response_data = response.json()
         hq_permissions = response_data['permissions']
         roles = response_data['roles'] or []
 
-        # Map between HQ and CCA
-        permissions = {
-            CAN_WRITE_PERMISSION: hq_permissions["can_edit"],
-            CAN_READ_PERMISSION: hq_permissions["can_view"],
-        }
-        return permissions, roles
-
-    @staticmethod
-    def _user_has_no_access(permissions: dict):
-        user_has_access = any([permissions[p] for p in permissions])
-        return not user_has_access
+        return hq_permissions["can_view"], hq_permissions["can_edit"], roles
 
     def _get_platform_roles(self, roles_names):
         platform_roles = []