From 3db16c7314391e22a23b2c871dbb64e5160e5e43 Mon Sep 17 00:00:00 2001 From: Simon Briere Date: Tue, 5 Mar 2024 14:14:32 -0500 Subject: [PATCH] Refs #1. Completed backend Dashboard API. --- API/user/QueryDashboard.py | 79 ++++++-- .../db/models/DashDashboardProjects.py | 4 + libDashboards/db/models/DashDashboardSites.py | 4 + libDashboards/db/models/DashDashboards.py | 8 +- tests/api/user/test_QueryDashboard.py | 189 +++++++++++++++++- 5 files changed, 258 insertions(+), 26 deletions(-) diff --git a/API/user/QueryDashboard.py b/API/user/QueryDashboard.py index 44dff5a..218d3e9 100644 --- a/API/user/QueryDashboard.py +++ b/API/user/QueryDashboard.py @@ -138,8 +138,8 @@ def post(self): user_info = current_user_client.get_user_info() accessible_project_ids = [role['id_project'] for role in user_info['projects'] - if role['id_project'] == 'admin'] - accessible_site_ids = [role['id_site'] for role in user_info['sites'] if role['id_site'] == 'admin'] + if role['project_role'] == 'admin'] + accessible_site_ids = [role['id_site'] for role in user_info['sites'] if role['site_role'] == 'admin'] dashboard = None if updating: # Load dashboard @@ -198,20 +198,29 @@ def post(self): if 'dashboard_sites' in json_dashboard: dashboard_sites = json_dashboard.pop('dashboard_sites') dashboard_sites_ids = [site['id_site'] for site in dashboard_sites] - if len(set(accessible_site_ids).intersection(dashboard_sites_ids)) == 0: - return gettext('No admin access to all sites for that dashboard'), 403 + if len(set(accessible_site_ids).intersection(dashboard_sites_ids)) == 0 and dashboard_sites_ids: + return gettext('No admin access to that dashboard'), 403 + for site in dashboard_sites: + if site['id_site'] not in accessible_site_ids: + return gettext('At least one site isn\'t accessible'), 403 if 'dashboard_projects' in json_dashboard: dashboard_projects = json_dashboard.pop('dashboard_projects') dashboard_projects_ids = [proj['id_project'] for proj in dashboard_projects] - if len(set(accessible_project_ids).intersection(dashboard_projects_ids)) == 0: - return gettext('No admin access to all projects for that dashboard'), 403 + if len(set(accessible_project_ids).intersection(dashboard_projects_ids)) == 0 and dashboard_projects_ids: + return gettext('No admin access to that dashboard'), 403 + for proj in dashboard_projects: + if proj['id_project'] not in accessible_project_ids: + return gettext('At least one project isn\'t accessible'), 403 if not updating and not dashboard_sites and not dashboard_projects: # Global new dashboard - only super admin can create if not current_user_client.user_superadmin: return gettext('Forbidden'), 403 + if dashboard_sites and dashboard_projects: + return gettext('A dashboard can\'t be associated to both sites and projects'), 400 + # Pop dashboard definition and version dashboard_version = json_dashboard.pop('dashboard_version') dashboard_definition = None @@ -251,28 +260,56 @@ def post(self): return gettext('Database error'), 400 # Manage sites - if dashboard_sites: + if dashboard_sites or dashboard_sites == []: # Add / update existing sites current_sites_ids = [site.id_site for site in dashboard.dashboard_sites] + for site in dashboard_sites: # Check if already there if site['id_site'] in current_sites_ids: # Update - dashboard.dashboard_sites[current_sites_ids.index(site['id_site'])].from_json(site) + dds = DashDashboardSites.get_for_site_and_dashboard(site['id_site'], dashboard.id_dashboard) + DashDashboardSites.update(dds.id_dashboard_site, site) + else: # Add - dds = DashDashboardSites + dds = DashDashboardSites() dds.from_json(site) - dashboard.dashboard_sites.append(dds) - - # Remove sites not present in the list - to_remove_ids = set(accessible_site_ids).intersection(dashboard_sites_ids) - for remove_id in to_remove_ids: - if remove_id in current_sites_ids: - del dashboard.dashboard_sites[current_sites_ids.index(remove_id)] - dashboard.commit() - - # TODO Manage dashboard projects + dds.id_dashboard = dashboard.id_dashboard + DashDashboardSites.insert(dds) + + # Remove sites not present in the list + to_remove_ids = set(accessible_site_ids).difference(dashboard_sites_ids) + for remove_id in to_remove_ids: + if remove_id in current_sites_ids: + dds = DashDashboardSites.get_for_site_and_dashboard(remove_id, dashboard.id_dashboard) + DashDashboardSites.delete(dds.id_dashboard_site) + + # Manage dashboard projects + if dashboard_projects or dashboard_projects == []: + # Add / update existing projects + current_proj_ids = [proj.id_project for proj in dashboard.dashboard_projects] + + for proj in dashboard_projects: + # Check if already there + if proj['id_project'] in current_proj_ids: + # Update + ddp = DashDashboardProjects.get_for_project_and_dashboard(proj['id_project'], + dashboard.id_dashboard) + DashDashboardProjects.update(ddp.id_dashboard_project, proj) + else: + # Add + ddp = DashDashboardProjects() + ddp.from_json(proj) + ddp.id_dashboard = dashboard.id_dashboard + DashDashboardProjects.insert(ddp) + + # Remove not present in the list + to_remove_ids = set(accessible_project_ids).difference(dashboard_projects_ids) + for remove_id in to_remove_ids: + if remove_id in current_proj_ids: + ddp = DashDashboardProjects.get_for_project_and_dashboard(remove_id, dashboard.id_dashboard) + DashDashboardProjects.delete(ddp.id_dashboard_project) # Manage versions if dashboard_version and dashboard_definition: @@ -329,8 +366,8 @@ def delete(self): # Check deletion access user_info = current_user_client.get_user_info() accessible_project_ids = [role['id_project'] for role in user_info['projects'] - if role['id_project'] == 'admin'] - accessible_site_ids = [role['id_site'] for role in user_info['sites'] if role['id_site'] == 'admin'] + if role['project_role'] == 'admin'] + accessible_site_ids = [role['id_site'] for role in user_info['sites'] if role['site_role'] == 'admin'] dashboard_sites_ids = [dash_site.id_site for dash_site in dashboard.dashboard_sites] dashboard_projects_ids = [dash_proj.id_project for dash_proj in dashboard.dashboard_projects] diff --git a/libDashboards/db/models/DashDashboardProjects.py b/libDashboards/db/models/DashDashboardProjects.py index f0be263..3888075 100644 --- a/libDashboards/db/models/DashDashboardProjects.py +++ b/libDashboards/db/models/DashDashboardProjects.py @@ -64,6 +64,10 @@ def get_dashboards_for_project(project_id: int, enabled_only=True) -> []: return ddp + @staticmethod + def get_for_project_and_dashboard(project_id: int, dashboard_id: int): + return DashDashboardProjects.query.filter_by(id_project=project_id, id_dashboard=dashboard_id).first() + @staticmethod def create_defaults(test=False): from libDashboards.db.models.DashDashboards import DashDashboards diff --git a/libDashboards/db/models/DashDashboardSites.py b/libDashboards/db/models/DashDashboardSites.py index 5eb265f..61e7876 100644 --- a/libDashboards/db/models/DashDashboardSites.py +++ b/libDashboards/db/models/DashDashboardSites.py @@ -62,6 +62,10 @@ def get_dashboards_for_site(site_id: int, enabled_only=True) -> []: return dds + @staticmethod + def get_for_site_and_dashboard(site_id: int, dashboard_id: int): + return DashDashboardSites.query.filter_by(id_site=site_id, id_dashboard=dashboard_id).first() + @staticmethod def create_defaults(test=False): if test: diff --git a/libDashboards/db/models/DashDashboards.py b/libDashboards/db/models/DashDashboards.py index 0ac1420..8aa21f5 100644 --- a/libDashboards/db/models/DashDashboards.py +++ b/libDashboards/db/models/DashDashboards.py @@ -13,9 +13,11 @@ class DashDashboards(BaseModel): dashboard_description = Column(String, nullable=True) # Dashboard user-visible description dashboard_versions = relationship("DashDashboardVersions", back_populates="dashboard_version_dashboard", - order_by="DashDashboardVersions.dashboard_version", viewonly=True) - dashboard_sites = relationship("DashDashboardSites", back_populates="dashboard_site_dashboard") - dashboard_projects = relationship("DashDashboardProjects", back_populates="dashboard_project_dashboard") + order_by="DashDashboardVersions.dashboard_version", cascade="all,delete") + dashboard_sites = relationship("DashDashboardSites", cascade="all,delete", + back_populates="dashboard_site_dashboard") + dashboard_projects = relationship("DashDashboardProjects", cascade="all,delete", + back_populates="dashboard_project_dashboard") def to_json(self, ignore_fields=None, minimal=False, latest=True): if ignore_fields is None: diff --git a/tests/api/user/test_QueryDashboard.py b/tests/api/user/test_QueryDashboard.py index 59e62c8..1704b01 100644 --- a/tests/api/user/test_QueryDashboard.py +++ b/tests/api/user/test_QueryDashboard.py @@ -1,6 +1,8 @@ from tests.BaseDashboardsAPITest import BaseDashboardsAPITest from libDashboards.db.models import DashDashboards from libDashboards.db.models.DashDashboardVersions import DashDashboardVersions +from libDashboards.db.models.DashDashboardSites import DashDashboardSites +from libDashboards.db.models.DashDashboardProjects import DashDashboardProjects class QueryDashboardTest(BaseDashboardsAPITest): @@ -230,7 +232,7 @@ def test_post_and_delete_new_global(self): params=delete_params) self.assertEqual(200, response.status_code) - def test_post_and_update_new_global(self): + def test_post_and_delete_update_global(self): with self.app_context(): global_dashs = DashDashboards.get_dashboards_globals() global_dash_id = global_dashs[0].id_dashboard @@ -251,7 +253,7 @@ def test_post_and_update_new_global(self): json=dashboard) self.assertEqual(400, response.status_code) # Can't change id if uuid - dashboard = {'dashboard': {'id_dashboard':global_dash_id}} + dashboard = {'dashboard': {'id_dashboard': global_dash_id}} response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], json=dashboard) self.assertEqual(403, response.status_code) # Can't access global dashboard @@ -280,6 +282,189 @@ def test_post_and_update_new_global(self): # Manually remove latest version DashDashboardVersions.delete(updated.dashboard_versions[-1].id_dashboard_version) + def test_post_and_delete_new_site(self): + with self.app_context(): + # Don't retest what was tested before... + dashboard = {'dashboard': {'dashboard_name': 'Test Site Dashboard', + 'dashboard_definition': '{"bla": "bla"}', + 'dashboard_sites': [{'id_site': 3}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(403, response.status_code) # Unavailable site for the user + + dashboard['dashboard']['dashboard_sites'] = [{'id_site': 1}] + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + self._check_json(response.json) + updated = DashDashboards.get_by_id(response.json['id_dashboard']) + self.assertTrue(len(updated.dashboard_sites) == 1) + self.assertEqual(1, updated.dashboard_sites[-1].id_site) + + # Remove currently added dashboard + id_to_del = response.json['id_dashboard'] + delete_params = {'id': id_to_del} + response = self._delete_with_user_token_auth(self.test_client, token=self._users['superadmin'], + params=delete_params) + self.assertEqual(200, response.status_code) + + def test_post_and_delete_update_site(self): + with self.app_context(): + # Note: Tests common for updating are done in the "global dashboard" tests + site_dashboards = [dash.dashboard_site_dashboard for dash in DashDashboardSites.get_dashboards_for_site(1)] + site_id_dashboard = site_dashboards[0].id_dashboard + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(403, response.status_code) # Can't access this site dashboard + + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard, 'dashboard_name': 'Site New Name'}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) # Was updated + updated = DashDashboards.get_by_id(site_id_dashboard) + self.assertEqual('Site New Name', updated.dashboard_name) + + # Add new site to dashboard + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard, + 'dashboard_sites': [{'id_site': 1}, {'id_site': 2}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(403, response.status_code) # No access to site 2 + + response = self._post_with_user_token_auth(self.test_client, token=self._users['superadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(site_id_dashboard) + self.assertEqual(2, len(updated.dashboard_sites)) + for dds in updated.dashboard_sites: + self.assertTrue(dds.id_site in [1, 2]) + + # Try to associate both site and projects + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard, + 'dashboard_sites': [{'id_site': 1}], + 'dashboard_projects': [{'id_project': 1}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(400, response.status_code) # No can do + + # Try to disable one site dashboard + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard, + 'dashboard_sites': [{'id_site': 1, 'dashboard_site_enabled': False}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(site_id_dashboard) + for dds in updated.dashboard_sites: + if dds.id_site == 1: + self.assertFalse(dds.dashboard_site_enabled) + + # Remove one site from dashboard + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard, + 'dashboard_sites': []}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(site_id_dashboard) + self.assertEqual(1, len(updated.dashboard_sites)) # Site #2 not removed + self.assertEqual(2, updated.dashboard_sites[0].id_site) + dashboard = {'dashboard': {'id_dashboard': site_id_dashboard, + 'dashboard_sites': [{'id_site': 1, 'dashboard_site_enabled': True}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['superadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(site_id_dashboard) + self.assertEqual(1, len(updated.dashboard_sites)) # Back to normal + self.assertEqual(1, updated.dashboard_sites[0].id_site) + + def test_post_and_delete_new_project(self): + with self.app_context(): + # Don't retest what was tested before... + dashboard = {'dashboard': {'dashboard_name': 'Test Project Dashboard', + 'dashboard_definition': '{"bla": "bla"}', + 'dashboard_projects': [{'id_project': 3}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(403, response.status_code) # Unavailable for the user + + dashboard['dashboard']['dashboard_projects'] = [{'id_project': 1}] + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + self._check_json(response.json) + updated = DashDashboards.get_by_id(response.json['id_dashboard']) + self.assertTrue(len(updated.dashboard_projects) == 1) + self.assertEqual(1, updated.dashboard_projects[-1].id_project) + + # Remove currently added dashboard + id_to_del = response.json['id_dashboard'] + delete_params = {'id': id_to_del} + response = self._delete_with_user_token_auth(self.test_client, token=self._users['superadmin'], + params=delete_params) + self.assertEqual(200, response.status_code) + + def test_post_and_delete_update_project(self): + with self.app_context(): + # Note: Tests common for updating are done in the "global dashboard" tests + proj_dashboards = [dash.dashboard_project_dashboard + for dash in DashDashboardProjects.get_dashboards_for_project(1)] + proj_id_dashboard = proj_dashboards[0].id_dashboard + dashboard = {'dashboard': {'id_dashboard': proj_id_dashboard}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['user'], + json=dashboard) + self.assertEqual(403, response.status_code) # Can't access this dashboard + + dashboard = {'dashboard': {'id_dashboard': proj_id_dashboard, 'dashboard_name': 'Project New Name'}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) # Was updated + updated = DashDashboards.get_by_id(proj_id_dashboard) + self.assertEqual('Project New Name', updated.dashboard_name) + + # Add new project to dashboard + dashboard = {'dashboard': {'id_dashboard': proj_id_dashboard, + 'dashboard_projects': [{'id_project': 1}, {'id_project': 2}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(403, response.status_code) # No access to site 2 + + response = self._post_with_user_token_auth(self.test_client, token=self._users['siteadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(proj_id_dashboard) + self.assertEqual(2, len(updated.dashboard_projects)) + for ddp in updated.dashboard_projects: + self.assertTrue(ddp.id_project in [1, 2]) + + # Try to force a version dashboard + dashboard = {'dashboard': {'id_dashboard': proj_id_dashboard, + 'dashboard_projects': [{'id_project': 1, 'dashboard_project_version': 3}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(proj_id_dashboard) + for ddp in updated.dashboard_projects: + if ddp.id_project == 1: + self.assertEqual(3, ddp.dashboard_project_version) + + # Remove one site from dashboard + dashboard = {'dashboard': {'id_dashboard': proj_id_dashboard, + 'dashboard_projects': []}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['projectadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(proj_id_dashboard) + self.assertEqual(1, len(updated.dashboard_projects)) # Project #2 not removed + self.assertEqual(2, updated.dashboard_projects[0].id_project) + dashboard = {'dashboard': {'id_dashboard': proj_id_dashboard, + 'dashboard_projects': [{'id_project': 1, 'dashboard_project_version': None}]}} + response = self._post_with_user_token_auth(self.test_client, token=self._users['superadmin'], + json=dashboard) + self.assertEqual(200, response.status_code) + updated = DashDashboards.get_by_id(proj_id_dashboard) + self.assertEqual(1, len(updated.dashboard_projects)) # Back to normal + self.assertEqual(1, updated.dashboard_projects[0].id_project) + def _check_json(self, json_data: str, minimal=False): self.assertTrue(json_data.__contains__('id_dashboard')) self.assertTrue(json_data.__contains__('dashboard_uuid'))