Expire user permissions

hq_superset/hq_domain.py

@@ -4,7 +4,16 @@
 
 
 def before_request_hook():
-    return ensure_domain_selected()
+    """
+    Call all hooks functions set in sequence and
+    if any hook returns a response,
+    break the chain and return that response
+    """
+    hooks = [ensure_domain_selected]
+    for _function in hooks:
+        response = _function()
+        if response:
+            return response
 
 
 def after_request_hook(response):

hq_superset/tests/base_test.py

@@ -2,14 +2,15 @@
 """
 Base TestCase class
 """
-
 import os
 import shutil
 
+import jwt
 from flask_testing import TestCase
 from sqlalchemy.sql import text
 from superset.app import create_app
 
+from hq_superset.tests.utils import OAuthMock
 from hq_superset.utils import DOMAIN_PREFIX, get_hq_database
 
 superset_test_home = os.path.join(os.path.dirname(__file__), ".test_superset")
@@ -48,3 +49,33 @@ def tearDown(self):
                     sql = "; ".join(domain_schemas) + ";"
                     connection.execute(text(sql))
         super(HQDBTestCase, self).tearDown()
+
+
+class LoginUserTestMixin(object):
+    """
+    Use this mixin by calling login function with client
+    & then logout once done for clearing the session
+    """
+    def login(self, client):
+        self._setup_user()
+        # bypass oauth-workflow by skipping login and oauth flow
+        with client.session_transaction() as session_:
+            session_["oauth_state"] = "mock_state"
+        state = jwt.encode({}, "mock_state", algorithm="HS256")
+        return client.get(f"/oauth-authorized/commcare?state={state}", follow_redirects=True)
+
+    def _setup_user(self):
+        self.app.appbuilder.add_permissions(update_perms=True)
+        self.app.appbuilder.sm.sync_role_definitions()
+
+        self.oauth_mock = OAuthMock()
+        self.app.appbuilder.sm.oauth_remotes = {"commcare": self.oauth_mock}
+
+        gamma_role = self.app.appbuilder.sm.find_role('Gamma')
+        self.user = self.app.appbuilder.sm.find_user(self.oauth_mock.user_json['username'])
+        if not self.user:
+            self.user = self.app.appbuilder.sm.add_user(**self.oauth_mock.user_json, role=[gamma_role])
+
+    @staticmethod
+    def logout(client):
+        return client.get("/logout/")

hq_superset/tests/const.py

@@ -108,3 +108,16 @@
   "id": "test1_ucr1",
   "resource_uri": "/a/demo/api/v0.5/ucr_data_source/52a134da12c9b801bd85d2122901b30c/"
 }
+
+TEST_UCR_CSV_V1 = """\
+doc_id,inserted_at,data_visit_date_eaece89e,data_visit_number_33d63739,data_lmp_date_5e24b993,data_visit_comment_fb984fda
+a1, 2021-12-20, 2022-01-19, 100, 2022-02-20, some_text
+a2, 2021-12-22, 2022-02-19, 10, 2022-03-20, some_other_text
+"""
+
+TEST_UCR_CSV_V2 = """\
+doc_id,inserted_at,data_visit_date_eaece89e,data_visit_number_33d63739,data_lmp_date_5e24b993,data_visit_comment_fb984fda
+a1, 2021-12-20, 2022-01-19, 100, 2022-02-20, some_text
+a2, 2021-12-22, 2022-02-19, 10, 2022-03-20, some_other_text
+a3, 2021-11-22, 2022-01-19, 10, 2022-03-20, some_other_text2
+"""

hq_superset/tests/test_utils.py

@@ -1,8 +1,10 @@
 import doctest
+
+from flask import session
 from unittest.mock import patch
 
-from hq_superset.utils import get_column_dtypes, DomainSyncUtil
-from .base_test import SupersetTestCase
+from hq_superset.utils import get_column_dtypes, DomainSyncUtil, SESSION_DOMAIN_ROLE_LAST_SYNCED_AT
+from .base_test import SupersetTestCase, LoginUserTestMixin
 from .const import TEST_DATASOURCE
 
 
@@ -27,7 +29,7 @@ def test_doctests():
     assert results.failed == 0
 
 
-class TestDomainSyncUtil(SupersetTestCase):
+class TestDomainSyncUtil(LoginUserTestMixin, SupersetTestCase):
     PLATFORM_ROLE_NAMES = ["Gamma", "sql_lab", "dataset_editor"]
 
     @patch.object(DomainSyncUtil, "_get_domain_access")
