diff --git a/loko/assets/permissions.py b/loko/assets/permissions.py index 942adea..e5074f2 100644 --- a/loko/assets/permissions.py +++ b/loko/assets/permissions.py @@ -385,12 +385,16 @@ def can_delete_location(user, location): return _check_instance_permission(user, location, 'can_delete', _fallback) -def _get_asset_thematic(asset): +def _get_asset_thematic(asset, thematics_by_code=None): """Get the thematic for an asset instance or model class.""" if asset is None: return None - from common.models import Thematic + def _resolve_thematic(code): + if thematics_by_code is not None: + return thematics_by_code.get(code) + from common.models import Thematic + return Thematic.objects.filter(code=code).first() thematic_mappings = { 'naturerwiz': 'water', @@ -422,10 +426,9 @@ def _get_asset_thematic(asset): class_name = asset.__name__.lower() for prefix, thematic_code in thematic_mappings.items(): if class_name.startswith(prefix): - try: - return Thematic.objects.get(code=thematic_code) - except Thematic.DoesNotExist: - pass + th = _resolve_thematic(thematic_code) + if th: + return th # Fallback: check AssetCategory via ContentType from django.contrib.contenttypes.models import ContentType from assets.models import AssetCategory @@ -468,10 +471,9 @@ def _get_asset_thematic(asset): class_name = asset.__class__.__name__.lower() for prefix, thematic_code in thematic_mappings.items(): if class_name.startswith(prefix): - try: - return Thematic.objects.get(code=thematic_code) - except Thematic.DoesNotExist: - pass + th = _resolve_thematic(thematic_code) + if th: + return th return None diff --git a/loko/inspections/apps.py b/loko/inspections/apps.py index 8b07c6f..87093e8 100644 --- a/loko/inspections/apps.py +++ b/loko/inspections/apps.py @@ -17,7 +17,7 @@ class InspectionsConfig(AppConfig): if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False): return True user_config = getattr(user, 'config', None) - if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern): + if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or (user_config.is_intern and not user_config.has_role('viewer'))): return True from inspections.permissions import can_view_inspection return can_view_inspection(user, getattr(obj, 'inspection', None)) diff --git a/loko/inspections/permissions.py b/loko/inspections/permissions.py index d46ac6c..9ed3758 100644 --- a/loko/inspections/permissions.py +++ b/loko/inspections/permissions.py @@ -9,13 +9,17 @@ from django.contrib.contenttypes.models import ContentType def is_user_internal_or_admin(user): """ Détermine si un utilisateur est un utilisateur interne ou un administrateur ayant un accès global. + Les utilisateurs ayant uniquement le rôle 'viewer' ne disposent pas d'un accès global inconditionnel + et doivent être filtrés par leurs thématiques autorisées. """ if not user or not user.is_authenticated: return False if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False): return True user_config = getattr(user, 'config', None) - if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern): + if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')): + return True + if user_config and user_config.is_intern and not user_config.has_role('viewer'): return True return False @@ -45,10 +49,52 @@ def can_edit_inspection(user, inspection): return False +def get_inspection_thematic(inspection): + """ + Détermine la thématique associée à une inspection. + """ + if not inspection: + return None + + from assets.permissions import _get_asset_thematic + + # 1. Cible directe (Asset ou Intervention) + if inspection.asset: + th = _get_asset_thematic(inspection.asset) + if th: + return th + + if inspection.asset_content_type_id: + if inspection.asset_content_type.app_label == 'interventions': + from interventions.models import Intervention + interv = Intervention.objects.filter(pk=inspection.asset_object_id).first() + if interv and interv.thematic: + return interv.thematic + else: + ModelClass = inspection.asset_content_type.model_class() + if ModelClass: + th = _get_asset_thematic(ModelClass) + if th: + return th + + # 2. Mission d'intervention + if inspection.mission_intervention_id and inspection.mission_intervention: + if inspection.mission_intervention.thematic: + return inspection.mission_intervention.thematic + + # 3. Intervention contrôlée + if inspection.checked_intervention_id and inspection.checked_intervention: + if inspection.checked_intervention.thematic: + return inspection.checked_intervention.thematic + + return None + + def can_view_inspection(user, inspection): """ Vérifie si un utilisateur a le droit de visualiser une inspection. - - Administrateurs et utilisateurs internes (is_intern=True) : accès complet. + - Administrateurs et utilisateurs internes (is_intern=True, hors Viewer) : accès complet. + - Utilisateurs ayant le rôle Viewer : accès restreint aux thématiques autorisées. - Utilisateurs externes : seules les inspections qui les concernent sont visibles : 1. L'utilisateur est l'auteur/inspecteur de l'inspection. 2. L'inspection est liée à une mission d'intervention visible par l'utilisateur. @@ -64,6 +110,19 @@ def can_view_inspection(user, inspection): if not inspection: return False + user_config = getattr(user, 'config', None) + if user_config and user_config.has_role('viewer'): + from common.models import UserThematics + viewable_thematic_ids = set( + UserThematics.objects.filter(user_config=user_config) + .filter(Q(can_view_assets=True) | Q(can_view_interventions=True)) + .values_list('thematic_id', flat=True) + ) + if not viewable_thematic_ids: + return False + insp_thematic = get_inspection_thematic(inspection) + return bool(insp_thematic and insp_thematic.id in viewable_thematic_ids) + # 1. Auteur / Inspecteur de l'inspection if inspection.inspector_id == user.id: return True @@ -198,6 +257,63 @@ def filter_viewable_inspections_for_user(user, qs=None): return qs user_config = getattr(user, 'config', None) + if not user_config: + return qs.none() + + if user_config.has_role('viewer'): + from common.models import UserThematics + from assets.permissions import _get_asset_thematic + from interventions.models import Intervention + + viewable_thematic_ids = set( + UserThematics.objects.filter(user_config=user_config) + .filter(Q(can_view_assets=True) | Q(can_view_interventions=True)) + .values_list('thematic_id', flat=True) + ) + if not viewable_thematic_ids: + return qs.none() + + viewer_conditions = [] + + # 1. Missions ou interventions contrôlées + viewer_conditions.append(Q(mission_intervention__thematic_id__in=viewable_thematic_ids)) + viewer_conditions.append(Q(checked_intervention__thematic_id__in=viewable_thematic_ids)) + + # 2. Cible intervention + interv_ct = ContentType.objects.filter(app_label='interventions', model='intervention').first() + if interv_ct: + matching_intervs = Intervention.objects.filter(thematic_id__in=viewable_thematic_ids) + viewer_conditions.append(Q(asset_content_type=interv_ct, asset_object_id__in=matching_intervs.values('pk'))) + + # 3. Cible asset + ct_ids = set(qs.values_list('asset_content_type_id', flat=True).distinct()) + for ct_id in ct_ids: + if not ct_id: + continue + try: + ct = ContentType.objects.get(pk=ct_id) + except ContentType.DoesNotExist: + continue + if ct.app_label == 'interventions': + continue + ModelClass = ct.model_class() + if not ModelClass: + continue + + th = _get_asset_thematic(ModelClass) + if th and th.id in viewable_thematic_ids: + viewer_conditions.append(Q(asset_content_type=ct)) + elif hasattr(ModelClass, 'category'): + matching_ids = list(ModelClass.objects.filter(category__thematic_id__in=viewable_thematic_ids).values_list('pk', flat=True)[:1000]) + if matching_ids: + viewer_conditions.append(Q(asset_content_type=ct, asset_object_id__in=matching_ids)) + + if not viewer_conditions: + return qs.none() + + q = reduce(or_, viewer_conditions) + return qs.filter(q).distinct() + conditions = [] # 1. Auteur / Inspecteur de l'inspection diff --git a/loko/inspections/tests.py b/loko/inspections/tests.py index 2b7e1f7..f4db302 100644 --- a/loko/inspections/tests.py +++ b/loko/inspections/tests.py @@ -1022,3 +1022,122 @@ class ExternalUserInspectionVisibilityTestCase(TestCase): self.assertEqual(detail_res.status_code, 200) data = detail_res.json() self.assertTrue(data['is_signpanel']) + + +class ViewerInspectionPermissionsAndGeojsonTestCase(TestCase): + def setUp(self): + self.client = Client() + self.viewer_user = User.objects.create_user(username='viewer_user', password='password123') + from common.models import UserConfig, Role, UserThematics + viewer_role, _ = Role.objects.get_or_create(name='viewer') + self.viewer_config = UserConfig.objects.create(user=self.viewer_user, is_intern=False) + self.viewer_config.roles.add(viewer_role) + + self.water_thematic, _ = Thematic.objects.get_or_create( + code='water', + defaults={'name_fr': "Gestion de l'Eau"} + ) + self.parking_thematic, _ = Thematic.objects.get_or_create( + code='parking', + defaults={'name_fr': "Stationnement"} + ) + + # Viewer only has access to water + UserThematics.objects.create( + user_config=self.viewer_config, + thematic=self.water_thematic, + can_view_assets=True, + can_view_interventions=True, + can_view_projects=True + ) + + # Create assets + from assets.models import NatureRWIZ, ParkingSpot + from django.contrib.gis.geos import Point, Polygon, MultiPolygon + poly = Polygon(((0, 0), (0, 10), (10, 10), (10, 0), (0, 0)), srid=3812) + self.rwiz = NatureRWIZ.objects.create( + code='RWIZ-001', + geom=MultiPolygon(poly, srid=3812) + ) + self.spot = ParkingSpot.objects.create( + code='SPOT-001', + geom=poly + ) + self.rwiz_ct = ContentType.objects.get_for_model(NatureRWIZ) + self.spot_ct = ContentType.objects.get_for_model(ParkingSpot) + + inspector = User.objects.create_user(username='insp_user', password='password123') + self.insp_water = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=inspector, + result_status='compliant' + ) + self.insp_parking = Inspection.objects.create( + asset_content_type=self.spot_ct, + asset_object_id=self.spot.id, + inspector=inspector, + result_status='compliant' + ) + + from django.core.files.uploadedfile import SimpleUploadedFile + dummy_file = SimpleUploadedFile("doc.txt", b"content", content_type="text/plain") + InspectionDocument.objects.create(inspection=self.insp_water, file=dummy_file) + + def test_viewer_thematic_permissions(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + + self.assertTrue(can_view_inspection(self.viewer_user, self.insp_water)) + self.assertFalse(can_view_inspection(self.viewer_user, self.insp_parking)) + + qs = filter_viewable_inspections_for_user(self.viewer_user) + self.assertIn(self.insp_water, qs) + self.assertNotIn(self.insp_parking, qs) + + def test_inspections_geojson_for_viewer(self): + self.client.login(username='viewer_user', password='password123') + url = reverse('inspections:inspections_geojson') + + # Test GET response correctness + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertEqual(data['type'], 'FeatureCollection') + + feature_ids = [f['id'] for f in data['features']] + self.assertIn(self.insp_water.id, feature_ids) + self.assertNotIn(self.insp_parking.id, feature_ids) + + water_feature = next(f for f in data['features'] if f['id'] == self.insp_water.id) + self.assertEqual(water_feature['properties']['documents_count'], 1) + self.assertFalse(water_feature['properties']['can_edit']) + + def test_inspections_geojson_query_efficiency(self): + # Create multiple water inspections with documents + inspector = self.insp_water.inspector + from django.core.files.uploadedfile import SimpleUploadedFile + from django.test.utils import CaptureQueriesContext + from django.db import connection + + for i in range(5): + insp = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=inspector, + result_status='compliant' + ) + dummy = SimpleUploadedFile(f"doc_{i}.txt", b"content", content_type="text/plain") + InspectionDocument.objects.create(inspection=insp, file=dummy) + + self.client.login(username='viewer_user', password='password123') + url = reverse('inspections:inspections_geojson') + + # Ensure query count doesn't blow up (O(1) database queries instead of O(N)) + with CaptureQueriesContext(connection) as ctx: + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + + # Number of queries should be bounded and independent of inspection count + # (O(1) queries instead of O(N), unaffected by inspection count) + self.assertLessEqual(len(ctx), 28) + diff --git a/loko/inspections/views.py b/loko/inspections/views.py index e651524..71a1107 100644 --- a/loko/inspections/views.py +++ b/loko/inspections/views.py @@ -44,7 +44,8 @@ def _filter_inspections_queryset(request): from .permissions import filter_viewable_inspections_for_user qs = filter_viewable_inspections_for_user(request.user) qs = qs.select_related( - 'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention' + 'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention', + 'parkingspotinspection', 'giepinspection', 'signpanelinspection' ).prefetch_related('documents') thematics = request.GET.getlist('thematic') @@ -182,7 +183,8 @@ def inspections_list(request): can_inspect = bool(insp_thematics and insp_thematics.exists()) if is_mobile: - filter_type = request.GET.get('tab', 'my') + default_tab = 'my' if can_inspect else 'all' + filter_type = request.GET.get('tab', default_tab) thematic_code = request.GET.get('thematic', '') status_filter = request.GET.get('status', '') nature_filter = request.GET.get('nature', '') @@ -193,7 +195,8 @@ def inspections_list(request): from .permissions import filter_viewable_inspections_for_user qs = filter_viewable_inspections_for_user(request.user) qs = qs.select_related( - 'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention' + 'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention', + 'parkingspotinspection', 'giepinspection', 'signpanelinspection' ).prefetch_related('documents').order_by('-inspection_date') if filter_type == 'my': @@ -1004,7 +1007,10 @@ def inspections_geojson(request): from .permissions import filter_viewable_inspections_for_user, get_contract_filtered_asset_qs qs = filter_viewable_inspections_for_user(request.user) - qs = qs.select_related('asset_content_type', 'inspector', 'mission_intervention').prefetch_related('documents') + qs = qs.select_related( + 'asset_content_type', 'inspector', 'mission_intervention', + 'parkingspotinspection', 'giepinspection', 'signpanelinspection' + ).prefetch_related('documents') if mission_id: qs = qs.filter(mission_intervention_id=mission_id) @@ -1056,6 +1062,11 @@ def inspections_geojson(request): features = [] user_config = getattr(request.user, 'config', None) + is_admin_user = ( + request.user.is_superuser or + request.user.is_staff or + (user_config and (user_config.has_role('admin') or user_config.has_role('top_manager'))) + ) is_external = (user_config is None) or (not user_config.is_intern) must_limit = is_external and (not user_config or user_config.limit_assets_to_contracts) accessible_contract_ids = [] @@ -1091,7 +1102,9 @@ def inspections_geojson(request): wgs84_json=AsGeoJSON(Transform(Centroid('geom'), 4326)) ) - asset_geom_map = dict(asset_qs.values_list('id', 'wgs84_json')) + asset_objs = list(asset_qs) + asset_geom_map = {obj.id: obj.wgs84_json for obj in asset_objs} + asset_obj_map = {obj.id: obj for obj in asset_objs} if not asset_geom_map: continue @@ -1111,7 +1124,7 @@ def inspections_geojson(request): active_insps = [i for i in insp_list if i.result_status != 'cancelled'] top_insp = active_insps[0] if active_insps else insp_list[0] - target_obj = top_insp.asset + target_obj = asset_obj_map.get(asset_id) inspections_data = [] for insp in insp_list: @@ -1124,6 +1137,8 @@ def inspections_geojson(request): 'is_image': doc.is_image }) + can_edit = is_admin_user or (insp.inspector_id == request.user.id) + inspections_data.append({ 'id': insp.id, 'date': insp.inspection_date.strftime('%d/%m/%Y %H:%M') if insp.inspection_date else '', @@ -1139,7 +1154,7 @@ def inspections_geojson(request): 'notes': insp.notes or '', 'target_display': insp.target_display, 'mobile_asset_url': insp.mobile_asset_url, - 'can_edit': insp.detailed_inspection.can_edit(request.user), + 'can_edit': can_edit, 'documents': docs_data, }) @@ -1170,6 +1185,14 @@ def inspections_geojson(request): qs = _filter_inspections_queryset(request)[:5000] inspections_list = list(qs) + user_config = getattr(request.user, 'config', None) + is_admin_user = ( + request.user.is_superuser or + request.user.is_staff or + (user_config and (user_config.has_role('admin') or user_config.has_role('top_manager'))) + ) + thematics_by_code = {t.code: t for t in Thematic.objects.all()} + ct_to_ids = {} for insp in inspections_list: ct_to_ids.setdefault(insp.asset_content_type_id, set()).add(insp.asset_object_id) @@ -1186,7 +1209,7 @@ def inspections_geojson(request): wgs84_geojson=AsGeoJSON(Transform(Centroid('geom'), 4326)) ) if hasattr(ModelClass, 'category'): - asset_qs = asset_qs.select_related('category') + asset_qs = asset_qs.select_related('category', 'category__thematic') for asset_obj in asset_qs: assets_by_ct_and_id[(ct_id, asset_obj.id)] = asset_obj except Exception: @@ -1208,7 +1231,7 @@ def inspections_geojson(request): if lon is not None and lat is not None: geometry = {"type": "Point", "coordinates": [lon, lat]} - th = _get_asset_thematic(asset) if asset else None + th = _get_asset_thematic(asset, thematics_by_code=thematics_by_code) if asset else None thematic_name = th.name if th else '' thematic_code = th.code if th else '' thematic_icon = th.icon if th and th.icon else 'bi-geo-alt' @@ -1222,6 +1245,8 @@ def inspections_geojson(request): target_display = str(asset) if asset else f"Cible #{insp.asset_object_id}" asset_url = getattr(asset, 'get_absolute_url', lambda: '')() if asset and hasattr(asset, 'get_absolute_url') else '' + can_edit = is_admin_user or (insp.inspector_id == request.user.id) + features.append({ "type": "Feature", "id": insp.id, @@ -1254,8 +1279,8 @@ def inspections_geojson(request): "details_summary": insp.details_summary if insp.details_summary and insp.details_summary != '-' else '', "verified_obsolescence": insp.verified_obsolescence or '', "is_cancelled": insp.is_cancelled, - "documents_count": insp.documents.count(), - "can_edit": insp.detailed_inspection.can_edit(request.user), + "documents_count": len(insp.documents.all()), + "can_edit": can_edit, } }) @@ -1279,6 +1304,7 @@ def inspections_export_excel(request): qs = _filter_inspections_queryset(request)[:5000] inspections_list = list(qs) + thematics_by_code = {t.code: t for t in Thematic.objects.all()} ct_to_ids = {} for insp in inspections_list: @@ -1292,7 +1318,7 @@ def inspections_export_excel(request): if ModelClass: asset_qs = ModelClass.objects.filter(id__in=obj_ids) if hasattr(ModelClass, 'category'): - asset_qs = asset_qs.select_related('category') + asset_qs = asset_qs.select_related('category', 'category__thematic') for asset_obj in asset_qs: assets_by_ct_and_id[(ct_id, asset_obj.id)] = asset_obj except Exception: diff --git a/loko/mobile/templates/mobile/mobile_index.html b/loko/mobile/templates/mobile/mobile_index.html index 62b7e02..d70cf79 100644 --- a/loko/mobile/templates/mobile/mobile_index.html +++ b/loko/mobile/templates/mobile/mobile_index.html @@ -100,7 +100,7 @@ {% endif %} - {% if user.config|has_any_role:"admin,manager,controller,external_manager,top_manager" %} + {% if user.config|has_any_role:"admin,manager,controller,external_manager,top_manager,viewer" %}