From 37aedbab4f1e0788d2e7e6b256631608e4cda14b Mon Sep 17 00:00:00 2001 From: Saqib Ansari Date: Sun, 10 May 2026 15:20:00 +0530 Subject: [PATCH 1/6] feat: add permission tests --- insights/api/workbooks.py | 30 +- insights/patches/normalize_workbook.py | 27 +- insights/permissions.py | 7 +- insights/tests/test_permissions.py | 542 ++++++++++++++++++++----- 4 files changed, 490 insertions(+), 116 deletions(-) diff --git a/insights/api/workbooks.py b/insights/api/workbooks.py index ce48556fe..2a735d56d 100644 --- a/insights/api/workbooks.py +++ b/insights/api/workbooks.py @@ -74,7 +74,7 @@ def import_workbook(workbook: dict): @insights_whitelist() -def get_share_permissions(workbook_name:str): +def get_share_permissions(workbook_name: str): if not frappe.has_permission("Insights Workbook", ptype="share", doc=workbook_name): frappe.throw(_("You do not have permission to share this workbook")) @@ -127,10 +127,26 @@ def get_share_permissions(workbook_name:str): @insights_whitelist() -def update_share_permissions(workbook_name:str, user_permissions: dict, organization_access: str | None = None): +def update_share_permissions( + workbook_name: str, user_permissions: dict, organization_access: str | None = None +): if not frappe.has_permission("Insights Workbook", ptype="share", doc=workbook_name): frappe.throw(_("You do not have permission to share this workbook")) + existing_shares = frappe.get_all( + "DocShare", + filters={ + "share_doctype": "Insights Workbook", + "share_name": workbook_name, + }, + fields=["name", "user", "everyone"], + ) + + allowed_users = {permission["user"] for permission in user_permissions} + for share in existing_shares: + if share.user and share.user not in allowed_users: + frappe.delete_doc("DocShare", share.name, ignore_permissions=True) + for permission in user_permissions: doc = DocShare.get_or_create_doc( share_doctype="Insights Workbook", @@ -158,16 +174,14 @@ def update_share_permissions(workbook_name:str, user_permissions: dict, organiza # folder Management APIs + @insights_whitelist() def create_folder(workbook: str, title: str, folder_type: str): """Create a new folder in workbook""" if not frappe.has_permission("Insights Workbook", ptype="write", doc=workbook): frappe.throw(_("You do not have permission to modify this workbook")) - current_folders = frappe.db.count( - "Insights Folder", - filters={"workbook": workbook, "type": folder_type} - ) + current_folders = frappe.db.count("Insights Folder", filters={"workbook": workbook, "type": folder_type}) folder = frappe.new_doc("Insights Folder") folder.workbook = workbook @@ -178,6 +192,7 @@ def create_folder(workbook: str, title: str, folder_type: str): return folder.name + @insights_whitelist() def rename_folder(folder_name: str, new_title: str): """Rename a folder""" @@ -190,6 +205,7 @@ def rename_folder(folder_name: str, new_title: str): return folder.name + @insights_whitelist() def delete_folder(folder_name: str, move_items_to_root: bool = True): """Delete folder and move items to root""" @@ -217,6 +233,7 @@ def delete_folder(folder_name: str, move_items_to_root: bool = True): frappe.delete_doc("Insights Folder", folder_name) + @insights_whitelist() def toggle_folder_expanded(folder_name: str, is_expanded: bool): """Toggle folder expanded state""" @@ -226,6 +243,7 @@ def toggle_folder_expanded(folder_name: str, is_expanded: bool): folder.db_set("is_expanded", is_expanded, update_modified=False) + @insights_whitelist() def move_item_to_folder(item_type: str, item_name: str, folder_name: str | None = None): """Move a query/chart to a folder""" diff --git a/insights/patches/normalize_workbook.py b/insights/patches/normalize_workbook.py index 11f7bd7aa..eabe63af4 100644 --- a/insights/patches/normalize_workbook.py +++ b/insights/patches/normalize_workbook.py @@ -5,9 +5,9 @@ def execute(): - should_run_patch = frappe.db.exists( - "DocType", "Insights Workbook" - ) and frappe.db.count("Insights Workbook") + should_run_patch = frappe.db.exists("DocType", "Insights Workbook") and frappe.db.count( + "Insights Workbook" + ) if not should_run_patch: return @@ -21,9 +21,18 @@ def execute(): for wb in workbooks: workbook = frappe.get_doc("Insights Workbook", wb) - queries = frappe.parse_json(workbook.queries) or [] - charts = frappe.parse_json(workbook.charts) or [] - dashboards = frappe.parse_json(workbook.dashboards) or [] + legacy_queries = getattr(workbook, "queries", None) + legacy_charts = getattr(workbook, "charts", None) + legacy_dashboards = getattr(workbook, "dashboards", None) + + # Sites that already use linked child doctypes no longer have these + # legacy JSON fields on the workbook itself, so this patch becomes a no-op. + if legacy_queries is None and legacy_charts is None and legacy_dashboards is None: + continue + + queries = frappe.parse_json(legacy_queries) or [] + charts = frappe.parse_json(legacy_charts) or [] + dashboards = frappe.parse_json(legacy_dashboards) or [] query_name_to_doc = {} chart_name_to_doc = {} @@ -93,11 +102,7 @@ def execute(): item.links = item.links or {} for chart_name, field in item.links.items(): - if ( - chart_name not in chart_name_to_doc - or not field - or "`.`" not in field - ): + if chart_name not in chart_name_to_doc or not field or "`.`" not in field: continue chart = chart_name_to_doc[chart_name] diff --git a/insights/permissions.py b/insights/permissions.py index be6dee932..cf566e599 100644 --- a/insights/permissions.py +++ b/insights/permissions.py @@ -129,12 +129,15 @@ def _build_source_permission_query(self, ptype): return frappe.qb.from_(frappe.qb.DocType("Insights Data Source v3")).select("name") # if team permissions are enabled, allow data sources of allowed tables + DataSource = frappe.qb.DocType("Insights Data Source v3") Table = frappe.qb.DocType("Insights Table v3") AllowedTables = self._build_table_permission_query(ptype) return ( - frappe.qb.from_(Table) - .select(Table.data_source) + frappe.qb.from_(DataSource) + .select(DataSource.name) + .left_join(Table) + .on(Table.data_source == DataSource.name) .left_join(AllowedTables) .on(Table.name == AllowedTables.name) .where(AllowedTables.name.isnotnull()) diff --git a/insights/tests/test_permissions.py b/insights/tests/test_permissions.py index 340a9fdbf..7677d7c06 100644 --- a/insights/tests/test_permissions.py +++ b/insights/tests/test_permissions.py @@ -1,106 +1,339 @@ -import unittest - import frappe +import frappe.share +from frappe.tests import IntegrationTestCase + +from insights.api.workbooks import get_share_permissions, update_share_permissions +from insights.decorators import insights_whitelist +from insights.insights.doctype.insights_table_v3.insights_table_v3 import get_table_name +from insights.insights.doctype.insights_team.insights_team import clear_cache as clear_team_cache +from insights.permissions import PERMISSION_DOCTYPES + +TEST_DS_TITLE = "Test DuckDB" +TEST_DS_NAME = frappe.scrub(TEST_DS_TITLE) +TEST_TABLE1_NAME = get_table_name(TEST_DS_NAME, "table1") +TEST_TABLE2_NAME = get_table_name(TEST_DS_NAME, "table2") +TEST_TABLE3_NAME = get_table_name(TEST_DS_NAME, "table3") + + +@insights_whitelist() +def protected_insights_call(): + return True + + +class DT: + DATA_SOURCE = "Insights Data Source v3" + TABLE = "Insights Table v3" + WORKBOOK = "Insights Workbook" + QUERY = "Insights Query v3" + CHART = "Insights Chart v3" + DASHBOARD = "Insights Dashboard v3" + TEAM = "Insights Team" + SETTINGS = "Insights Settings" + + +class TestInsightsPermissions(IntegrationTestCase): + SAVEPOINT = "test_insights_permissions" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.original_enable_permissions = frappe.db.get_single_value(DT.SETTINGS, "enable_permissions") + frappe.set_user("Administrator") + cleanup_test_fixtures() + create_test_users() + clear_team_cache() + frappe.db.commit() + + @classmethod + def tearDownClass(cls): + frappe.set_user("Administrator") + frappe.db.set_single_value(DT.SETTINGS, "enable_permissions", cls.original_enable_permissions) + clear_team_cache() + cleanup_test_fixtures() + frappe.db.commit() + super().tearDownClass() - -class TestInsightsPermissions(unittest.TestCase): def setUp(self): - create_test_users() - create_test_data_sources() - create_test_tables() - create_test_teams() + super().setUp() + self.original_user = frappe.session.user + frappe.set_user("Administrator") + clear_team_cache() + frappe.db.savepoint(self.SAVEPOINT) def tearDown(self): - delete_test_teams() - delete_test_tables() - delete_test_data_sources() - delete_test_users() + frappe.set_user("Administrator") + frappe.db.rollback(save_point=self.SAVEPOINT) + clear_team_cache() + frappe.set_user(self.original_user) + super().tearDown() def toggle_team_permissions(self, enable): - frappe.db.set_value( - "Insights Settings", None, "enable_team_permissions", enable - ) + frappe.db.set_single_value(DT.SETTINGS, "enable_permissions", enable) + clear_team_cache() def test_permissions_for_non_insights_user(self): - # check if a non insights user can access a insights doctypes - pass + frappe.set_user("non_insights_user@test.com") + + for doctype in PERMISSION_DOCTYPES: + self.assertFalse( + frappe.has_permission(doctype, ptype="read"), + f"{doctype} should not be readable without an Insights role", + ) + + with self.assertRaises(frappe.PermissionError): + protected_insights_call() def test_permissions_on_team_based_doctype_with_team_permissions_disabled(self): + create_test_data_sources() + create_test_tables() + create_test_teams() self.toggle_team_permissions(False) - # check if a insights user can access all teams - pass + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) + self.assertTrue(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) def test_permission_on_team_based_doctype_with_team_permissions_enabled(self): + create_test_data_sources() + create_test_tables() + team = create_test_teams() self.toggle_team_permissions(True) - # check if insights user can access a team, table, data source - # add user to a team, give access to a table, data source - # check if insights user can access a team, table, data source - pass + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) + self.assertFalse(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) + + frappe.set_user("Administrator") + team.append( + "team_permissions", + {"resource_type": DT.DATA_SOURCE, "resource_name": TEST_DS_NAME}, + ) + team.append( + "team_permissions", + { + "resource_type": DT.TABLE, + "resource_name": TEST_TABLE1_NAME, + }, + ) + team.save(ignore_permissions=True) + clear_team_cache() + + frappe.set_user("insights_user1@test.com") + self.assertTrue(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) + self.assertTrue(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) def test_permission_for_admin_on_team_based_doctype_with_team_permissions_enabled( self, ): + create_test_data_sources() + create_test_tables() self.toggle_team_permissions(True) - # check if admin can access all teams, tables, data sources - pass + + frappe.set_user("insights_admin@test.com") + self.assertTrue(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) + self.assertTrue(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) def test_permission_for_workbook(self): - # check if insights user has access to no workbooks - # check if insights user can create a workbook - # check if insights user can share a workbook - # check if another insights user can access a shared workbook - # unshare the workbook and check if another insights user can access it - pass + workbook = create_test_workbook("insights_user1@test.com") + + frappe.set_user("insights_user1@test.com") + self.assertTrue(is_allowed(DT.WORKBOOK, workbook.name)) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.WORKBOOK, workbook.name)) + + frappe.set_user("insights_user1@test.com") + update_share_permissions( + workbook.name, + [{"user": "insights_user2@test.com", "read": 1, "write": 0}], + ) + share_permissions = get_share_permissions(workbook.name) + self.assertIn( + "insights_user2@test.com", + [permission["user"] for permission in share_permissions["user_permissions"]], + ) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.WORKBOOK, workbook.name)) + + frappe.set_user("insights_user1@test.com") + update_share_permissions(workbook.name, []) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.WORKBOOK, workbook.name)) def test_permission_for_dashboard(self): - # check if insights user has access to no dashboard - # create a workbook - # check if insights user can create a dashboard - # check if insights user can share a dashboard - # check if another insights user can access the shared dashboard - # unshare the dashboard and check if another insights user can access it - # share the workbook with read access - # check if another insights user can access the workbook dashboard - # check if another user cannot create a dashboard for the workbook - pass + workbook = create_test_workbook("insights_user1@test.com") + dashboard = create_test_dashboard("insights_user1@test.com", workbook.name) + + frappe.set_user("insights_user1@test.com") + self.assertTrue(is_allowed(DT.DASHBOARD, dashboard.name)) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.DASHBOARD, dashboard.name)) + + frappe.set_user("insights_user1@test.com") + update_dashboard_access(dashboard.name, ["insights_user2@test.com"]) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.DASHBOARD, dashboard.name)) + + frappe.set_user("insights_user1@test.com") + update_dashboard_access(dashboard.name, []) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.DASHBOARD, dashboard.name)) + + frappe.set_user("insights_user1@test.com") + update_share_permissions( + workbook.name, + [{"user": "insights_user2@test.com", "read": 1, "write": 0}], + ) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.DASHBOARD, dashboard.name)) + self.assertFalse(frappe.has_permission(DT.DASHBOARD, ptype="write", doc=dashboard.name)) + with self.assertRaises(frappe.PermissionError): + create_test_dashboard( + "insights_user2@test.com", + workbook.name, + title="Permissions Test Dashboard Read Only", + ) def test_permission_for_chart(self): - # check if insights user has access to no chart - # create a workbook - # create a chart - # check if insights user can share a chart - # check if another insights user can access the shared chart - # unshare the chart and check if another insights user can access it - # share the workbook with read access - # check if another insights user can access the workbook chart - # check if another user cannot create a chart for the workbook - # unshare the workbook and chart - # create a dashboard and add a chart - # share the dashboard - # check if another insights user can access the dashboard chart - pass + workbook = create_test_workbook("insights_user1@test.com") + query = create_test_query("insights_user1@test.com", workbook.name) + chart = create_test_chart("insights_user1@test.com", workbook.name, query.name) + + frappe.set_user("insights_user1@test.com") + self.assertTrue(is_allowed(DT.CHART, chart.name)) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.CHART, chart.name)) + + frappe.set_user("insights_user1@test.com") + share_chart(chart.name, "insights_user2@test.com") + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.CHART, chart.name)) + + frappe.set_user("insights_user1@test.com") + unshare_chart(chart.name, "insights_user2@test.com") + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.CHART, chart.name)) + + frappe.set_user("insights_user1@test.com") + update_share_permissions( + workbook.name, + [{"user": "insights_user2@test.com", "read": 1, "write": 0}], + ) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.CHART, chart.name)) + self.assertFalse(frappe.has_permission(DT.CHART, ptype="write", doc=chart.name)) + with self.assertRaises(frappe.PermissionError): + create_test_chart( + "insights_user2@test.com", + workbook.name, + query.name, + title="Permissions Test Chart Read Only", + ) + + frappe.set_user("insights_user1@test.com") + update_share_permissions(workbook.name, []) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.CHART, chart.name)) + + dashboard = create_test_dashboard( + "insights_user1@test.com", + workbook.name, + chart.name, + title="Permissions Test Dashboard For Chart", + ) + frappe.set_user("insights_user1@test.com") + update_dashboard_access(dashboard.name, ["insights_user2@test.com"]) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.CHART, chart.name)) def test_permission_for_query(self): - # check if insights user has access to no query - # create a workbook - # create a query - # share the workbook with read access - # check if another insights user can access the workbook query - # check if another user cannot create a query for the workbook - # unshare the workbook and query - # create a chart and select the query - # share the chart - # check if another insights user can access the chart query & data_query - # unshare the chart - # create a dashboard and add a chart - # share the dashboard - # check if another insights user can access the dashboard chart query & data_query - - # drilldown cases - # if a non-insights user creates a query, it should fail, because role permissions will not allow it - # if an insights user creates a query against a workbook, then it should check workbook permissions - # if an insights user creates a query without a workbook, then it should check permissions of the sources - pass + workbook = create_test_workbook("insights_user1@test.com") + query = create_test_query("insights_user1@test.com", workbook.name) + + frappe.set_user("insights_user1@test.com") + self.assertTrue(is_allowed(DT.QUERY, query.name)) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.QUERY, query.name)) + + frappe.set_user("insights_user1@test.com") + update_share_permissions( + workbook.name, + [{"user": "insights_user2@test.com", "read": 1, "write": 0}], + ) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.QUERY, query.name)) + with self.assertRaises(frappe.PermissionError): + create_test_query( + "insights_user2@test.com", + workbook.name, + title="Permissions Test Query Read Only", + ) + + frappe.set_user("insights_user1@test.com") + update_share_permissions(workbook.name, []) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.QUERY, query.name)) + + chart = create_test_chart( + "insights_user1@test.com", + workbook.name, + query.name, + title="Permissions Test Chart For Query", + ) + chart = frappe.get_doc(DT.CHART, chart.name) + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.QUERY, query.name)) + self.assertFalse(is_allowed(DT.QUERY, chart.data_query)) + + frappe.set_user("insights_user1@test.com") + share_chart(chart.name, "insights_user2@test.com") + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.QUERY, query.name)) + self.assertTrue(is_allowed(DT.QUERY, chart.data_query)) + + frappe.set_user("insights_user1@test.com") + unshare_chart(chart.name, "insights_user2@test.com") + + frappe.set_user("insights_user2@test.com") + self.assertFalse(is_allowed(DT.QUERY, query.name)) + self.assertFalse(is_allowed(DT.QUERY, chart.data_query)) + + dashboard = create_test_dashboard( + "insights_user1@test.com", + workbook.name, + chart.name, + title="Permissions Test Dashboard For Query", + ) + frappe.set_user("insights_user1@test.com") + update_dashboard_access(dashboard.name, ["insights_user2@test.com"]) + + frappe.set_user("insights_user2@test.com") + self.assertTrue(is_allowed(DT.QUERY, query.name)) + self.assertTrue(is_allowed(DT.QUERY, chart.data_query)) + + with self.assertRaises(frappe.PermissionError): + create_test_query( + "non_insights_user@test.com", + workbook.name, + title="Permissions Test Query Non Insights", + ) def create_test_users(): @@ -173,64 +406,179 @@ def create_test_users(): def delete_test_users(): - frappe.delete_doc("User", "web_user@test.com", force=True) - frappe.delete_doc("User", "non_insights_user@test.com", force=True) - frappe.delete_doc("User", "insights_user1@test.com", force=True) - frappe.delete_doc("User", "insights_user2@test.com", force=True) - frappe.delete_doc("User", "insights_admin@test.com", force=True) + delete_doc_if_exists("User", "web_user@test.com") + delete_doc_if_exists("User", "non_insights_user@test.com") + delete_doc_if_exists("User", "insights_user1@test.com") + delete_doc_if_exists("User", "insights_user2@test.com") + delete_doc_if_exists("User", "insights_admin@test.com") def create_test_data_sources(): frappe.get_doc( { - "doctype": "Insights Data Source v3", + "doctype": DT.DATA_SOURCE, + "title": TEST_DS_TITLE, "database_type": "DuckDB", - "database_name": "Test DuckDB", + "database_name": "test_duckdb", } ).insert() def delete_test_data_sources(): - frappe.delete_doc("Insights Data Source v3", "Test DuckDB", force=True) + delete_doc_if_exists(DT.DATA_SOURCE, TEST_DS_NAME) def create_test_tables(): frappe.get_doc( { - "doctype": "Insights Table v3", - "table_name": "table1", - "data_source": "Test DuckDB", + "doctype": DT.TABLE, + "table": "table1", + "label": "table1", + "data_source": TEST_DS_NAME, + "sync_mode": "Full", } ).insert() frappe.get_doc( { - "doctype": "Insights Table v3", - "table_name": "table2", - "data_source": "Test DuckDB", + "doctype": DT.TABLE, + "table": "table2", + "label": "table2", + "data_source": TEST_DS_NAME, + "sync_mode": "Full", } ).insert() frappe.get_doc( { - "doctype": "Insights Table v3", - "table_name": "table3", - "data_source": "Test DuckDB", + "doctype": DT.TABLE, + "table": "table3", + "label": "table3", + "data_source": TEST_DS_NAME, + "sync_mode": "Full", } ).insert() def delete_test_tables(): - frappe.delete_doc("Insights Table v3", "table1", force=True) - frappe.delete_doc("Insights Table v3", "table2", force=True) - frappe.delete_doc("Insights Table v3", "table3", force=True) + delete_doc_if_exists(DT.TABLE, TEST_TABLE1_NAME) + delete_doc_if_exists(DT.TABLE, TEST_TABLE2_NAME) + delete_doc_if_exists(DT.TABLE, TEST_TABLE3_NAME) def create_test_teams(): - team1 = frappe.get_doc({"doctype": "Insights Team", "team_name": "team1"}) + team1 = frappe.get_doc({"doctype": DT.TEAM, "team_name": "team1"}) team1.append("team_members", {"user": "insights_user1@test.com"}) team1.save() + clear_team_cache() + return team1 def delete_test_teams(): - frappe.delete_doc("Insights Team", "team1", force=True) + delete_doc_if_exists(DT.TEAM, "team1") + + +def create_test_workbook(owner, title="Permissions Test Workbook"): + frappe.set_user(owner) + return frappe.get_doc({"doctype": DT.WORKBOOK, "title": title}).insert() + + +def create_test_query(owner, workbook, title="Permissions Test Query"): + frappe.set_user(owner) + return frappe.get_doc( + { + "doctype": DT.QUERY, + "title": title, + "workbook": workbook, + "is_builder_query": 1, + "use_live_connection": 1, + "operations": [], + } + ).insert() + + +def create_test_chart(owner, workbook, query=None, title="Permissions Test Chart"): + frappe.set_user(owner) + chart = frappe.get_doc( + { + "doctype": DT.CHART, + "title": title, + "workbook": workbook, + "query": query, + "chart_type": "Bar", + "config": {}, + } + ).insert() + return frappe.get_doc(DT.CHART, chart.name) + + +def create_test_dashboard(owner, workbook, chart=None, title="Permissions Test Dashboard"): + frappe.set_user(owner) + items = [] + if chart: + items.append({"id": "chart-1", "type": "chart", "chart": chart}) + + dashboard = frappe.get_doc( + { + "doctype": DT.DASHBOARD, + "title": title, + "workbook": workbook, + "items": items, + } + ).insert() + return frappe.get_doc(DT.DASHBOARD, dashboard.name) + + +def delete_test_workbooks(): + workbooks = frappe.get_all( + DT.WORKBOOK, + filters={"title": ["like", "Permissions Test Workbook%"]}, + pluck="name", + ) + for workbook in workbooks: + frappe.delete_doc(DT.WORKBOOK, workbook, force=True) + + +def is_allowed(doctype, docname): + return bool(frappe.get_list(doctype, filters={"name": docname}, pluck="name", limit=1)) + + +def update_dashboard_access(dashboard_name, people_with_access): + frappe.get_doc(DT.DASHBOARD, dashboard_name).update_access( + { + "is_public": 0, + "is_shared_with_organization": 0, + "people_with_access": people_with_access, + } + ) + + +def share_chart(chart_name, user): + frappe.share.add(DT.CHART, chart_name, user=user, read=1, notify=0) + + +def unshare_chart(chart_name, user): + share_name = frappe.db.get_value( + "DocShare", + { + "share_doctype": DT.CHART, + "share_name": chart_name, + "user": user, + }, + ) + if share_name: + frappe.delete_doc("DocShare", share_name, ignore_permissions=True) + + +def cleanup_test_fixtures(): + delete_test_workbooks() + delete_test_teams() + delete_test_tables() + delete_test_data_sources() + delete_test_users() + clear_team_cache() + + +def delete_doc_if_exists(doctype, name): + if frappe.db.exists(doctype, name): + frappe.delete_doc(doctype, name, force=True) From c89cc32086802a1db4837b887a899cae8c426464 Mon Sep 17 00:00:00 2001 From: Saqib Ansari Date: Sun, 10 May 2026 15:54:25 +0530 Subject: [PATCH 2/6] chore: exclude legacy v2 code from backend coverage --- .coveragerc | 6 + .github/helpers/legacy_v2_python_globs.txt | 38 +++++ .github/helpers/run_backend_tests.py | 171 +++++++++++++++++++++ .github/workflows/server-tests.yml | 34 ++-- codecov.yml | 31 ++++ pyproject.toml | 8 + 6 files changed, 273 insertions(+), 15 deletions(-) create mode 100644 .coveragerc create mode 100644 .github/helpers/legacy_v2_python_globs.txt create mode 100644 .github/helpers/run_backend_tests.py create mode 100644 codecov.yml diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 000000000..904219c6e --- /dev/null +++ b/.coveragerc @@ -0,0 +1,6 @@ +[report] +omit = + */tests/* + */config/* + */insights/setup.py + */coverage.py diff --git a/.github/helpers/legacy_v2_python_globs.txt b/.github/helpers/legacy_v2_python_globs.txt new file mode 100644 index 000000000..fb5d717f7 --- /dev/null +++ b/.github/helpers/legacy_v2_python_globs.txt @@ -0,0 +1,38 @@ +# Paths are relative to apps/insights. +# Keep this list aligned with the Python files deleted by the remove-v2-code branch. + +insights/api/home.py +insights/api/notebooks.py +insights/api/permissions.py +insights/api/public.py +insights/api/queries.py +insights/api/setup.py +insights/api/subscription.py + +insights/insights/doctype/insights_chart/** +insights/insights/doctype/insights_dashboard/** +insights/insights/doctype/insights_dashboard_item/** +insights/insights/doctype/insights_data_source/** +insights/insights/doctype/insights_notebook/** +insights/insights/doctype/insights_notebook_page/** +insights/insights/doctype/insights_query/** +insights/insights/doctype/insights_query_chart/** +insights/insights/doctype/insights_query_column/** +insights/insights/doctype/insights_query_result/** +insights/insights/doctype/insights_query_table/** +insights/insights/doctype/insights_query_transform/** +insights/insights/doctype/insights_table/** +insights/insights/doctype/insights_table_column/** +insights/insights/doctype/insights_table_link/** +insights/insights/doctype/insights_team/insights_team_client.py +insights/insights/doctype/insights_team/test_insights_team.py + +insights/insights/query_builders/__init__.py +insights/insights/query_builders/legacy_query_builder.py +insights/insights/query_builders/postgresql/builder.py +insights/insights/query_builders/sql_builder.py +insights/insights/query_builders/sqlite/sqlite_query_builder.py +insights/insights/query_builders/test_sql_builder.py +insights/insights/query_builders/utils.py + +insights/www/insights_v2.py diff --git a/.github/helpers/run_backend_tests.py b/.github/helpers/run_backend_tests.py new file mode 100644 index 000000000..a631c8dab --- /dev/null +++ b/.github/helpers/run_backend_tests.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import argparse +import os +from fnmatch import fnmatch +from pathlib import Path + +from coverage import Coverage +from coverage.exceptions import NoDataError + +APP_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_APP = APP_ROOT.name +LEGACY_GLOBS_FILE = Path(__file__).with_name("legacy_v2_python_globs.txt") +COVERAGE_CONFIG_FILE = APP_ROOT / ".coveragerc" +SKIP_DIRS = {"node_modules", "locals", "public", "__pycache__"} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run Insights backend tests while excluding legacy v2 code from discovery and coverage." + ) + parser.add_argument("--site", help="Site to run tests against") + parser.add_argument("--app", default=DEFAULT_APP, help="App to run tests for") + parser.add_argument( + "--coverage-file", + default="sites/coverage.xml", + help="Coverage XML output path, relative to the current working directory", + ) + parser.add_argument( + "--legacy-globs-file", + default=str(LEGACY_GLOBS_FILE), + help="Path to the legacy v2 path glob list", + ) + parser.add_argument( + "--list-only", + action="store_true", + help="Print the included and excluded test modules without running tests", + ) + return parser.parse_args() + + +def dedupe(items: list[str]) -> list[str]: + return list(dict.fromkeys(items)) + + +def load_globs(path: Path) -> list[str]: + patterns: list[str] = [] + for line in path.read_text().splitlines(): + stripped = line.strip() + if stripped and not stripped.startswith("#"): + patterns.append(stripped) + return patterns + + +def matches_any(path: str, patterns: list[str]) -> bool: + return any(fnmatch(path, pattern) for pattern in patterns) + + +def discover_test_modules(app_root: Path, legacy_globs: list[str]) -> tuple[list[str], list[str]]: + modules: list[str] = [] + excluded: list[str] = [] + + for file_path in sorted(app_root.rglob("test_*.py")): + rel_path = file_path.relative_to(app_root) + rel_path_str = rel_path.as_posix() + + if file_path.name == "test_runner.py": + continue + + if any(part.startswith(".") for part in rel_path.parts): + continue + + if any(skip_dir in rel_path.parts for skip_dir in SKIP_DIRS): + continue + + if "doctype/doctype/boilerplate" in rel_path_str: + continue + + module_name = ".".join(file_path.relative_to(app_root).with_suffix("").parts) + if matches_any(rel_path_str, legacy_globs): + excluded.append(module_name) + continue + + modules.append(module_name) + + return modules, excluded + + +def load_base_omit_patterns(config_file: Path) -> list[str]: + if not config_file.exists(): + return [] + + coverage = Coverage(config_file=str(config_file)) + return dedupe(list(coverage.config.run_omit) + list(coverage.config.report_omit)) + + +def build_coverage_omit_patterns(base_patterns: list[str], legacy_globs: list[str]) -> list[str]: + from frappe.coverage import STANDARD_EXCLUSIONS + + legacy_patterns = [f"*/{pattern}" for pattern in legacy_globs] + return dedupe(STANDARD_EXCLUSIONS + base_patterns + legacy_patterns) + + +def run_tests(site: str, app: str, modules: list[str], coverage_file: Path, omit_patterns: list[str]) -> None: + import frappe + from frappe.commands.testing import main as run_tests_main + + bench_root = Path.cwd() + coverage_file = coverage_file if coverage_file.is_absolute() else bench_root / coverage_file + sites_dir = bench_root / "sites" + original_cwd = Path.cwd() + source_root = APP_ROOT / app + coverage = Coverage( + source=[str(source_root)], + omit=omit_patterns, + ) + + coverage_file.parent.mkdir(parents=True, exist_ok=True) + coverage_started = False + + try: + os.chdir(sites_dir) + frappe.init(site) + coverage.start() + coverage_started = True + run_tests_main(site=site, app=app, module=modules) + finally: + if coverage_started: + coverage.stop() + coverage.save() + try: + coverage.xml_report(outfile=str(coverage_file)) + except NoDataError: + print("No coverage data collected") + frappe.destroy() + os.chdir(original_cwd) + print(f"Saved Coverage: {coverage_file}") + + +def main() -> None: + args = parse_args() + legacy_globs = load_globs(Path(args.legacy_globs_file)) + modules, excluded_modules = discover_test_modules(APP_ROOT, legacy_globs) + + print(f"Included test modules: {len(modules)}") + print(f"Excluded legacy test modules: {len(excluded_modules)}") + + if excluded_modules: + for module_name in excluded_modules: + print(f"- {module_name}") + + if args.list_only: + return + + if not args.site: + raise SystemExit("--site is required unless --list-only is used") + + if not modules: + raise SystemExit("No backend test modules remain after applying the legacy v2 exclusions") + + omit_patterns = build_coverage_omit_patterns( + load_base_omit_patterns(COVERAGE_CONFIG_FILE), + legacy_globs, + ) + run_tests(args.site, args.app, modules, Path(args.coverage_file), omit_patterns) + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/server-tests.yml b/.github/workflows/server-tests.yml index a2661b2db..666d48647 100644 --- a/.github/workflows/server-tests.yml +++ b/.github/workflows/server-tests.yml @@ -1,12 +1,14 @@ -name: Server Tests +name: Server on: + pull_request: + workflow_dispatch: schedule: # Run everday at midnight UTC / 5:30 IST - cron: "0 0 * * *" concurrency: - group: develop-insights-${{ github.event.number }} + group: server-${{ github.event_name }}-${{ github.event.number }} cancel-in-progress: true permissions: @@ -30,7 +32,7 @@ jobs: steps: - name: Clone - uses: actions/checkout@v2 + uses: actions/checkout@v6 - name: Check for valid Python & Merge Conflicts run: | @@ -41,18 +43,18 @@ jobs: fi - name: Setup Python - uses: actions/setup-python@v2 + uses: actions/setup-python@v6 with: python-version: "3.10" - name: Setup Node - uses: actions/setup-node@v2 + uses: actions/setup-node@v5 with: node-version: 18 check-latest: true - name: Cache pip - uses: actions/cache@v2 + uses: actions/cache@v4 with: path: ~/.cache/pip key: ${{ runner.os }}-pip-${{ hashFiles('**/*requirements.txt', '**/pyproject.toml', '**/setup.py', '**/setup.cfg') }} @@ -62,9 +64,9 @@ jobs: - name: Get yarn cache directory path id: yarn-cache-dir-path - run: 'echo "::set-output name=dir::$(yarn cache dir)"' + run: echo "dir=$(yarn cache dir)" >> $GITHUB_OUTPUT - - uses: actions/cache@v2 + - uses: actions/cache@v4 id: yarn-cache with: path: ${{ steps.yarn-cache-dir-path.outputs.dir }} @@ -94,14 +96,14 @@ jobs: working-directory: /home/runner/frappe-bench run: | bench --site test_site set-config allow_tests true - bench --site test_site run-tests --app insights --coverage + env/bin/python apps/insights/.github/helpers/run_backend_tests.py --site test_site env: TYPE: server - name: Upload coverage data - uses: actions/upload-artifact@v3 + uses: actions/upload-artifact@v4 with: - name: coverage-${{ matrix.container }} + name: coverage-server path: /home/runner/frappe-bench/sites/coverage.xml coverage: @@ -110,14 +112,16 @@ jobs: runs-on: ubuntu-latest steps: - name: Clone - uses: actions/checkout@v3 + uses: actions/checkout@v6 - name: Download artifacts - uses: actions/download-artifact@v3 + uses: actions/download-artifact@v4 - name: Upload coverage data - uses: codecov/codecov-action@v3 + uses: codecov/codecov-action@v6 with: - name: coverage-unittests + name: Server + token: ${{ secrets.CODECOV_TOKEN }} fail_ci_if_error: true verbose: true + flags: server diff --git a/codecov.yml b/codecov.yml new file mode 100644 index 000000000..5eee5f5b0 --- /dev/null +++ b/codecov.yml @@ -0,0 +1,31 @@ +codecov: + require_ci_to_pass: yes + +coverage: + range: 60..90 + status: + project: + default: + target: auto + threshold: 0.5% + flags: + - server + patch: + default: + target: 85% + threshold: 0% + only_pulls: true + if_ci_failed: ignore + flags: + - server + +comment: + layout: "diff, flags" + require_changes: true + show_critical_paths: true + +flags: + server: + paths: + - "**/*.py" + carryforward: true diff --git a/pyproject.toml b/pyproject.toml index a387c2907..2a16976d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,10 +24,18 @@ dependencies = [ # "ibis-framework[bigquery]", ] +[project.optional-dependencies] +test = [ + "coverage~=7.10.0", +] + [build-system] requires = ["flit_core >=3.4,<4"] build-backend = "flit_core.buildapi" +[tool.bench.dev-dependencies] +coverage = "~=7.10.0" + [tool.ruff] line-length = 110 From 644e2d2cb9001af7362fb37289ee9d8aa8cd7e09 Mon Sep 17 00:00:00 2001 From: Saqib Ansari Date: Sun, 10 May 2026 16:06:13 +0530 Subject: [PATCH 3/6] ci: run server tests on frappe develop --- .github/workflows/server-tests.yml | 35 ++++++++++++++++++++---------- 1 file changed, 23 insertions(+), 12 deletions(-) diff --git a/.github/workflows/server-tests.yml b/.github/workflows/server-tests.yml index 666d48647..582382d1c 100644 --- a/.github/workflows/server-tests.yml +++ b/.github/workflows/server-tests.yml @@ -20,6 +20,12 @@ jobs: strategy: fail-fast: false name: Server + env: + BENCH_PATH: /home/runner/frappe-bench + SITE_NAME: test_site + DB_ROOT_PASSWORD: root + ADMIN_PASSWORD: admin + FRAPPE_BRANCH: develop services: mariadb: @@ -45,12 +51,12 @@ jobs: - name: Setup Python uses: actions/setup-python@v6 with: - python-version: "3.10" + python-version: "3.14" - name: Setup Node uses: actions/setup-node@v5 with: - node-version: 18 + node-version: 24 check-latest: true - name: Cache pip @@ -77,26 +83,31 @@ jobs: - name: Setup run: | pip install frappe-bench - bench init --skip-redis-config-generation --skip-assets --python "$(which python)" ~/frappe-bench - mysql --host 127.0.0.1 --port 3306 -u root -proot -e "SET GLOBAL character_set_server = 'utf8mb4'" - mysql --host 127.0.0.1 --port 3306 -u root -proot -e "SET GLOBAL collation_server = 'utf8mb4_unicode_ci'" + bench init \ + --frappe-branch "$FRAPPE_BRANCH" \ + --skip-redis-config-generation \ + --skip-assets \ + --python "$(which python)" \ + "$BENCH_PATH" + mysql --host 127.0.0.1 --port 3306 -u root -p"$DB_ROOT_PASSWORD" -e "SET GLOBAL character_set_server = 'utf8mb4'" + mysql --host 127.0.0.1 --port 3306 -u root -p"$DB_ROOT_PASSWORD" -e "SET GLOBAL collation_server = 'utf8mb4_unicode_ci'" - name: Install - working-directory: /home/runner/frappe-bench + working-directory: ${{ env.BENCH_PATH }} run: | bench get-app insights $GITHUB_WORKSPACE bench setup requirements --dev - bench new-site --db-root-password root --admin-password admin test_site - bench --site test_site install-app insights + bench new-site --db-root-password "$DB_ROOT_PASSWORD" --admin-password "$ADMIN_PASSWORD" "$SITE_NAME" + bench --site "$SITE_NAME" install-app insights bench build env: CI: "Yes" - name: Run Tests - working-directory: /home/runner/frappe-bench + working-directory: ${{ env.BENCH_PATH }} run: | - bench --site test_site set-config allow_tests true - env/bin/python apps/insights/.github/helpers/run_backend_tests.py --site test_site + bench --site "$SITE_NAME" set-config allow_tests true + env/bin/python apps/insights/.github/helpers/run_backend_tests.py --site "$SITE_NAME" env: TYPE: server @@ -104,7 +115,7 @@ jobs: uses: actions/upload-artifact@v4 with: name: coverage-server - path: /home/runner/frappe-bench/sites/coverage.xml + path: ${{ env.BENCH_PATH }}/sites/coverage.xml coverage: name: Coverage Wrap Up From 33daf3040e4a89bcee8ba7a96450cc528700c118 Mon Sep 17 00:00:00 2001 From: Saqib Ansari Date: Sun, 10 May 2026 16:16:35 +0530 Subject: [PATCH 4/6] chore: align server tests with frappe develop setup --- .github/workflows/server-tests.yml | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/.github/workflows/server-tests.yml b/.github/workflows/server-tests.yml index 582382d1c..313cb15d9 100644 --- a/.github/workflows/server-tests.yml +++ b/.github/workflows/server-tests.yml @@ -80,12 +80,16 @@ jobs: restore-keys: | ${{ runner.os }}-yarn- + - name: Install system dependencies + run: | + sudo apt -qq update + sudo apt -qq install -y redis-server + - name: Setup run: | pip install frappe-bench bench init \ --frappe-branch "$FRAPPE_BRANCH" \ - --skip-redis-config-generation \ --skip-assets \ --python "$(which python)" \ "$BENCH_PATH" @@ -95,18 +99,27 @@ jobs: - name: Install working-directory: ${{ env.BENCH_PATH }} run: | - bench get-app insights $GITHUB_WORKSPACE + bench get-app insights $GITHUB_WORKSPACE --skip-assets bench setup requirements --dev bench new-site --db-root-password "$DB_ROOT_PASSWORD" --admin-password "$ADMIN_PASSWORD" "$SITE_NAME" + bench --site "$SITE_NAME" set-config server_script_enabled 1 --parse bench --site "$SITE_NAME" install-app insights bench build env: CI: "Yes" + - name: Start Bench Services + working-directory: ${{ env.BENCH_PATH }} + run: | + sed -i '/^watch:/d' Procfile + sed -i '/^schedule:/d' Procfile + sed -i '/^socketio:/d' Procfile + bench start &> bench_start.log & + - name: Run Tests working-directory: ${{ env.BENCH_PATH }} run: | - bench --site "$SITE_NAME" set-config allow_tests true + bench --site "$SITE_NAME" set-config allow_tests 1 --parse env/bin/python apps/insights/.github/helpers/run_backend_tests.py --site "$SITE_NAME" env: TYPE: server From 8f7c4782401b23214dfc9fd21ce0aba669280f71 Mon Sep 17 00:00:00 2001 From: Saqib Ansari Date: Mon, 11 May 2026 18:26:18 +0530 Subject: [PATCH 5/6] chore: add workbook tests --- insights/api/__init__.py | 2 +- insights/api/workbooks.py | 19 +- insights/tests/workbook/helpers.py | 135 ++++++++ insights/tests/workbook/test_workbook.py | 390 +++++++++++++++++++++++ 4 files changed, 536 insertions(+), 10 deletions(-) create mode 100644 insights/tests/workbook/helpers.py create mode 100644 insights/tests/workbook/test_workbook.py diff --git a/insights/api/__init__.py b/insights/api/__init__.py index af078b083..3f4dce255 100644 --- a/insights/api/__init__.py +++ b/insights/api/__init__.py @@ -214,7 +214,7 @@ def _execute_doc_method(doc, method: str, args: dict | None = None, ignore_permi is_whitelisted(fn) is_valid_http_method(fn) - new_kwargs = frappe.get_newargs(fn, args) + new_kwargs = frappe.get_newargs(fn, args or {}) response = doc.run_method(method, **new_kwargs) frappe.response.docs.append(doc) frappe.response["message"] = response diff --git a/insights/api/workbooks.py b/insights/api/workbooks.py index 2a735d56d..5df6e29c3 100644 --- a/insights/api/workbooks.py +++ b/insights/api/workbooks.py @@ -1,5 +1,5 @@ import frappe -from ibis import _ +from frappe import _ from insights.decorators import insights_whitelist from insights.utils import DocShare @@ -76,7 +76,7 @@ def import_workbook(workbook: dict): @insights_whitelist() def get_share_permissions(workbook_name: str): if not frappe.has_permission("Insights Workbook", ptype="share", doc=workbook_name): - frappe.throw(_("You do not have permission to share this workbook")) + frappe.throw(_("You do not have permission to share this workbook"), frappe.PermissionError) DocShare = frappe.qb.DocType("DocShare") User = frappe.qb.DocType("User") @@ -94,6 +94,7 @@ def get_share_permissions(workbook_name: str): ) .where(DocShare.share_doctype == "Insights Workbook") .where(DocShare.share_name == workbook_name) + .where(DocShare.everyone == 0) .run(as_dict=True) ) owner = frappe.db.get_value("Insights Workbook", workbook_name, "owner") @@ -131,7 +132,7 @@ def update_share_permissions( workbook_name: str, user_permissions: dict, organization_access: str | None = None ): if not frappe.has_permission("Insights Workbook", ptype="share", doc=workbook_name): - frappe.throw(_("You do not have permission to share this workbook")) + frappe.throw(_("You do not have permission to share this workbook"), frappe.PermissionError) existing_shares = frappe.get_all( "DocShare", @@ -179,7 +180,7 @@ def update_share_permissions( def create_folder(workbook: str, title: str, folder_type: str): """Create a new folder in workbook""" if not frappe.has_permission("Insights Workbook", ptype="write", doc=workbook): - frappe.throw(_("You do not have permission to modify this workbook")) + frappe.throw(_("You do not have permission to modify this workbook"), frappe.PermissionError) current_folders = frappe.db.count("Insights Folder", filters={"workbook": workbook, "type": folder_type}) @@ -198,7 +199,7 @@ def rename_folder(folder_name: str, new_title: str): """Rename a folder""" folder = frappe.get_doc("Insights Folder", folder_name) if not frappe.has_permission("Insights Workbook", ptype="write", doc=folder.workbook): - frappe.throw(_("You do not have permission to modify this workbook")) + frappe.throw(_("You do not have permission to modify this workbook"), frappe.PermissionError) folder.title = new_title folder.save() @@ -211,7 +212,7 @@ def delete_folder(folder_name: str, move_items_to_root: bool = True): """Delete folder and move items to root""" folder = frappe.get_doc("Insights Folder", folder_name) if not frappe.has_permission("Insights Workbook", ptype="write", doc=folder.workbook): - frappe.throw(_("You do not have permission to modify this workbook")) + frappe.throw(_("You do not have permission to modify this workbook"), frappe.PermissionError) if move_items_to_root: # move all queries to root @@ -239,7 +240,7 @@ def toggle_folder_expanded(folder_name: str, is_expanded: bool): """Toggle folder expanded state""" folder = frappe.get_doc("Insights Folder", folder_name) if not frappe.has_permission("Insights Workbook", ptype="read", doc=folder.workbook): - frappe.throw(_("You do not have permission to modify this workbook")) + frappe.throw(_("You do not have permission to modify this workbook"), frappe.PermissionError) folder.db_set("is_expanded", is_expanded, update_modified=False) @@ -251,7 +252,7 @@ def move_item_to_folder(item_type: str, item_name: str, folder_name: str | None item = frappe.get_doc(doctype, item_name) if not frappe.has_permission("Insights Workbook", ptype="write", doc=item.workbook): - frappe.throw(_("You do not have permission to modify this workbook")) + frappe.throw(_("You do not have permission to modify this workbook"), frappe.PermissionError) if folder_name: folder = frappe.get_doc("Insights Folder", folder_name) @@ -265,7 +266,7 @@ def move_item_to_folder(item_type: str, item_name: str, folder_name: str | None def update_sort_orders(workbook: str, items: list): """Bulk update sort orders""" if not frappe.has_permission("Insights Workbook", ptype="write", doc=workbook): - frappe.throw(_("You do not have permission to modify this workbook")) + frappe.throw(_("You do not have permission to modify this workbook"), frappe.PermissionError) for item in items: if item["type"] == "folder": diff --git a/insights/tests/workbook/helpers.py b/insights/tests/workbook/helpers.py new file mode 100644 index 000000000..18296557b --- /dev/null +++ b/insights/tests/workbook/helpers.py @@ -0,0 +1,135 @@ +import frappe + +from insights.insights.doctype.insights_data_source_v3.insights_data_source_v3 import db_connections + + +class DT: + WORKBOOK = "Insights Workbook" + QUERY = "Insights Query v3" + CHART = "Insights Chart v3" + DASHBOARD = "Insights Dashboard v3" + USER = "User" + + +TEST_USER_EMAIL = "workbook_flow_user@test.com" +TEST_WORKBOOK_TITLE = "Workbook Flow Test Workbook" +TEST_QUERY_TITLE = "Workbook Flow Test Query" +TEST_CHART_TITLE = "Workbook Flow Test Chart" +TEST_DASHBOARD_TITLE = "Workbook Flow Test Dashboard" + + +def create_test_user(email=TEST_USER_EMAIL, role="Insights User"): + if frappe.db.exists(DT.USER, email): + user = frappe.get_doc(DT.USER, email) + else: + user = frappe.get_doc( + { + "doctype": DT.USER, + "email": email, + "first_name": "Workbook", + "last_name": "Flow User", + "send_welcome_email": 0, + "user_type": "System User", + "enabled": 1, + } + ).insert(ignore_permissions=True) + + if role and not frappe.db.exists("Has Role", {"parent": email, "role": role}): + user.add_roles(role) + + return user + + +def delete_test_users(): + delete_doc_if_exists(DT.USER, TEST_USER_EMAIL) + + +def create_test_workbook(owner, title=TEST_WORKBOOK_TITLE): + frappe.set_user(owner) + return frappe.get_doc({"doctype": DT.WORKBOOK, "title": title}).insert() + + +def create_test_query(owner, workbook, title=TEST_QUERY_TITLE, operations=None): + frappe.set_user(owner) + return frappe.get_doc( + { + "doctype": DT.QUERY, + "title": title, + "workbook": workbook, + "use_live_connection": 1, + "is_builder_query": 1, + "operations": operations + or [ + { + "type": "source", + "table": { + "type": "table", + "data_source": "Site DB", + "table_name": "tabToDo", + }, + } + ], + } + ).insert() + + +def create_test_chart(owner, workbook, query, title=TEST_CHART_TITLE): + frappe.set_user(owner) + chart = frappe.get_doc( + { + "doctype": DT.CHART, + "title": title, + "workbook": workbook, + "query": query, + "chart_type": "Bar", + "config": {}, + } + ).insert() + return frappe.get_doc(DT.CHART, chart.name) + + +def create_test_dashboard(owner, workbook, chart, title=TEST_DASHBOARD_TITLE): + frappe.set_user(owner) + dashboard = frappe.get_doc( + { + "doctype": DT.DASHBOARD, + "title": title, + "workbook": workbook, + "items": [{"id": "chart-1", "type": "chart", "chart": chart}], + } + ).insert() + return frappe.get_doc(DT.DASHBOARD, dashboard.name) + + +def execute_test_query(query_name): + query = frappe.get_doc(DT.QUERY, query_name) + with db_connections(): + return query.execute() + + +def doc_exists(doctype, name): + return bool(frappe.db.exists(doctype, name)) + + +def is_visible(doctype, name): + return bool(frappe.get_list(doctype, filters={"name": name}, pluck="name", limit=1)) + + +def delete_test_workbooks(): + workbooks = frappe.get_all( + DT.WORKBOOK, + filters={"title": ["like", f"{TEST_WORKBOOK_TITLE}%"]}, + pluck="name", + ) + for workbook in workbooks: + frappe.delete_doc(DT.WORKBOOK, workbook, force=True) + + +def cleanup_workbook_flow_fixtures(): + delete_test_workbooks() + delete_test_users() + + +def delete_doc_if_exists(doctype, name): + if frappe.db.exists(doctype, name): + frappe.delete_doc(doctype, name, force=True) diff --git a/insights/tests/workbook/test_workbook.py b/insights/tests/workbook/test_workbook.py new file mode 100644 index 000000000..77d3dd3d0 --- /dev/null +++ b/insights/tests/workbook/test_workbook.py @@ -0,0 +1,390 @@ +import frappe +from frappe.tests import IntegrationTestCase +from frappe.utils import set_request + +from insights.api import get_doc as get_public_doc +from insights.api import run_doc_method as run_public_doc_method +from insights.api.workbooks import ( + create_folder, + delete_folder, + get_share_permissions, + get_workbooks, + import_workbook, + move_item_to_folder, + update_share_permissions, + update_sort_orders, +) +from insights.tests.workbook.helpers import ( + DT, + TEST_USER_EMAIL, + create_test_chart, + create_test_dashboard, + create_test_query, + create_test_user, + create_test_workbook, + delete_doc_if_exists, + doc_exists, +) + +COLLABORATOR_EMAIL = "workbook_flow_collaborator@test.com" + + +def cleanup_test_workbooks(*owners): + workbooks = frappe.get_all( + DT.WORKBOOK, + filters={"owner": ["in", list(owners)]}, + pluck="name", + ) + for workbook_name in workbooks: + frappe.delete_doc(DT.WORKBOOK, workbook_name, force=True) + + +class TestWorkbook(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + frappe.set_user("Administrator") + cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) + delete_doc_if_exists(DT.USER, TEST_USER_EMAIL) + delete_doc_if_exists(DT.USER, COLLABORATOR_EMAIL) + create_test_user(TEST_USER_EMAIL) + create_test_user(COLLABORATOR_EMAIL) + frappe.db.commit() + + @classmethod + def tearDownClass(cls): + frappe.set_user("Administrator") + cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) + delete_doc_if_exists(DT.USER, TEST_USER_EMAIL) + delete_doc_if_exists(DT.USER, COLLABORATOR_EMAIL) + frappe.db.commit() + super().tearDownClass() + + def setUp(self): + super().setUp() + self.original_user = frappe.session.user + frappe.set_user("Administrator") + cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) + frappe.db.commit() + + def tearDown(self): + frappe.set_user("Administrator") + cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) + frappe.db.commit() + frappe.set_user(self.original_user) + super().tearDown() + + def get_doc(self, doctype, name): + set_request(method="GET", path="/api/method/insights.api.get_doc") + return get_public_doc(doctype, name) + + def run_doc_method(self, method, doc, args=None): + set_request(method="POST", path="/api/method/insights.api.run_doc_method") + return run_public_doc_method(method, doc, args=args) + + def get_workbook(self, name): + workbook = self.get_doc(DT.WORKBOOK, name) + for field in ("folders", "queries", "charts", "dashboards"): + workbook[field] = frappe.parse_json(workbook.get(field)) or [] + return workbook + + def create_workbook_bundle(self, title, include_secondary_items=False, include_folders=False): + workbook = create_test_workbook(TEST_USER_EMAIL, title=title) + query = create_test_query(TEST_USER_EMAIL, workbook.name, title=f"{title} Query 1") + chart = create_test_chart(TEST_USER_EMAIL, workbook.name, query.name, title=f"{title} Chart 1") + dashboard = create_test_dashboard( + TEST_USER_EMAIL, + workbook.name, + chart.name, + title=f"{title} Dashboard 1", + ) + + bundle = { + "workbook": workbook, + "query": query, + "chart": chart, + "dashboard": dashboard, + "folders": {}, + } + + if include_secondary_items: + bundle["secondary_query"] = create_test_query( + TEST_USER_EMAIL, + workbook.name, + title=f"{title} Query 2", + ) + bundle["secondary_chart"] = create_test_chart( + TEST_USER_EMAIL, + workbook.name, + bundle["secondary_query"].name, + title=f"{title} Chart 2", + ) + + if include_folders: + frappe.set_user(TEST_USER_EMAIL) + bundle["folders"]["query"] = create_folder( + workbook.name, + f"{title} Query Folder", + "query", + ) + bundle["folders"]["chart"] = create_folder( + workbook.name, + f"{title} Chart Folder", + "chart", + ) + move_item_to_folder("query", query.name, bundle["folders"]["query"]) + move_item_to_folder("chart", chart.name, bundle["folders"]["chart"]) + + return bundle + + def test_owner_can_build_workbook_query_chart_dashboard_flow(self): + bundle = self.create_workbook_bundle("Workbook Flow Test Authoring") + + frappe.set_user(TEST_USER_EMAIL) + workbook = self.get_workbook(bundle["workbook"].name) + chart = self.get_doc(DT.CHART, bundle["chart"].name) + dashboard = self.get_doc(DT.DASHBOARD, bundle["dashboard"].name) + query_result = self.run_doc_method("execute", self.get_doc(DT.QUERY, bundle["query"].name)) + dashboard_items = frappe.parse_json(dashboard["items"]) or [] + + self.assertEqual([row["name"] for row in workbook["queries"]], [bundle["query"].name]) + self.assertEqual([row["name"] for row in workbook["charts"]], [bundle["chart"].name]) + self.assertEqual([row["name"] for row in workbook["dashboards"]], [bundle["dashboard"].name]) + + self.assertIn("sql", query_result) + self.assertGreater(len(query_result["columns"]), 0) + self.assertEqual(chart["query"], bundle["query"].name) + self.assertTrue(chart["data_query"]) + self.assertTrue(doc_exists(DT.QUERY, chart["data_query"])) + self.assertTrue(any(item.get("chart") == bundle["chart"].name for item in dashboard_items)) + + def test_deleting_workbook_removes_the_user_visible_tree(self): + bundle = self.create_workbook_bundle("Workbook Flow Test Delete") + data_query_name = self.get_doc(DT.CHART, bundle["chart"].name)["data_query"] + + frappe.set_user("Administrator") + frappe.delete_doc(DT.WORKBOOK, bundle["workbook"].name, force=True) + + self.assertFalse(doc_exists(DT.WORKBOOK, bundle["workbook"].name)) + self.assertFalse(doc_exists(DT.QUERY, bundle["query"].name)) + self.assertFalse(doc_exists(DT.CHART, bundle["chart"].name)) + self.assertFalse(doc_exists(DT.QUERY, data_query_name)) + self.assertFalse(doc_exists(DT.DASHBOARD, bundle["dashboard"].name)) + + def test_owner_can_organize_and_reorder_workbook_contents(self): + bundle = self.create_workbook_bundle( + "Workbook Flow Test Organization", + include_secondary_items=True, + ) + + frappe.set_user(TEST_USER_EMAIL) + query_folder = create_folder( + bundle["workbook"].name, + "Workbook Flow Test Organization Query Folder", + "query", + ) + chart_folder = create_folder( + bundle["workbook"].name, + "Workbook Flow Test Organization Chart Folder", + "chart", + ) + move_item_to_folder("query", bundle["query"].name, query_folder) + move_item_to_folder("chart", bundle["chart"].name, chart_folder) + update_sort_orders( + bundle["workbook"].name, + [ + {"type": "folder", "name": chart_folder, "sort_order": 0}, + {"type": "folder", "name": query_folder, "sort_order": 1}, + { + "type": "query", + "name": bundle["secondary_query"].name, + "sort_order": 0, + "folder": None, + }, + { + "type": "query", + "name": bundle["query"].name, + "sort_order": 1, + "folder": query_folder, + }, + { + "type": "chart", + "name": bundle["secondary_chart"].name, + "sort_order": 0, + "folder": None, + }, + { + "type": "chart", + "name": bundle["chart"].name, + "sort_order": 1, + "folder": chart_folder, + }, + ], + ) + + workbook = self.get_workbook(bundle["workbook"].name) + + self.assertEqual([row["name"] for row in workbook["folders"]], [chart_folder, query_folder]) + self.assertEqual( + [row["name"] for row in workbook["queries"]], + [bundle["secondary_query"].name, bundle["query"].name], + ) + self.assertEqual( + [row["name"] for row in workbook["charts"]], + [bundle["secondary_chart"].name, bundle["chart"].name], + ) + self.assertEqual(workbook["queries"][1]["folder"], query_folder) + self.assertEqual(workbook["charts"][1]["folder"], chart_folder) + + delete_folder(query_folder, move_items_to_root=True) + delete_folder(chart_folder, move_items_to_root=True) + + workbook = self.get_workbook(bundle["workbook"].name) + + self.assertEqual(workbook["folders"], []) + self.assertTrue(all(not row["folder"] for row in workbook["queries"])) + self.assertTrue(all(not row["folder"] for row in workbook["charts"])) + + def test_duplicate_workbook_preserves_a_usable_copy(self): + bundle = self.create_workbook_bundle( + "Workbook Flow Test Duplicate", + include_folders=True, + ) + + frappe.set_user(TEST_USER_EMAIL) + original_workbook = self.get_workbook(bundle["workbook"].name) + original_chart = self.get_doc(DT.CHART, bundle["chart"].name) + duplicate_name = self.run_doc_method( + "duplicate", + self.get_doc(DT.WORKBOOK, bundle["workbook"].name), + ) + duplicate_workbook = self.get_workbook(duplicate_name) + duplicate_query_name = duplicate_workbook["queries"][0]["name"] + duplicate_chart_name = duplicate_workbook["charts"][0]["name"] + duplicate_chart = self.get_doc(DT.CHART, duplicate_chart_name) + duplicate_dashboard = self.get_doc( + DT.DASHBOARD, + duplicate_workbook["dashboards"][0]["name"], + ) + duplicate_dashboard_items = frappe.parse_json(duplicate_dashboard["items"]) or [] + duplicate_result = self.run_doc_method("execute", self.get_doc(DT.QUERY, duplicate_query_name)) + + self.assertEqual(len(duplicate_workbook["folders"]), 2) + self.assertEqual(len(duplicate_workbook["queries"]), 1) + self.assertEqual(len(duplicate_workbook["charts"]), 1) + self.assertEqual(len(duplicate_workbook["dashboards"]), 1) + self.assertNotEqual(duplicate_query_name, bundle["query"].name) + self.assertEqual(duplicate_chart["query"], duplicate_query_name) + self.assertTrue(duplicate_chart["data_query"]) + self.assertNotEqual(duplicate_chart["data_query"], original_chart["data_query"]) + self.assertTrue(any(item.get("chart") == duplicate_chart_name for item in duplicate_dashboard_items)) + self.assertTrue( + {row["name"] for row in duplicate_workbook["folders"]}.isdisjoint( + {row["name"] for row in original_workbook["folders"]} + ) + ) + self.assertIn( + duplicate_workbook["queries"][0]["folder"], {row["name"] for row in duplicate_workbook["folders"]} + ) + self.assertIn( + duplicate_workbook["charts"][0]["folder"], {row["name"] for row in duplicate_workbook["folders"]} + ) + self.assertGreater(len(duplicate_result["columns"]), 0) + + def test_export_and_import_preserve_a_usable_workflow(self): + bundle = self.create_workbook_bundle( + "Workbook Flow Test Import", + include_folders=True, + ) + + frappe.set_user(TEST_USER_EMAIL) + original_workbook = self.get_workbook(bundle["workbook"].name) + original_chart = self.get_doc(DT.CHART, bundle["chart"].name) + exported_workbook = self.run_doc_method( + "export", + self.get_doc(DT.WORKBOOK, bundle["workbook"].name), + ) + imported_name = import_workbook(exported_workbook) + imported_workbook = self.get_workbook(imported_name) + imported_query_name = imported_workbook["queries"][0]["name"] + imported_chart_name = imported_workbook["charts"][0]["name"] + imported_chart = self.get_doc(DT.CHART, imported_chart_name) + imported_dashboard = self.get_doc( + DT.DASHBOARD, + imported_workbook["dashboards"][0]["name"], + ) + imported_dashboard_items = frappe.parse_json(imported_dashboard["items"]) or [] + imported_result = self.run_doc_method("execute", self.get_doc(DT.QUERY, imported_query_name)) + + self.assertEqual(len(imported_workbook["folders"]), 2) + self.assertEqual(len(imported_workbook["queries"]), 1) + self.assertEqual(len(imported_workbook["charts"]), 1) + self.assertEqual(len(imported_workbook["dashboards"]), 1) + self.assertNotEqual(imported_query_name, bundle["query"].name) + self.assertEqual(imported_chart["query"], imported_query_name) + self.assertTrue(imported_chart["data_query"]) + self.assertNotEqual(imported_chart["data_query"], original_chart["data_query"]) + self.assertTrue(any(item.get("chart") == imported_chart_name for item in imported_dashboard_items)) + self.assertTrue( + {row["name"] for row in imported_workbook["folders"]}.isdisjoint( + {row["name"] for row in original_workbook["folders"]} + ) + ) + self.assertIn( + imported_workbook["queries"][0]["folder"], {row["name"] for row in imported_workbook["folders"]} + ) + self.assertIn( + imported_workbook["charts"][0]["folder"], {row["name"] for row in imported_workbook["folders"]} + ) + self.assertGreater(len(imported_result["columns"]), 0) + + def test_shared_workbook_supports_read_only_public_access_but_blocks_structure_changes(self): + bundle = self.create_workbook_bundle("Workbook Flow Test Shared") + + frappe.set_user(TEST_USER_EMAIL) + self.run_doc_method("track_view", self.get_doc(DT.WORKBOOK, bundle["workbook"].name)) + update_share_permissions(bundle["workbook"].name, [], organization_access="view") + share_permissions = get_share_permissions(bundle["workbook"].name) + owner_workbooks = get_workbooks(search_term="Workbook Flow Test Shared") + + self.assertEqual(share_permissions["organization_access"], "view") + self.assertEqual( + {permission["user"] for permission in share_permissions["user_permissions"]}, + {TEST_USER_EMAIL}, + ) + self.assertEqual(len(owner_workbooks), 1) + self.assertEqual(owner_workbooks[0]["name"], bundle["workbook"].name) + self.assertEqual(owner_workbooks[0]["views"], 1) + self.assertTrue(owner_workbooks[0]["shared_with_organization"]) + + frappe.set_user(COLLABORATOR_EMAIL) + collaborator_workbooks = get_workbooks(search_term="Workbook Flow Test Shared") + workbook = self.get_workbook(bundle["workbook"].name) + query = self.get_doc(DT.QUERY, bundle["query"].name) + chart = self.get_doc(DT.CHART, bundle["chart"].name) + dashboard = self.get_doc(DT.DASHBOARD, bundle["dashboard"].name) + + self.assertEqual([row["name"] for row in collaborator_workbooks], [bundle["workbook"].name]) + self.assertTrue(workbook["read_only"]) + self.assertTrue(query["read_only"]) + self.assertTrue(chart["read_only"]) + self.assertTrue(dashboard["read_only"]) + self.assertEqual([row["name"] for row in workbook["queries"]], [bundle["query"].name]) + self.assertEqual([row["name"] for row in workbook["charts"]], [bundle["chart"].name]) + self.assertEqual([row["name"] for row in workbook["dashboards"]], [bundle["dashboard"].name]) + + with self.assertRaises(frappe.PermissionError): + create_folder(bundle["workbook"].name, "Blocked Query Folder", "query") + + with self.assertRaises(frappe.PermissionError): + update_sort_orders( + bundle["workbook"].name, + [ + { + "type": "query", + "name": bundle["query"].name, + "sort_order": 0, + "folder": None, + } + ], + ) From 69338d6c763dd3071f007f1f0dd6ba5dbcc05cfd Mon Sep 17 00:00:00 2001 From: Saqib Ansari Date: Mon, 11 May 2026 18:52:40 +0530 Subject: [PATCH 6/6] chore: add query tests & reduce boilerplate code --- .../insights_data_source_v3/ibis_utils.py | 8 +- insights/tests/base.py | 83 +++ insights/tests/factories.py | 190 ++++++ insights/tests/permissions_utils.py | 146 ++++ insights/tests/test_permissions.py | 638 +++++------------- insights/tests/workbook/helpers.py | 135 ---- insights/tests/workbook/test_querying.py | 383 +++++++++++ insights/tests/workbook/test_workbook.py | 388 +++++------ insights/tests/workbook_utils.py | 86 +++ 9 files changed, 1214 insertions(+), 843 deletions(-) create mode 100644 insights/tests/base.py create mode 100644 insights/tests/factories.py create mode 100644 insights/tests/permissions_utils.py delete mode 100644 insights/tests/workbook/helpers.py create mode 100644 insights/tests/workbook/test_querying.py create mode 100644 insights/tests/workbook_utils.py diff --git a/insights/insights/doctype/insights_data_source_v3/ibis_utils.py b/insights/insights/doctype/insights_data_source_v3/ibis_utils.py index 857052b27..10e7ddcf2 100644 --- a/insights/insights/doctype/insights_data_source_v3/ibis_utils.py +++ b/insights/insights/doctype/insights_data_source_v3/ibis_utils.py @@ -61,6 +61,7 @@ def __init__(self, doc, active_operation_idx=None): def set_operations(self): operations = frappe.parse_json(self.operations) + adhoc_filters_by_query = getattr(frappe.local, "insights_adhoc_filters", None) or {} if ( self.active_operation_idx is not None @@ -69,11 +70,8 @@ def set_operations(self): ): operations = operations[: self.active_operation_idx + 1] - if ( - hasattr(frappe.local, "insights_adhoc_filters") - and self.doc.name in frappe.local.insights_adhoc_filters - ): - adhoc_filters = frappe.local.insights_adhoc_filters[self.doc.name] + if self.doc.name in adhoc_filters_by_query: + adhoc_filters = adhoc_filters_by_query[self.doc.name] if ( adhoc_filters and isinstance(adhoc_filters, dict) diff --git a/insights/tests/base.py b/insights/tests/base.py new file mode 100644 index 000000000..0623baa03 --- /dev/null +++ b/insights/tests/base.py @@ -0,0 +1,83 @@ +import frappe +from frappe.tests import IntegrationTestCase + +from insights.tests.factories import as_user, is_visible + + +class InsightsIntegrationTestCase(IntegrationTestCase): + SAVEPOINT = None + COMMIT_AFTER_CLASS_SETUP = True + COMMIT_AFTER_CLASS_TEARDOWN = True + COMMIT_AFTER_TEST_SETUP = False + COMMIT_AFTER_TEST_TEARDOWN = False + + @classmethod + def before_class(cls): + pass + + @classmethod + def after_class(cls): + pass + + def before_test(self): + pass + + def after_test(self): + pass + + @classmethod + def setUpClass(cls): + super().setUpClass() + with as_user("Administrator"): + cls.before_class() + if cls.COMMIT_AFTER_CLASS_SETUP: + frappe.db.commit() + + @classmethod + def tearDownClass(cls): + try: + with as_user("Administrator"): + cls.after_class() + if cls.COMMIT_AFTER_CLASS_TEARDOWN: + frappe.db.commit() + finally: + super().tearDownClass() + + def setUp(self): + super().setUp() + self.original_user = frappe.session.user + self.addCleanup(frappe.set_user, self.original_user) + frappe.set_user("Administrator") + self.before_test() + if self.SAVEPOINT: + frappe.db.savepoint(self.SAVEPOINT) + elif self.COMMIT_AFTER_TEST_SETUP: + frappe.db.commit() + + def tearDown(self): + try: + frappe.set_user("Administrator") + if self.SAVEPOINT: + frappe.db.rollback(save_point=self.SAVEPOINT) + self.after_test() + if self.COMMIT_AFTER_TEST_TEARDOWN and not self.SAVEPOINT: + frappe.db.commit() + finally: + super().tearDown() + + def as_user(self, user): + return as_user(user) + + def assert_visible_to(self, user, doctype, name, message=None): + with self.as_user(user): + self.assertTrue( + is_visible(doctype, name), + message or f"{doctype} {name} should be visible to {user}", + ) + + def assert_not_visible_to(self, user, doctype, name, message=None): + with self.as_user(user): + self.assertFalse( + is_visible(doctype, name), + message or f"{doctype} {name} should not be visible to {user}", + ) diff --git a/insights/tests/factories.py b/insights/tests/factories.py new file mode 100644 index 000000000..260e5222d --- /dev/null +++ b/insights/tests/factories.py @@ -0,0 +1,190 @@ +from contextlib import contextmanager + +import frappe + +from insights.insights.doctype.insights_data_source_v3.insights_data_source_v3 import db_connections + + +class DT: + DATA_SOURCE = "Insights Data Source v3" + TABLE = "Insights Table v3" + WORKBOOK = "Insights Workbook" + QUERY = "Insights Query v3" + CHART = "Insights Chart v3" + DASHBOARD = "Insights Dashboard v3" + TEAM = "Insights Team" + SETTINGS = "Insights Settings" + USER = "User" + + +USER_1 = "workbook_flow_user@test.com" +TEST_WORKBOOK_TITLE = "Workbook Flow Test Workbook" +TEST_QUERY_TITLE = "Workbook Flow Test Query" +TEST_CHART_TITLE = "Workbook Flow Test Chart" +TEST_DASHBOARD_TITLE = "Workbook Flow Test Dashboard" + + +@contextmanager +def as_user(user): + original_user = frappe.session.user + frappe.set_user(user) + try: + yield + finally: + frappe.set_user(original_user) + + +def create_user( + email, + first_name="Test", + last_name="User", + roles=None, + user_type="System User", + **kwargs, +): + if frappe.db.exists(DT.USER, email): + user = frappe.get_doc(DT.USER, email) + else: + user = frappe.get_doc( + { + "doctype": DT.USER, + "email": email, + "first_name": first_name, + "last_name": last_name, + "send_welcome_email": 0, + "user_type": user_type, + "enabled": 1, + **kwargs, + } + ).insert(ignore_permissions=True) + + if roles and not isinstance(roles, list | tuple | set): + roles = [roles] + + for role in roles or []: + if not frappe.db.exists("Has Role", {"parent": email, "role": role}): + user.add_roles(role) + + return frappe.get_doc(DT.USER, email) + + +def create_test_user(email=USER_1, role="Insights User"): + return create_user( + email, + first_name="Workbook", + last_name="Flow User", + roles=role, + ) + + +def delete_doc_if_exists(doctype, name): + if frappe.db.exists(doctype, name): + frappe.delete_doc(doctype, name, force=True) + + +def delete_users(*emails): + for email in emails: + delete_doc_if_exists(DT.USER, email) + + +def delete_test_users(): + delete_users(USER_1) + + +def create_test_workbook(owner, title=TEST_WORKBOOK_TITLE): + with as_user(owner): + return frappe.get_doc({"doctype": DT.WORKBOOK, "title": title}).insert() + + +def create_test_query(owner, workbook, title=TEST_QUERY_TITLE, operations=None): + with as_user(owner): + return frappe.get_doc( + { + "doctype": DT.QUERY, + "title": title, + "workbook": workbook, + "use_live_connection": 1, + "is_builder_query": 1, + "operations": operations + or [ + { + "type": "source", + "table": { + "type": "table", + "data_source": "Site DB", + "table_name": "tabToDo", + }, + } + ], + } + ).insert() + + +def create_test_chart(owner, workbook, query=None, title=TEST_CHART_TITLE): + with as_user(owner): + chart = frappe.get_doc( + { + "doctype": DT.CHART, + "title": title, + "workbook": workbook, + "query": query, + "chart_type": "Bar", + "config": {}, + } + ).insert() + return frappe.get_doc(DT.CHART, chart.name) + + +def create_test_dashboard(owner, workbook, chart=None, title=TEST_DASHBOARD_TITLE): + with as_user(owner): + items = [] + if chart: + items.append({"id": "chart-1", "type": "chart", "chart": chart}) + + dashboard = frappe.get_doc( + { + "doctype": DT.DASHBOARD, + "title": title, + "workbook": workbook, + "items": items, + } + ).insert() + return frappe.get_doc(DT.DASHBOARD, dashboard.name) + + +def execute_test_query(query_name): + query = frappe.get_doc(DT.QUERY, query_name) + with db_connections(): + return query.execute() + + +def doc_exists(doctype, name): + return bool(frappe.db.exists(doctype, name)) + + +def is_visible(doctype, name): + return bool(frappe.get_list(doctype, filters={"name": name}, pluck="name", limit=1)) + + +def delete_workbooks(title_prefix=None, owners=None): + if not title_prefix and not owners: + raise ValueError("delete_workbooks requires a title_prefix or owners") + + filters = {} + if title_prefix: + filters["title"] = ["like", f"{title_prefix}%"] + if owners: + filters["owner"] = ["in", list(owners)] + + workbooks = frappe.get_all(DT.WORKBOOK, filters=filters, pluck="name") + for workbook in workbooks: + frappe.delete_doc(DT.WORKBOOK, workbook, force=True) + + +def delete_test_workbooks(): + delete_workbooks(title_prefix=TEST_WORKBOOK_TITLE) + + +def cleanup_workbook_flow_fixtures(): + delete_test_workbooks() + delete_test_users() diff --git a/insights/tests/permissions_utils.py b/insights/tests/permissions_utils.py new file mode 100644 index 000000000..9b302f0da --- /dev/null +++ b/insights/tests/permissions_utils.py @@ -0,0 +1,146 @@ +import frappe +import frappe.share + +from insights.insights.doctype.insights_table_v3.insights_table_v3 import get_table_name +from insights.insights.doctype.insights_team.insights_team import clear_cache as clear_team_cache +from insights.tests.factories import DT, create_user, delete_doc_if_exists, delete_users, delete_workbooks + +WEB_USER_EMAIL = "web_user@test.com" +NON_INSIGHTS_USER = "non_insights_user@test.com" +USER_1 = "insights_user1@test.com" +USER_2 = "insights_user2@test.com" +ADMIN = "insights_admin@test.com" + +TEST_DS_TITLE = "Test DuckDB" +TEST_DS = frappe.scrub(TEST_DS_TITLE) +TEST_TABLE1 = get_table_name(TEST_DS, "table1") +TEST_TABLE2_NAME = get_table_name(TEST_DS, "table2") +TEST_TABLE3_NAME = get_table_name(TEST_DS, "table3") + + +def create_test_users(): + create_user( + WEB_USER_EMAIL, + first_name="Web", + last_name="User", + user_type="Website User", + ) + create_user( + NON_INSIGHTS_USER, + first_name="Non", + last_name="Insights User", + ) + create_user( + USER_1, + first_name="Insights", + last_name="User", + roles="Insights User", + ) + create_user( + USER_2, + first_name="Insights", + last_name="User", + roles="Insights User", + ) + create_user( + ADMIN, + first_name="Insights", + last_name="Admin", + roles="Insights Admin", + ) + + +def delete_test_users(): + delete_users( + WEB_USER_EMAIL, + NON_INSIGHTS_USER, + USER_1, + USER_2, + ADMIN, + ) + + +def create_test_data_sources(): + frappe.get_doc( + { + "doctype": DT.DATA_SOURCE, + "title": TEST_DS_TITLE, + "database_type": "DuckDB", + "database_name": "test_duckdb", + } + ).insert() + + +def delete_test_data_sources(): + delete_doc_if_exists(DT.DATA_SOURCE, TEST_DS) + + +def create_test_tables(): + for table_name in ("table1", "table2", "table3"): + frappe.get_doc( + { + "doctype": DT.TABLE, + "table": table_name, + "label": table_name, + "data_source": TEST_DS, + "sync_mode": "Full", + } + ).insert() + + +def delete_test_tables(): + delete_doc_if_exists(DT.TABLE, TEST_TABLE1) + delete_doc_if_exists(DT.TABLE, TEST_TABLE2_NAME) + delete_doc_if_exists(DT.TABLE, TEST_TABLE3_NAME) + + +def create_test_teams(): + team = frappe.get_doc({"doctype": DT.TEAM, "team_name": "team1"}) + team.append("team_members", {"user": USER_1}) + team.save() + clear_team_cache() + return team + + +def delete_test_teams(): + delete_doc_if_exists(DT.TEAM, "team1") + + +def update_dashboard_access(dashboard_name, people_with_access): + frappe.get_doc(DT.DASHBOARD, dashboard_name).update_access( + { + "is_public": 0, + "is_shared_with_organization": 0, + "people_with_access": people_with_access, + } + ) + + +def share_chart(chart_name, user): + frappe.share.add(DT.CHART, chart_name, user=user, read=1, notify=0) + + +def unshare_chart(chart_name, user): + share_name = frappe.db.get_value( + "DocShare", + { + "share_doctype": DT.CHART, + "share_name": chart_name, + "user": user, + }, + ) + if share_name: + frappe.delete_doc("DocShare", share_name, ignore_permissions=True) + + +def delete_test_workbooks(): + delete_workbooks(owners=[USER_1, USER_2, ADMIN]) + + +def cleanup_test_fixtures(): + delete_test_workbooks() + delete_test_teams() + delete_test_tables() + delete_test_data_sources() + delete_test_users() + clear_team_cache() diff --git a/insights/tests/test_permissions.py b/insights/tests/test_permissions.py index 7677d7c06..c027e709c 100644 --- a/insights/tests/test_permissions.py +++ b/insights/tests/test_permissions.py @@ -1,18 +1,33 @@ import frappe -import frappe.share -from frappe.tests import IntegrationTestCase from insights.api.workbooks import get_share_permissions, update_share_permissions from insights.decorators import insights_whitelist -from insights.insights.doctype.insights_table_v3.insights_table_v3 import get_table_name -from insights.insights.doctype.insights_team.insights_team import clear_cache as clear_team_cache from insights.permissions import PERMISSION_DOCTYPES - -TEST_DS_TITLE = "Test DuckDB" -TEST_DS_NAME = frappe.scrub(TEST_DS_TITLE) -TEST_TABLE1_NAME = get_table_name(TEST_DS_NAME, "table1") -TEST_TABLE2_NAME = get_table_name(TEST_DS_NAME, "table2") -TEST_TABLE3_NAME = get_table_name(TEST_DS_NAME, "table3") +from insights.tests.base import InsightsIntegrationTestCase +from insights.tests.factories import ( + DT, + create_test_chart, + create_test_dashboard, + create_test_query, + create_test_workbook, +) +from insights.tests.permissions_utils import ( + ADMIN, + NON_INSIGHTS_USER, + TEST_DS, + TEST_TABLE1, + USER_1, + USER_2, + cleanup_test_fixtures, + clear_team_cache, + create_test_data_sources, + create_test_tables, + create_test_teams, + create_test_users, + share_chart, + unshare_chart, + update_dashboard_access, +) @insights_whitelist() @@ -20,68 +35,42 @@ def protected_insights_call(): return True -class DT: - DATA_SOURCE = "Insights Data Source v3" - TABLE = "Insights Table v3" - WORKBOOK = "Insights Workbook" - QUERY = "Insights Query v3" - CHART = "Insights Chart v3" - DASHBOARD = "Insights Dashboard v3" - TEAM = "Insights Team" - SETTINGS = "Insights Settings" - - -class TestInsightsPermissions(IntegrationTestCase): +class TestInsightsPermissions(InsightsIntegrationTestCase): SAVEPOINT = "test_insights_permissions" @classmethod - def setUpClass(cls): - super().setUpClass() + def before_class(cls): cls.original_enable_permissions = frappe.db.get_single_value(DT.SETTINGS, "enable_permissions") - frappe.set_user("Administrator") cleanup_test_fixtures() create_test_users() clear_team_cache() - frappe.db.commit() @classmethod - def tearDownClass(cls): - frappe.set_user("Administrator") + def after_class(cls): frappe.db.set_single_value(DT.SETTINGS, "enable_permissions", cls.original_enable_permissions) clear_team_cache() cleanup_test_fixtures() - frappe.db.commit() - super().tearDownClass() - def setUp(self): - super().setUp() - self.original_user = frappe.session.user - frappe.set_user("Administrator") + def before_test(self): clear_team_cache() - frappe.db.savepoint(self.SAVEPOINT) - def tearDown(self): - frappe.set_user("Administrator") - frappe.db.rollback(save_point=self.SAVEPOINT) + def after_test(self): clear_team_cache() - frappe.set_user(self.original_user) - super().tearDown() def toggle_team_permissions(self, enable): frappe.db.set_single_value(DT.SETTINGS, "enable_permissions", enable) clear_team_cache() def test_permissions_for_non_insights_user(self): - frappe.set_user("non_insights_user@test.com") - - for doctype in PERMISSION_DOCTYPES: - self.assertFalse( - frappe.has_permission(doctype, ptype="read"), - f"{doctype} should not be readable without an Insights role", - ) + with self.as_user(NON_INSIGHTS_USER): + for doctype in PERMISSION_DOCTYPES: + self.assertFalse( + frappe.has_permission(doctype, ptype="read"), + f"{doctype} should not be readable without an Insights role", + ) - with self.assertRaises(frappe.PermissionError): - protected_insights_call() + with self.assertRaises(frappe.PermissionError): + protected_insights_call() def test_permissions_on_team_based_doctype_with_team_permissions_disabled(self): create_test_data_sources() @@ -89,9 +78,8 @@ def test_permissions_on_team_based_doctype_with_team_permissions_disabled(self): create_test_teams() self.toggle_team_permissions(False) - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) - self.assertTrue(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) + self.assert_visible_to(USER_2, DT.DATA_SOURCE, TEST_DS) + self.assert_visible_to(USER_2, DT.TABLE, TEST_TABLE1) def test_permission_on_team_based_doctype_with_team_permissions_enabled(self): create_test_data_sources() @@ -99,28 +87,26 @@ def test_permission_on_team_based_doctype_with_team_permissions_enabled(self): team = create_test_teams() self.toggle_team_permissions(True) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) - self.assertFalse(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) + self.assert_not_visible_to(USER_2, DT.DATA_SOURCE, TEST_DS) + self.assert_not_visible_to(USER_2, DT.TABLE, TEST_TABLE1) - frappe.set_user("Administrator") - team.append( - "team_permissions", - {"resource_type": DT.DATA_SOURCE, "resource_name": TEST_DS_NAME}, - ) - team.append( - "team_permissions", - { - "resource_type": DT.TABLE, - "resource_name": TEST_TABLE1_NAME, - }, - ) - team.save(ignore_permissions=True) - clear_team_cache() + with self.as_user("Administrator"): + team.append( + "team_permissions", + {"resource_type": DT.DATA_SOURCE, "resource_name": TEST_DS}, + ) + team.append( + "team_permissions", + { + "resource_type": DT.TABLE, + "resource_name": TEST_TABLE1, + }, + ) + team.save(ignore_permissions=True) + clear_team_cache() - frappe.set_user("insights_user1@test.com") - self.assertTrue(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) - self.assertTrue(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) + self.assert_visible_to(USER_1, DT.DATA_SOURCE, TEST_DS) + self.assert_visible_to(USER_1, DT.TABLE, TEST_TABLE1) def test_permission_for_admin_on_team_based_doctype_with_team_permissions_enabled( self, @@ -129,456 +115,176 @@ def test_permission_for_admin_on_team_based_doctype_with_team_permissions_enable create_test_tables() self.toggle_team_permissions(True) - frappe.set_user("insights_admin@test.com") - self.assertTrue(is_allowed(DT.DATA_SOURCE, TEST_DS_NAME)) - self.assertTrue(is_allowed(DT.TABLE, TEST_TABLE1_NAME)) + self.assert_visible_to(ADMIN, DT.DATA_SOURCE, TEST_DS) + self.assert_visible_to(ADMIN, DT.TABLE, TEST_TABLE1) def test_permission_for_workbook(self): - workbook = create_test_workbook("insights_user1@test.com") + workbook = create_test_workbook(USER_1) - frappe.set_user("insights_user1@test.com") - self.assertTrue(is_allowed(DT.WORKBOOK, workbook.name)) + self.assert_visible_to(USER_1, DT.WORKBOOK, workbook.name) + self.assert_not_visible_to(USER_2, DT.WORKBOOK, workbook.name) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.WORKBOOK, workbook.name)) - - frappe.set_user("insights_user1@test.com") - update_share_permissions( - workbook.name, - [{"user": "insights_user2@test.com", "read": 1, "write": 0}], - ) - share_permissions = get_share_permissions(workbook.name) + with self.as_user(USER_1): + update_share_permissions( + workbook.name, + [{"user": USER_2, "read": 1, "write": 0}], + ) + share_permissions = get_share_permissions(workbook.name) self.assertIn( - "insights_user2@test.com", + USER_2, [permission["user"] for permission in share_permissions["user_permissions"]], ) - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.WORKBOOK, workbook.name)) + self.assert_visible_to(USER_2, DT.WORKBOOK, workbook.name) - frappe.set_user("insights_user1@test.com") - update_share_permissions(workbook.name, []) + with self.as_user(USER_1): + update_share_permissions(workbook.name, []) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.WORKBOOK, workbook.name)) + self.assert_not_visible_to(USER_2, DT.WORKBOOK, workbook.name) def test_permission_for_dashboard(self): - workbook = create_test_workbook("insights_user1@test.com") - dashboard = create_test_dashboard("insights_user1@test.com", workbook.name) + workbook = create_test_workbook(USER_1) + dashboard = create_test_dashboard(USER_1, workbook.name) - frappe.set_user("insights_user1@test.com") - self.assertTrue(is_allowed(DT.DASHBOARD, dashboard.name)) + self.assert_visible_to(USER_1, DT.DASHBOARD, dashboard.name) + self.assert_not_visible_to(USER_2, DT.DASHBOARD, dashboard.name) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.DASHBOARD, dashboard.name)) + with self.as_user(USER_1): + update_dashboard_access(dashboard.name, [USER_2]) + self.assert_visible_to(USER_2, DT.DASHBOARD, dashboard.name) - frappe.set_user("insights_user1@test.com") - update_dashboard_access(dashboard.name, ["insights_user2@test.com"]) - - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.DASHBOARD, dashboard.name)) - - frappe.set_user("insights_user1@test.com") - update_dashboard_access(dashboard.name, []) - - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.DASHBOARD, dashboard.name)) - - frappe.set_user("insights_user1@test.com") - update_share_permissions( - workbook.name, - [{"user": "insights_user2@test.com", "read": 1, "write": 0}], - ) + with self.as_user(USER_1): + update_dashboard_access(dashboard.name, []) + self.assert_not_visible_to(USER_2, DT.DASHBOARD, dashboard.name) - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.DASHBOARD, dashboard.name)) - self.assertFalse(frappe.has_permission(DT.DASHBOARD, ptype="write", doc=dashboard.name)) - with self.assertRaises(frappe.PermissionError): - create_test_dashboard( - "insights_user2@test.com", + with self.as_user(USER_1): + update_share_permissions( workbook.name, - title="Permissions Test Dashboard Read Only", + [{"user": USER_2, "read": 1, "write": 0}], ) - def test_permission_for_chart(self): - workbook = create_test_workbook("insights_user1@test.com") - query = create_test_query("insights_user1@test.com", workbook.name) - chart = create_test_chart("insights_user1@test.com", workbook.name, query.name) - - frappe.set_user("insights_user1@test.com") - self.assertTrue(is_allowed(DT.CHART, chart.name)) - - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.CHART, chart.name)) - - frappe.set_user("insights_user1@test.com") - share_chart(chart.name, "insights_user2@test.com") + self.assert_visible_to(USER_2, DT.DASHBOARD, dashboard.name) + with self.as_user(USER_2): + self.assertFalse(frappe.has_permission(DT.DASHBOARD, ptype="write", doc=dashboard.name)) + with self.assertRaises(frappe.PermissionError): + create_test_dashboard( + USER_2, + workbook.name, + title="Permissions Test Dashboard Read Only", + ) - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.CHART, chart.name)) + def test_permission_for_chart(self): + workbook = create_test_workbook(USER_1) + query = create_test_query(USER_1, workbook.name) + chart = create_test_chart(USER_1, workbook.name, query.name) - frappe.set_user("insights_user1@test.com") - unshare_chart(chart.name, "insights_user2@test.com") + self.assert_visible_to(USER_1, DT.CHART, chart.name) + self.assert_not_visible_to(USER_2, DT.CHART, chart.name) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.CHART, chart.name)) + with self.as_user(USER_1): + share_chart(chart.name, USER_2) + self.assert_visible_to(USER_2, DT.CHART, chart.name) - frappe.set_user("insights_user1@test.com") - update_share_permissions( - workbook.name, - [{"user": "insights_user2@test.com", "read": 1, "write": 0}], - ) + with self.as_user(USER_1): + unshare_chart(chart.name, USER_2) + self.assert_not_visible_to(USER_2, DT.CHART, chart.name) - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.CHART, chart.name)) - self.assertFalse(frappe.has_permission(DT.CHART, ptype="write", doc=chart.name)) - with self.assertRaises(frappe.PermissionError): - create_test_chart( - "insights_user2@test.com", + with self.as_user(USER_1): + update_share_permissions( workbook.name, - query.name, - title="Permissions Test Chart Read Only", + [{"user": USER_2, "read": 1, "write": 0}], ) - frappe.set_user("insights_user1@test.com") - update_share_permissions(workbook.name, []) - - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.CHART, chart.name)) + self.assert_visible_to(USER_2, DT.CHART, chart.name) + with self.as_user(USER_2): + self.assertFalse(frappe.has_permission(DT.CHART, ptype="write", doc=chart.name)) + with self.assertRaises(frappe.PermissionError): + create_test_chart( + USER_2, + workbook.name, + query.name, + title="Permissions Test Chart Read Only", + ) + + with self.as_user(USER_1): + update_share_permissions(workbook.name, []) + self.assert_not_visible_to(USER_2, DT.CHART, chart.name) dashboard = create_test_dashboard( - "insights_user1@test.com", + USER_1, workbook.name, chart.name, title="Permissions Test Dashboard For Chart", ) - frappe.set_user("insights_user1@test.com") - update_dashboard_access(dashboard.name, ["insights_user2@test.com"]) - - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.CHART, chart.name)) + with self.as_user(USER_1): + update_dashboard_access(dashboard.name, [USER_2]) + self.assert_visible_to(USER_2, DT.CHART, chart.name) def test_permission_for_query(self): - workbook = create_test_workbook("insights_user1@test.com") - query = create_test_query("insights_user1@test.com", workbook.name) + workbook = create_test_workbook(USER_1) + query = create_test_query(USER_1, workbook.name) - frappe.set_user("insights_user1@test.com") - self.assertTrue(is_allowed(DT.QUERY, query.name)) + self.assert_visible_to(USER_1, DT.QUERY, query.name) + self.assert_not_visible_to(USER_2, DT.QUERY, query.name) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.QUERY, query.name)) - - frappe.set_user("insights_user1@test.com") - update_share_permissions( - workbook.name, - [{"user": "insights_user2@test.com", "read": 1, "write": 0}], - ) - - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.QUERY, query.name)) - with self.assertRaises(frappe.PermissionError): - create_test_query( - "insights_user2@test.com", + with self.as_user(USER_1): + update_share_permissions( workbook.name, - title="Permissions Test Query Read Only", + [{"user": USER_2, "read": 1, "write": 0}], ) - frappe.set_user("insights_user1@test.com") - update_share_permissions(workbook.name, []) + self.assert_visible_to(USER_2, DT.QUERY, query.name) + with self.as_user(USER_2): + with self.assertRaises(frappe.PermissionError): + create_test_query( + USER_2, + workbook.name, + title="Permissions Test Query Read Only", + ) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.QUERY, query.name)) + with self.as_user(USER_1): + update_share_permissions(workbook.name, []) + self.assert_not_visible_to(USER_2, DT.QUERY, query.name) chart = create_test_chart( - "insights_user1@test.com", + USER_1, workbook.name, query.name, title="Permissions Test Chart For Query", ) chart = frappe.get_doc(DT.CHART, chart.name) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.QUERY, query.name)) - self.assertFalse(is_allowed(DT.QUERY, chart.data_query)) + self.assert_not_visible_to(USER_2, DT.QUERY, query.name) + self.assert_not_visible_to(USER_2, DT.QUERY, chart.data_query) - frappe.set_user("insights_user1@test.com") - share_chart(chart.name, "insights_user2@test.com") + with self.as_user(USER_1): + share_chart(chart.name, USER_2) - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.QUERY, query.name)) - self.assertTrue(is_allowed(DT.QUERY, chart.data_query)) + self.assert_visible_to(USER_2, DT.QUERY, query.name) + self.assert_visible_to(USER_2, DT.QUERY, chart.data_query) - frappe.set_user("insights_user1@test.com") - unshare_chart(chart.name, "insights_user2@test.com") + with self.as_user(USER_1): + unshare_chart(chart.name, USER_2) - frappe.set_user("insights_user2@test.com") - self.assertFalse(is_allowed(DT.QUERY, query.name)) - self.assertFalse(is_allowed(DT.QUERY, chart.data_query)) + self.assert_not_visible_to(USER_2, DT.QUERY, query.name) + self.assert_not_visible_to(USER_2, DT.QUERY, chart.data_query) dashboard = create_test_dashboard( - "insights_user1@test.com", + USER_1, workbook.name, chart.name, title="Permissions Test Dashboard For Query", ) - frappe.set_user("insights_user1@test.com") - update_dashboard_access(dashboard.name, ["insights_user2@test.com"]) - - frappe.set_user("insights_user2@test.com") - self.assertTrue(is_allowed(DT.QUERY, query.name)) - self.assertTrue(is_allowed(DT.QUERY, chart.data_query)) - - with self.assertRaises(frappe.PermissionError): - create_test_query( - "non_insights_user@test.com", - workbook.name, - title="Permissions Test Query Non Insights", - ) - - -def create_test_users(): - # create a website user - user = frappe.get_doc( - { - "doctype": "User", - "email": "web_user@test.com", - "first_name": "Web", - "last_name": "User", - "send_welcome_email": 0, - "user_type": "Website User", - "enabled": 1, - } - ).insert() - - # create a non insights user - user = frappe.get_doc( - { - "doctype": "User", - "email": "non_insights_user@test.com", - "first_name": "Non", - "last_name": "Insights User", - "send_welcome_email": 0, - "user_type": "System User", - "enabled": 1, - } - ).insert() - - # create a insights user - user = frappe.get_doc( - { - "doctype": "User", - "email": "insights_user1@test.com", - "first_name": "Insights", - "last_name": "User", - "send_welcome_email": 0, - "user_type": "System User", - "enabled": 1, - } - ).insert() - user.add_roles("Insights User") - - user = frappe.get_doc( - { - "doctype": "User", - "email": "insights_user2@test.com", - "first_name": "Insights", - "last_name": "User", - "send_welcome_email": 0, - "user_type": "System User", - "enabled": 1, - } - ).insert() - user.add_roles("Insights User") - - # create a insights admin - user = frappe.get_doc( - { - "doctype": "User", - "email": "insights_admin@test.com", - "first_name": "Insights", - "last_name": "Admin", - "send_welcome_email": 0, - "user_type": "System User", - "enabled": 1, - } - ).insert() - user.add_roles("Insights Admin") - - -def delete_test_users(): - delete_doc_if_exists("User", "web_user@test.com") - delete_doc_if_exists("User", "non_insights_user@test.com") - delete_doc_if_exists("User", "insights_user1@test.com") - delete_doc_if_exists("User", "insights_user2@test.com") - delete_doc_if_exists("User", "insights_admin@test.com") - - -def create_test_data_sources(): - frappe.get_doc( - { - "doctype": DT.DATA_SOURCE, - "title": TEST_DS_TITLE, - "database_type": "DuckDB", - "database_name": "test_duckdb", - } - ).insert() - - -def delete_test_data_sources(): - delete_doc_if_exists(DT.DATA_SOURCE, TEST_DS_NAME) - - -def create_test_tables(): - frappe.get_doc( - { - "doctype": DT.TABLE, - "table": "table1", - "label": "table1", - "data_source": TEST_DS_NAME, - "sync_mode": "Full", - } - ).insert() - - frappe.get_doc( - { - "doctype": DT.TABLE, - "table": "table2", - "label": "table2", - "data_source": TEST_DS_NAME, - "sync_mode": "Full", - } - ).insert() - - frappe.get_doc( - { - "doctype": DT.TABLE, - "table": "table3", - "label": "table3", - "data_source": TEST_DS_NAME, - "sync_mode": "Full", - } - ).insert() - - -def delete_test_tables(): - delete_doc_if_exists(DT.TABLE, TEST_TABLE1_NAME) - delete_doc_if_exists(DT.TABLE, TEST_TABLE2_NAME) - delete_doc_if_exists(DT.TABLE, TEST_TABLE3_NAME) - - -def create_test_teams(): - team1 = frappe.get_doc({"doctype": DT.TEAM, "team_name": "team1"}) - team1.append("team_members", {"user": "insights_user1@test.com"}) - team1.save() - clear_team_cache() - return team1 - - -def delete_test_teams(): - delete_doc_if_exists(DT.TEAM, "team1") - - -def create_test_workbook(owner, title="Permissions Test Workbook"): - frappe.set_user(owner) - return frappe.get_doc({"doctype": DT.WORKBOOK, "title": title}).insert() - - -def create_test_query(owner, workbook, title="Permissions Test Query"): - frappe.set_user(owner) - return frappe.get_doc( - { - "doctype": DT.QUERY, - "title": title, - "workbook": workbook, - "is_builder_query": 1, - "use_live_connection": 1, - "operations": [], - } - ).insert() - - -def create_test_chart(owner, workbook, query=None, title="Permissions Test Chart"): - frappe.set_user(owner) - chart = frappe.get_doc( - { - "doctype": DT.CHART, - "title": title, - "workbook": workbook, - "query": query, - "chart_type": "Bar", - "config": {}, - } - ).insert() - return frappe.get_doc(DT.CHART, chart.name) - - -def create_test_dashboard(owner, workbook, chart=None, title="Permissions Test Dashboard"): - frappe.set_user(owner) - items = [] - if chart: - items.append({"id": "chart-1", "type": "chart", "chart": chart}) - - dashboard = frappe.get_doc( - { - "doctype": DT.DASHBOARD, - "title": title, - "workbook": workbook, - "items": items, - } - ).insert() - return frappe.get_doc(DT.DASHBOARD, dashboard.name) - - -def delete_test_workbooks(): - workbooks = frappe.get_all( - DT.WORKBOOK, - filters={"title": ["like", "Permissions Test Workbook%"]}, - pluck="name", - ) - for workbook in workbooks: - frappe.delete_doc(DT.WORKBOOK, workbook, force=True) - - -def is_allowed(doctype, docname): - return bool(frappe.get_list(doctype, filters={"name": docname}, pluck="name", limit=1)) - - -def update_dashboard_access(dashboard_name, people_with_access): - frappe.get_doc(DT.DASHBOARD, dashboard_name).update_access( - { - "is_public": 0, - "is_shared_with_organization": 0, - "people_with_access": people_with_access, - } - ) - - -def share_chart(chart_name, user): - frappe.share.add(DT.CHART, chart_name, user=user, read=1, notify=0) - - -def unshare_chart(chart_name, user): - share_name = frappe.db.get_value( - "DocShare", - { - "share_doctype": DT.CHART, - "share_name": chart_name, - "user": user, - }, - ) - if share_name: - frappe.delete_doc("DocShare", share_name, ignore_permissions=True) - - -def cleanup_test_fixtures(): - delete_test_workbooks() - delete_test_teams() - delete_test_tables() - delete_test_data_sources() - delete_test_users() - clear_team_cache() - - -def delete_doc_if_exists(doctype, name): - if frappe.db.exists(doctype, name): - frappe.delete_doc(doctype, name, force=True) + with self.as_user(USER_1): + update_dashboard_access(dashboard.name, [USER_2]) + + self.assert_visible_to(USER_2, DT.QUERY, query.name) + self.assert_visible_to(USER_2, DT.QUERY, chart.data_query) + + with self.as_user(NON_INSIGHTS_USER): + with self.assertRaises(frappe.PermissionError): + create_test_query( + NON_INSIGHTS_USER, + workbook.name, + title="Permissions Test Query Non Insights", + ) diff --git a/insights/tests/workbook/helpers.py b/insights/tests/workbook/helpers.py deleted file mode 100644 index 18296557b..000000000 --- a/insights/tests/workbook/helpers.py +++ /dev/null @@ -1,135 +0,0 @@ -import frappe - -from insights.insights.doctype.insights_data_source_v3.insights_data_source_v3 import db_connections - - -class DT: - WORKBOOK = "Insights Workbook" - QUERY = "Insights Query v3" - CHART = "Insights Chart v3" - DASHBOARD = "Insights Dashboard v3" - USER = "User" - - -TEST_USER_EMAIL = "workbook_flow_user@test.com" -TEST_WORKBOOK_TITLE = "Workbook Flow Test Workbook" -TEST_QUERY_TITLE = "Workbook Flow Test Query" -TEST_CHART_TITLE = "Workbook Flow Test Chart" -TEST_DASHBOARD_TITLE = "Workbook Flow Test Dashboard" - - -def create_test_user(email=TEST_USER_EMAIL, role="Insights User"): - if frappe.db.exists(DT.USER, email): - user = frappe.get_doc(DT.USER, email) - else: - user = frappe.get_doc( - { - "doctype": DT.USER, - "email": email, - "first_name": "Workbook", - "last_name": "Flow User", - "send_welcome_email": 0, - "user_type": "System User", - "enabled": 1, - } - ).insert(ignore_permissions=True) - - if role and not frappe.db.exists("Has Role", {"parent": email, "role": role}): - user.add_roles(role) - - return user - - -def delete_test_users(): - delete_doc_if_exists(DT.USER, TEST_USER_EMAIL) - - -def create_test_workbook(owner, title=TEST_WORKBOOK_TITLE): - frappe.set_user(owner) - return frappe.get_doc({"doctype": DT.WORKBOOK, "title": title}).insert() - - -def create_test_query(owner, workbook, title=TEST_QUERY_TITLE, operations=None): - frappe.set_user(owner) - return frappe.get_doc( - { - "doctype": DT.QUERY, - "title": title, - "workbook": workbook, - "use_live_connection": 1, - "is_builder_query": 1, - "operations": operations - or [ - { - "type": "source", - "table": { - "type": "table", - "data_source": "Site DB", - "table_name": "tabToDo", - }, - } - ], - } - ).insert() - - -def create_test_chart(owner, workbook, query, title=TEST_CHART_TITLE): - frappe.set_user(owner) - chart = frappe.get_doc( - { - "doctype": DT.CHART, - "title": title, - "workbook": workbook, - "query": query, - "chart_type": "Bar", - "config": {}, - } - ).insert() - return frappe.get_doc(DT.CHART, chart.name) - - -def create_test_dashboard(owner, workbook, chart, title=TEST_DASHBOARD_TITLE): - frappe.set_user(owner) - dashboard = frappe.get_doc( - { - "doctype": DT.DASHBOARD, - "title": title, - "workbook": workbook, - "items": [{"id": "chart-1", "type": "chart", "chart": chart}], - } - ).insert() - return frappe.get_doc(DT.DASHBOARD, dashboard.name) - - -def execute_test_query(query_name): - query = frappe.get_doc(DT.QUERY, query_name) - with db_connections(): - return query.execute() - - -def doc_exists(doctype, name): - return bool(frappe.db.exists(doctype, name)) - - -def is_visible(doctype, name): - return bool(frappe.get_list(doctype, filters={"name": name}, pluck="name", limit=1)) - - -def delete_test_workbooks(): - workbooks = frappe.get_all( - DT.WORKBOOK, - filters={"title": ["like", f"{TEST_WORKBOOK_TITLE}%"]}, - pluck="name", - ) - for workbook in workbooks: - frappe.delete_doc(DT.WORKBOOK, workbook, force=True) - - -def cleanup_workbook_flow_fixtures(): - delete_test_workbooks() - delete_test_users() - - -def delete_doc_if_exists(doctype, name): - if frappe.db.exists(doctype, name): - frappe.delete_doc(doctype, name, force=True) diff --git a/insights/tests/workbook/test_querying.py b/insights/tests/workbook/test_querying.py new file mode 100644 index 000000000..92bbb41e4 --- /dev/null +++ b/insights/tests/workbook/test_querying.py @@ -0,0 +1,383 @@ +import frappe +from frappe.utils import add_days, nowdate + +from insights.insights.doctype.insights_data_source_v3.ibis_utils import CircularQueryReferenceError +from insights.insights.doctype.insights_data_source_v3.insights_data_source_v3 import db_connections +from insights.tests.base import InsightsIntegrationTestCase +from insights.tests.factories import ( + DT, + USER_1, + cleanup_workbook_flow_fixtures, + create_test_query, + create_test_user, + create_test_workbook, + execute_test_query, +) + +TODO_PREFIX = "Insights Querying Test" + + +def column(column_name): + return {"type": "column", "column_name": column_name} + + +def table_source(table_name="tabToDo"): + return { + "type": "source", + "table": { + "type": "table", + "data_source": "Site DB", + "table_name": table_name, + }, + } + + +def query_source(query_name): + return { + "type": "source", + "table": { + "type": "query", + "query_name": query_name, + }, + } + + +class TestQuerying(InsightsIntegrationTestCase): + COMMIT_AFTER_TEST_SETUP = True + COMMIT_AFTER_TEST_TEARDOWN = True + + @classmethod + def before_class(cls): + cleanup_workbook_flow_fixtures() + cls.delete_test_todos() + create_test_user(USER_1) + + @classmethod + def after_class(cls): + cls.delete_test_todos() + cleanup_workbook_flow_fixtures() + + def before_test(self): + self.delete_test_todos() + cleanup_workbook_flow_fixtures() + create_test_user(USER_1) + + def after_test(self): + self.delete_test_todos() + cleanup_workbook_flow_fixtures() + + @staticmethod + def delete_test_todos(): + todo_names = frappe.get_all( + "ToDo", + filters={"description": ["like", f"{TODO_PREFIX}%"]}, + pluck="name", + ) + for todo_name in todo_names: + frappe.delete_doc("ToDo", todo_name, force=True) + + def seed_todos(self): + records = [ + { + "description": f"{TODO_PREFIX} Open Alpha", + "status": "Open", + "date": add_days(nowdate(), 1), + "allocated_to": USER_1, + }, + { + "description": f"{TODO_PREFIX} Closed Beta", + "status": "Closed", + "date": add_days(nowdate(), 2), + "allocated_to": USER_1, + }, + { + "description": f"{TODO_PREFIX} Open Gamma", + "status": "Open", + "date": add_days(nowdate(), 3), + "allocated_to": USER_1, + }, + ] + + todo_names = [] + for record in records: + todo = frappe.get_doc({"doctype": "ToDo", **record}).insert(ignore_permissions=True) + todo_names.append(todo.name) + + frappe.db.commit() + return todo_names + + def get_query(self, query_name): + return frappe.get_doc(DT.QUERY, query_name) + + def test_builder_query_executes_from_source_operation(self): + data_sources = set(frappe.get_all("Insights Data Source v3", pluck="name")) + + self.assertIn("Site DB", data_sources) + + workbook = create_test_workbook(USER_1) + query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Smoke", + ) + + result = execute_test_query(query.name) + + self.assertIn("sql", result) + self.assertGreater(len(result["columns"]), 0) + self.assertIsInstance(result["rows"], list) + + def test_query_pipeline_supports_filter_group_mutate_order_and_limit(self): + self.seed_todos() + workbook = create_test_workbook(USER_1) + query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Pipeline", + operations=[ + table_source(), + { + "type": "filter_group", + "logical_operator": "And", + "filters": [ + { + "column": column("description"), + "operator": "contains", + "value": TODO_PREFIX, + }, + { + "column": column("status"), + "operator": "=", + "value": "Open", + }, + ], + }, + { + "type": "mutate", + "new_name": "docstatus_plus_one", + "expression": {"type": "expression", "expression": "docstatus + 1"}, + "data_type": "Integer", + }, + { + "type": "order_by", + "column": column("description"), + "direction": "asc", + }, + {"type": "limit", "limit": 2}, + ], + ) + + with db_connections(): + preview_columns = self.get_query(query.name).get_columns_for_selection(active_operation_idx=2) + + result = execute_test_query(query.name) + preview_column_names = {col["name"] for col in preview_columns} + + self.assertIn("docstatus_plus_one", preview_column_names) + self.assertEqual(len(result["rows"]), 2) + self.assertEqual( + [row["description"] for row in result["rows"]], + [ + f"{TODO_PREFIX} Open Alpha", + f"{TODO_PREFIX} Open Gamma", + ], + ) + self.assertTrue(all(row["docstatus_plus_one"] == 1 for row in result["rows"])) + + def test_query_pipeline_supports_filter_and_join(self): + self.seed_todos() + workbook = create_test_workbook(USER_1) + query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Join", + operations=[ + table_source(), + { + "type": "filter", + "column": column("description"), + "operator": "contains", + "value": TODO_PREFIX, + }, + { + "type": "join", + "join_type": "left", + "table": { + "type": "table", + "data_source": "Site DB", + "table_name": "tabUser", + }, + "select_columns": [column("full_name")], + "join_condition": { + "left_column": column("allocated_to"), + "right_column": column("name"), + }, + }, + { + "type": "order_by", + "column": column("description"), + "direction": "asc", + }, + ], + ) + + with db_connections(): + preview_columns = self.get_query(query.name).get_columns_for_selection(active_operation_idx=2) + + result = execute_test_query(query.name) + preview_column_names = {col["name"] for col in preview_columns} + + self.assertIn("full_name", preview_column_names) + self.assertEqual(len(result["rows"]), 3) + self.assertEqual({row["full_name"] for row in result["rows"]}, {"Workbook Flow User"}) + + def test_query_summary_groups_filtered_rows_by_status(self): + self.seed_todos() + workbook = create_test_workbook(USER_1) + query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Summary", + operations=[ + table_source(), + { + "type": "filter", + "column": column("description"), + "operator": "contains", + "value": TODO_PREFIX, + }, + { + "type": "summarize", + "measures": [ + { + "measure_name": "todo_count", + "column_name": "name", + "aggregation": "count", + } + ], + "dimensions": [ + { + "column_name": "status", + "data_type": "String", + "dimension_name": "status", + } + ], + }, + { + "type": "order_by", + "column": column("status"), + "direction": "asc", + }, + ], + ) + + result = execute_test_query(query.name) + summary = {row["status"]: row["todo_count"] for row in result["rows"]} + + self.assertEqual(summary, {"Closed": 1, "Open": 2}) + + def test_query_helpers_return_count_distinct_values_and_csv_results(self): + self.seed_todos() + workbook = create_test_workbook(USER_1) + query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Helpers", + operations=[ + table_source(), + { + "type": "filter", + "column": column("description"), + "operator": "contains", + "value": TODO_PREFIX, + }, + ], + ) + query_doc = self.get_query(query.name) + + with db_connections(): + total_count = query_doc.get_count() + distinct_statuses = query_doc.get_distinct_column_values("status") + csv_data = query_doc.download_results(format="csv") + + self.assertEqual(total_count, 3) + self.assertEqual(set(distinct_statuses), {"Closed", "Open"}) + self.assertIn("description", csv_data) + self.assertIn(f"{TODO_PREFIX} Open Alpha", csv_data) + self.assertIn(f"{TODO_PREFIX} Closed Beta", csv_data) + + def test_query_can_use_another_query_as_its_source(self): + self.seed_todos() + workbook = create_test_workbook(USER_1) + base_query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Base", + operations=[ + table_source(), + { + "type": "filter", + "column": column("description"), + "operator": "contains", + "value": TODO_PREFIX, + }, + ], + ) + derived_query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Derived", + operations=[ + query_source(base_query.name), + { + "type": "summarize", + "measures": [ + { + "measure_name": "todo_count", + "column_name": "name", + "aggregation": "count", + } + ], + "dimensions": [ + { + "column_name": "status", + "data_type": "String", + "dimension_name": "status", + } + ], + }, + ], + ) + + frappe.db.commit() + result = execute_test_query(derived_query.name) + + with db_connections(): + source_tables = self.get_query(derived_query.name).get_source_tables() + + self.assertEqual( + {row["status"]: row["todo_count"] for row in result["rows"]}, {"Closed": 1, "Open": 2} + ) + self.assertEqual(source_tables, [{"data_source": "Site DB", "table_name": "tabToDo"}]) + + def test_query_validation_blocks_circular_query_references(self): + workbook = create_test_workbook(USER_1) + first_query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Circular First", + operations=[table_source()], + ) + second_query = create_test_query( + USER_1, + workbook.name, + title="Workbook Flow Test Query Circular Second", + operations=[query_source(first_query.name)], + ) + + frappe.db.commit() + first_query_doc = self.get_query(first_query.name) + first_query_doc.operations = [query_source(second_query.name)] + + with self.assertRaises(CircularQueryReferenceError): + first_query_doc.save() diff --git a/insights/tests/workbook/test_workbook.py b/insights/tests/workbook/test_workbook.py index 77d3dd3d0..c8c16c6ae 100644 --- a/insights/tests/workbook/test_workbook.py +++ b/insights/tests/workbook/test_workbook.py @@ -1,150 +1,57 @@ import frappe -from frappe.tests import IntegrationTestCase -from frappe.utils import set_request -from insights.api import get_doc as get_public_doc -from insights.api import run_doc_method as run_public_doc_method from insights.api.workbooks import ( create_folder, delete_folder, get_share_permissions, get_workbooks, import_workbook, - move_item_to_folder, update_share_permissions, update_sort_orders, ) -from insights.tests.workbook.helpers import ( - DT, - TEST_USER_EMAIL, - create_test_chart, - create_test_dashboard, - create_test_query, - create_test_user, - create_test_workbook, - delete_doc_if_exists, - doc_exists, +from insights.tests.base import InsightsIntegrationTestCase +from insights.tests.factories import DT, USER_1, create_test_user, delete_users, doc_exists +from insights.tests.workbook_utils import ( + cleanup_test_workbooks, + create_workbook_bundle, + get_doc, + get_workbook, + run_doc_method, ) -COLLABORATOR_EMAIL = "workbook_flow_collaborator@test.com" +COLLABORATOR = "workbook_flow_collaborator@test.com" -def cleanup_test_workbooks(*owners): - workbooks = frappe.get_all( - DT.WORKBOOK, - filters={"owner": ["in", list(owners)]}, - pluck="name", - ) - for workbook_name in workbooks: - frappe.delete_doc(DT.WORKBOOK, workbook_name, force=True) +class TestWorkbook(InsightsIntegrationTestCase): + COMMIT_AFTER_TEST_SETUP = True + COMMIT_AFTER_TEST_TEARDOWN = True - -class TestWorkbook(IntegrationTestCase): @classmethod - def setUpClass(cls): - super().setUpClass() - frappe.set_user("Administrator") - cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) - delete_doc_if_exists(DT.USER, TEST_USER_EMAIL) - delete_doc_if_exists(DT.USER, COLLABORATOR_EMAIL) - create_test_user(TEST_USER_EMAIL) - create_test_user(COLLABORATOR_EMAIL) - frappe.db.commit() + def before_class(cls): + cleanup_test_workbooks(USER_1, COLLABORATOR) + delete_users(USER_1, COLLABORATOR) + create_test_user(USER_1) + create_test_user(COLLABORATOR) @classmethod - def tearDownClass(cls): - frappe.set_user("Administrator") - cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) - delete_doc_if_exists(DT.USER, TEST_USER_EMAIL) - delete_doc_if_exists(DT.USER, COLLABORATOR_EMAIL) - frappe.db.commit() - super().tearDownClass() - - def setUp(self): - super().setUp() - self.original_user = frappe.session.user - frappe.set_user("Administrator") - cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) - frappe.db.commit() - - def tearDown(self): - frappe.set_user("Administrator") - cleanup_test_workbooks(TEST_USER_EMAIL, COLLABORATOR_EMAIL) - frappe.db.commit() - frappe.set_user(self.original_user) - super().tearDown() - - def get_doc(self, doctype, name): - set_request(method="GET", path="/api/method/insights.api.get_doc") - return get_public_doc(doctype, name) - - def run_doc_method(self, method, doc, args=None): - set_request(method="POST", path="/api/method/insights.api.run_doc_method") - return run_public_doc_method(method, doc, args=args) - - def get_workbook(self, name): - workbook = self.get_doc(DT.WORKBOOK, name) - for field in ("folders", "queries", "charts", "dashboards"): - workbook[field] = frappe.parse_json(workbook.get(field)) or [] - return workbook - - def create_workbook_bundle(self, title, include_secondary_items=False, include_folders=False): - workbook = create_test_workbook(TEST_USER_EMAIL, title=title) - query = create_test_query(TEST_USER_EMAIL, workbook.name, title=f"{title} Query 1") - chart = create_test_chart(TEST_USER_EMAIL, workbook.name, query.name, title=f"{title} Chart 1") - dashboard = create_test_dashboard( - TEST_USER_EMAIL, - workbook.name, - chart.name, - title=f"{title} Dashboard 1", - ) + def after_class(cls): + cleanup_test_workbooks(USER_1, COLLABORATOR) + delete_users(USER_1, COLLABORATOR) - bundle = { - "workbook": workbook, - "query": query, - "chart": chart, - "dashboard": dashboard, - "folders": {}, - } - - if include_secondary_items: - bundle["secondary_query"] = create_test_query( - TEST_USER_EMAIL, - workbook.name, - title=f"{title} Query 2", - ) - bundle["secondary_chart"] = create_test_chart( - TEST_USER_EMAIL, - workbook.name, - bundle["secondary_query"].name, - title=f"{title} Chart 2", - ) - - if include_folders: - frappe.set_user(TEST_USER_EMAIL) - bundle["folders"]["query"] = create_folder( - workbook.name, - f"{title} Query Folder", - "query", - ) - bundle["folders"]["chart"] = create_folder( - workbook.name, - f"{title} Chart Folder", - "chart", - ) - move_item_to_folder("query", query.name, bundle["folders"]["query"]) - move_item_to_folder("chart", chart.name, bundle["folders"]["chart"]) + def before_test(self): + cleanup_test_workbooks(USER_1, COLLABORATOR) - return bundle + def after_test(self): + cleanup_test_workbooks(USER_1, COLLABORATOR) def test_owner_can_build_workbook_query_chart_dashboard_flow(self): - bundle = self.create_workbook_bundle("Workbook Flow Test Authoring") + bundle = create_workbook_bundle(USER_1, "Workbook Flow Test Authoring") - frappe.set_user(TEST_USER_EMAIL) - workbook = self.get_workbook(bundle["workbook"].name) - chart = self.get_doc(DT.CHART, bundle["chart"].name) - dashboard = self.get_doc(DT.DASHBOARD, bundle["dashboard"].name) - query_result = self.run_doc_method("execute", self.get_doc(DT.QUERY, bundle["query"].name)) + with self.as_user(USER_1): + workbook = get_workbook(bundle["workbook"].name) + chart = get_doc(DT.CHART, bundle["chart"].name) + dashboard = get_doc(DT.DASHBOARD, bundle["dashboard"].name) + query_result = run_doc_method("execute", get_doc(DT.QUERY, bundle["query"].name)) dashboard_items = frappe.parse_json(dashboard["items"]) or [] self.assertEqual([row["name"] for row in workbook["queries"]], [bundle["query"].name]) @@ -159,11 +66,13 @@ def test_owner_can_build_workbook_query_chart_dashboard_flow(self): self.assertTrue(any(item.get("chart") == bundle["chart"].name for item in dashboard_items)) def test_deleting_workbook_removes_the_user_visible_tree(self): - bundle = self.create_workbook_bundle("Workbook Flow Test Delete") - data_query_name = self.get_doc(DT.CHART, bundle["chart"].name)["data_query"] + bundle = create_workbook_bundle(USER_1, "Workbook Flow Test Delete") - frappe.set_user("Administrator") - frappe.delete_doc(DT.WORKBOOK, bundle["workbook"].name, force=True) + with self.as_user(USER_1): + data_query_name = get_doc(DT.CHART, bundle["chart"].name)["data_query"] + + with self.as_user("Administrator"): + frappe.delete_doc(DT.WORKBOOK, bundle["workbook"].name, force=True) self.assertFalse(doc_exists(DT.WORKBOOK, bundle["workbook"].name)) self.assertFalse(doc_exists(DT.QUERY, bundle["query"].name)) @@ -172,57 +81,55 @@ def test_deleting_workbook_removes_the_user_visible_tree(self): self.assertFalse(doc_exists(DT.DASHBOARD, bundle["dashboard"].name)) def test_owner_can_organize_and_reorder_workbook_contents(self): - bundle = self.create_workbook_bundle( + bundle = create_workbook_bundle( + USER_1, "Workbook Flow Test Organization", include_secondary_items=True, ) - frappe.set_user(TEST_USER_EMAIL) - query_folder = create_folder( - bundle["workbook"].name, - "Workbook Flow Test Organization Query Folder", - "query", - ) - chart_folder = create_folder( - bundle["workbook"].name, - "Workbook Flow Test Organization Chart Folder", - "chart", - ) - move_item_to_folder("query", bundle["query"].name, query_folder) - move_item_to_folder("chart", bundle["chart"].name, chart_folder) - update_sort_orders( - bundle["workbook"].name, - [ - {"type": "folder", "name": chart_folder, "sort_order": 0}, - {"type": "folder", "name": query_folder, "sort_order": 1}, - { - "type": "query", - "name": bundle["secondary_query"].name, - "sort_order": 0, - "folder": None, - }, - { - "type": "query", - "name": bundle["query"].name, - "sort_order": 1, - "folder": query_folder, - }, - { - "type": "chart", - "name": bundle["secondary_chart"].name, - "sort_order": 0, - "folder": None, - }, - { - "type": "chart", - "name": bundle["chart"].name, - "sort_order": 1, - "folder": chart_folder, - }, - ], - ) - - workbook = self.get_workbook(bundle["workbook"].name) + with self.as_user(USER_1): + query_folder = create_folder( + bundle["workbook"].name, + "Workbook Flow Test Organization Query Folder", + "query", + ) + chart_folder = create_folder( + bundle["workbook"].name, + "Workbook Flow Test Organization Chart Folder", + "chart", + ) + update_sort_orders( + bundle["workbook"].name, + [ + {"type": "folder", "name": chart_folder, "sort_order": 0}, + {"type": "folder", "name": query_folder, "sort_order": 1}, + { + "type": "query", + "name": bundle["secondary_query"].name, + "sort_order": 0, + "folder": None, + }, + { + "type": "query", + "name": bundle["query"].name, + "sort_order": 1, + "folder": query_folder, + }, + { + "type": "chart", + "name": bundle["secondary_chart"].name, + "sort_order": 0, + "folder": None, + }, + { + "type": "chart", + "name": bundle["chart"].name, + "sort_order": 1, + "folder": chart_folder, + }, + ], + ) + workbook = get_workbook(bundle["workbook"].name) self.assertEqual([row["name"] for row in workbook["folders"]], [chart_folder, query_folder]) self.assertEqual( @@ -236,38 +143,41 @@ def test_owner_can_organize_and_reorder_workbook_contents(self): self.assertEqual(workbook["queries"][1]["folder"], query_folder) self.assertEqual(workbook["charts"][1]["folder"], chart_folder) - delete_folder(query_folder, move_items_to_root=True) - delete_folder(chart_folder, move_items_to_root=True) - - workbook = self.get_workbook(bundle["workbook"].name) + with self.as_user(USER_1): + delete_folder(query_folder, move_items_to_root=True) + delete_folder(chart_folder, move_items_to_root=True) + workbook = get_workbook(bundle["workbook"].name) self.assertEqual(workbook["folders"], []) self.assertTrue(all(not row["folder"] for row in workbook["queries"])) self.assertTrue(all(not row["folder"] for row in workbook["charts"])) def test_duplicate_workbook_preserves_a_usable_copy(self): - bundle = self.create_workbook_bundle( + bundle = create_workbook_bundle( + USER_1, "Workbook Flow Test Duplicate", include_folders=True, ) - frappe.set_user(TEST_USER_EMAIL) - original_workbook = self.get_workbook(bundle["workbook"].name) - original_chart = self.get_doc(DT.CHART, bundle["chart"].name) - duplicate_name = self.run_doc_method( - "duplicate", - self.get_doc(DT.WORKBOOK, bundle["workbook"].name), - ) - duplicate_workbook = self.get_workbook(duplicate_name) + with self.as_user(USER_1): + original_workbook = get_workbook(bundle["workbook"].name) + original_chart = get_doc(DT.CHART, bundle["chart"].name) + duplicate_name = run_doc_method( + "duplicate", + get_doc(DT.WORKBOOK, bundle["workbook"].name), + ) + duplicate_workbook = get_workbook(duplicate_name) duplicate_query_name = duplicate_workbook["queries"][0]["name"] duplicate_chart_name = duplicate_workbook["charts"][0]["name"] - duplicate_chart = self.get_doc(DT.CHART, duplicate_chart_name) - duplicate_dashboard = self.get_doc( - DT.DASHBOARD, - duplicate_workbook["dashboards"][0]["name"], - ) + with self.as_user(USER_1): + duplicate_chart = get_doc(DT.CHART, duplicate_chart_name) + duplicate_dashboard = get_doc( + DT.DASHBOARD, + duplicate_workbook["dashboards"][0]["name"], + ) duplicate_dashboard_items = frappe.parse_json(duplicate_dashboard["items"]) or [] - duplicate_result = self.run_doc_method("execute", self.get_doc(DT.QUERY, duplicate_query_name)) + with self.as_user(USER_1): + duplicate_result = run_doc_method("execute", get_doc(DT.QUERY, duplicate_query_name)) self.assertEqual(len(duplicate_workbook["folders"]), 2) self.assertEqual(len(duplicate_workbook["queries"]), 1) @@ -292,29 +202,32 @@ def test_duplicate_workbook_preserves_a_usable_copy(self): self.assertGreater(len(duplicate_result["columns"]), 0) def test_export_and_import_preserve_a_usable_workflow(self): - bundle = self.create_workbook_bundle( + bundle = create_workbook_bundle( + USER_1, "Workbook Flow Test Import", include_folders=True, ) - frappe.set_user(TEST_USER_EMAIL) - original_workbook = self.get_workbook(bundle["workbook"].name) - original_chart = self.get_doc(DT.CHART, bundle["chart"].name) - exported_workbook = self.run_doc_method( - "export", - self.get_doc(DT.WORKBOOK, bundle["workbook"].name), - ) - imported_name = import_workbook(exported_workbook) - imported_workbook = self.get_workbook(imported_name) + with self.as_user(USER_1): + original_workbook = get_workbook(bundle["workbook"].name) + original_chart = get_doc(DT.CHART, bundle["chart"].name) + exported_workbook = run_doc_method( + "export", + get_doc(DT.WORKBOOK, bundle["workbook"].name), + ) + imported_name = import_workbook(exported_workbook) + imported_workbook = get_workbook(imported_name) imported_query_name = imported_workbook["queries"][0]["name"] imported_chart_name = imported_workbook["charts"][0]["name"] - imported_chart = self.get_doc(DT.CHART, imported_chart_name) - imported_dashboard = self.get_doc( - DT.DASHBOARD, - imported_workbook["dashboards"][0]["name"], - ) + with self.as_user(USER_1): + imported_chart = get_doc(DT.CHART, imported_chart_name) + imported_dashboard = get_doc( + DT.DASHBOARD, + imported_workbook["dashboards"][0]["name"], + ) imported_dashboard_items = frappe.parse_json(imported_dashboard["items"]) or [] - imported_result = self.run_doc_method("execute", self.get_doc(DT.QUERY, imported_query_name)) + with self.as_user(USER_1): + imported_result = run_doc_method("execute", get_doc(DT.QUERY, imported_query_name)) self.assertEqual(len(imported_workbook["folders"]), 2) self.assertEqual(len(imported_workbook["queries"]), 1) @@ -339,30 +252,30 @@ def test_export_and_import_preserve_a_usable_workflow(self): self.assertGreater(len(imported_result["columns"]), 0) def test_shared_workbook_supports_read_only_public_access_but_blocks_structure_changes(self): - bundle = self.create_workbook_bundle("Workbook Flow Test Shared") + bundle = create_workbook_bundle(USER_1, "Workbook Flow Test Shared") - frappe.set_user(TEST_USER_EMAIL) - self.run_doc_method("track_view", self.get_doc(DT.WORKBOOK, bundle["workbook"].name)) - update_share_permissions(bundle["workbook"].name, [], organization_access="view") - share_permissions = get_share_permissions(bundle["workbook"].name) - owner_workbooks = get_workbooks(search_term="Workbook Flow Test Shared") + with self.as_user(USER_1): + run_doc_method("track_view", get_doc(DT.WORKBOOK, bundle["workbook"].name)) + update_share_permissions(bundle["workbook"].name, [], organization_access="view") + share_permissions = get_share_permissions(bundle["workbook"].name) + owner_workbooks = get_workbooks(search_term="Workbook Flow Test Shared") self.assertEqual(share_permissions["organization_access"], "view") self.assertEqual( {permission["user"] for permission in share_permissions["user_permissions"]}, - {TEST_USER_EMAIL}, + {USER_1}, ) self.assertEqual(len(owner_workbooks), 1) self.assertEqual(owner_workbooks[0]["name"], bundle["workbook"].name) self.assertEqual(owner_workbooks[0]["views"], 1) self.assertTrue(owner_workbooks[0]["shared_with_organization"]) - frappe.set_user(COLLABORATOR_EMAIL) - collaborator_workbooks = get_workbooks(search_term="Workbook Flow Test Shared") - workbook = self.get_workbook(bundle["workbook"].name) - query = self.get_doc(DT.QUERY, bundle["query"].name) - chart = self.get_doc(DT.CHART, bundle["chart"].name) - dashboard = self.get_doc(DT.DASHBOARD, bundle["dashboard"].name) + with self.as_user(COLLABORATOR): + collaborator_workbooks = get_workbooks(search_term="Workbook Flow Test Shared") + workbook = get_workbook(bundle["workbook"].name) + query = get_doc(DT.QUERY, bundle["query"].name) + chart = get_doc(DT.CHART, bundle["chart"].name) + dashboard = get_doc(DT.DASHBOARD, bundle["dashboard"].name) self.assertEqual([row["name"] for row in collaborator_workbooks], [bundle["workbook"].name]) self.assertTrue(workbook["read_only"]) @@ -373,18 +286,19 @@ def test_shared_workbook_supports_read_only_public_access_but_blocks_structure_c self.assertEqual([row["name"] for row in workbook["charts"]], [bundle["chart"].name]) self.assertEqual([row["name"] for row in workbook["dashboards"]], [bundle["dashboard"].name]) - with self.assertRaises(frappe.PermissionError): - create_folder(bundle["workbook"].name, "Blocked Query Folder", "query") - - with self.assertRaises(frappe.PermissionError): - update_sort_orders( - bundle["workbook"].name, - [ - { - "type": "query", - "name": bundle["query"].name, - "sort_order": 0, - "folder": None, - } - ], - ) + with self.as_user(COLLABORATOR): + with self.assertRaises(frappe.PermissionError): + create_folder(bundle["workbook"].name, "Blocked Query Folder", "query") + + with self.assertRaises(frappe.PermissionError): + update_sort_orders( + bundle["workbook"].name, + [ + { + "type": "query", + "name": bundle["query"].name, + "sort_order": 0, + "folder": None, + } + ], + ) diff --git a/insights/tests/workbook_utils.py b/insights/tests/workbook_utils.py new file mode 100644 index 000000000..c432139c3 --- /dev/null +++ b/insights/tests/workbook_utils.py @@ -0,0 +1,86 @@ +import frappe +from frappe.utils import set_request + +from insights.api import get_doc as get_public_doc +from insights.api import run_doc_method as run_public_doc_method +from insights.api.workbooks import create_folder, move_item_to_folder +from insights.tests.factories import ( + as_user, + create_test_chart, + create_test_dashboard, + create_test_query, + create_test_workbook, + delete_workbooks, +) + + +def cleanup_test_workbooks(*owners): + if owners: + delete_workbooks(owners=owners) + + +def get_doc(doctype, name): + set_request(method="GET", path="/api/method/insights.api.get_doc") + return get_public_doc(doctype, name) + + +def run_doc_method(method, doc, args=None): + set_request(method="POST", path="/api/method/insights.api.run_doc_method") + return run_public_doc_method(method, doc, args=args) + + +def get_workbook(name): + workbook = get_doc("Insights Workbook", name) + for field in ("folders", "queries", "charts", "dashboards"): + workbook[field] = frappe.parse_json(workbook.get(field)) or [] + return workbook + + +def create_workbook_bundle(owner, title, include_secondary_items=False, include_folders=False): + workbook = create_test_workbook(owner, title=title) + query = create_test_query(owner, workbook.name, title=f"{title} Query 1") + chart = create_test_chart(owner, workbook.name, query.name, title=f"{title} Chart 1") + dashboard = create_test_dashboard( + owner, + workbook.name, + chart.name, + title=f"{title} Dashboard 1", + ) + + bundle = { + "workbook": workbook, + "query": query, + "chart": chart, + "dashboard": dashboard, + "folders": {}, + } + + if include_secondary_items: + bundle["secondary_query"] = create_test_query( + owner, + workbook.name, + title=f"{title} Query 2", + ) + bundle["secondary_chart"] = create_test_chart( + owner, + workbook.name, + bundle["secondary_query"].name, + title=f"{title} Chart 2", + ) + + if include_folders: + with as_user(owner): + bundle["folders"]["query"] = create_folder( + workbook.name, + f"{title} Query Folder", + "query", + ) + bundle["folders"]["chart"] = create_folder( + workbook.name, + f"{title} Chart Folder", + "chart", + ) + move_item_to_folder("query", query.name, bundle["folders"]["query"]) + move_item_to_folder("chart", chart.name, bundle["folders"]["chart"]) + + return bundle