@@ -102,6 +104,26 @@ def test_permissions_change_updates_user_role(self, get_domain_access_mock, doma
         additional_roles = DomainSyncUtil(security_manager)._get_additional_user_roles("test-domain")
         assert not additional_roles
 
+    @patch('hq_superset.utils.datetime_utcnow_naive')
+    @patch.object(DomainSyncUtil, "_get_domain_access")
+    def test_sync_domain_role(self, get_domain_access_mock, utcnow_mock):
+        client = self.app.test_client()
+        self.login(client)
+
+        utcnow_mock_return = "2024-11-01 14:30:04.323000+00:00"
+        utcnow_mock.return_value = utcnow_mock_return
+        get_domain_access_mock.return_value = self._to_permissions_response(
+            can_write=False,
+            can_read=True,
+            roles=[],
+        )
+        security_manager = self.app.appbuilder.sm
+
+        self.assertIsNone(session.get(SESSION_DOMAIN_ROLE_LAST_SYNCED_AT))
+        DomainSyncUtil(security_manager).sync_domain_role("test-domain")
+        self.assertEqual(session[SESSION_DOMAIN_ROLE_LAST_SYNCED_AT], utcnow_mock_return)
+        self.logout(client)
+
     def _ensure_platform_roles_exist(self, sm):
         for role_name in self.PLATFORM_ROLE_NAMES:
             sm.add_role(role_name)

hq_superset/tests/test_views.py

@@ -9,112 +9,19 @@
 from sqlalchemy.sql import text
 
 from hq_superset.exceptions import HQAPIException
+from hq_superset.tests.base_test import HQDBTestCase
+from hq_superset.tests.const import (
+    TEST_DATASOURCE,
+    TEST_UCR_CSV_V1,
+    TEST_UCR_CSV_V2,
+)
+from hq_superset.tests.utils import MockResponse, OAuthMock, UserMock
 from hq_superset.utils import (
     SESSION_USER_DOMAINS_KEY,
-    get_schema_name_for_domain,
     DomainSyncUtil,
+    get_schema_name_for_domain,
 )
 
-from .base_test import HQDBTestCase
-from .const import TEST_DATASOURCE
-
-
-class MockResponse:
-    def __init__(self, json_data, status_code):
-        self.json_data = json_data
-        self.status_code = status_code
-
-    def json(self):
-        return self.json_data
-
-    @property
-    def content(self):
-        return pickle.dumps(self.json_data)
-
-
-class UserMock():
-    user_id = '123'
-
-    def get_id(self):
-        return self.user_id
-
-
-class OAuthMock():
-
-    def __init__(self):
-        self.user_json = {
-            'username': 'testuser1',
-            'first_name': 'user',
-            'last_name': '1',
-            'email': '[email protected]',
-        }
-        self.domain_json = {
-            "objects": [
-                {
-                    "domain_name":"test1",
-                    "project_name":"test1"
-                },
-                {
-                    "domain_name":"test2",
-                    "project_name":"test 1"
-                },
-            ]
-        }
-        self.test1_datasources = {
-            "objects": [
-                {
-                    "id": 'test1_ucr1',
-                    "display_name": 'Test1 UCR1',
-                },
-                {
-                    "id": 'test1_ucr2',
-                    "display_name": 'Test1 UCR2',
-                },
-            ]
-        }
-        self.test2_datasources = {
-            "objects": [
-                {
-                    "id": 'test2_ucr1',
-                    "display_name": 'Test2 UCR1',
-                }
-            ]
-        }
-        self.api_base_url = "https://cchq.org/"
-        self.user_domain_roles = {
-            "permissions": {"can_view": True, "can_edit": True},
-            "roles": ["Gamma", "sql_lab"],
-        }
-
-    def authorize_access_token(self):
-        return {"access_token": "some-key"}
-
-    def get(self, url, token):
-        return {
-            'api/v0.5/identity/': MockResponse(self.user_json, 200),
-            'api/v0.5/user_domains?feature_flag=superset-analytics&can_view_reports=true': MockResponse(self.domain_json, 200),
-            'a/test1/api/v0.5/ucr_data_source/': MockResponse(self.test1_datasources, 200),
-            'a/test2/api/v0.5/ucr_data_source/': MockResponse(self.test2_datasources, 200),
-            'a/test1/api/v0.5/ucr_data_source/test1_ucr1/': MockResponse(TEST_DATASOURCE, 200),
-            'a/test1/configurable_reports/data_sources/export/test1_ucr1/?format=csv': MockResponse(TEST_UCR_CSV_V1, 200),
-            'a/test1/api/v0.5/analytics-roles/': MockResponse(self.user_domain_roles, 200),
-            'a/test2/api/v0.5/analytics-roles/': MockResponse(self.user_domain_roles, 200),
-        }[url]
-
-
-TEST_UCR_CSV_V1 = """\
-doc_id,inserted_at,data_visit_date_eaece89e,data_visit_number_33d63739,data_lmp_date_5e24b993,data_visit_comment_fb984fda
-a1, 2021-12-20, 2022-01-19, 100, 2022-02-20, some_text
-a2, 2021-12-22, 2022-02-19, 10, 2022-03-20, some_other_text
-"""
-
-TEST_UCR_CSV_V2 = """\
-doc_id,inserted_at,data_visit_date_eaece89e,data_visit_number_33d63739,data_lmp_date_5e24b993,data_visit_comment_fb984fda
-a1, 2021-12-20, 2022-01-19, 100, 2022-02-20, some_text
-a2, 2021-12-22, 2022-02-19, 10, 2022-03-20, some_other_text
-a3, 2021-11-22, 2022-01-19, 10, 2022-03-20, some_other_text2
-"""
-
 
 class TestViews(HQDBTestCase):
 

hq_superset/tests/utils.py

@@ -0,0 +1,91 @@
+import pickle
+
+from hq_superset.tests.const import TEST_DATASOURCE, TEST_UCR_CSV_V1
+
+
+class OAuthMock(object):
+
+    def __init__(self):
+        self.user_json = {
+            'username': 'testuser1',
+            'first_name': 'user',
+            'last_name': '1',
+            'email': '[email protected]',
+        }
+        self.domain_json = {
+            "objects": [
+                {
+                    "domain_name":"test1",
+                    "project_name":"test1"
+                },
+                {
+                    "domain_name":"test2",
+                    "project_name":"test 1"
+                },
+            ]
+        }
+        self.test1_datasources = {
+            "objects": [
+                {
+                    "id": 'test1_ucr1',
+                    "display_name": 'Test1 UCR1',
+                },
+                {
+                    "id": 'test1_ucr2',
+                    "display_name": 'Test1 UCR2',
+                },
+            ]
+        }
+        self.test2_datasources = {
+            "objects": [
+                {
+                    "id": 'test2_ucr1',
+                    "display_name": 'Test2 UCR1',
+                }
+            ]
+        }
+        self.api_base_url = "https://cchq.org/"
+        self.user_domain_roles = {
+            "permissions": {"can_view": True, "can_edit": True},
+            "roles": ["Gamma", "sql_lab"],
+        }
+
+    def authorize_access_token(self):
+        return {"access_token": "some-key"}
+
+    def get(self, url, token):
+        return {
+            'api/v0.5/identity/': MockResponse(self.user_json, 200),
+            'api/v0.5/user_domains?feature_flag=superset-analytics&can_view_reports=true': MockResponse(
+                self.domain_json, 200
+            ),
+            'a/test1/api/v0.5/ucr_data_source/': MockResponse(self.test1_datasources, 200),
+            'a/test2/api/v0.5/ucr_data_source/': MockResponse(self.test2_datasources, 200),
+            'a/test1/api/v0.5/ucr_data_source/test1_ucr1/': MockResponse(TEST_DATASOURCE, 200),
+            'a/test1/configurable_reports/data_sources/export/test1_ucr1/?format=csv': MockResponse(
+                TEST_UCR_CSV_V1, 200
+            ),
+            'a/test1/api/v0.5/analytics-roles/': MockResponse(self.user_domain_roles, 200),
+            'a/test2/api/v0.5/analytics-roles/': MockResponse(self.user_domain_roles, 200),
+        }[url]
+
+
+class UserMock(object):
+    user_id = '123'
+
+    def get_id(self):
+        return self.user_id
+
+
+class MockResponse:
+    def __init__(self, json_data, status_code):
+        self.json_data = json_data
+        self.status_code = status_code
+
+    def json(self):
+        return self.json_data
+
+    @property
+    def content(self):
+        return pickle.dumps(self.json_data)
+

hq_superset/utils.py

@@ -9,9 +9,10 @@
 from zipfile import ZipFile
 
 import pandas
+import pytz
 import sqlalchemy
 from cryptography.fernet import Fernet
-from flask import current_app
+from flask import current_app, session
 from flask_login import current_user
 from sqlalchemy.sql import TableClause
 from superset.utils.database import get_or_create_db
@@ -30,6 +31,7 @@
 DOMAIN_PREFIX = "hqdomain_"
 SESSION_USER_DOMAINS_KEY = "user_hq_domains"
 SESSION_OAUTH_RESPONSE_KEY = "oauth_response"
+SESSION_DOMAIN_ROLE_LAST_SYNCED_AT = "domain_role_last_synced_at"
 
 
 def get_hq_database():
@@ -151,6 +153,7 @@ def sync_domain_role(self, domain):
 
         self.sm.get_session.add(current_user)
         self.sm.get_session.commit()
+        session['domain_role_last_synced_at'] = datetime_utcnow_naive()
         return True
 
     def _ensure_hq_user_role(self):
@@ -415,3 +418,7 @@ def cast_data_for_table(
 def generate_secret():
     alphabet = string.ascii_letters + string.digits
     return ''.join(secrets.choice(alphabet) for __ in range(64))
+
+
+def datetime_utcnow_naive():
+    return datetime.utcnow().replace(tzinfo=pytz.UTC)