From 15403b8604e95ca74069d4366c46baacd18418e2 Mon Sep 17 00:00:00 2001 From: kdeterme Date: Fri, 9 Oct 2026 22:59:10 +0200 Subject: [PATCH] fix(inspections): scope viewable inspections and assets by contract and municipality --- loko/assets/permissions.py | 31 ++++++++- loko/inspections/permissions.py | 120 ++++++++++++++++++++++++++++---- loko/inspections/tests.py | 56 +++++++++++++++ 3 files changed, 191 insertions(+), 16 deletions(-) diff --git a/loko/assets/permissions.py b/loko/assets/permissions.py index 62ac89f..3adcbe2 100644 --- a/loko/assets/permissions.py +++ b/loko/assets/permissions.py @@ -258,9 +258,9 @@ def can_view_asset(user, asset): if not can_view: return False - # Vérifier la restriction par contrat pour les utilisateurs externes + # Vérifier la restriction par contrat et commune pour les utilisateurs externes ou limités aux contrats 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) + must_limit = is_external or (user_config and user_config.limit_assets_to_contracts) if must_limit and user_config: asset_cids = get_asset_contract_ids(asset) if asset_cids is not None: @@ -271,8 +271,35 @@ def can_view_asset(user, asset): can_view_assets=True ).values_list('contract_id', flat=True) ) + if user_config.default_contract_id: + accessible_cids.add(user_config.default_contract_id) if not (asset_cids & accessible_cids): return False + else: + # Vérifier la restriction par commune pour les assets sans contrat direct + accessible_munis = get_user_accessible_municipalities(user) + accessible_muni_ids = set(accessible_munis.values_list('id', flat=True)) + if accessible_muni_ids: + muni_id = getattr(asset, 'municipality_id', None) + if not muni_id and hasattr(asset, 'location') and asset.location: + muni_id = getattr(asset.location, 'municipality_id', None) + if not muni_id and hasattr(asset, 'street') and asset.street: + muni_id = getattr(asset.street, 'municipality_id', None) + if not muni_id and hasattr(asset, 'pole') and asset.pole: + muni_id = getattr(asset.pole, 'municipality_id', None) + if not muni_id and hasattr(asset.pole, 'street') and asset.pole.street: + muni_id = getattr(asset.pole.street, 'municipality_id', None) + if not muni_id and hasattr(asset, 'intersection') and asset.intersection: + muni_id = getattr(asset.intersection, 'municipality_id', None) + if not muni_id and hasattr(asset, 'building') and asset.building: + muni_id = getattr(asset.building, 'municipality_id', None) + if muni_id is not None: + if muni_id not in accessible_muni_ids: + return False + elif getattr(asset, 'geom', None): + from common.models import Municipality + if not Municipality.objects.filter(id__in=accessible_muni_ids, geom__intersects=asset.geom).exists(): + return False return True return _check_instance_permission(user, asset, 'can_view', _fallback) diff --git a/loko/inspections/permissions.py b/loko/inspections/permissions.py index f61d0d8..eab3416 100644 --- a/loko/inspections/permissions.py +++ b/loko/inspections/permissions.py @@ -8,9 +8,11 @@ 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. + Détermine si un utilisateur est un administrateur ayant un accès global inconditionnel + ou un utilisateur interne non restreint aux contrats/communes. + Les utilisateurs limités aux contrats (limit_assets_to_contracts=True) ou ayant + le rôle 'viewer' ne disposent pas d'un accès global inconditionnel et doivent être + filtrés par leur périmètre autorisé. """ if not user or not user.is_authenticated: return False @@ -18,9 +20,11 @@ def is_user_internal_or_admin(user): return True user_config = getattr(user, 'config', None) if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')): - return True + if not user_config.limit_assets_to_contracts: + return True if user_config and user_config.is_intern and not user_config.has_role('viewer'): - return True + if not user_config.limit_assets_to_contracts: + return True return False @@ -136,9 +140,11 @@ def can_view_inspection(user, inspection): insp_thematic = get_inspection_thematic(inspection) return bool(insp_thematic and insp_thematic.id in viewable_thematic_ids) - # 0. Équipe assignée + # 0. Équipe ou membre assigné if inspection.assigned_team_id and inspection.assigned_team.members.filter(user=user).exists(): return True + if inspection.assigned_member_id and inspection.assigned_member.user_id == user.id: + return True # 1. Auteur / Inspecteur de l'inspection if inspection.inspector_id == user.id: @@ -222,26 +228,30 @@ def get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids): if not accessible_contract_ids: return ModelClass.objects.none() return ModelClass.objects.filter(trafficlight_contracts__status='active', trafficlight_contracts__contract_id__in=accessible_contract_ids) - elif class_name in ('trafficlightpole', 'trafficlightlantern', 'trafficlightcable', 'trafficlightdetector', + elif class_name in ('trafficlightpole', 'trafficlightcable', 'trafficlightdetector', 'trafficlightcontroller', 'trafficlightelectricalcabinet', 'trafficlightradar', 'trafficlightaccessory'): if not accessible_contract_ids: return ModelClass.objects.none() return ModelClass.objects.filter(intersection__trafficlight_contracts__status='active', intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids) - elif class_name == 'trafficlightradarpole': + elif class_name == 'trafficlightlantern': if not accessible_contract_ids: return ModelClass.objects.none() return ModelClass.objects.filter(pole__intersection__trafficlight_contracts__status='active', pole__intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'trafficlightradarpole': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(intersection__trafficlight_contracts__status='active', intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids) # Clean elif class_name == 'cleanlocation': if not accessible_contract_ids: return ModelClass.objects.none() - return ModelClass.objects.filter(clean_location_contracts__contract_id__in=accessible_contract_ids) + return ModelClass.objects.filter(contracts__contract_id__in=accessible_contract_ids) elif class_name in ('cleanlitterbin', 'cleanglasscontainer'): if not accessible_contract_ids: return ModelClass.objects.none() - return ModelClass.objects.filter(location__clean_location_contracts__contract_id__in=accessible_contract_ids) + return ModelClass.objects.filter(location__contracts__contract_id__in=accessible_contract_ids) # Artworks elif class_name in ('artwork', 'monument', 'fountain'): @@ -249,7 +259,82 @@ def get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids): return ModelClass.objects.none() return ModelClass.objects.filter(maintenance_contract_id__in=accessible_contract_ids) - # Modèles sans notion de contrat (ex: ParkingSpot, NatureRWIZ, Road, PublicLighting, Sign, etc.) + # Modèles sans notion de contrat direct (ex: ParkingSpot, NatureTree, Road, PublicLighting, Sign, PublicBuilding, etc.) + return None + + +def get_scoped_asset_qs(ModelClass, accessible_contract_ids, accessible_muni_ids): + """ + Retourne un QuerySet filtré des instances de ModelClass selon : + 1. Les contrats autorisés (pour les modèles gérant les contrats). + 2. Les communes autorisées (pour les modèles localisés par commune, rue, emplacement ou bâtiment). + Retourne None si ModelClass n'a pas de restriction applicable. + """ + # 1. Filtre contrat si le modèle dispose d'une gestion de contrats + contract_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids) + if contract_qs is not None: + return contract_qs + + # 2. Filtre commune pour les modèles sans contrat direct (uniquement si des communes sont définies) + if not accessible_muni_ids: + return None + + field_names = [f.name for f in ModelClass._meta.get_fields()] + + if 'municipality' in field_names: + return ModelClass.objects.filter(municipality_id__in=accessible_muni_ids) + + if 'location' in field_names: + loc_field = ModelClass._meta.get_field('location') + if hasattr(loc_field, 'related_model') and loc_field.related_model: + loc_fields = [f.name for f in loc_field.related_model._meta.get_fields()] + if 'municipality' in loc_fields: + return ModelClass.objects.filter(location__municipality_id__in=accessible_muni_ids) + + if 'street' in field_names: + street_field = ModelClass._meta.get_field('street') + if hasattr(street_field, 'related_model') and street_field.related_model: + st_fields = [f.name for f in street_field.related_model._meta.get_fields()] + if 'municipality' in st_fields: + return ModelClass.objects.filter(street__municipality_id__in=accessible_muni_ids) + + if 'pole' in field_names: + pole_field = ModelClass._meta.get_field('pole') + if hasattr(pole_field, 'related_model') and pole_field.related_model: + p_fields = [f.name for f in pole_field.related_model._meta.get_fields()] + if 'municipality' in p_fields: + return ModelClass.objects.filter(pole__municipality_id__in=accessible_muni_ids) + if 'street' in p_fields: + return ModelClass.objects.filter(pole__street__municipality_id__in=accessible_muni_ids) + + if 'building' in field_names: + b_field = ModelClass._meta.get_field('building') + if hasattr(b_field, 'related_model') and b_field.related_model: + b_fields = [f.name for f in b_field.related_model._meta.get_fields()] + if 'municipality' in b_fields: + return ModelClass.objects.filter(building__municipality_id__in=accessible_muni_ids) + + if 'rwiz' in field_names: + rw_field = ModelClass._meta.get_field('rwiz') + if hasattr(rw_field, 'related_model') and rw_field.related_model: + rw_fields = [f.name for f in rw_field.related_model._meta.get_fields()] + if 'municipality' in rw_fields: + return ModelClass.objects.filter(rwiz__municipality_id__in=accessible_muni_ids) + + if 'road' in field_names: + r_field = ModelClass._meta.get_field('road') + if hasattr(r_field, 'related_model') and r_field.related_model: + r_fields = [f.name for f in r_field.related_model._meta.get_fields()] + if 'municipality' in r_fields: + return ModelClass.objects.filter(road__municipality_id__in=accessible_muni_ids) + + if 'geom' in field_names: + from common.models import Municipality + muni_geoms = list(Municipality.objects.filter(id__in=accessible_muni_ids, geom__isnull=False).values_list('geom', flat=True)) + if muni_geoms: + geom_q = reduce(or_, [Q(geom__intersects=mg) for mg in muni_geoms]) + return ModelClass.objects.filter(geom_q) + return None @@ -336,6 +421,7 @@ def filter_viewable_inspections_for_user(user, qs=None): # 1. Auteur / Inspecteur de l'inspection ou membre de l'équipe assignée conditions.append(Q(inspector=user)) conditions.append(Q(assigned_team__members__user=user)) + conditions.append(Q(assigned_member__user=user)) # 2. Interventions liées visibles (missions d'intervention, contrôles post-travaux, ou cible intervention) from interventions.permissions import filter_viewable_interventions_for_user @@ -351,15 +437,21 @@ def filter_viewable_inspections_for_user(user, qs=None): # 3. Assets cibles visibles if user_config: from common.models import UserThematics, UserContractAccess, UserAssetAccess, UserAssetTypeAccess, AssetTypePermissionConfig - from assets.permissions import _get_asset_thematic + from assets.permissions import _get_asset_thematic, get_user_accessible_municipalities viewable_thematic_ids = set( UserThematics.objects.filter(user_config=user_config, can_view_assets=True).values_list('thematic_id', flat=True) ) - must_limit = not user_config.is_intern and user_config.limit_assets_to_contracts + is_external = not user_config.is_intern + must_limit = is_external or user_config.limit_assets_to_contracts accessible_contract_ids = list( UserContractAccess.objects.filter(user_config=user_config, can_view_assets=True).values_list('contract_id', flat=True) ) if must_limit else [] + if user_config.default_contract_id and user_config.default_contract_id not in accessible_contract_ids: + accessible_contract_ids.append(user_config.default_contract_id) + + accessible_munis = get_user_accessible_municipalities(user) if must_limit else [] + accessible_muni_ids = list(accessible_munis.values_list('id', flat=True)) if must_limit else [] explicit_denies = defaultdict(set) for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=False).values_list('content_type_id', 'object_id'): @@ -413,7 +505,7 @@ def filter_viewable_inspections_for_user(user, qs=None): # Thématique autorisée (ou sans thématique assignée) if must_limit: - filtered_asset_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids) + filtered_asset_qs = get_scoped_asset_qs(ModelClass, accessible_contract_ids, accessible_muni_ids) if filtered_asset_qs is not None: ct_q = Q(asset_content_type=ct, asset_object_id__in=filtered_asset_qs.values('pk')) else: diff --git a/loko/inspections/tests.py b/loko/inspections/tests.py index 1e255e9..6e31e62 100644 --- a/loko/inspections/tests.py +++ b/loko/inspections/tests.py @@ -1109,6 +1109,62 @@ class ExternalUserInspectionVisibilityTestCase(TestCase): self.assertIn(insp_water, qs) self.assertIn(insp_struct_b, qs) + def test_internal_user_contract_limited_inspection_filtering(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + from common.models import UserConfig, UserContractAccess, UserThematics, Role + from django.contrib.auth import get_user_model + User = get_user_model() + + # Utilisateur interne restreint au contrat A (ex: type WBF) + user_a = User.objects.create_user(username='internal_a', password='pwd') + cfg_a = UserConfig.objects.create( + user=user_a, + is_intern=True, + limit_assets_to_contracts=True, + limit_interventions_to_contracts=True + ) + role_mgr, _ = Role.objects.get_or_create(name='manager') + cfg_a.roles.add(role_mgr) + UserThematics.objects.create(user_config=cfg_a, thematic=self.thematic_structures, can_view_assets=True) + UserContractAccess.objects.create(user_config=cfg_a, contract=self.contract_a, can_view_assets=True) + + # Utilisateur interne restreint au contrat B (ex: type CSE) + user_b = User.objects.create_user(username='internal_b', password='pwd') + cfg_b = UserConfig.objects.create( + user=user_b, + is_intern=True, + limit_assets_to_contracts=True, + limit_interventions_to_contracts=True + ) + cfg_b.roles.add(role_mgr) + UserThematics.objects.create(user_config=cfg_b, thematic=self.thematic_structures, can_view_assets=True) + UserContractAccess.objects.create(user_config=cfg_b, contract=self.contract_b, can_view_assets=True) + + insp_a = Inspection.objects.create( + asset_content_type=self.struct_ct, + asset_object_id=self.struct_a.id, + result_status='compliant' + ) + insp_b = Inspection.objects.create( + asset_content_type=self.struct_ct, + asset_object_id=self.struct_b.id, + result_status='compliant' + ) + + # User A ne doit voir que l'inspection du contrat A + self.assertTrue(can_view_inspection(user_a, insp_a)) + self.assertFalse(can_view_inspection(user_a, insp_b)) + qs_a = filter_viewable_inspections_for_user(user_a) + self.assertIn(insp_a, qs_a) + self.assertNotIn(insp_b, qs_a) + + # User B ne doit voir que l'inspection du contrat B + self.assertTrue(can_view_inspection(user_b, insp_b)) + self.assertFalse(can_view_inspection(user_b, insp_a)) + qs_b = filter_viewable_inspections_for_user(user_b) + self.assertIn(insp_b, qs_b) + self.assertNotIn(insp_a, qs_b) + def test_signpanel_inspection_creation_and_asset_sync(self): from assets.models import SignPanel, SignPanelType, SignPole, SignStreet from inspections.models import SignPanelInspection